aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/job-queue
diff options
context:
space:
mode:
authorGérald Niel <gerald.niel@gegeweb.org>2018-04-19 07:54:51 +0200
committerGérald Niel <gerald.niel@gegeweb.org>2018-04-19 07:54:51 +0200
commit0030284b0df2983914291d6fe83675e2aa892e6a (patch)
tree2379f6f0c8a5e6ec1cb86d4bc808c7c48a4c4422 /server/lib/job-queue
parentea5cd0fa2cdc7655ed5ecf134dedd52400837ef2 (diff)
parentf55e5a7bf81c2c27db1514273e3366511aabf4ae (diff)
downloadPeerTube-0030284b0df2983914291d6fe83675e2aa892e6a.tar.gz
PeerTube-0030284b0df2983914291d6fe83675e2aa892e6a.tar.zst
PeerTube-0030284b0df2983914291d6fe83675e2aa892e6a.zip
Merge branch 'develop' of framagit.org:chocobozzz/PeerTube into develop
Diffstat (limited to 'server/lib/job-queue')
-rw-r--r--server/lib/job-queue/handlers/activitypub-follow.ts68
-rw-r--r--server/lib/job-queue/handlers/activitypub-http-broadcast.ts17
-rw-r--r--server/lib/job-queue/job-queue.ts5
3 files changed, 79 insertions, 11 deletions
diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts
new file mode 100644
index 000000000..6764a4037
--- /dev/null
+++ b/server/lib/job-queue/handlers/activitypub-follow.ts
@@ -0,0 +1,68 @@
1import * as kue from 'kue'
2import { logger } from '../../../helpers/logger'
3import { getServerActor } from '../../../helpers/utils'
4import { REMOTE_SCHEME, sequelizeTypescript, SERVER_ACTOR_NAME } from '../../../initializers'
5import { sendFollow } from '../../activitypub/send'
6import { sanitizeHost } from '../../../helpers/core-utils'
7import { loadActorUrlOrGetFromWebfinger } from '../../../helpers/webfinger'
8import { getOrCreateActorAndServerAndModel } from '../../activitypub/actor'
9import { retryTransactionWrapper } from '../../../helpers/database-utils'
10import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
11import { ActorModel } from '../../../models/activitypub/actor'
12
13export type ActivitypubFollowPayload = {
14 host: string
15}
16
17async function processActivityPubFollow (job: kue.Job) {
18 const payload = job.data as ActivitypubFollowPayload
19 const host = payload.host
20
21 logger.info('Processing ActivityPub follow in job %d.', job.id)
22
23 const sanitizedHost = sanitizeHost(host, REMOTE_SCHEME.HTTP)
24
25 const actorUrl = await loadActorUrlOrGetFromWebfinger(SERVER_ACTOR_NAME, sanitizedHost)
26 const targetActor = await getOrCreateActorAndServerAndModel(actorUrl)
27
28 const fromActor = await getServerActor()
29 const options = {
30 arguments: [ fromActor, targetActor ],
31 errorMessage: 'Cannot follow with many retries.'
32 }
33
34 return retryTransactionWrapper(follow, options)
35}
36// ---------------------------------------------------------------------------
37
38export {
39 processActivityPubFollow
40}
41
42// ---------------------------------------------------------------------------
43
44function follow (fromActor: ActorModel, targetActor: ActorModel) {
45 if (fromActor.id === targetActor.id) {
46 throw new Error('Follower is the same than target actor.')
47 }
48
49 return sequelizeTypescript.transaction(async t => {
50 const [ actorFollow ] = await ActorFollowModel.findOrCreate({
51 where: {
52 actorId: fromActor.id,
53 targetActorId: targetActor.id
54 },
55 defaults: {
56 state: 'pending',
57 actorId: fromActor.id,
58 targetActorId: targetActor.id
59 },
60 transaction: t
61 })
62 actorFollow.ActorFollowing = targetActor
63 actorFollow.ActorFollower = fromActor
64
65 // Send a notification to remote server
66 await sendFollow(actorFollow)
67 })
68}
diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
index 78878fc01..38b8393f4 100644
--- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -1,8 +1,10 @@
1import * as kue from 'kue' 1import * as kue from 'kue'
2import * as Bluebird from 'bluebird'
2import { logger } from '../../../helpers/logger' 3import { logger } from '../../../helpers/logger'
3import { doRequest } from '../../../helpers/requests' 4import { doRequest } from '../../../helpers/requests'
4import { ActorFollowModel } from '../../../models/activitypub/actor-follow' 5import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
5import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils' 6import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
7import { BROADCAST_CONCURRENCY } from '../../../initializers'
6 8
7export type ActivitypubHttpBroadcastPayload = { 9export type ActivitypubHttpBroadcastPayload = {
8 uris: string[] 10 uris: string[]
@@ -28,16 +30,11 @@ async function processActivityPubHttpBroadcast (job: kue.Job) {
28 const badUrls: string[] = [] 30 const badUrls: string[] = []
29 const goodUrls: string[] = [] 31 const goodUrls: string[] = []
30 32
31 for (const uri of payload.uris) { 33 await Bluebird.map(payload.uris, uri => {
32 options.uri = uri 34 return doRequest(Object.assign({}, options, { uri }))
33 35 .then(() => goodUrls.push(uri))
34 try { 36 .catch(() => badUrls.push(uri))
35 await doRequest(options) 37 }, { concurrency: BROADCAST_CONCURRENCY })
36 goodUrls.push(uri)
37 } catch (err) {
38 badUrls.push(uri)
39 }
40 }
41 38
42 return ActorFollowModel.updateActorFollowsScore(goodUrls, badUrls, undefined) 39 return ActorFollowModel.updateActorFollowsScore(goodUrls, badUrls, undefined)
43} 40}
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 1dc28755e..bf40a9206 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -8,11 +8,13 @@ import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './
8import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast' 8import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
9import { EmailPayload, processEmail } from './handlers/email' 9import { EmailPayload, processEmail } from './handlers/email'
10import { processVideoFile, VideoFilePayload } from './handlers/video-file' 10import { processVideoFile, VideoFilePayload } from './handlers/video-file'
11import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow'
11 12
12type CreateJobArgument = 13type CreateJobArgument =
13 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | 14 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
14 { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | 15 { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
15 { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } | 16 { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
17 { type: 'activitypub-follow', payload: ActivitypubFollowPayload } |
16 { type: 'video-file', payload: VideoFilePayload } | 18 { type: 'video-file', payload: VideoFilePayload } |
17 { type: 'email', payload: EmailPayload } 19 { type: 'email', payload: EmailPayload }
18 20
@@ -20,6 +22,7 @@ const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
20 'activitypub-http-broadcast': processActivityPubHttpBroadcast, 22 'activitypub-http-broadcast': processActivityPubHttpBroadcast,
21 'activitypub-http-unicast': processActivityPubHttpUnicast, 23 'activitypub-http-unicast': processActivityPubHttpUnicast,
22 'activitypub-http-fetcher': processActivityPubHttpFetcher, 24 'activitypub-http-fetcher': processActivityPubHttpFetcher,
25 'activitypub-follow': processActivityPubFollow,
23 'video-file': processVideoFile, 26 'video-file': processVideoFile,
24 'email': processEmail 27 'email': processEmail
25} 28}
@@ -50,7 +53,7 @@ class JobQueue {
50 } 53 }
51 }) 54 })
52 55
53 this.jobQueue.setMaxListeners(15) 56 this.jobQueue.setMaxListeners(20)
54 57
55 this.jobQueue.on('error', err => { 58 this.jobQueue.on('error', err => {
56 logger.error('Error in job queue.', { err }) 59 logger.error('Error in job queue.', { err })