Cleanup invalid rates/comments/shares

server/controllers/api/users/my-blocklist.ts

@@ -17,11 +17,9 @@ import {
   serversBlocklistSortValidator,
   unblockServerByAccountValidator
 } from '../../../middlewares/validators'
-import { AccountModel } from '../../../models/account/account'
 import { AccountBlocklistModel } from '../../../models/account/account-blocklist'
 import { addAccountInBlocklist, addServerInBlocklist, removeAccountFromBlocklist, removeServerFromBlocklist } from '../../../lib/blocklist'
 import { ServerBlocklistModel } from '../../../models/server/server-blocklist'
-import { ServerModel } from '../../../models/server/server'
 
 const myBlocklistRouter = express.Router()
 

server/controllers/api/videos/watching.ts

@@ -2,7 +2,6 @@ import * as express from 'express'
 import { UserWatchingVideo } from '../../../../shared'
 import { asyncMiddleware, asyncRetryTransactionMiddleware, authenticate, videoWatchingValidator } from '../../../middlewares'
 import { UserVideoHistoryModel } from '../../../models/account/user-video-history'
-import { UserModel } from '../../../models/account/user'
 
 const watchingRouter = express.Router()
 

server/helpers/audit-logger.ts

@@ -6,13 +6,12 @@ import * as flatten from 'flat'
 import * as winston from 'winston'
 import { CONFIG } from '../initializers'
 import { jsonLoggerFormat, labelFormatter } from './logger'
-import { VideoDetails, User, VideoChannel, VideoAbuse, VideoImport } from '../../shared'
+import { User, VideoAbuse, VideoChannel, VideoDetails, VideoImport } from '../../shared'
 import { VideoComment } from '../../shared/models/videos/video-comment.model'
 import { CustomConfig } from '../../shared/models/server/custom-config.model'
