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

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