mirror of https://github.com/Chocobozzz/PeerTube
Fetch outbox to grab old activities
parent
e71bcc0f4b
commit
c986175d68
|
@ -6,7 +6,7 @@ engines:
|
||||||
enabled: true
|
enabled: true
|
||||||
config:
|
config:
|
||||||
languages:
|
languages:
|
||||||
- javascript
|
- typescript
|
||||||
eslint:
|
eslint:
|
||||||
enabled: true
|
enabled: true
|
||||||
fixme:
|
fixme:
|
||||||
|
@ -18,7 +18,6 @@ ratings:
|
||||||
exclude_paths:
|
exclude_paths:
|
||||||
- config/
|
- config/
|
||||||
- node_modules/
|
- node_modules/
|
||||||
- client
|
|
||||||
- scripts/
|
- scripts/
|
||||||
- server/tests/
|
- server/tests/
|
||||||
- .tmp/
|
- .tmp/
|
||||||
|
|
|
@ -8,7 +8,6 @@
|
||||||
>
|
>
|
||||||
<p-column field="id" header="ID"></p-column>
|
<p-column field="id" header="ID"></p-column>
|
||||||
<p-column field="follower.host" header="Host"></p-column>
|
<p-column field="follower.host" header="Host"></p-column>
|
||||||
<p-column field="email" header="Email"></p-column>
|
|
||||||
<p-column field="follower.score" header="Score"></p-column>
|
<p-column field="follower.score" header="Score"></p-column>
|
||||||
<p-column field="state" header="State"></p-column>
|
<p-column field="state" header="State"></p-column>
|
||||||
<p-column field="createdAt" header="Created date" [sortable]="true"></p-column>
|
<p-column field="createdAt" header="Created date" [sortable]="true"></p-column>
|
||||||
|
|
|
@ -8,7 +8,6 @@
|
||||||
>
|
>
|
||||||
<p-column field="id" header="ID"></p-column>
|
<p-column field="id" header="ID"></p-column>
|
||||||
<p-column field="following.host" header="Host"></p-column>
|
<p-column field="following.host" header="Host"></p-column>
|
||||||
<p-column field="email" header="Email"></p-column>
|
|
||||||
<p-column field="state" header="State"></p-column>
|
<p-column field="state" header="State"></p-column>
|
||||||
<p-column field="createdAt" header="Created date" [sortable]="true"></p-column>
|
<p-column field="createdAt" header="Created date" [sortable]="true"></p-column>
|
||||||
<p-column header="Unfollow" styleClass="action-cell">
|
<p-column header="Unfollow" styleClass="action-cell">
|
||||||
|
|
|
@ -1,27 +1,10 @@
|
||||||
import * as express from 'express'
|
import * as express from 'express'
|
||||||
import { Activity, ActivityPubCollection, ActivityPubOrderedCollection, ActivityType, RootActivity } from '../../../shared'
|
import { Activity, ActivityPubCollection, ActivityPubOrderedCollection, RootActivity } from '../../../shared'
|
||||||
import { logger } from '../../helpers'
|
import { logger } from '../../helpers'
|
||||||
import { isActivityValid } from '../../helpers/custom-validators/activitypub/activity'
|
import { isActivityValid } from '../../helpers/custom-validators/activitypub/activity'
|
||||||
import { processCreateActivity, processUpdateActivity, processUndoActivity } from '../../lib'
|
import { processActivities } from '../../lib/activitypub/process/process'
|
||||||
import { processAcceptActivity } from '../../lib/activitypub/process/process-accept'
|
|
||||||
import { processAddActivity } from '../../lib/activitypub/process/process-add'
|
|
||||||
import { processAnnounceActivity } from '../../lib/activitypub/process/process-announce'
|
|
||||||
import { processDeleteActivity } from '../../lib/activitypub/process/process-delete'
|
|
||||||
import { processFollowActivity } from '../../lib/activitypub/process/process-follow'
|
|
||||||
import { asyncMiddleware, checkSignature, localAccountValidator, signatureValidator } from '../../middlewares'
|
import { asyncMiddleware, checkSignature, localAccountValidator, signatureValidator } from '../../middlewares'
|
||||||
import { activityPubValidator } from '../../middlewares/validators/activitypub/activity'
|
import { activityPubValidator } from '../../middlewares/validators/activitypub/activity'
|
||||||
import { AccountInstance } from '../../models/account/account-interface'
|
|
||||||
|
|
||||||
const processActivity: { [ P in ActivityType ]: (activity: Activity, inboxAccount?: AccountInstance) => Promise<any> } = {
|
|
||||||
Create: processCreateActivity,
|
|
||||||
Add: processAddActivity,
|
|
||||||
Update: processUpdateActivity,
|
|
||||||
Delete: processDeleteActivity,
|
|
||||||
Follow: processFollowActivity,
|
|
||||||
Accept: processAcceptActivity,
|
|
||||||
Announce: processAnnounceActivity,
|
|
||||||
Undo: processUndoActivity
|
|
||||||
}
|
|
||||||
|
|
||||||
const inboxRouter = express.Router()
|
const inboxRouter = express.Router()
|
||||||
|
|
||||||
|
@ -69,15 +52,3 @@ async function inboxController (req: express.Request, res: express.Response, nex
|
||||||
|
|
||||||
res.status(204).end()
|
res.status(204).end()
|
||||||
}
|
}
|
||||||
|
|
||||||
async function processActivities (activities: Activity[], inboxAccount?: AccountInstance) {
|
|
||||||
for (const activity of activities) {
|
|
||||||
const activityProcessor = processActivity[activity.type]
|
|
||||||
if (activityProcessor === undefined) {
|
|
||||||
logger.warn('Unknown activity type %s.', activity.type, { activityId: activity.id })
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
await activityProcessor(activity, inboxAccount)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
|
@ -34,8 +34,6 @@ async function outboxController (req: express.Request, res: express.Response, ne
|
||||||
const data = await db.Video.listAllAndSharedByAccountForOutbox(account.id, start, count)
|
const data = await db.Video.listAllAndSharedByAccountForOutbox(account.id, start, count)
|
||||||
const activities: Activity[] = []
|
const activities: Activity[] = []
|
||||||
|
|
||||||
console.log(account.url)
|
|
||||||
|
|
||||||
for (const video of data.data) {
|
for (const video of data.data) {
|
||||||
const videoObject = video.toActivityPubObject()
|
const videoObject = video.toActivityPubObject()
|
||||||
let addActivity: ActivityAdd = await addActivityData(video.url, account, video, video.VideoChannel.url, videoObject)
|
let addActivity: ActivityAdd = await addActivityData(video.url, account, video, video.VideoChannel.url, videoObject)
|
||||||
|
|
|
@ -19,6 +19,7 @@ import { sendUndoFollow } from '../../../lib/activitypub/send/send-undo'
|
||||||
import { AccountInstance } from '../../../models/account/account-interface'
|
import { AccountInstance } from '../../../models/account/account-interface'
|
||||||
import { retryTransactionWrapper } from '../../../helpers/database-utils'
|
import { retryTransactionWrapper } from '../../../helpers/database-utils'
|
||||||
import { saveAccountAndServerIfNotExist } from '../../../lib/activitypub/account'
|
import { saveAccountAndServerIfNotExist } from '../../../lib/activitypub/account'
|
||||||
|
import { addFetchOutboxJob } from '../../../lib/activitypub/fetch'
|
||||||
|
|
||||||
const serverFollowsRouter = express.Router()
|
const serverFollowsRouter = express.Router()
|
||||||
|
|
||||||
|
@ -136,6 +137,8 @@ async function follow (fromAccount: AccountInstance, targetAccount: AccountInsta
|
||||||
if (accountFollow.state === 'pending') {
|
if (accountFollow.state === 'pending') {
|
||||||
await sendFollow(accountFollow, t)
|
await sendFollow(accountFollow, t)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
await addFetchOutboxJob(targetAccount, t)
|
||||||
})
|
})
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
// Reset target account
|
// Reset target account
|
||||||
|
|
|
@ -46,14 +46,16 @@ function activityPubCollectionPagination (url: string, page: number, result: Res
|
||||||
orderedItems: result.data
|
orderedItems: result.data
|
||||||
}
|
}
|
||||||
|
|
||||||
const obj = {
|
if (page === 1) {
|
||||||
|
return activityPubContextify({
|
||||||
id: url,
|
id: url,
|
||||||
type: 'OrderedCollection',
|
type: 'OrderedCollection',
|
||||||
totalItems: result.total,
|
totalItems: result.total,
|
||||||
orderedItems: orderedCollectionPagination
|
first: orderedCollectionPagination
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
return activityPubContextify(obj)
|
return orderedCollectionPagination
|
||||||
}
|
}
|
||||||
|
|
||||||
function buildSignedActivity (byAccount: AccountInstance, data: Object) {
|
function buildSignedActivity (byAccount: AccountInstance, data: Object) {
|
||||||
|
|
|
@ -228,6 +228,7 @@ const ACTIVITY_PUB = {
|
||||||
ACCEPT_HEADER: 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"',
|
ACCEPT_HEADER: 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"',
|
||||||
PUBLIC: 'https://www.w3.org/ns/activitystreams#Public',
|
PUBLIC: 'https://www.w3.org/ns/activitystreams#Public',
|
||||||
COLLECTION_ITEMS_PER_PAGE: 10,
|
COLLECTION_ITEMS_PER_PAGE: 10,
|
||||||
|
FETCH_PAGE_LIMIT: 100,
|
||||||
URL_MIME_TYPES: {
|
URL_MIME_TYPES: {
|
||||||
VIDEO: [ 'video/mp4', 'video/webm', 'video/ogg' ], // TODO: Merge with VIDEO_MIMETYPE_EXT
|
VIDEO: [ 'video/mp4', 'video/webm', 'video/ogg' ], // TODO: Merge with VIDEO_MIMETYPE_EXT
|
||||||
TORRENT: [ 'application/x-bittorrent' ],
|
TORRENT: [ 'application/x-bittorrent' ],
|
||||||
|
|
|
@ -0,0 +1,15 @@
|
||||||
|
import { Transaction } from 'sequelize'
|
||||||
|
import { AccountInstance } from '../../models/account/account-interface'
|
||||||
|
import { activitypubHttpJobScheduler, ActivityPubHttpPayload } from '../jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler'
|
||||||
|
|
||||||
|
async function addFetchOutboxJob (account: AccountInstance, t: Transaction) {
|
||||||
|
const jobPayload: ActivityPubHttpPayload = {
|
||||||
|
uris: [ account.outboxUrl ]
|
||||||
|
}
|
||||||
|
|
||||||
|
return activitypubHttpJobScheduler.createJob(t, 'activitypubHttpFetcherHandler', jobPayload)
|
||||||
|
}
|
||||||
|
|
||||||
|
export {
|
||||||
|
addFetchOutboxJob
|
||||||
|
}
|
|
@ -1,6 +1,7 @@
|
||||||
export * from './process'
|
export * from './process'
|
||||||
export * from './send'
|
export * from './send'
|
||||||
export * from './account'
|
export * from './account'
|
||||||
|
export * from './fetch'
|
||||||
export * from './share'
|
export * from './share'
|
||||||
export * from './video-channels'
|
export * from './video-channels'
|
||||||
export * from './videos'
|
export * from './videos'
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
export * from './process'
|
||||||
export * from './process-accept'
|
export * from './process-accept'
|
||||||
export * from './process-add'
|
export * from './process-add'
|
||||||
export * from './process-announce'
|
export * from './process-announce'
|
||||||
|
|
|
@ -0,0 +1,38 @@
|
||||||
|
import { Activity, ActivityType } from '../../../../shared/models/activitypub/activity'
|
||||||
|
import { AccountInstance } from '../../../models/account/account-interface'
|
||||||
|
import { processAcceptActivity } from './process-accept'
|
||||||
|
import { processAddActivity } from './process-add'
|
||||||
|
import { processAnnounceActivity } from './process-announce'
|
||||||
|
import { processCreateActivity } from './process-create'
|
||||||
|
import { processDeleteActivity } from './process-delete'
|
||||||
|
import { processFollowActivity } from './process-follow'
|
||||||
|
import { processUndoActivity } from './process-undo'
|
||||||
|
import { processUpdateActivity } from './process-update'
|
||||||
|
import { logger } from '../../../helpers/logger'
|
||||||
|
|
||||||
|
const processActivity: { [ P in ActivityType ]: (activity: Activity, inboxAccount?: AccountInstance) => Promise<any> } = {
|
||||||
|
Create: processCreateActivity,
|
||||||
|
Add: processAddActivity,
|
||||||
|
Update: processUpdateActivity,
|
||||||
|
Delete: processDeleteActivity,
|
||||||
|
Follow: processFollowActivity,
|
||||||
|
Accept: processAcceptActivity,
|
||||||
|
Announce: processAnnounceActivity,
|
||||||
|
Undo: processUndoActivity
|
||||||
|
}
|
||||||
|
|
||||||
|
async function processActivities (activities: Activity[], inboxAccount?: AccountInstance) {
|
||||||
|
for (const activity of activities) {
|
||||||
|
const activityProcessor = processActivity[activity.type]
|
||||||
|
if (activityProcessor === undefined) {
|
||||||
|
logger.warn('Unknown activity type %s.', activity.type, { activityId: activity.id })
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
await activityProcessor(activity, inboxAccount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export {
|
||||||
|
processActivities
|
||||||
|
}
|
|
@ -21,6 +21,8 @@ async function sendVideoAbuse (byAccount: AccountInstance, videoAbuse: VideoAbus
|
||||||
return unicastTo(data, byAccount, video.VideoChannel.Account.sharedInboxUrl, t)
|
return unicastTo(data, byAccount, video.VideoChannel.Account.sharedInboxUrl, t)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// async function sendCreateView ()
|
||||||
|
|
||||||
async function createActivityData (url: string, byAccount: AccountInstance, object: any) {
|
async function createActivityData (url: string, byAccount: AccountInstance, object: any) {
|
||||||
const { to, cc } = await getAudience(byAccount)
|
const { to, cc } = await getAudience(byAccount)
|
||||||
const activity: ActivityCreate = {
|
const activity: ActivityCreate = {
|
||||||
|
|
|
@ -0,0 +1,69 @@
|
||||||
|
import { logger } from '../../../helpers'
|
||||||
|
import { buildSignedActivity } from '../../../helpers/activitypub'
|
||||||
|
import { doRequest } from '../../../helpers/requests'
|
||||||
|
import { database as db } from '../../../initializers'
|
||||||
|
import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler'
|
||||||
|
import { processActivities } from '../../activitypub/process/process'
|
||||||
|
import { ACTIVITY_PUB } from '../../../initializers/constants'
|
||||||
|
|
||||||
|
async function process (payload: ActivityPubHttpPayload, jobId: number) {
|
||||||
|
logger.info('Processing ActivityPub fetcher in job %d.', jobId)
|
||||||
|
|
||||||
|
const options = {
|
||||||
|
method: 'GET',
|
||||||
|
uri: '',
|
||||||
|
json: true
|
||||||
|
}
|
||||||
|
|
||||||
|
for (const uri of payload.uris) {
|
||||||
|
options.uri = uri
|
||||||
|
logger.info('Fetching ActivityPub data on %s.', uri)
|
||||||
|
|
||||||
|
const response = await doRequest(options)
|
||||||
|
const firstBody = response.body
|
||||||
|
|
||||||
|
if (firstBody.first && Array.isArray(firstBody.first.orderedItems)) {
|
||||||
|
const activities = firstBody.first.orderedItems
|
||||||
|
|
||||||
|
logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, uri)
|
||||||
|
|
||||||
|
await processActivities(activities)
|
||||||
|
}
|
||||||
|
|
||||||
|
let limit = ACTIVITY_PUB.FETCH_PAGE_LIMIT
|
||||||
|
let i = 0
|
||||||
|
let nextLink = firstBody.first.next
|
||||||
|
while (nextLink && i < limit) {
|
||||||
|
options.uri = nextLink
|
||||||
|
|
||||||
|
const { body } = await doRequest(options)
|
||||||
|
nextLink = body.nextLink
|
||||||
|
i++
|
||||||
|
|
||||||
|
if (Array.isArray(body.orderedItems)) {
|
||||||
|
const activities = body.orderedItems
|
||||||
|
logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, uri)
|
||||||
|
|
||||||
|
await processActivities(activities)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function onError (err: Error, jobId: number) {
|
||||||
|
logger.error('Error when broadcasting ActivityPub request in job %d.', jobId, err)
|
||||||
|
return Promise.resolve()
|
||||||
|
}
|
||||||
|
|
||||||
|
function onSuccess (jobId: number) {
|
||||||
|
logger.info('Job %d is a success.', jobId)
|
||||||
|
return Promise.resolve()
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
export {
|
||||||
|
process,
|
||||||
|
onError,
|
||||||
|
onSuccess
|
||||||
|
}
|
|
@ -2,16 +2,18 @@ import { JobScheduler, JobHandler } from '../job-scheduler'
|
||||||
|
|
||||||
import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler'
|
import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler'
|
||||||
import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler'
|
import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler'
|
||||||
|
import * as activitypubHttpFetcherHandler from './activitypub-http-fetcher-handler'
|
||||||
import { JobCategory } from '../../../../shared'
|
import { JobCategory } from '../../../../shared'
|
||||||
|
|
||||||
type ActivityPubHttpPayload = {
|
type ActivityPubHttpPayload = {
|
||||||
uris: string[]
|
uris: string[]
|
||||||
signatureAccountId: number
|
signatureAccountId?: number
|
||||||
body: any
|
body?: any
|
||||||
}
|
}
|
||||||
const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = {
|
const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = {
|
||||||
activitypubHttpBroadcastHandler,
|
activitypubHttpBroadcastHandler,
|
||||||
activitypubHttpUnicastHandler
|
activitypubHttpUnicastHandler,
|
||||||
|
activitypubHttpFetcherHandler
|
||||||
}
|
}
|
||||||
const jobCategory: JobCategory = 'activitypub-http'
|
const jobCategory: JobCategory = 'activitypub-http'
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue