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

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