diff options
author | Chocobozzz <me@florianbigard.com> | 2018-01-25 15:05:18 +0100 |
---|---|---|
committer | Chocobozzz <me@florianbigard.com> | 2018-01-25 18:41:17 +0100 |
commit | 94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4 (patch) | |
tree | 32a9148e0e4567f0c4ffae0412cbed20b84e8873 /server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts | |
parent | d765fafc3faf0db9818eb1a07161df1cb1bc0efa (diff) | |
download | PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.tar.gz PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.tar.zst PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.zip |
Move job queue to redis
We'll use it as cache in the future.
/!\ You'll loose your old jobs (pending jobs too) so upgrade only when
you don't have pending job anymore.
Diffstat (limited to 'server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts')
-rw-r--r-- | server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts | 94 |
1 files changed, 0 insertions, 94 deletions
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 deleted file mode 100644 index 4459152db..000000000 --- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts +++ /dev/null | |||
@@ -1,94 +0,0 @@ | |||
1 | import { JobCategory } from '../../../../shared' | ||
2 | import { buildSignedActivity } from '../../../helpers/activitypub' | ||
3 | import { logger } from '../../../helpers/logger' | ||
4 | import { getServerActor } from '../../../helpers/utils' | ||
5 | import { ACTIVITY_PUB } from '../../../initializers' | ||
6 | import { ActorModel } from '../../../models/activitypub/actor' | ||
7 | import { ActorFollowModel } from '../../../models/activitypub/actor-follow' | ||
8 | import { JobHandler, JobScheduler } from '../job-scheduler' | ||
9 | |||
10 | import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler' | ||
11 | import * as activitypubHttpFetcherHandler from './activitypub-http-fetcher-handler' | ||
12 | import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler' | ||
13 | |||
14 | type ActivityPubHttpPayload = { | ||
15 | uris: string[] | ||
16 | signatureActorId?: number | ||
17 | body?: any | ||
18 | attemptNumber?: number | ||
19 | } | ||
20 | |||
21 | const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = { | ||
22 | activitypubHttpBroadcastHandler, | ||
23 | activitypubHttpUnicastHandler, | ||
24 | activitypubHttpFetcherHandler | ||
25 | } | ||
26 | const jobCategory: JobCategory = 'activitypub-http' | ||
27 | |||
28 | const activitypubHttpJobScheduler = new JobScheduler(jobCategory, jobHandlers) | ||
29 | |||
30 | async function maybeRetryRequestLater (err: Error, payload: ActivityPubHttpPayload, uri: string) { | ||
31 | logger.warn('Cannot make request to %s.', uri, err) | ||
32 | |||
33 | let attemptNumber = payload.attemptNumber || 1 | ||
34 | attemptNumber += 1 | ||
35 | |||
36 | if (attemptNumber < ACTIVITY_PUB.MAX_HTTP_ATTEMPT) { | ||
37 | logger.debug('Retrying request to %s (attempt %d/%d).', uri, attemptNumber, ACTIVITY_PUB.MAX_HTTP_ATTEMPT, err) | ||
38 | |||
39 | const actor = await ActorFollowModel.loadByFollowerInbox(uri, undefined) | ||
40 | if (!actor) { | ||
41 | logger.debug('Actor %s is not a follower, do not retry the request.', uri) | ||
42 | return false | ||
43 | } | ||
44 | |||
45 | const newPayload = Object.assign(payload, { | ||
46 | uris: [ uri ], | ||
47 | attemptNumber | ||
48 | }) | ||
49 | await activitypubHttpJobScheduler.createJob(undefined, 'activitypubHttpUnicastHandler', newPayload) | ||
50 | |||
51 | return true | ||
52 | } | ||
53 | |||
54 | return false | ||
55 | } | ||
56 | |||
57 | async function computeBody (payload: ActivityPubHttpPayload) { | ||
58 | let body = payload.body | ||
59 | |||
60 | if (payload.signatureActorId) { | ||
61 | const actorSignature = await ActorModel.load(payload.signatureActorId) | ||
62 | if (!actorSignature) throw new Error('Unknown signature actor id.') | ||
63 | body = await buildSignedActivity(actorSignature, payload.body) | ||
64 | } | ||
65 | |||
66 | return body | ||
67 | } | ||
68 | |||
69 | async function buildSignedRequestOptions (payload: ActivityPubHttpPayload) { | ||
70 | let actor: ActorModel | ||
71 | if (payload.signatureActorId) { | ||
72 | actor = await ActorModel.load(payload.signatureActorId) | ||
73 | if (!actor) throw new Error('Unknown signature actor id.') | ||
74 | } else { | ||
75 | // We need to sign the request, so use the server | ||
76 | actor = await getServerActor() | ||
77 | } | ||
78 | |||
79 | const keyId = actor.getWebfingerUrl() | ||
80 | return { | ||
81 | algorithm: 'rsa-sha256', | ||
82 | authorizationHeaderName: 'Signature', | ||
83 | keyId, | ||
84 | key: actor.privateKey | ||
85 | } | ||
86 | } | ||
87 | |||
88 | export { | ||
89 | ActivityPubHttpPayload, | ||
90 | activitypubHttpJobScheduler, | ||
91 | maybeRetryRequestLater, | ||
92 | computeBody, | ||
93 | buildSignedRequestOptions | ||
94 | } | ||