-import { UserModel } from '../models/account/user'
 
 function getAuditIdFromRes (res: express.Response) {
-  return (res.locals.oauth.token.User as UserModel).username
+  return res.locals.oauth.token.User.username
 }
 
 enum AUDIT_TYPE {

server/lib/activitypub/crawl.ts

@@ -4,7 +4,10 @@ import { logger } from '../../helpers/logger'
 import * as Bluebird from 'bluebird'
 import { ActivityPubOrderedCollection } from '../../../shared/models/activitypub'
 
-async function crawlCollectionPage <T> (uri: string, handler: (items: T[]) => (Promise<any> | Bluebird<any>)) {
+type HandlerFunction<T> = (items: T[]) => (Promise<any> | Bluebird<any>)
+type CleanerFunction = (startedDate: Date) => (Promise<any> | Bluebird<any>)
+
+async function crawlCollectionPage <T> (uri: string, handler: HandlerFunction<T>, cleaner?: CleanerFunction) {
   logger.info('Crawling ActivityPub data on %s.', uri)
 
   const options = {
@@ -15,6 +18,8 @@ async function crawlCollectionPage <T> (uri: string, handler: (items: T[]) => (P
     timeout: JOB_REQUEST_TIMEOUT
   }
 
+  const startDate = new Date()
+
   const response = await doRequest<ActivityPubOrderedCollection<T>>(options)
   const firstBody = response.body
 
@@ -35,6 +40,8 @@ async function crawlCollectionPage <T> (uri: string, handler: (items: T[]) => (P
       await handler(items)
     }
   }
+
+  if (cleaner) await cleaner(startDate)
 }
 
 export {

server/lib/activitypub/share.ts

@@ -54,12 +54,7 @@ async function addVideoShares (shareUrls: string[], instance: VideoModel) {
         url: shareUrl
       }
 
-      await VideoShareModel.findOrCreate({
+      await VideoShareModel.upsert(entry)
-        where: {
-          url: shareUrl
-        },
-        defaults: entry
-      })
     } catch (err) {
       logger.warn('Cannot add share %s.', shareUrl, { err })
     }

server/lib/activitypub/video-comments.ts

@@ -34,8 +34,7 @@ async function videoCommentActivityObjectToDBAttributes (video: VideoModel, acto
     accountId: actor.Account.id,
     inReplyToCommentId,
     originCommentId,
-    createdAt: new Date(comment.published),
+    createdAt: new Date(comment.published)
-    updatedAt: new Date(comment.updated)
   }
 }
 
@@ -74,12 +73,7 @@ async function addVideoComment (videoInstance: VideoModel, commentUrl: string) {
   const entry = await videoCommentActivityObjectToDBAttributes(videoInstance, actor, body)
   if (!entry) return { created: false }
 
-  const [ comment, created ] = await VideoCommentModel.findOrCreate({
+  const [ comment, created ] = await VideoCommentModel.upsert<VideoCommentModel>(entry, { returning: true })
-    where: {
-      url: body.id
-    },
-    defaults: entry
-  })
   comment.Account = actor.Account
   comment.Video = videoInstance
 

server/lib/activitypub/video-rates.ts

@@ -38,19 +38,14 @@ async function createRates (ratesUrl: string[], video: VideoModel, rate: VideoRa
 
       const actor = await getOrCreateActorAndServerAndModel(actorUrl)
 
-      const [ , created ] = await AccountVideoRateModel
+      const entry = {
-        .findOrCreate({
-          where: {
-            videoId: video.id,
-            accountId: actor.Account.id
-          },
-          defaults: {
         videoId: video.id,
         accountId: actor.Account.id,
         type: rate,
         url: body.id
       }
-        })
+
+      const created = await AccountVideoRateModel.upsert(entry)
 
       if (created) rateCounts += 1
     } catch (err) {

server/lib/activitypub/videos.ts

@@ -40,6 +40,9 @@ import { Notifier } from '../notifier'
 import { VideoStreamingPlaylistModel } from '../../models/video/video-streaming-playlist'
 import { VideoStreamingPlaylistType } from '../../../shared/models/videos/video-streaming-playlist.type'
 import { FilteredModelAttributes } from 'sequelize-typescript/lib/models/Model'
+import { AccountVideoRateModel } from '../../models/account/account-video-rate'
+import { VideoShareModel } from '../../models/video/video-share'
+import { VideoCommentModel } from '../../models/video/video-comment'
 
 async function federateVideoIfNeeded (video: VideoModel, isNewVideo: boolean, transaction?: sequelize.Transaction) {
   // If the video is not private and published, we federate it
@@ -134,31 +137,43 @@ async function syncVideoExternalAttributes (video: VideoModel, fetchedVideo: Vid
   const jobPayloads: ActivitypubHttpFetcherPayload[] = []
 
   if (syncParam.likes === true) {
-    await crawlCollectionPage<string>(fetchedVideo.likes, items => createRates(items, video, 'like'))
+    const handler = items => createRates(items, video, 'like')
+    const cleaner = crawlStartDate => AccountVideoRateModel.cleanOldRatesOf(video.id, 'like' as 'like', crawlStartDate)
+
+    await crawlCollectionPage<string>(fetchedVideo.likes, handler, cleaner)
       .catch(err => logger.error('Cannot add likes of video %s.', video.uuid, { err }))
   } else {
     jobPayloads.push({ uri: fetchedVideo.likes, videoId: video.id, type: 'video-likes' as 'video-likes' })
   }
 
   if (syncParam.dislikes === true) {
-    await crawlCollectionPage<string>(fetchedVideo.dislikes, items => createRates(items, video, 'dislike'))
+    const handler = items => createRates(items, video, 'dislike')
+    const cleaner = crawlStartDate => AccountVideoRateModel.cleanOldRatesOf(video.id, 'dislike' as 'dislike', crawlStartDate)
+
+    await crawlCollectionPage<string>(fetchedVideo.dislikes, handler, cleaner)
       .catch(err => logger.error('Cannot add dislikes of video %s.', video.uuid, { err }))
   } else {
     jobPayloads.push({ uri: fetchedVideo.dislikes, videoId: video.id, type: 'video-dislikes' as 'video-dislikes' })
   }
 
   if (syncParam.shares === true) {
-    await crawlCollectionPage<string>(fetchedVideo.shares, items => addVideoShares(items, video))
+    const handler = items => addVideoShares(items, video)
+    const cleaner = crawlStartDate => VideoShareModel.cleanOldSharesOf(video.id, crawlStartDate)
+
+    await crawlCollectionPage<string>(fetchedVideo.shares, handler, cleaner)
       .catch(err => logger.error('Cannot add shares of video %s.', video.uuid, { err }))
   } else {
     jobPayloads.push({ uri: fetchedVideo.shares, videoId: video.id, type: 'video-shares' as 'video-shares' })
   }
 
   if (syncParam.comments === true) {
-    await crawlCollectionPage<string>(fetchedVideo.comments, items => addVideoComments(items, video))
+    const handler = items => addVideoComments(items, video)
+    const cleaner = crawlStartDate => VideoCommentModel.cleanOldCommentsOf(video.id, crawlStartDate)
+
+    await crawlCollectionPage<string>(fetchedVideo.comments, handler, cleaner)
       .catch(err => logger.error('Cannot add comments of video %s.', video.uuid, { err }))
   } else {
-    jobPayloads.push({ uri: fetchedVideo.shares, videoId: video.id, type: 'video-shares' as 'video-shares' })
+    jobPayloads.push({ uri: fetchedVideo.comments, videoId: video.id, type: 'video-comments' as 'video-comments' })
   }
 
   await Bluebird.map(jobPayloads, payload => JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload }))

server/lib/job-queue/handlers/activitypub-http-fetcher.ts

@@ -1,4 +1,5 @@
 import * as Bull from 'bull'
+import * as Bluebird from 'bluebird'
 import { logger } from '../../../helpers/logger'
 import { processActivities } from '../../activitypub/process'
 import { addVideoComments } from '../../activitypub/video-comments'
@@ -7,6 +8,9 @@ import { VideoModel } from '../../../models/video/video'
 import { addVideoShares, createRates } from '../../activitypub'
 import { createAccountPlaylists } from '../../activitypub/playlist'
 import { AccountModel } from '../../../models/account/account'
+import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
+import { VideoShareModel } from '../../../models/video/video-share'
+import { VideoCommentModel } from '../../../models/video/video-comment'
 
 type FetchType = 'activity' | 'video-likes' | 'video-dislikes' | 'video-shares' | 'video-comments' | 'account-playlists'
 
@@ -37,7 +41,14 @@ async function processActivityPubHttpFetcher (job: Bull.Job) {
     'account-playlists': items => createAccountPlaylists(items, account)
   }
 
-  return crawlCollectionPage(payload.uri, fetcherType[payload.type])
+  const cleanerType: { [ id in FetchType ]?: (crawlStartDate: Date) => Bluebird<any> } = {
+    'video-likes': crawlStartDate => AccountVideoRateModel.cleanOldRatesOf(video.id, 'like' as 'like', crawlStartDate),
+    'video-dislikes': crawlStartDate => AccountVideoRateModel.cleanOldRatesOf(video.id, 'dislike' as 'dislike', crawlStartDate),
+    'video-shares': crawlStartDate => VideoShareModel.cleanOldSharesOf(video.id, crawlStartDate),
+    'video-comments': crawlStartDate => VideoCommentModel.cleanOldCommentsOf(video.id, crawlStartDate)
+  }
+
+  return crawlCollectionPage(payload.uri, fetcherType[payload.type], cleanerType[payload.type])
 }
 
 // ---------------------------------------------------------------------------

server/models/account/account-video-rate.ts

@@ -1,5 +1,5 @@
 import { values } from 'lodash'
-import { Transaction } from 'sequelize'
+import { Transaction, Op } from 'sequelize'
 import { AllowNull, BelongsTo, Column, CreatedAt, DataType, ForeignKey, Is, Model, Table, UpdatedAt } from 'sequelize-typescript'
 import { IFindOptions } from 'sequelize-typescript/lib/interfaces/IFindOptions'
 import { VideoRateType } from '../../../shared/models/videos'
@@ -158,4 +158,31 @@ export class AccountVideoRateModel extends Model<AccountVideoRateModel> {
 
     return AccountVideoRateModel.findAndCountAll(query)
   }
+
+  static cleanOldRatesOf (videoId: number, type: VideoRateType, beforeUpdatedAt: Date) {
+    return AccountVideoRateModel.sequelize.transaction(async t => {
+      const query = {
+        where: {
+          updatedAt: {
+            [Op.lt]: beforeUpdatedAt
+          },
+          videoId,
+          type
+        },
+        transaction: t
+      }
+
+      const deleted = await AccountVideoRateModel.destroy(query)
+
+      const options = {
+        transaction: t,
+        where: {
+          id: videoId
+        }
+      }
+
+      if (type === 'like') await VideoModel.increment({ likes: -deleted }, options)
+      else if (type === 'dislike') await VideoModel.increment({ dislikes: -deleted }, options)
+    })
+  }
 }

server/models/video/video-comment.ts

@@ -1,4 +1,5 @@
 import * as Sequelize from 'sequelize'
+import { Op } from 'sequelize'
 import {
   AllowNull,
   BeforeDestroy,
@@ -453,6 +454,19 @@ export class VideoCommentModel extends Model<VideoCommentModel> {
     }
   }
 
+  static cleanOldCommentsOf (videoId: number, beforeUpdatedAt: Date) {
+    const query = {
+      where: {
+        updatedAt: {
+          [Op.lt]: beforeUpdatedAt
+        },
+        videoId
+      }
+    }
+
+    return VideoCommentModel.destroy(query)
+  }
+
   getCommentStaticPath () {
     return this.Video.getWatchStaticPath() + ';threadId=' + this.getThreadId()
   }

server/models/video/video-share.ts

@@ -1,4 +1,5 @@
 import * as Sequelize from 'sequelize'
+import { Op } from 'sequelize'
 import * as Bluebird from 'bluebird'
 import { AllowNull, BelongsTo, Column, CreatedAt, DataType, ForeignKey, Is, Model, Scopes, Table, UpdatedAt } from 'sequelize-typescript'
 import { isActivityPubUrlValid } from '../../helpers/custom-validators/activitypub/misc'
@@ -200,4 +201,17 @@ export class VideoShareModel extends Model<VideoShareModel> {
 
     return VideoShareModel.findAndCountAll(query)
   }
+
+  static cleanOldSharesOf (videoId: number, beforeUpdatedAt: Date) {
+    const query = {
+      where: {
+        updatedAt: {
+          [Op.lt]: beforeUpdatedAt
+        },
+        videoId
+      }
+    }
+
+    return VideoShareModel.destroy(query)
+  }
 }

server/models/video/video.ts

@@ -1547,7 +1547,7 @@ export class VideoModel extends Model<VideoModel> {
       attributes: query.attributes,
       order: [ // Keep original order
         Sequelize.literal(
-          ids.map(id => `"VideoModel".id = ${id}`).join(', ')
+          ids.map(id => `"VideoModel".id = ${id} DESC`).join(', ')
         )
       ]
     }

server/tests/api/activitypub/refresher.ts

@@ -8,7 +8,7 @@ import {
   generateUserAccessToken,
   getVideo,
   getVideoPlaylist,
-  killallServers,
+  killallServers, rateVideo,
   reRunServer,
   ServerInfo,
   setAccessTokensToServers,

server/tests/api/videos/multiple-servers.ts

@@ -579,15 +579,15 @@ describe('Test multiple servers', function () {
       this.timeout(20000)
 
       await rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'like')
-      await wait(200)
+      await wait(500)
       await rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'dislike')
-      await wait(200)
+      await wait(500)
       await rateVideo(servers[0].url, servers[0].accessToken, remoteVideosServer1[0], 'like')
       await rateVideo(servers[2].url, servers[2].accessToken, localVideosServer3[1], 'like')
-      await wait(200)
+      await wait(500)
       await rateVideo(servers[2].url, servers[2].accessToken, localVideosServer3[1], 'dislike')
       await rateVideo(servers[2].url, servers[2].accessToken, remoteVideosServer3[1], 'dislike')
-      await wait(200)
+      await wait(500)
       await rateVideo(servers[2].url, servers[2].accessToken, remoteVideosServer3[0], 'like')
 
       await waitJobs(servers)