server/lib/job-queue/handlers/activitypub-cleaner.ts

   1 import { map } from 'bluebird'
   2 import { Job } from 'bull'
   3 import {
   4   isAnnounceActivityValid,
   5   isDislikeActivityValid,
   6   isLikeActivityValid
   7 } from '@server/helpers/custom-validators/activitypub/activity'
   8 import { sanitizeAndCheckVideoCommentObject } from '@server/helpers/custom-validators/activitypub/video-comments'
   9 import { doJSONRequest, PeerTubeRequestError } from '@server/helpers/requests'
  10 import { AP_CLEANER } from '@server/initializers/constants'
  11 import { checkUrlsSameHost } from '@server/lib/activitypub/url'
  12 import { Redis } from '@server/lib/redis'
  13 import { VideoModel } from '@server/models/video/video'
  14 import { VideoCommentModel } from '@server/models/video/video-comment'
  15 import { VideoShareModel } from '@server/models/video/video-share'
  16 import { HttpStatusCode } from '@shared/models'
  17 import { logger, loggerTagsFactory } from '../../../helpers/logger'
  18 import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
  19
  20 const lTags = loggerTagsFactory('ap-cleaner')
  21
  22 // Job to clean remote interactions off local videos
  23
  24 async function processActivityPubCleaner (_job: Job) {
  25   logger.info('Processing ActivityPub cleaner.', lTags())
  26
  27   {
  28     const rateUrls = await AccountVideoRateModel.listRemoteRateUrlsOfLocalVideos()
  29     const { bodyValidator, deleter, updater } = rateOptionsFactory()
  30
  31     await map(rateUrls, async rateUrl => {
  32       const result = await updateObjectIfNeeded({ url: rateUrl, bodyValidator, updater, deleter })
  33
  34       if (result?.status === 'deleted') {
  35         const { videoId, type } = result.data
  36
  37         await VideoModel.syncLocalRates(videoId, type, undefined)
  38       }
  39     }, { concurrency: AP_CLEANER.CONCURRENCY })
  40   }
  41
  42   {
  43     const shareUrls = await VideoShareModel.listRemoteShareUrlsOfLocalVideos()
  44     const { bodyValidator, deleter, updater } = shareOptionsFactory()
  45
  46     await map(shareUrls, async shareUrl => {
  47       await updateObjectIfNeeded({ url: shareUrl, bodyValidator, updater, deleter })
  48     }, { concurrency: AP_CLEANER.CONCURRENCY })
  49   }
  50
  51   {
  52     const commentUrls = await VideoCommentModel.listRemoteCommentUrlsOfLocalVideos()
  53     const { bodyValidator, deleter, updater } = commentOptionsFactory()
  54
  55     await map(commentUrls, async commentUrl => {
  56       await updateObjectIfNeeded({ url: commentUrl, bodyValidator, updater, deleter })
  57     }, { concurrency: AP_CLEANER.CONCURRENCY })
  58   }
  59 }
  60
  61 // ---------------------------------------------------------------------------
  62
  63 export {
  64   processActivityPubCleaner
  65 }
  66
  67 // ---------------------------------------------------------------------------
  68
  69 async function updateObjectIfNeeded <T> (options: {
  70   url: string
  71   bodyValidator: (body: any) => boolean
  72   updater: (url: string, newUrl: string) => Promise<T>
  73   deleter: (url: string) => Promise<T> }
  74 ): Promise<{ data: T, status: 'deleted' | 'updated' } | null> {
  75   const { url, bodyValidator, updater, deleter } = options
  76
  77   const on404OrTombstone = async () => {
  78     logger.info('Removing remote AP object %s.', url, lTags(url))
  79     const data = await deleter(url)
  80
  81     return { status: 'deleted' as 'deleted', data }
  82   }
  83
  84   try {
  85     const { body } = await doJSONRequest<any>(url, { activityPub: true })
  86
  87     // If not same id, check same host and update
  88     if (!body || !body.id || !bodyValidator(body)) throw new Error(`Body or body id of ${url} is invalid`)
  89
  90     if (body.type === 'Tombstone') {
  91       return on404OrTombstone()
  92     }
  93
  94     const newUrl = body.id
  95     if (newUrl !== url) {
  96       if (checkUrlsSameHost(newUrl, url) !== true) {
  97         throw new Error(`New url ${newUrl} has not the same host than old url ${url}`)
  98       }
  99
 100       logger.info('Updating remote AP object %s.', url, lTags(url))
 101       const data = await updater(url, newUrl)
 102
 103       return { status: 'updated', data }
 104     }
 105
 106     return null
 107   } catch (err) {
 108     // Does not exist anymore, remove entry
 109     if ((err as PeerTubeRequestError).statusCode === HttpStatusCode.NOT_FOUND_404) {
 110       return on404OrTombstone()
 111     }
 112
 113     logger.debug('Remote AP object %s is unavailable.', url, lTags(url))
 114
 115     const unavailability = await Redis.Instance.addAPUnavailability(url)
 116     if (unavailability >= AP_CLEANER.UNAVAILABLE_TRESHOLD) {
 117       logger.info('Removing unavailable AP resource %s.', url, lTags(url))
 118       return on404OrTombstone()
 119     }
 120
 121     return null
 122   }
 123 }
 124
 125 function rateOptionsFactory () {
 126   return {
 127     bodyValidator: (body: any) => isLikeActivityValid(body) || isDislikeActivityValid(body),
 128
 129     updater: async (url: string, newUrl: string) => {
 130       const rate = await AccountVideoRateModel.loadByUrl(url, undefined)
 131       rate.url = newUrl
 132
 133       const videoId = rate.videoId
 134       const type = rate.type
 135
 136       await rate.save()
 137
 138       return { videoId, type }
 139     },
 140
 141     deleter: async (url) => {
 142       const rate = await AccountVideoRateModel.loadByUrl(url, undefined)
 143
 144       const videoId = rate.videoId
 145       const type = rate.type
 146
 147       await rate.destroy()
 148
 149       return { videoId, type }
 150     }
 151   }
 152 }
 153
 154 function shareOptionsFactory () {
 155   return {
 156     bodyValidator: (body: any) => isAnnounceActivityValid(body),
 157
 158     updater: async (url: string, newUrl: string) => {
 159       const share = await VideoShareModel.loadByUrl(url, undefined)
 160       share.url = newUrl
 161
 162       await share.save()
 163
 164       return undefined
 165     },
 166
 167     deleter: async (url) => {
 168       const share = await VideoShareModel.loadByUrl(url, undefined)
 169
 170       await share.destroy()
 171
 172       return undefined
 173     }
 174   }
 175 }
 176
 177 function commentOptionsFactory () {
 178   return {
 179     bodyValidator: (body: any) => sanitizeAndCheckVideoCommentObject(body),
 180
 181     updater: async (url: string, newUrl: string) => {
 182       const comment = await VideoCommentModel.loadByUrlAndPopulateAccountAndVideo(url)
 183       comment.url = newUrl
 184
 185       await comment.save()
 186
 187       return undefined
 188     },
 189
 190     deleter: async (url) => {
 191       const comment = await VideoCommentModel.loadByUrlAndPopulateAccountAndVideo(url)
 192
 193       await comment.destroy()
 194
 195       return undefined
 196     }
 197   }
 198 }