aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/jobs
diff options
context:
space:
mode:
authorChocobozzz <florian.bigard@gmail.com>2017-11-22 10:29:55 +0100
committerChocobozzz <florian.bigard@gmail.com>2017-11-27 19:40:53 +0100
commitc986175d68a18e96fbd41537a05c7796a2c64f38 (patch)
treee8a8a02abb57e637451afbf3b1b0e7a0dbd4206c /server/lib/jobs
parente71bcc0f4b31ecfd84a786411febfc6d18a85258 (diff)
downloadPeerTube-c986175d68a18e96fbd41537a05c7796a2c64f38.tar.gz
PeerTube-c986175d68a18e96fbd41537a05c7796a2c64f38.tar.zst
PeerTube-c986175d68a18e96fbd41537a05c7796a2c64f38.zip
Fetch outbox to grab old activities
Diffstat (limited to 'server/lib/jobs')
-rw-r--r--server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts69
-rw-r--r--server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts8
2 files changed, 74 insertions, 3 deletions
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts
new file mode 100644
index 000000000..b8ead32a4
--- /dev/null
+++ b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts
@@ -0,0 +1,69 @@
1import { logger } from '../../../helpers'
2import { buildSignedActivity } from '../../../helpers/activitypub'
3import { doRequest } from '../../../helpers/requests'
4import { database as db } from '../../../initializers'
5import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler'
6import { processActivities } from '../../activitypub/process/process'
7import { ACTIVITY_PUB } from '../../../initializers/constants'
8
9async 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, 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.nextLink
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, uri)
46
47 await processActivities(activities)
48 }
49 }
50 }
51}
52
53function onError (err: Error, jobId: number) {
54 logger.error('Error when broadcasting ActivityPub request in job %d.', jobId, err)
55 return Promise.resolve()
56}
57
58function onSuccess (jobId: number) {
59 logger.info('Job %d is a success.', jobId)
60 return Promise.resolve()
61}
62
63// ---------------------------------------------------------------------------
64
65export {
66 process,
67 onError,
68 onSuccess
69}
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts
index e4f6c94a5..aef217ce7 100644
--- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts
+++ b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts
@@ -2,16 +2,18 @@ import { JobScheduler, JobHandler } from '../job-scheduler'
2 2
3import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler' 3import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler'
4import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler' 4import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler'
5import * as activitypubHttpFetcherHandler from './activitypub-http-fetcher-handler'
5import { JobCategory } from '../../../../shared' 6import { JobCategory } from '../../../../shared'
6 7
7type ActivityPubHttpPayload = { 8type ActivityPubHttpPayload = {
8 uris: string[] 9 uris: string[]
9 signatureAccountId: number 10 signatureAccountId?: number
10 body: any 11 body?: any
11} 12}
12const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = { 13const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = {
13 activitypubHttpBroadcastHandler, 14 activitypubHttpBroadcastHandler,
14 activitypubHttpUnicastHandler 15 activitypubHttpUnicastHandler,
16 activitypubHttpFetcherHandler
15} 17}
16const jobCategory: JobCategory = 'activitypub-http' 18const jobCategory: JobCategory = 'activitypub-http'
17 19