diff options
Diffstat (limited to 'server/lib/job-queue/handlers/move-to-object-storage.ts')
-rw-r--r-- | server/lib/job-queue/handlers/move-to-object-storage.ts | 114 |
1 files changed, 114 insertions, 0 deletions
diff --git a/server/lib/job-queue/handlers/move-to-object-storage.ts b/server/lib/job-queue/handlers/move-to-object-storage.ts new file mode 100644 index 000000000..a0c58d211 --- /dev/null +++ b/server/lib/job-queue/handlers/move-to-object-storage.ts | |||
@@ -0,0 +1,114 @@ | |||
1 | import * as Bull from 'bull' | ||
2 | import { remove } from 'fs-extra' | ||
3 | import { join } from 'path' | ||
4 | import { logger } from '@server/helpers/logger' | ||
5 | import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent' | ||
6 | import { CONFIG } from '@server/initializers/config' | ||
7 | import { storeHLSFile, storeWebTorrentFile } from '@server/lib/object-storage' | ||
8 | import { getHLSDirectory, getHlsResolutionPlaylistFilename } from '@server/lib/paths' | ||
9 | import { moveToNextState } from '@server/lib/video-state' | ||
10 | import { VideoModel } from '@server/models/video/video' | ||
11 | import { VideoJobInfoModel } from '@server/models/video/video-job-info' | ||
12 | import { MStreamingPlaylistVideo, MVideo, MVideoFile, MVideoWithAllFiles } from '@server/types/models' | ||
13 | import { MoveObjectStoragePayload, VideoStorage } from '../../../../shared' | ||
14 | |||
15 | export async function processMoveToObjectStorage (job: Bull.Job) { | ||
16 | const payload = job.data as MoveObjectStoragePayload | ||
17 | logger.info('Moving video %s in job %d.', payload.videoUUID, job.id) | ||
18 | |||
19 | const video = await VideoModel.loadWithFiles(payload.videoUUID) | ||
20 | // No video, maybe deleted? | ||
21 | if (!video) { | ||
22 | logger.info('Can\'t process job %d, video does not exist.', job.id) | ||
23 | return undefined | ||
24 | } | ||
25 | |||
26 | if (video.VideoFiles) { | ||
27 | await moveWebTorrentFiles(video) | ||
28 | } | ||
29 | |||
30 | if (video.VideoStreamingPlaylists) { | ||
31 | await moveHLSFiles(video) | ||
32 | } | ||
33 | |||
34 | const pendingMove = await VideoJobInfoModel.decrease(video.uuid, 'pendingMove') | ||
35 | if (pendingMove === 0) { | ||
36 | logger.info('Running cleanup after moving files to object storage (video %s in job %d)', video.uuid, job.id) | ||
37 | await doAfterLastJob(video, payload.isNewVideo) | ||
38 | } | ||
39 | |||
40 | return payload.videoUUID | ||
41 | } | ||
42 | |||
43 | // --------------------------------------------------------------------------- | ||
44 | |||
45 | async function moveWebTorrentFiles (video: MVideoWithAllFiles) { | ||
46 | for (const file of video.VideoFiles) { | ||
47 | if (file.storage !== VideoStorage.FILE_SYSTEM) continue | ||
48 | |||
49 | const fileUrl = await storeWebTorrentFile(file.filename) | ||
50 | |||
51 | const oldPath = join(CONFIG.STORAGE.VIDEOS_DIR, file.filename) | ||
52 | await onFileMoved({ videoOrPlaylist: video, file, fileUrl, oldPath }) | ||
53 | } | ||
54 | } | ||
55 | |||
56 | async function moveHLSFiles (video: MVideoWithAllFiles) { | ||
57 | for (const playlist of video.VideoStreamingPlaylists) { | ||
58 | |||
59 | for (const file of playlist.VideoFiles) { | ||
60 | if (file.storage !== VideoStorage.FILE_SYSTEM) continue | ||
61 | |||
62 | // Resolution playlist | ||
63 | const playlistFilename = getHlsResolutionPlaylistFilename(file.filename) | ||
64 | await storeHLSFile(playlist, video, playlistFilename) | ||
65 | |||
66 | // Resolution fragmented file | ||
67 | const fileUrl = await storeHLSFile(playlist, video, file.filename) | ||
68 | |||
69 | const oldPath = join(getHLSDirectory(video), file.filename) | ||
70 | |||
71 | await onFileMoved({ videoOrPlaylist: Object.assign(playlist, { Video: video }), file, fileUrl, oldPath }) | ||
72 | } | ||
73 | } | ||
74 | } | ||
75 | |||
76 | async function doAfterLastJob (video: MVideoWithAllFiles, isNewVideo: boolean) { | ||
77 | for (const playlist of video.VideoStreamingPlaylists) { | ||
78 | if (playlist.storage === VideoStorage.OBJECT_STORAGE) continue | ||
79 | |||
80 | // Master playlist | ||
81 | playlist.playlistUrl = await storeHLSFile(playlist, video, playlist.playlistFilename) | ||
82 | // Sha256 segments file | ||
83 | playlist.segmentsSha256Url = await storeHLSFile(playlist, video, playlist.segmentsSha256Filename) | ||
84 | |||
85 | playlist.storage = VideoStorage.OBJECT_STORAGE | ||
86 | |||
87 | await playlist.save() | ||
88 | } | ||
89 | |||
90 | // Remove empty hls video directory | ||
91 | if (video.VideoStreamingPlaylists) { | ||
92 | await remove(getHLSDirectory(video)) | ||
93 | } | ||
94 | |||
95 | await moveToNextState(video, isNewVideo) | ||
96 | } | ||
97 | |||
98 | async function onFileMoved (options: { | ||
99 | videoOrPlaylist: MVideo | MStreamingPlaylistVideo | ||
100 | file: MVideoFile | ||
101 | fileUrl: string | ||
102 | oldPath: string | ||
103 | }) { | ||
104 | const { videoOrPlaylist, file, fileUrl, oldPath } = options | ||
105 | |||
106 | file.fileUrl = fileUrl | ||
107 | file.storage = VideoStorage.OBJECT_STORAGE | ||
108 | |||
109 | await createTorrentAndSetInfoHash(videoOrPlaylist, file) | ||
110 | await file.save() | ||
111 | |||
112 | logger.debug('Removing %s because it\'s now on object storage', oldPath) | ||
113 | await remove(oldPath) | ||
114 | } | ||