diff options
author | Florent <florent.git@zeteo.me> | 2022-08-10 09:53:39 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2022-08-10 09:53:39 +0200 |
commit | 2a491182e483b97afb1b65c908b23cb48d591807 (patch) | |
tree | ec13503216ad72a3ea8f1ce3b659899f8167fb47 /server/lib/job-queue/handlers | |
parent | 06ac128958c489efe1008eeca1df683819bd2f18 (diff) | |
download | PeerTube-2a491182e483b97afb1b65c908b23cb48d591807.tar.gz PeerTube-2a491182e483b97afb1b65c908b23cb48d591807.tar.zst PeerTube-2a491182e483b97afb1b65c908b23cb48d591807.zip |
Channel sync (#5135)
* Add external channel URL for channel update / creation (#754)
* Disallow synchronisation if user has no video quota (#754)
* More constraints serverside (#754)
* Disable sync if server configuration does not allow HTTP import (#754)
* Working version synchronizing videos with a job (#754)
TODO: refactoring, too much code duplication
* More logs and try/catch (#754)
* Fix eslint error (#754)
* WIP: support synchronization time change (#754)
* New frontend #754
* WIP: Create sync front (#754)
* Enhance UI, sync creation form (#754)
* Warning message when HTTP upload is disallowed
* More consistent names (#754)
* Binding Front with API (#754)
* Add a /me API (#754)
* Improve list UI (#754)
* Implement creation and deletion routes (#754)
* Lint (#754)
* Lint again (#754)
* WIP: UI for triggering import existing videos (#754)
* Implement jobs for syncing and importing channels
* Don't sync videos before sync creation + avoid concurrency issue (#754)
* Cleanup (#754)
* Cleanup: OpenAPI + API rework (#754)
* Remove dead code (#754)
* Eslint (#754)
* Revert the mess with whitespaces in constants.ts (#754)
* Some fixes after rebase (#754)
* Several fixes after PR remarks (#754)
* Front + API: Rename video-channels-sync to video-channel-syncs (#754)
* Allow enabling channel sync through UI (#754)
* getChannelInfo (#754)
* Minor fixes: openapi + model + sql (#754)
* Simplified API validators (#754)
* Rename MChannelSync to MChannelSyncChannel (#754)
* Add command for VideoChannelSync (#754)
* Use synchronization.enabled config (#754)
* Check parameters test + some fixes (#754)
* Fix conflict mistake (#754)
* Restrict access to video channel sync list API (#754)
* Start adding unit test for synchronization (#754)
* Continue testing (#754)
* Tests finished + convertion of job to scheduler (#754)
* Add lastSyncAt field (#754)
* Fix externalRemoteUrl sort + creation date not well formatted (#754)
* Small fix (#754)
* Factorize addYoutubeDLImport and buildVideo (#754)
* Check duplicates on channel not on users (#754)
* factorize thumbnail generation (#754)
* Fetch error should return status 400 (#754)
* Separate video-channel-import and video-channel-sync-latest (#754)
* Bump DB migration version after rebase (#754)
* Prettier states in UI table (#754)
* Add DefaultScope in VideoChannelSyncModel (#754)
* Fix audit logs (#754)
* Ensure user can upload when importing channel + minor fixes (#754)
* Mark synchronization as failed on exception + typos (#754)
* Change REST API for importing videos into channel (#754)
* Add option for fully synchronize a chnanel (#754)
* Return a whole sync object on creation to avoid tricks in Front (#754)
* Various remarks (#754)
* Single quotes by default (#754)
* Rename synchronization to video_channel_synchronization
* Add check.latest_videos_count and max_per_user options (#754)
* Better channel rendering in list #754
* Allow sorting with channel name and state (#754)
* Add missing tests for channel imports (#754)
* Prefer using a parent job for channel sync
* Styling
* Client styling
Co-authored-by: Chocobozzz <me@florianbigard.com>
Diffstat (limited to 'server/lib/job-queue/handlers')
-rw-r--r-- | server/lib/job-queue/handlers/after-video-channel-import.ts | 37 | ||||
-rw-r--r-- | server/lib/job-queue/handlers/video-channel-import.ts | 36 | ||||
-rw-r--r-- | server/lib/job-queue/handlers/video-import.ts | 20 |
3 files changed, 88 insertions, 5 deletions
diff --git a/server/lib/job-queue/handlers/after-video-channel-import.ts b/server/lib/job-queue/handlers/after-video-channel-import.ts new file mode 100644 index 000000000..ffdd8c5b5 --- /dev/null +++ b/server/lib/job-queue/handlers/after-video-channel-import.ts | |||
@@ -0,0 +1,37 @@ | |||
1 | import { Job } from 'bullmq' | ||
2 | import { logger } from '@server/helpers/logger' | ||
3 | import { VideoChannelSyncModel } from '@server/models/video/video-channel-sync' | ||
4 | import { AfterVideoChannelImportPayload, VideoChannelSyncState, VideoImportPreventExceptionResult } from '@shared/models' | ||
5 | |||
6 | export async function processAfterVideoChannelImport (job: Job) { | ||
7 | const payload = job.data as AfterVideoChannelImportPayload | ||
8 | if (!payload.channelSyncId) return | ||
9 | |||
10 | logger.info('Processing after video channel import in job %s.', job.id) | ||
11 | |||
12 | const sync = await VideoChannelSyncModel.loadWithChannel(payload.channelSyncId) | ||
13 | if (!sync) { | ||
14 | logger.error('Unknown sync id %d.', payload.channelSyncId) | ||
15 | return | ||
16 | } | ||
17 | |||
18 | const childrenValues = await job.getChildrenValues<VideoImportPreventExceptionResult>() | ||
19 | |||
20 | let errors = 0 | ||
21 | let successes = 0 | ||
22 | |||
23 | for (const value of Object.values(childrenValues)) { | ||
24 | if (value.resultType === 'success') successes++ | ||
25 | else if (value.resultType === 'error') errors++ | ||
26 | } | ||
27 | |||
28 | if (errors > 0) { | ||
29 | sync.state = VideoChannelSyncState.FAILED | ||
30 | logger.error(`Finished synchronizing "${sync.VideoChannel.Actor.preferredUsername}" with failures.`, { errors, successes }) | ||
31 | } else { | ||
32 | sync.state = VideoChannelSyncState.SYNCED | ||
33 | logger.info(`Finished synchronizing "${sync.VideoChannel.Actor.preferredUsername}" successfully.`, { successes }) | ||
34 | } | ||
35 | |||
36 | await sync.save() | ||
37 | } | ||
diff --git a/server/lib/job-queue/handlers/video-channel-import.ts b/server/lib/job-queue/handlers/video-channel-import.ts new file mode 100644 index 000000000..9bdb2d269 --- /dev/null +++ b/server/lib/job-queue/handlers/video-channel-import.ts | |||
@@ -0,0 +1,36 @@ | |||
1 | import { Job } from 'bullmq' | ||
2 | import { logger } from '@server/helpers/logger' | ||
3 | import { CONFIG } from '@server/initializers/config' | ||
4 | import { synchronizeChannel } from '@server/lib/sync-channel' | ||
5 | import { VideoChannelModel } from '@server/models/video/video-channel' | ||
6 | import { VideoChannelImportPayload } from '@shared/models' | ||
7 | |||
8 | export async function processVideoChannelImport (job: Job) { | ||
9 | const payload = job.data as VideoChannelImportPayload | ||
10 | |||
11 | logger.info('Processing video channel import in job %s.', job.id) | ||
12 | |||
13 | // Channel import requires only http upload to be allowed | ||
14 | if (!CONFIG.IMPORT.VIDEOS.HTTP.ENABLED) { | ||
15 | logger.error('Cannot import channel as the HTTP upload is disabled') | ||
16 | return | ||
17 | } | ||
18 | |||
19 | if (!CONFIG.IMPORT.VIDEO_CHANNEL_SYNCHRONIZATION.ENABLED) { | ||
20 | logger.error('Cannot import channel as the synchronization is disabled') | ||
21 | return | ||
22 | } | ||
23 | |||
24 | const videoChannel = await VideoChannelModel.loadAndPopulateAccount(payload.videoChannelId) | ||
25 | |||
26 | try { | ||
27 | logger.info(`Starting importing videos from external channel "${payload.externalChannelUrl}" to "${videoChannel.name}" `) | ||
28 | |||
29 | await synchronizeChannel({ | ||
30 | channel: videoChannel, | ||
31 | externalChannelUrl: payload.externalChannelUrl | ||
32 | }) | ||
33 | } catch (err) { | ||
34 | logger.error(`Failed to import channel ${videoChannel.name}`, { err }) | ||
35 | } | ||
36 | } | ||
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts index f4629159c..9901b878c 100644 --- a/server/lib/job-queue/handlers/video-import.ts +++ b/server/lib/job-queue/handlers/video-import.ts | |||
@@ -8,7 +8,7 @@ import { generateWebTorrentVideoFilename } from '@server/lib/paths' | |||
8 | import { Hooks } from '@server/lib/plugins/hooks' | 8 | import { Hooks } from '@server/lib/plugins/hooks' |
9 | import { ServerConfigManager } from '@server/lib/server-config-manager' | 9 | import { ServerConfigManager } from '@server/lib/server-config-manager' |
10 | import { isAbleToUploadVideo } from '@server/lib/user' | 10 | import { isAbleToUploadVideo } from '@server/lib/user' |
11 | import { buildOptimizeOrMergeAudioJob, buildMoveToObjectStorageJob } from '@server/lib/video' | 11 | import { buildMoveToObjectStorageJob, buildOptimizeOrMergeAudioJob } from '@server/lib/video' |
12 | import { VideoPathManager } from '@server/lib/video-path-manager' | 12 | import { VideoPathManager } from '@server/lib/video-path-manager' |
13 | import { buildNextVideoState } from '@server/lib/video-state' | 13 | import { buildNextVideoState } from '@server/lib/video-state' |
14 | import { ThumbnailModel } from '@server/models/video/thumbnail' | 14 | import { ThumbnailModel } from '@server/models/video/thumbnail' |
@@ -18,6 +18,7 @@ import { isAudioFile } from '@shared/extra-utils' | |||
18 | import { | 18 | import { |
19 | ThumbnailType, | 19 | ThumbnailType, |
20 | VideoImportPayload, | 20 | VideoImportPayload, |
21 | VideoImportPreventExceptionResult, | ||
21 | VideoImportState, | 22 | VideoImportState, |
22 | VideoImportTorrentPayload, | 23 | VideoImportTorrentPayload, |
23 | VideoImportTorrentPayloadType, | 24 | VideoImportTorrentPayloadType, |
@@ -41,20 +42,29 @@ import { Notifier } from '../../notifier' | |||
41 | import { generateVideoMiniature } from '../../thumbnail' | 42 | import { generateVideoMiniature } from '../../thumbnail' |
42 | import { JobQueue } from '../job-queue' | 43 | import { JobQueue } from '../job-queue' |
43 | 44 | ||
44 | async function processVideoImport (job: Job) { | 45 | async function processVideoImport (job: Job): Promise<VideoImportPreventExceptionResult> { |
45 | const payload = job.data as VideoImportPayload | 46 | const payload = job.data as VideoImportPayload |
46 | 47 | ||
47 | const videoImport = await getVideoImportOrDie(payload) | 48 | const videoImport = await getVideoImportOrDie(payload) |
48 | if (videoImport.state === VideoImportState.CANCELLED) { | 49 | if (videoImport.state === VideoImportState.CANCELLED) { |
49 | logger.info('Do not process import since it has been cancelled', { payload }) | 50 | logger.info('Do not process import since it has been cancelled', { payload }) |
50 | return | 51 | return { resultType: 'success' } |
51 | } | 52 | } |
52 | 53 | ||
53 | videoImport.state = VideoImportState.PROCESSING | 54 | videoImport.state = VideoImportState.PROCESSING |
54 | await videoImport.save() | 55 | await videoImport.save() |
55 | 56 | ||
56 | if (payload.type === 'youtube-dl') return processYoutubeDLImport(job, videoImport, payload) | 57 | try { |
57 | if (payload.type === 'magnet-uri' || payload.type === 'torrent-file') return processTorrentImport(job, videoImport, payload) | 58 | if (payload.type === 'youtube-dl') await processYoutubeDLImport(job, videoImport, payload) |
59 | if (payload.type === 'magnet-uri' || payload.type === 'torrent-file') await processTorrentImport(job, videoImport, payload) | ||
60 | |||
61 | return { resultType: 'success' } | ||
62 | } catch (err) { | ||
63 | if (!payload.preventException) throw err | ||
64 | |||
65 | logger.warn('Catch error in video import to send value to parent job.', { payload, err }) | ||
66 | return { resultType: 'error' } | ||
67 | } | ||
58 | } | 68 | } |
59 | 69 | ||
60 | // --------------------------------------------------------------------------- | 70 | // --------------------------------------------------------------------------- |