Verified Commit 6b9c966f authored by Chocobozzz's avatar Chocobozzz
Browse files

Automatically remove bad followings

parent 466e3f20
Pipeline #156656 passed with stage
in 61 minutes and 49 seconds
......@@ -254,14 +254,14 @@ async function refreshActorIfNeeded (
await actor.save({ transaction: t })
if (actor.Account) {
actor.Account.set('name', result.name)
actor.Account.set('description', result.summary)
actor.Account.name = result.name
actor.Account.description = result.summary
await actor.Account.save({ transaction: t })
} else if (actor.VideoChannel) {
actor.VideoChannel.set('name', result.name)
actor.VideoChannel.set('description', result.summary)
actor.VideoChannel.set('support', result.support)
actor.VideoChannel.name = result.name
actor.VideoChannel.description = result.summary
actor.VideoChannel.support = result.support
await actor.VideoChannel.save({ transaction: t })
}
......
......@@ -4,7 +4,7 @@ import { retryTransactionWrapper } from '../../../helpers/database-utils'
import { logger } from '../../../helpers/logger'
import { sequelizeTypescript } from '../../../initializers'
import { ActorModel } from '../../../models/activitypub/actor'
import { addVideoComment, resolveThread } from '../video-comments'
import { resolveThread } from '../video-comments'
import { getOrCreateVideoAndAccountAndChannel } from '../videos'
import { forwardVideoRelatedActivity } from '../send/utils'
import { createOrUpdateCacheFile } from '../cache-file'
......@@ -13,6 +13,7 @@ import { PlaylistObject } from '../../../../shared/models/activitypub/objects/pl
import { createOrUpdateVideoPlaylist } from '../playlist'
import { VideoModel } from '../../../models/video/video'
import { APProcessorOptions } from '../../../typings/activitypub-processor.model'
import { VideoCommentModel } from '../../../models/video/video-comment'
async function processCreateActivity (options: APProcessorOptions<ActivityCreate>) {
const { activity, byActor } = options
......@@ -83,9 +84,13 @@ async function processCreateVideoComment (activity: ActivityCreate, byActor: Act
if (!byAccount) throw new Error('Cannot create video comment with the non account actor ' + byActor.url)
let video: VideoModel
let created: boolean
let comment: VideoCommentModel
try {
const resolveThreadResult = await resolveThread(commentObject.inReplyTo)
const resolveThreadResult = await resolveThread({ url: commentObject.id, isVideo: false })
video = resolveThreadResult.video
created = resolveThreadResult.commentCreated
comment = resolveThreadResult.comment
} catch (err) {
logger.debug(
'Cannot process video comment because we could not resolve thread %s. Maybe it was not a video thread, so skip it.',
......@@ -95,8 +100,6 @@ async function processCreateVideoComment (activity: ActivityCreate, byActor: Act
return
}
const { comment, created } = await addVideoComment(video, commentObject.id)
if (video.isOwned() && created === true) {
// Don't resend the activity to the sender
const exceptions = [ byActor ]
......
import { VideoCommentObject } from '../../../shared/models/activitypub/objects/video-comment-object'
import { sanitizeAndCheckVideoCommentObject } from '../../helpers/custom-validators/activitypub/video-comments'
import { logger } from '../../helpers/logger'
import { doRequest } from '../../helpers/requests'
import { ACTIVITY_PUB, CRAWL_REQUEST_CONCURRENCY } from '../../initializers/constants'
import { ActorModel } from '../../models/activitypub/actor'
import { VideoModel } from '../../models/video/video'
import { VideoCommentModel } from '../../models/video/video-comment'
import { getOrCreateActorAndServerAndModel } from './actor'
......@@ -11,79 +9,53 @@ import { getOrCreateVideoAndAccountAndChannel } from './videos'
import * as Bluebird from 'bluebird'
import { checkUrlsSameHost } from '../../helpers/activitypub'
async function videoCommentActivityObjectToDBAttributes (video: VideoModel, actor: ActorModel, comment: VideoCommentObject) {
let originCommentId: number = null
let inReplyToCommentId: number = null
// If this is not a reply to the video (thread), create or get the parent comment
if (video.url !== comment.inReplyTo) {
const { comment: parent } = await addVideoComment(video, comment.inReplyTo)
if (!parent) {
logger.warn('Cannot fetch or get parent comment %s of comment %s.', comment.inReplyTo, comment.id)
return undefined
}
originCommentId = parent.originCommentId || parent.id
inReplyToCommentId = parent.id
}
return {
url: comment.id,
text: comment.content,
videoId: video.id,
accountId: actor.Account.id,
inReplyToCommentId,
originCommentId,
createdAt: new Date(comment.published)
}
type ResolveThreadParams = {
url: string,
comments?: VideoCommentModel[],
isVideo?: boolean,
commentCreated?: boolean
}
type ResolveThreadResult = Promise<{ video: VideoModel, comment: VideoCommentModel, commentCreated: boolean }>
async function addVideoComments (commentUrls: string[], instance: VideoModel) {
async function addVideoComments (commentUrls: string[]) {
return Bluebird.map(commentUrls, commentUrl => {
return addVideoComment(instance, commentUrl)
return resolveThread({ url: commentUrl, isVideo: false })
}, { concurrency: CRAWL_REQUEST_CONCURRENCY })
}
async function addVideoComment (videoInstance: VideoModel, commentUrl: string) {
logger.info('Fetching remote video comment %s.', commentUrl)
async function resolveThread (params: ResolveThreadParams): ResolveThreadResult {
const { url, isVideo } = params
if (params.commentCreated === undefined) params.commentCreated = false
if (params.comments === undefined) params.comments = []
const { body } = await doRequest({
uri: commentUrl,
json: true,
activityPub: true
})
if (sanitizeAndCheckVideoCommentObject(body) === false) {
logger.debug('Remote video comment JSON %s is not valid.', commentUrl, { body })
return { created: false }
// Already have this comment?
if (isVideo !== true) {
const result = await resolveCommentFromDB(params)
if (result) return result
}
const actorUrl = body.attributedTo
if (!actorUrl) return { created: false }
try {
if (isVideo !== false) return await tryResolveThreadFromVideo(params)
if (checkUrlsSameHost(commentUrl, actorUrl) !== true) {
throw new Error(`Actor url ${actorUrl} has not the same host than the comment url ${commentUrl}`)
}
return resolveParentComment(params)
} catch (err) {
logger.debug('Cannot get or create account and video and channel for reply %s, fetch comment', url, { err })
if (checkUrlsSameHost(body.id, commentUrl) !== true) {
throw new Error(`Comment url ${commentUrl} host is different from the AP object id ${body.id}`)
return resolveParentComment(params)
}
}
const actor = await getOrCreateActorAndServerAndModel(actorUrl, 'all')
const entry = await videoCommentActivityObjectToDBAttributes(videoInstance, actor, body)
if (!entry) return { created: false }
export {
addVideoComments,
resolveThread
}
const [ comment, created ] = await VideoCommentModel.upsert<VideoCommentModel>(entry, { returning: true })
comment.Account = actor.Account
comment.Video = videoInstance
// ---------------------------------------------------------------------------
return { comment, created }
}
async function resolveCommentFromDB (params: ResolveThreadParams) {
const { url, comments, commentCreated } = params
type ResolveThreadResult = Promise<{ video: VideoModel, parents: VideoCommentModel[] }>
async function resolveThread (url: string, comments: VideoCommentModel[] = []): ResolveThreadResult {
// Already have this comment?
const commentFromDatabase = await VideoCommentModel.loadByUrlAndPopulateReplyAndVideo(url)
const commentFromDatabase = await VideoCommentModel.loadByUrlAndPopulateReplyAndVideoUrlAndAccount(url)
if (commentFromDatabase) {
let parentComments = comments.concat([ commentFromDatabase ])
......@@ -94,79 +66,97 @@ async function resolveThread (url: string, comments: VideoCommentModel[] = []):
parentComments = parentComments.concat(data)
}
return resolveThread(commentFromDatabase.Video.url, parentComments)
return resolveThread({
url: commentFromDatabase.Video.url,
comments: parentComments,
isVideo: true,
commentCreated
})
}
try {
// Maybe it's a reply to a video?
// If yes, it's done: we resolved all the thread
const { video } = await getOrCreateVideoAndAccountAndChannel({ videoObject: url })
if (comments.length !== 0) {
const firstReply = comments[ comments.length - 1 ]
firstReply.inReplyToCommentId = null
firstReply.originCommentId = null
firstReply.videoId = video.id
comments[comments.length - 1] = await firstReply.save()
for (let i = comments.length - 2; i >= 0; i--) {
const comment = comments[ i ]
comment.originCommentId = firstReply.id
comment.inReplyToCommentId = comments[ i + 1 ].id
comment.videoId = video.id
comments[i] = await comment.save()
}
return undefined
}
async function tryResolveThreadFromVideo (params: ResolveThreadParams) {
const { url, comments, commentCreated } = params
// Maybe it's a reply to a video?
// If yes, it's done: we resolved all the thread
const syncParam = { likes: true, dislikes: true, shares: true, comments: false, thumbnail: true, refreshVideo: false }
const { video } = await getOrCreateVideoAndAccountAndChannel({ videoObject: url, syncParam })
let resultComment: VideoCommentModel
if (comments.length !== 0) {
const firstReply = comments[ comments.length - 1 ]
firstReply.inReplyToCommentId = null
firstReply.originCommentId = null
firstReply.videoId = video.id
firstReply.changed('updatedAt', true)
firstReply.Video = video
comments[comments.length - 1] = await firstReply.save()
for (let i = comments.length - 2; i >= 0; i--) {
const comment = comments[ i ]
comment.originCommentId = firstReply.id
comment.inReplyToCommentId = comments[ i + 1 ].id
comment.videoId = video.id
comment.changed('updatedAt', true)
comment.Video = video
comments[i] = await comment.save()
}
return { video, parents: comments }
} catch (err) {
logger.debug('Cannot get or create account and video and channel for reply %s, fetch comment', url, { err })
resultComment = comments[0]
}
if (comments.length > ACTIVITY_PUB.MAX_RECURSION_COMMENTS) {
throw new Error('Recursion limit reached when resolving a thread')
}
return { video, comment: resultComment, commentCreated }
}
const { body } = await doRequest({
uri: url,
json: true,
activityPub: true
})
async function resolveParentComment (params: ResolveThreadParams) {
const { url, comments } = params
if (sanitizeAndCheckVideoCommentObject(body) === false) {
throw new Error('Remote video comment JSON is not valid:' + JSON.stringify(body))
}
if (comments.length > ACTIVITY_PUB.MAX_RECURSION_COMMENTS) {
throw new Error('Recursion limit reached when resolving a thread')
}
const actorUrl = body.attributedTo
if (!actorUrl) throw new Error('Miss attributed to in comment')
const { body } = await doRequest({
uri: url,
json: true,
activityPub: true
})
if (checkUrlsSameHost(url, actorUrl) !== true) {
throw new Error(`Actor url ${actorUrl} has not the same host than the comment url ${url}`)
}
if (sanitizeAndCheckVideoCommentObject(body) === false) {
throw new Error('Remote video comment JSON is not valid:' + JSON.stringify(body))
}
if (checkUrlsSameHost(body.id, url) !== true) {
throw new Error(`Comment url ${url} host is different from the AP object id ${body.id}`)
}
const actorUrl = body.attributedTo
if (!actorUrl) throw new Error('Miss attributed to in comment')
const actor = await getOrCreateActorAndServerAndModel(actorUrl)
const comment = new VideoCommentModel({
url: body.id,
text: body.content,
videoId: null,
accountId: actor.Account.id,
inReplyToCommentId: null,
originCommentId: null,
createdAt: new Date(body.published),
updatedAt: new Date(body.updated)
})
if (checkUrlsSameHost(url, actorUrl) !== true) {
throw new Error(`Actor url ${actorUrl} has not the same host than the comment url ${url}`)
}
return resolveThread(body.inReplyTo, comments.concat([ comment ]))
if (checkUrlsSameHost(body.id, url) !== true) {
throw new Error(`Comment url ${url} host is different from the AP object id ${body.id}`)
}
}
export {
videoCommentActivityObjectToDBAttributes,
addVideoComments,
addVideoComment,
resolveThread
const actor = await getOrCreateActorAndServerAndModel(actorUrl)
const comment = new VideoCommentModel({
url: body.id,
text: body.content,
videoId: null,
accountId: actor.Account.id,
inReplyToCommentId: null,
originCommentId: null,
createdAt: new Date(body.published),
updatedAt: new Date(body.updated)
})
comment.Account = actor.Account
return resolveThread({
url: body.inReplyTo,
comments: comments.concat([ comment ]),
commentCreated: true
})
}
......@@ -56,6 +56,7 @@ import { join } from 'path'
import { FilteredModelAttributes } from '../../typings/sequelize'
import { Hooks } from '../plugins/hooks'
import { autoBlacklistVideoIfNeeded } from '../video-blacklist'
import { ActorFollowScoreCache } from '../files-cache'
async function federateVideoIfNeeded (video: VideoModel, isNewVideo: boolean, transaction?: sequelize.Transaction) {
if (
......@@ -182,7 +183,7 @@ async function syncVideoExternalAttributes (video: VideoModel, fetchedVideo: Vid
}
if (syncParam.comments === true) {
const handler = items => addVideoComments(items, video)
const handler = items => addVideoComments(items)
const cleaner = crawlStartDate => VideoCommentModel.cleanOldCommentsOf(video.id, crawlStartDate)
await crawlCollectionPage<string>(fetchedVideo.comments, handler, cleaner)
......@@ -421,10 +422,14 @@ async function refreshVideoIfNeeded (options: {
await retryTransactionWrapper(updateVideoFromAP, updateOptions)
await syncVideoExternalAttributes(video, videoObject, options.syncParam)
ActorFollowScoreCache.Instance.addGoodServerId(video.VideoChannel.Actor.serverId)
return video
} catch (err) {
logger.warn('Cannot refresh video %s.', options.video.url, { err })
ActorFollowScoreCache.Instance.addBadServerId(video.VideoChannel.Actor.serverId)
// Don't refresh in loop
await video.setAsRefreshed()
return video
......@@ -500,7 +505,7 @@ async function createVideo (videoObject: VideoTorrentObject, channelActor: Actor
const videoStreamingPlaylists = streamingPlaylistActivityUrlToDBAttributes(videoCreated, videoObject, videoFiles)
const playlistPromises = videoStreamingPlaylists.map(p => VideoStreamingPlaylistModel.create(p, { transaction: t }))
await Promise.all(playlistPromises)
const streamingPlaylists = await Promise.all(playlistPromises)
// Process tags
const tags = videoObject.tag
......@@ -513,7 +518,12 @@ async function createVideo (videoObject: VideoTorrentObject, channelActor: Actor
const videoCaptionsPromises = videoObject.subtitleLanguage.map(c => {
return VideoCaptionModel.insertOrReplaceLanguage(videoCreated.id, c.identifier, t)
})
await Promise.all(videoCaptionsPromises)
const captions = await Promise.all(videoCaptionsPromises)
video.VideoFiles = videoFiles
video.VideoStreamingPlaylists = streamingPlaylists
video.Tags = tagInstances
video.VideoCaptions = captions
const autoBlacklisted = await autoBlacklistVideoIfNeeded({
video,
......
......@@ -7,6 +7,8 @@ class ActorFollowScoreCache {
private static instance: ActorFollowScoreCache
private pendingFollowsScore: { [ url: string ]: number } = {}
private pendingBadServer = new Set<number>()
private pendingGoodServer = new Set<number>()
private constructor () {}
......@@ -32,7 +34,31 @@ class ActorFollowScoreCache {
}
}
getPendingFollowsScoreCopy () {
addBadServerId (serverId: number) {
this.pendingBadServer.add(serverId)
}
getBadFollowingServerIds () {
return Array.from(this.pendingBadServer)
}
clearBadFollowingServerIds () {
this.pendingBadServer = new Set<number>()
}
addGoodServerId (serverId: number) {
this.pendingGoodServer.add(serverId)
}
getGoodFollowingServerIds () {
return Array.from(this.pendingGoodServer)
}
clearGoodFollowingServerIds () {
this.pendingGoodServer = new Set<number>()
}
getPendingFollowsScore () {
return this.pendingFollowsScore
}
......
......@@ -37,7 +37,7 @@ async function processActivityPubHttpFetcher (job: Bull.Job) {
'video-likes': items => createRates(items, video, 'like'),
'video-dislikes': items => createRates(items, video, 'dislike'),
'video-shares': items => addVideoShares(items, video),
'video-comments': items => addVideoComments(items, video),
'video-comments': items => addVideoComments(items),
'account-playlists': items => createAccountPlaylists(items, account)
}
......
......@@ -2,7 +2,7 @@ import { isTestInstance } from '../../helpers/core-utils'
import { logger } from '../../helpers/logger'
import { ActorFollowModel } from '../../models/activitypub/actor-follow'
import { AbstractScheduler } from './abstract-scheduler'
import { SCHEDULER_INTERVALS_MS } from '../../initializers/constants'
import { ACTOR_FOLLOW_SCORE, SCHEDULER_INTERVALS_MS } from '../../initializers/constants'
import { ActorFollowScoreCache } from '../files-cache'
export class ActorFollowScheduler extends AbstractScheduler {
......@@ -22,13 +22,20 @@ export class ActorFollowScheduler extends AbstractScheduler {
}
private async processPendingScores () {
const pendingScores = ActorFollowScoreCache.Instance.getPendingFollowsScoreCopy()
const pendingScores = ActorFollowScoreCache.Instance.getPendingFollowsScore()
const badServerIds = ActorFollowScoreCache.Instance.getBadFollowingServerIds()
const goodServerIds = ActorFollowScoreCache.Instance.getGoodFollowingServerIds()
ActorFollowScoreCache.Instance.clearPendingFollowsScore()
ActorFollowScoreCache.Instance.clearBadFollowingServerIds()
ActorFollowScoreCache.Instance.clearGoodFollowingServerIds()
for (const inbox of Object.keys(pendingScores)) {
await ActorFollowModel.updateFollowScore(inbox, pendingScores[inbox])
await ActorFollowModel.updateScore(inbox, pendingScores[inbox])
}
await ActorFollowModel.updateScoreByFollowingServers(badServerIds, ACTOR_FOLLOW_SCORE.PENALTY)
await ActorFollowModel.updateScoreByFollowingServers(goodServerIds, ACTOR_FOLLOW_SCORE.BONUS)
}
private async removeBadActorFollows () {
......
......@@ -6,7 +6,7 @@ import { CONSTRAINTS_FIELDS, VIDEO_RATE_TYPES } from '../../initializers/constan
import { VideoModel } from '../video/video'
import { AccountModel } from './account'
import { ActorModel } from '../activitypub/actor'
import { getSort, throwIfNotValid } from '../utils'
import { buildLocalAccountIdsIn, getSort, throwIfNotValid } from '../utils'
import { isActivityPubUrlValid } from '../../helpers/custom-validators/activitypub/misc'
import { AccountVideoRate } from '../../../shared'
import { ScopeNames as VideoChannelScopeNames, SummaryOptions, VideoChannelModel } from '../video/video-channel'
......@@ -219,25 +219,11 @@ export class AccountVideoRateModel extends Model<AccountVideoRateModel> {
[Op.lt]: beforeUpdatedAt
},
videoId,
type
},
include: [
{
model: AccountModel.unscoped(),
required: true,
include: [
{
model: ActorModel.unscoped(),
required: true,
where: {
serverId: {
[Op.ne]: null
}
}
}
]
type,
accountId: {
[Op.notIn]: buildLocalAccountIdsIn()
}
],
},
transaction: t
}
......
......@@ -23,7 +23,7 @@ import { logger } from '../../helpers/logger'
import { getServerActor } from '../../helpers/utils'
import { ACTOR_FOLLOW_SCORE, FOLLOW_STATES } from '../../initializers/constants'
import { ServerModel } from '../server/server'
import { getSort } from '../utils'
import { createSafeIn, getSort } from '../utils'
import { ActorModel, unusedActorAttributesForAPI } from './actor'
import { VideoChannelModel } from '../video/video-channel'
import { AccountModel } from '../account/account'
......@@ -464,7 +464,7 @@ export class ActorFollowModel extends Model<ActorFollowModel> {
}
}
static updateFollowScore (inboxUrl: string, value: number, t?: Transaction) {
static updateScore (inboxUrl: string, value: number, t?: Transaction) {
const query = `UPDATE "actorFollow" SET "score" = LEAST("score" + ${value}, ${ACTOR_FOLLOW_SCORE.MAX}) ` +
'WHERE id IN (' +
'SELECT "actorFollow"."id" FROM "actorFollow" ' +
......@@ -480,6 +480,28 @@ export class ActorFollowModel extends Model<ActorFollowModel> {
return ActorFollowModel.sequelize.query(query, options)
}
static async updateScoreByFollowingServers (serverIds: number[], value: number, t?: Transaction) {
if (serverIds.length === 0) return
const me = await getServerActor()
const serverIdsString = createSafeIn(ActorFollowModel, serverIds)
const query = `UPDATE "actorFollow" SET "score" = "score" + ${value} ` +
'WHERE id IN (' +
'SELECT "actorFollow"."id" FROM "actorFollow" ' +
'INNER JOIN "actor" ON "actor"."id" = "actorFollow"."targetActorId" ' +
`WHERE "actorFollow"."actorId" = ${me.Account.actorId} ` + // I'm the follower
`AND "actor"."serverId" IN (${serverIdsString})` + // Criteria on followings
')'
const options = {
type: QueryTypes.BULKUPDATE,
transaction: t
}
return ActorFollowModel.sequelize.query(query, options)
}
private static async createListAcceptedFollowForApiQuery (
type: 'followers' | 'following',
actorIds: number[],
......
import { Model, Sequelize } from 'sequelize-typescript'
import * as validator from 'validator'
import { Col } from 'sequelize/types/lib/utils'
import { OrderItem } from 'sequelize/types'
import { OrderItem, literal } from 'sequelize'
type SortType = { sortModel: any, sortValue: string }
......@@ -129,16 +129,30 @@ function parseAggregateResult (result: any) {
return total
}
const createSafeIn = (model: typeof Model, stringArr: string[]) => {
return stringArr.map(t => model.sequelize.escape(t))
const createSafeIn = (model: typeof Model, stringArr: (string | number)[]) => {