server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts

   1 import { logger } from '../../../helpers'
   2 import { buildSignedActivity } from '../../../helpers/activitypub'
   3 import { doRequest } from '../../../helpers/requests'
   4 import { database as db } from '../../../initializers'
   5 import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler'
   6 import { processActivities } from '../../activitypub/process/process'
   7 import { ACTIVITY_PUB } from '../../../initializers/constants'
   8
   9 async function process (payload: ActivityPubHttpPayload, jobId: number) {
  10   logger.info('Processing ActivityPub fetcher in job %d.', jobId)
  11
  12   const options = {
  13     method: 'GET',
  14     uri: '',
  15     json: true
  16   }
  17
  18   for (const uri of payload.uris) {
  19     options.uri = uri
  20     logger.info('Fetching ActivityPub data on %s.', uri)
  21
  22     const response = await doRequest(options)
  23     const firstBody = response.body
  24
  25     if (firstBody.first && Array.isArray(firstBody.first.orderedItems)) {
  26       const activities = firstBody.first.orderedItems
  27
  28       logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri)
  29
  30       await processActivities(activities)
  31     }
  32
  33     let limit = ACTIVITY_PUB.FETCH_PAGE_LIMIT
  34     let i = 0
  35     let nextLink = firstBody.first.next
  36     while (nextLink && i < limit) {
  37       options.uri = nextLink
  38
  39       const { body } = await doRequest(options)
  40       nextLink = body.next
  41       i++
  42
  43       if (Array.isArray(body.orderedItems)) {
  44         const activities = body.orderedItems
  45         logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri)
  46
  47         await processActivities(activities)
  48       }
  49     }
  50   }
  51 }
  52
  53 function onError (err: Error, jobId: number) {
  54   logger.error('Error when broadcasting ActivityPub request in job %d.', jobId, err)
  55   return Promise.resolve()
  56 }
  57
  58 function onSuccess (jobId: number) {
  59   logger.info('Job %d is a success.', jobId)
  60   return Promise.resolve()
  61 }
  62
  63 // ---------------------------------------------------------------------------
  64
  65 export {
  66   process,
  67   onError,
  68   onSuccess
  69 }