aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/job-queue
diff options
context:
space:
mode:
authorChocobozzz <me@florianbigard.com>2019-08-22 10:43:11 +0200
committerChocobozzz <me@florianbigard.com>2019-08-22 10:43:11 +0200
commit5c5e587307a27e173333789b5b5167d35f468b01 (patch)
tree94e3721caf2e11d38fd5f4112c0fc98da89ac535 /server/lib/job-queue
parent1b42d73f44811eec1b7ddd72dd0d640a57c3376c (diff)
parentb5fecbf44192144d1ca27c23a0b53922de288c10 (diff)
downloadPeerTube-5c5e587307a27e173333789b5b5167d35f468b01.tar.gz
PeerTube-5c5e587307a27e173333789b5b5167d35f468b01.tar.zst
PeerTube-5c5e587307a27e173333789b5b5167d35f468b01.zip
Merge branch 'feature/strong-model-types' into develop
Diffstat (limited to 'server/lib/job-queue')
-rw-r--r--server/lib/job-queue/handlers/activitypub-follow.ts20
-rw-r--r--server/lib/job-queue/handlers/activitypub-http-fetcher.ts5
-rw-r--r--server/lib/job-queue/handlers/utils/activitypub-http-utils.ts4
-rw-r--r--server/lib/job-queue/handlers/video-file-import.ts11
-rw-r--r--server/lib/job-queue/handlers/video-import.ts52
-rw-r--r--server/lib/job-queue/handlers/video-transcoding.ts7
6 files changed, 59 insertions, 40 deletions
diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts
index 4ae66cd01..5cb55cad6 100644
--- a/server/lib/job-queue/handlers/activitypub-follow.ts
+++ b/server/lib/job-queue/handlers/activitypub-follow.ts
@@ -10,6 +10,7 @@ import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
10import { ActorModel } from '../../../models/activitypub/actor' 10import { ActorModel } from '../../../models/activitypub/actor'
11import { Notifier } from '../../notifier' 11import { Notifier } from '../../notifier'
12import { sequelizeTypescript } from '../../../initializers/database' 12import { sequelizeTypescript } from '../../../initializers/database'
13import { MAccount, MActor, MActorFollowActors, MActorFollowFull, MActorFull } from '../../../typings/models'
13 14
14export type ActivitypubFollowPayload = { 15export type ActivitypubFollowPayload = {
15 followerActorId: number 16 followerActorId: number
@@ -23,13 +24,13 @@ async function processActivityPubFollow (job: Bull.Job) {
23 24
24 logger.info('Processing ActivityPub follow in job %d.', job.id) 25 logger.info('Processing ActivityPub follow in job %d.', job.id)
25 26
26 let targetActor: ActorModel 27 let targetActor: MActorFull
27 if (!host || host === WEBSERVER.HOST) { 28 if (!host || host === WEBSERVER.HOST) {
28 targetActor = await ActorModel.loadLocalByName(payload.name) 29 targetActor = await ActorModel.loadLocalByName(payload.name)
29 } else { 30 } else {
30 const sanitizedHost = sanitizeHost(host, REMOTE_SCHEME.HTTP) 31 const sanitizedHost = sanitizeHost(host, REMOTE_SCHEME.HTTP)
31 const actorUrl = await loadActorUrlOrGetFromWebfinger(payload.name + '@' + sanitizedHost) 32 const actorUrl = await loadActorUrlOrGetFromWebfinger(payload.name + '@' + sanitizedHost)
32 targetActor = await getOrCreateActorAndServerAndModel(actorUrl) 33 targetActor = await getOrCreateActorAndServerAndModel(actorUrl, 'all')
33 } 34 }
34 35
35 const fromActor = await ActorModel.load(payload.followerActorId) 36 const fromActor = await ActorModel.load(payload.followerActorId)
@@ -44,7 +45,7 @@ export {
44 45
45// --------------------------------------------------------------------------- 46// ---------------------------------------------------------------------------
46 47
47async function follow (fromActor: ActorModel, targetActor: ActorModel) { 48async function follow (fromActor: MActor, targetActor: MActorFull) {
48 if (fromActor.id === targetActor.id) { 49 if (fromActor.id === targetActor.id) {
49 throw new Error('Follower is the same than target actor.') 50 throw new Error('Follower is the same than target actor.')
50 } 51 }
@@ -53,7 +54,7 @@ async function follow (fromActor: ActorModel, targetActor: ActorModel) {
53 const state = !fromActor.serverId && !targetActor.serverId ? 'accepted' : 'pending' 54 const state = !fromActor.serverId && !targetActor.serverId ? 'accepted' : 'pending'
54 55
55 const actorFollow = await sequelizeTypescript.transaction(async t => { 56 const actorFollow = await sequelizeTypescript.transaction(async t => {
56 const [ actorFollow ] = await ActorFollowModel.findOrCreate({ 57 const [ actorFollow ] = await ActorFollowModel.findOrCreate<MActorFollowActors>({
57 where: { 58 where: {
58 actorId: fromActor.id, 59 actorId: fromActor.id,
59 targetActorId: targetActor.id 60 targetActorId: targetActor.id
@@ -74,5 +75,14 @@ async function follow (fromActor: ActorModel, targetActor: ActorModel) {
74 return actorFollow 75 return actorFollow
75 }) 76 })
76 77
77 if (actorFollow.state === 'accepted') Notifier.Instance.notifyOfNewUserFollow(actorFollow) 78 if (actorFollow.state === 'accepted') {
79 const followerFull = Object.assign(fromActor, { Account: await actorFollow.ActorFollower.$get('Account') as MAccount })
80
81 const actorFollowFull = Object.assign(actorFollow, {
82 ActorFollowing: targetActor,
83 ActorFollower: followerFull
84 })
85
86 Notifier.Instance.notifyOfNewUserFollow(actorFollowFull)
87 }
78} 88}
diff --git a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
index c3f59dc77..0182c5169 100644
--- a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
@@ -11,6 +11,7 @@ import { AccountModel } from '../../../models/account/account'
11import { AccountVideoRateModel } from '../../../models/account/account-video-rate' 11import { AccountVideoRateModel } from '../../../models/account/account-video-rate'
12import { VideoShareModel } from '../../../models/video/video-share' 12import { VideoShareModel } from '../../../models/video/video-share'
13import { VideoCommentModel } from '../../../models/video/video-comment' 13import { VideoCommentModel } from '../../../models/video/video-comment'
14import { MAccountDefault, MVideoFullLight } from '../../../typings/models'
14 15
15type FetchType = 'activity' | 'video-likes' | 'video-dislikes' | 'video-shares' | 'video-comments' | 'account-playlists' 16type FetchType = 'activity' | 'video-likes' | 'video-dislikes' | 'video-shares' | 'video-comments' | 'account-playlists'
16 17
@@ -26,10 +27,10 @@ async function processActivityPubHttpFetcher (job: Bull.Job) {
26 27
27 const payload = job.data as ActivitypubHttpFetcherPayload 28 const payload = job.data as ActivitypubHttpFetcherPayload
28 29
29 let video: VideoModel 30 let video: MVideoFullLight
30 if (payload.videoId) video = await VideoModel.loadAndPopulateAccountAndServerAndTags(payload.videoId) 31 if (payload.videoId) video = await VideoModel.loadAndPopulateAccountAndServerAndTags(payload.videoId)
31 32
32 let account: AccountModel 33 let account: MAccountDefault
33 if (payload.accountId) account = await AccountModel.load(payload.accountId) 34 if (payload.accountId) account = await AccountModel.load(payload.accountId)
34 35
35 const fetcherType: { [ id in FetchType ]: (items: any[]) => Promise<any> } = { 36 const fetcherType: { [ id in FetchType ]: (items: any[]) => Promise<any> } = {
diff --git a/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts b/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
index cdee1f6fd..d3bde6e6a 100644
--- a/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
+++ b/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
@@ -3,6 +3,7 @@ import { getServerActor } from '../../../../helpers/utils'
3import { ActorModel } from '../../../../models/activitypub/actor' 3import { ActorModel } from '../../../../models/activitypub/actor'
4import { sha256 } from '../../../../helpers/core-utils' 4import { sha256 } from '../../../../helpers/core-utils'
5import { HTTP_SIGNATURE } from '../../../../initializers/constants' 5import { HTTP_SIGNATURE } from '../../../../initializers/constants'
6import { MActor } from '../../../../typings/models'
6 7
7type Payload = { body: any, signatureActorId?: number } 8type Payload = { body: any, signatureActorId?: number }
8 9
@@ -19,7 +20,8 @@ async function computeBody (payload: Payload) {
19} 20}
20 21
21async function buildSignedRequestOptions (payload: Payload) { 22async function buildSignedRequestOptions (payload: Payload) {
22 let actor: ActorModel | null 23 let actor: MActor | null
24
23 if (payload.signatureActorId) { 25 if (payload.signatureActorId) {
24 actor = await ActorModel.load(payload.signatureActorId) 26 actor = await ActorModel.load(payload.signatureActorId)
25 if (!actor) throw new Error('Unknown signature actor id.') 27 if (!actor) throw new Error('Unknown signature actor id.')
diff --git a/server/lib/job-queue/handlers/video-file-import.ts b/server/lib/job-queue/handlers/video-file-import.ts
index 8cacb0ef3..5c5b7dccb 100644
--- a/server/lib/job-queue/handlers/video-file-import.ts
+++ b/server/lib/job-queue/handlers/video-file-import.ts
@@ -6,6 +6,7 @@ import { getVideoFileFPS, getVideoFileResolution } from '../../../helpers/ffmpeg
6import { copy, stat } from 'fs-extra' 6import { copy, stat } from 'fs-extra'
7import { VideoFileModel } from '../../../models/video/video-file' 7import { VideoFileModel } from '../../../models/video/video-file'
8import { extname } from 'path' 8import { extname } from 'path'
9import { MVideoFile, MVideoWithFile } from '@server/typings/models'
9 10
10export type VideoFileImportPayload = { 11export type VideoFileImportPayload = {
11 videoUUID: string, 12 videoUUID: string,
@@ -37,7 +38,7 @@ export {
37 38
38// --------------------------------------------------------------------------- 39// ---------------------------------------------------------------------------
39 40
40async function updateVideoFile (video: VideoModel, inputFilePath: string) { 41async function updateVideoFile (video: MVideoWithFile, inputFilePath: string) {
41 const { videoFileResolution } = await getVideoFileResolution(inputFilePath) 42 const { videoFileResolution } = await getVideoFileResolution(inputFilePath)
42 const { size } = await stat(inputFilePath) 43 const { size } = await stat(inputFilePath)
43 const fps = await getVideoFileFPS(inputFilePath) 44 const fps = await getVideoFileFPS(inputFilePath)
@@ -48,7 +49,7 @@ async function updateVideoFile (video: VideoModel, inputFilePath: string) {
48 size, 49 size,
49 fps, 50 fps,
50 videoId: video.id 51 videoId: video.id
51 }) 52 }) as MVideoFile
52 53
53 const currentVideoFile = video.VideoFiles.find(videoFile => videoFile.resolution === updatedVideoFile.resolution) 54 const currentVideoFile = video.VideoFiles.find(videoFile => videoFile.resolution === updatedVideoFile.resolution)
54 55
@@ -60,9 +61,9 @@ async function updateVideoFile (video: VideoModel, inputFilePath: string) {
60 video.VideoFiles = video.VideoFiles.filter(f => f !== currentVideoFile) 61 video.VideoFiles = video.VideoFiles.filter(f => f !== currentVideoFile)
61 62
62 // Update the database 63 // Update the database
63 currentVideoFile.set('extname', updatedVideoFile.extname) 64 currentVideoFile.extname = updatedVideoFile.extname
64 currentVideoFile.set('size', updatedVideoFile.size) 65 currentVideoFile.size = updatedVideoFile.size
65 currentVideoFile.set('fps', updatedVideoFile.fps) 66 currentVideoFile.fps = updatedVideoFile.fps
66 67
67 updatedVideoFile = currentVideoFile 68 updatedVideoFile = currentVideoFile
68 } 69 }
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts
index 13b741180..ff8c93328 100644
--- a/server/lib/job-queue/handlers/video-import.ts
+++ b/server/lib/job-queue/handlers/video-import.ts
@@ -17,9 +17,10 @@ import { move, remove, stat } from 'fs-extra'
17import { Notifier } from '../../notifier' 17import { Notifier } from '../../notifier'
18import { CONFIG } from '../../../initializers/config' 18import { CONFIG } from '../../../initializers/config'
19import { sequelizeTypescript } from '../../../initializers/database' 19import { sequelizeTypescript } from '../../../initializers/database'
20import { ThumbnailModel } from '../../../models/video/thumbnail'
21import { createVideoMiniatureFromUrl, generateVideoMiniature } from '../../thumbnail' 20import { createVideoMiniatureFromUrl, generateVideoMiniature } from '../../thumbnail'
22import { ThumbnailType } from '../../../../shared/models/videos/thumbnail.type' 21import { ThumbnailType } from '../../../../shared/models/videos/thumbnail.type'
22import { MThumbnail } from '../../../typings/models/video/thumbnail'
23import { MVideoImportDefault, MVideoImportDefaultFiles, MVideoImportVideo } from '@server/typings/models/video/video-import'
23 24
24type VideoImportYoutubeDLPayload = { 25type VideoImportYoutubeDLPayload = {
25 type: 'youtube-dl' 26 type: 'youtube-dl'
@@ -110,7 +111,7 @@ type ProcessFileOptions = {
110 generateThumbnail: boolean 111 generateThumbnail: boolean
111 generatePreview: boolean 112 generatePreview: boolean
112} 113}
113async function processFile (downloader: () => Promise<string>, videoImport: VideoImportModel, options: ProcessFileOptions) { 114async function processFile (downloader: () => Promise<string>, videoImport: MVideoImportDefault, options: ProcessFileOptions) {
114 let tempVideoPath: string 115 let tempVideoPath: string
115 let videoDestFile: string 116 let videoDestFile: string
116 let videoFile: VideoFileModel 117 let videoFile: VideoFileModel
@@ -139,41 +140,44 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
139 videoId: videoImport.videoId 140 videoId: videoImport.videoId
140 } 141 }
141 videoFile = new VideoFileModel(videoFileData) 142 videoFile = new VideoFileModel(videoFileData)
143
144 const videoWithFiles = Object.assign(videoImport.Video, { VideoFiles: [ videoFile ] })
142 // To clean files if the import fails 145 // To clean files if the import fails
143 videoImport.Video.VideoFiles = [ videoFile ] 146 const videoImportWithFiles: MVideoImportDefaultFiles = Object.assign(videoImport, { Video: videoWithFiles })
144 147
145 // Move file 148 // Move file
146 videoDestFile = join(CONFIG.STORAGE.VIDEOS_DIR, videoImport.Video.getVideoFilename(videoFile)) 149 videoDestFile = join(CONFIG.STORAGE.VIDEOS_DIR, videoImportWithFiles.Video.getVideoFilename(videoFile))
147 await move(tempVideoPath, videoDestFile) 150 await move(tempVideoPath, videoDestFile)
148 tempVideoPath = null // This path is not used anymore 151 tempVideoPath = null // This path is not used anymore
149 152
150 // Process thumbnail 153 // Process thumbnail
151 let thumbnailModel: ThumbnailModel 154 let thumbnailModel: MThumbnail
152 if (options.downloadThumbnail && options.thumbnailUrl) { 155 if (options.downloadThumbnail && options.thumbnailUrl) {
153 thumbnailModel = await createVideoMiniatureFromUrl(options.thumbnailUrl, videoImport.Video, ThumbnailType.MINIATURE) 156 thumbnailModel = await createVideoMiniatureFromUrl(options.thumbnailUrl, videoImportWithFiles.Video, ThumbnailType.MINIATURE)
154 } else if (options.generateThumbnail || options.downloadThumbnail) { 157 } else if (options.generateThumbnail || options.downloadThumbnail) {
155 thumbnailModel = await generateVideoMiniature(videoImport.Video, videoFile, ThumbnailType.MINIATURE) 158 thumbnailModel = await generateVideoMiniature(videoImportWithFiles.Video, videoFile, ThumbnailType.MINIATURE)
156 } 159 }
157 160
158 // Process preview 161 // Process preview
159 let previewModel: ThumbnailModel 162 let previewModel: MThumbnail
160 if (options.downloadPreview && options.thumbnailUrl) { 163 if (options.downloadPreview && options.thumbnailUrl) {
161 previewModel = await createVideoMiniatureFromUrl(options.thumbnailUrl, videoImport.Video, ThumbnailType.PREVIEW) 164 previewModel = await createVideoMiniatureFromUrl(options.thumbnailUrl, videoImportWithFiles.Video, ThumbnailType.PREVIEW)
162 } else if (options.generatePreview || options.downloadPreview) { 165 } else if (options.generatePreview || options.downloadPreview) {
163 previewModel = await generateVideoMiniature(videoImport.Video, videoFile, ThumbnailType.PREVIEW) 166 previewModel = await generateVideoMiniature(videoImportWithFiles.Video, videoFile, ThumbnailType.PREVIEW)
164 } 167 }
165 168
166 // Create torrent 169 // Create torrent
167 await videoImport.Video.createTorrentAndSetInfoHash(videoFile) 170 await videoImportWithFiles.Video.createTorrentAndSetInfoHash(videoFile)
171
172 const { videoImportUpdated, video } = await sequelizeTypescript.transaction(async t => {
173 const videoImportToUpdate = videoImportWithFiles as MVideoImportVideo
168 174
169 const videoImportUpdated: VideoImportModel = await sequelizeTypescript.transaction(async t => {
170 // Refresh video 175 // Refresh video
171 const video = await VideoModel.load(videoImport.videoId, t) 176 const video = await VideoModel.load(videoImportToUpdate.videoId, t)
172 if (!video) throw new Error('Video linked to import ' + videoImport.videoId + ' does not exist anymore.') 177 if (!video) throw new Error('Video linked to import ' + videoImportToUpdate.videoId + ' does not exist anymore.')
173 videoImport.Video = video
174 178
175 const videoFileCreated = await videoFile.save({ transaction: t }) 179 const videoFileCreated = await videoFile.save({ transaction: t })
176 video.VideoFiles = [ videoFileCreated ] 180 videoImportToUpdate.Video = Object.assign(video, { VideoFiles: [ videoFileCreated ] })
177 181
178 // Update video DB object 182 // Update video DB object
179 video.duration = duration 183 video.duration = duration
@@ -188,25 +192,25 @@ async function processFile (downloader: () => Promise<string>, videoImport: Vide
188 await federateVideoIfNeeded(videoForFederation, true, t) 192 await federateVideoIfNeeded(videoForFederation, true, t)
189 193
190 // Update video import object 194 // Update video import object
191 videoImport.state = VideoImportState.SUCCESS 195 videoImportToUpdate.state = VideoImportState.SUCCESS
192 const videoImportUpdated = await videoImport.save({ transaction: t }) 196 const videoImportUpdated = await videoImportToUpdate.save({ transaction: t }) as MVideoImportVideo
197 videoImportUpdated.Video = video
193 198
194 logger.info('Video %s imported.', video.uuid) 199 logger.info('Video %s imported.', video.uuid)
195 200
196 videoImportUpdated.Video = videoForFederation 201 return { videoImportUpdated, video: videoForFederation }
197 return videoImportUpdated
198 }) 202 })
199 203
200 Notifier.Instance.notifyOnFinishedVideoImport(videoImportUpdated, true) 204 Notifier.Instance.notifyOnFinishedVideoImport(videoImportUpdated, true)
201 205
202 if (videoImportUpdated.Video.isBlacklisted()) { 206 if (video.isBlacklisted()) {
203 Notifier.Instance.notifyOnVideoAutoBlacklist(videoImportUpdated.Video) 207 Notifier.Instance.notifyOnVideoAutoBlacklist(video)
204 } else { 208 } else {
205 Notifier.Instance.notifyOnNewVideoIfNeeded(videoImportUpdated.Video) 209 Notifier.Instance.notifyOnNewVideoIfNeeded(video)
206 } 210 }
207 211
208 // Create transcoding jobs? 212 // Create transcoding jobs?
209 if (videoImportUpdated.Video.state === VideoState.TO_TRANSCODE) { 213 if (video.state === VideoState.TO_TRANSCODE) {
210 // Put uuid because we don't have id auto incremented for now 214 // Put uuid because we don't have id auto incremented for now
211 const dataInput = { 215 const dataInput = {
212 type: 'optimize' as 'optimize', 216 type: 'optimize' as 'optimize',
diff --git a/server/lib/job-queue/handlers/video-transcoding.ts b/server/lib/job-queue/handlers/video-transcoding.ts
index 981daf9a1..2ebe15bcb 100644
--- a/server/lib/job-queue/handlers/video-transcoding.ts
+++ b/server/lib/job-queue/handlers/video-transcoding.ts
@@ -11,6 +11,7 @@ import { computeResolutionsToTranscode } from '../../../helpers/ffmpeg-utils'
11import { generateHlsPlaylist, optimizeVideofile, transcodeOriginalVideofile, mergeAudioVideofile } from '../../video-transcoding' 11import { generateHlsPlaylist, optimizeVideofile, transcodeOriginalVideofile, mergeAudioVideofile } from '../../video-transcoding'
12import { Notifier } from '../../notifier' 12import { Notifier } from '../../notifier'
13import { CONFIG } from '../../../initializers/config' 13import { CONFIG } from '../../../initializers/config'
14import { MVideoUUID, MVideoWithFile } from '@server/typings/models'
14 15
15interface BaseTranscodingPayload { 16interface BaseTranscodingPayload {
16 videoUUID: string 17 videoUUID: string
@@ -73,7 +74,7 @@ async function processVideoTranscoding (job: Bull.Job) {
73 return video 74 return video
74} 75}
75 76
76async function onHlsPlaylistGenerationSuccess (video: VideoModel) { 77async function onHlsPlaylistGenerationSuccess (video: MVideoUUID) {
77 if (video === undefined) return undefined 78 if (video === undefined) return undefined
78 79
79 await sequelizeTypescript.transaction(async t => { 80 await sequelizeTypescript.transaction(async t => {
@@ -87,7 +88,7 @@ async function onHlsPlaylistGenerationSuccess (video: VideoModel) {
87 }) 88 })
88} 89}
89 90
90async function publishNewResolutionIfNeeded (video: VideoModel, payload?: NewResolutionTranscodingPayload | MergeAudioTranscodingPayload) { 91async function publishNewResolutionIfNeeded (video: MVideoUUID, payload?: NewResolutionTranscodingPayload | MergeAudioTranscodingPayload) {
91 const { videoDatabase, videoPublished } = await sequelizeTypescript.transaction(async t => { 92 const { videoDatabase, videoPublished } = await sequelizeTypescript.transaction(async t => {
92 // Maybe the video changed in database, refresh it 93 // Maybe the video changed in database, refresh it
93 let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t) 94 let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
@@ -119,7 +120,7 @@ async function publishNewResolutionIfNeeded (video: VideoModel, payload?: NewRes
119 await createHlsJobIfEnabled(payload) 120 await createHlsJobIfEnabled(payload)
120} 121}
121 122
122async function onVideoFileOptimizerSuccess (videoArg: VideoModel, payload: OptimizeTranscodingPayload) { 123async function onVideoFileOptimizerSuccess (videoArg: MVideoWithFile, payload: OptimizeTranscodingPayload) {
123 if (videoArg === undefined) return undefined 124 if (videoArg === undefined) return undefined
124 125
125 // Outside the transaction (IO on disk) 126 // Outside the transaction (IO on disk)