1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
import { isBlockedByServerOrAccount } from '@server/lib/blocklist'
import { isRedundancyAccepted } from '@server/lib/redundancy'
import { VideoModel } from '@server/models/video/video'
import { ActivityCreate, CacheFileObject, PlaylistObject, VideoCommentObject, VideoObject, WatchActionObject } from '@shared/models'
import { retryTransactionWrapper } from '../../../helpers/database-utils'
import { logger } from '../../../helpers/logger'
import { sequelizeTypescript } from '../../../initializers/database'
import { APProcessorOptions } from '../../../types/activitypub-processor.model'
import { MActorSignature, MCommentOwnerVideo, MVideoAccountLightBlacklistAllFiles } from '../../../types/models'
import { Notifier } from '../../notifier'
import { createOrUpdateCacheFile } from '../cache-file'
import { createOrUpdateLocalVideoViewer } from '../local-video-viewer'
import { createOrUpdateVideoPlaylist } from '../playlists'
import { forwardVideoRelatedActivity } from '../send/shared/send-utils'
import { resolveThread } from '../video-comments'
import { getOrCreateAPVideo } from '../videos'
async function processCreateActivity (options: APProcessorOptions<ActivityCreate>) {
const { activity, byActor } = options
// Only notify if it is not from a fetcher job
const notify = options.fromFetch !== true
const activityObject = activity.object
const activityType = activityObject.type
if (activityType === 'Video') {
return processCreateVideo(activity, notify)
}
if (activityType === 'Note') {
// Comments will be fetched from videos
if (options.fromFetch) return
return retryTransactionWrapper(processCreateVideoComment, activity, byActor, notify)
}
if (activityType === 'WatchAction') {
return retryTransactionWrapper(processCreateWatchAction, activity)
}
if (activityType === 'CacheFile') {
return retryTransactionWrapper(processCreateCacheFile, activity, byActor)
}
if (activityType === 'Playlist') {
return retryTransactionWrapper(processCreatePlaylist, activity, byActor)
}
logger.warn('Unknown activity object type %s when creating activity.', activityType, { activity: activity.id })
return Promise.resolve(undefined)
}
// ---------------------------------------------------------------------------
export {
processCreateActivity
}
// ---------------------------------------------------------------------------
async function processCreateVideo (activity: ActivityCreate, notify: boolean) {
const videoToCreateData = activity.object as VideoObject
const syncParam = { rates: false, shares: false, comments: false, thumbnail: true, refreshVideo: false }
const { video, created } = await getOrCreateAPVideo({ videoObject: videoToCreateData, syncParam })
if (created && notify) Notifier.Instance.notifyOnNewVideoIfNeeded(video)
return video
}
async function processCreateCacheFile (activity: ActivityCreate, byActor: MActorSignature) {
if (await isRedundancyAccepted(activity, byActor) !== true) return
const cacheFile = activity.object as CacheFileObject
const { video } = await getOrCreateAPVideo({ videoObject: cacheFile.object })
await sequelizeTypescript.transaction(async t => {
return createOrUpdateCacheFile(cacheFile, video, byActor, t)
})
if (video.isOwned()) {
// Don't resend the activity to the sender
const exceptions = [ byActor ]
await forwardVideoRelatedActivity(activity, undefined, exceptions, video)
}
}
async function processCreateWatchAction (activity: ActivityCreate) {
const watchAction = activity.object as WatchActionObject
if (watchAction.actionStatus !== 'CompletedActionStatus') return
const video = await VideoModel.loadByUrl(watchAction.object)
if (video.remote) return
await sequelizeTypescript.transaction(async t => {
return createOrUpdateLocalVideoViewer(watchAction, video, t)
})
}
async function processCreateVideoComment (activity: ActivityCreate, byActor: MActorSignature, notify: boolean) {
const commentObject = activity.object as VideoCommentObject
const byAccount = byActor.Account
if (!byAccount) throw new Error('Cannot create video comment with the non account actor ' + byActor.url)
let video: MVideoAccountLightBlacklistAllFiles
let created: boolean
let comment: MCommentOwnerVideo
try {
const resolveThreadResult = await resolveThread({ url: commentObject.id, isVideo: false })
video = resolveThreadResult.video
created = resolveThreadResult.commentCreated
comment = resolveThreadResult.comment
} catch (err) {
logger.debug(
'Cannot process video comment because we could not resolve thread %s. Maybe it was not a video thread, so skip it.',
commentObject.inReplyTo,
{ err }
)
return
}
// Try to not forward unwanted commments on our videos
if (video.isOwned()) {
if (await isBlockedByServerOrAccount(comment.Account, video.VideoChannel.Account)) {
logger.info('Skip comment forward from blocked account or server %s.', comment.Account.Actor.url)
return
}
if (created === true) {
// Don't resend the activity to the sender
const exceptions = [ byActor ]
await forwardVideoRelatedActivity(activity, undefined, exceptions, video)
}
}
if (created && notify) Notifier.Instance.notifyOnNewComment(comment)
}
async function processCreatePlaylist (activity: ActivityCreate, byActor: MActorSignature) {
const playlistObject = activity.object as PlaylistObject
const byAccount = byActor.Account
if (!byAccount) throw new Error('Cannot create video playlist with the non account actor ' + byActor.url)
await createOrUpdateVideoPlaylist(playlistObject, activity.to)
}
|
