From 4ec52d04dcc5d664612331f8e08d7d90da990415 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Thu, 21 Apr 2022 09:06:52 +0200 Subject: Add ability to save replay of permanent lives --- server/lib/job-queue/handlers/video-live-ending.ts | 159 +++++++++++++++------ server/lib/live/live-manager.ts | 69 +++++---- server/lib/live/live-utils.ts | 4 +- server/lib/live/shared/muxing-session.ts | 33 ++--- server/lib/paths.ts | 7 +- server/lib/transcoding/transcoding.ts | 10 +- server/lib/video-blacklist.ts | 3 +- 7 files changed, 190 insertions(+), 95 deletions(-) (limited to 'server/lib') diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts index f4de4b47c..1e290338c 100644 --- a/server/lib/job-queue/handlers/video-live-ending.ts +++ b/server/lib/job-queue/handlers/video-live-ending.ts @@ -1,25 +1,33 @@ import { Job } from 'bull' import { pathExists, readdir, remove } from 'fs-extra' import { join } from 'path' -import { ffprobePromise, getAudioStream, getVideoStreamDuration, getVideoStreamDimensionsInfo } from '@server/helpers/ffmpeg' -import { VIDEO_LIVE } from '@server/initializers/constants' -import { buildConcatenatedName, cleanupLive, LiveSegmentShaStore } from '@server/lib/live' -import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getLiveDirectory } from '@server/lib/paths' +import { ffprobePromise, getAudioStream, getVideoStreamDimensionsInfo, getVideoStreamDuration } from '@server/helpers/ffmpeg' +import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url' +import { federateVideoIfNeeded } from '@server/lib/activitypub/videos' +import { cleanupLive, LiveSegmentShaStore } from '@server/lib/live' +import { + generateHLSMasterPlaylistFilename, + generateHlsSha256SegmentsFilename, + getLiveDirectory, + getLiveReplayBaseDirectory +} from '@server/lib/paths' import { generateVideoMiniature } from '@server/lib/thumbnail' import { generateHlsPlaylistResolutionFromTS } from '@server/lib/transcoding/transcoding' -import { VideoPathManager } from '@server/lib/video-path-manager' import { moveToNextState } from '@server/lib/video-state' import { VideoModel } from '@server/models/video/video' import { VideoFileModel } from '@server/models/video/video-file' import { VideoLiveModel } from '@server/models/video/video-live' import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist' -import { MStreamingPlaylist, MVideo, MVideoLive } from '@server/types/models' +import { MVideo, MVideoLive, MVideoWithAllFiles } from '@server/types/models' import { ThumbnailType, VideoLiveEndingPayload, VideoState } from '@shared/models' import { logger } from '../../../helpers/logger' +import { VideoBlacklistModel } from '@server/models/video/video-blacklist' async function processVideoLiveEnding (job: Job) { const payload = job.data as VideoLiveEndingPayload + logger.info('Processing video live ending for %s.', payload.videoId, { payload }) + function logError () { logger.warn('Video live %d does not exist anymore. Cannot process live ending.', payload.videoId) } @@ -32,19 +40,19 @@ async function processVideoLiveEnding (job: Job) { return } - const streamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id) - if (!streamingPlaylist) { - logError() - return - } - LiveSegmentShaStore.Instance.cleanupShaSegments(video.uuid) if (live.saveReplay !== true) { - return cleanupLive(video, streamingPlaylist) + return cleanupLiveAndFederate(video) } - return saveLive(video, live, streamingPlaylist) + if (live.permanentLive) { + await saveReplayToExternalVideo(video, payload.publishedAt, payload.replayDirectory) + + return cleanupLiveAndFederate(video) + } + + return replaceLiveByReplay(video, live, payload.replayDirectory) } // --------------------------------------------------------------------------- @@ -55,22 +63,66 @@ export { // --------------------------------------------------------------------------- -async function saveLive (video: MVideo, live: MVideoLive, streamingPlaylist: MStreamingPlaylist) { - const replayDirectory = VideoPathManager.Instance.getFSHLSOutputPath(video, VIDEO_LIVE.REPLAY_DIRECTORY) +async function saveReplayToExternalVideo (liveVideo: MVideo, publishedAt: string, replayDirectory: string) { + await cleanupTMPLiveFiles(getLiveDirectory(liveVideo)) + + const video = new VideoModel({ + name: `${liveVideo.name} - ${new Date(publishedAt).toLocaleString()}`, + isLive: false, + state: VideoState.TO_TRANSCODE, + duration: 0, + + remote: liveVideo.remote, + category: liveVideo.category, + licence: liveVideo.licence, + language: liveVideo.language, + commentsEnabled: liveVideo.commentsEnabled, + downloadEnabled: liveVideo.downloadEnabled, + waitTranscoding: liveVideo.waitTranscoding, + nsfw: liveVideo.nsfw, + description: liveVideo.description, + support: liveVideo.support, + privacy: liveVideo.privacy, + channelId: liveVideo.channelId + }) as MVideoWithAllFiles + + video.Thumbnails = [] + video.VideoFiles = [] + video.VideoStreamingPlaylists = [] + + video.url = getLocalVideoActivityPubUrl(video) + + await video.save() + + // If live is blacklisted, also blacklist the replay + const blacklist = await VideoBlacklistModel.loadByVideoId(liveVideo.id) + if (blacklist) { + await VideoBlacklistModel.create({ + videoId: video.id, + unfederated: blacklist.unfederated, + reason: blacklist.reason, + type: blacklist.type + }) + } + + await assignReplaysToVideo(video, replayDirectory) - const rootFiles = await readdir(getLiveDirectory(video)) + await remove(replayDirectory) + + for (const type of [ ThumbnailType.MINIATURE, ThumbnailType.PREVIEW ]) { + const image = await generateVideoMiniature({ video, videoFile: video.getMaxQualityFile(), type }) + await video.addAndSaveThumbnail(image) + } - const playlistFiles = rootFiles.filter(file => { - return file.endsWith('.m3u8') && file !== streamingPlaylist.playlistFilename - }) + await moveToNextState({ video, isNewVideo: true }) +} +async function replaceLiveByReplay (video: MVideo, live: MVideoLive, replayDirectory: string) { await cleanupTMPLiveFiles(getLiveDirectory(video)) await live.destroy() video.isLive = false - // Reinit views - video.views = 0 video.state = VideoState.TO_TRANSCODE await video.save() @@ -87,10 +139,38 @@ async function saveLive (video: MVideo, live: MVideoLive, streamingPlaylist: MSt hlsPlaylist.segmentsSha256Filename = generateHlsSha256SegmentsFilename() await hlsPlaylist.save() + await assignReplaysToVideo(videoWithFiles, replayDirectory) + + await remove(getLiveReplayBaseDirectory(videoWithFiles)) + + // Regenerate the thumbnail & preview? + if (videoWithFiles.getMiniature().automaticallyGenerated === true) { + const miniature = await generateVideoMiniature({ + video: videoWithFiles, + videoFile: videoWithFiles.getMaxQualityFile(), + type: ThumbnailType.MINIATURE + }) + await video.addAndSaveThumbnail(miniature) + } + + if (videoWithFiles.getPreview().automaticallyGenerated === true) { + const preview = await generateVideoMiniature({ + video: videoWithFiles, + videoFile: videoWithFiles.getMaxQualityFile(), + type: ThumbnailType.PREVIEW + }) + await video.addAndSaveThumbnail(preview) + } + + await moveToNextState({ video: videoWithFiles, isNewVideo: false }) +} + +async function assignReplaysToVideo (video: MVideo, replayDirectory: string) { let durationDone = false - for (const playlistFile of playlistFiles) { - const concatenatedTsFile = buildConcatenatedName(playlistFile) + const concatenatedTsFiles = await readdir(replayDirectory) + + for (const concatenatedTsFile of concatenatedTsFiles) { const concatenatedTsFilePath = join(replayDirectory, concatenatedTsFile) const probe = await ffprobePromise(concatenatedTsFilePath) @@ -99,7 +179,7 @@ async function saveLive (video: MVideo, live: MVideoLive, streamingPlaylist: MSt const { resolution, isPortraitMode } = await getVideoStreamDimensionsInfo(concatenatedTsFilePath, probe) const { resolutionPlaylistPath: outputPath } = await generateHlsPlaylistResolutionFromTS({ - video: videoWithFiles, + video, concatenatedTsFilePath, resolution, isPortraitMode, @@ -107,33 +187,22 @@ async function saveLive (video: MVideo, live: MVideoLive, streamingPlaylist: MSt }) if (!durationDone) { - videoWithFiles.duration = await getVideoStreamDuration(outputPath) - await videoWithFiles.save() + video.duration = await getVideoStreamDuration(outputPath) + await video.save() durationDone = true } } - await remove(replayDirectory) - - // Regenerate the thumbnail & preview? - if (videoWithFiles.getMiniature().automaticallyGenerated === true) { - await generateVideoMiniature({ - video: videoWithFiles, - videoFile: videoWithFiles.getMaxQualityFile(), - type: ThumbnailType.MINIATURE - }) - } + return video +} - if (videoWithFiles.getPreview().automaticallyGenerated === true) { - await generateVideoMiniature({ - video: videoWithFiles, - videoFile: videoWithFiles.getMaxQualityFile(), - type: ThumbnailType.PREVIEW - }) - } +async function cleanupLiveAndFederate (video: MVideo) { + const streamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id) + await cleanupLive(video, streamingPlaylist) - await moveToNextState({ video: videoWithFiles, isNewVideo: false }) + const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.id) + return federateVideoIfNeeded(fullVideo, false, undefined) } async function cleanupTMPLiveFiles (hlsDirectory: string) { diff --git a/server/lib/live/live-manager.ts b/server/lib/live/live-manager.ts index 920d3a5ec..5ffe41ee3 100644 --- a/server/lib/live/live-manager.ts +++ b/server/lib/live/live-manager.ts @@ -1,6 +1,7 @@ -import { readFile } from 'fs-extra' +import { readdir, readFile } from 'fs-extra' import { createServer, Server } from 'net' +import { join } from 'path' import { createServer as createServerTLS, Server as ServerTLS } from 'tls' import { computeLowerResolutionsToTranscode, @@ -18,10 +19,11 @@ import { VideoModel } from '@server/models/video/video' import { VideoLiveModel } from '@server/models/video/video-live' import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist' import { MStreamingPlaylistVideo, MVideo, MVideoLiveVideo } from '@server/types/models' +import { wait } from '@shared/core-utils' import { VideoState, VideoStreamingPlaylistType } from '@shared/models' import { federateVideoIfNeeded } from '../activitypub/videos' import { JobQueue } from '../job-queue' -import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename } from '../paths' +import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getLiveReplayBaseDirectory } from '../paths' import { PeerTubeSocket } from '../peertube-socket' import { LiveQuotaStore } from './live-quota-store' import { LiveSegmentShaStore } from './live-segment-sha-store' @@ -322,7 +324,7 @@ class LiveManager { muxingSession.destroy() - return this.onAfterMuxingCleanup(videoId) + return this.onAfterMuxingCleanup({ videoId }) .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags })) }) @@ -349,12 +351,15 @@ class LiveManager { live.Video = video - setTimeout(() => { - federateVideoIfNeeded(video, false) - .catch(err => logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags })) + await wait(getLiveSegmentTime(live.latencyMode) * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION) - PeerTubeSocket.Instance.sendVideoLiveNewState(video) - }, getLiveSegmentTime(live.latencyMode) * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION) + try { + await federateVideoIfNeeded(video, false) + } catch (err) { + logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags }) + } + + PeerTubeSocket.Instance.sendVideoLiveNewState(video) } catch (err) { logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags }) } @@ -364,25 +369,32 @@ class LiveManager { this.videoSessions.delete(videoId) } - private async onAfterMuxingCleanup (videoUUID: string, cleanupNow = false) { + private async onAfterMuxingCleanup (options: { + videoId: number | string + cleanupNow?: boolean // Default false + }) { + const { videoId, cleanupNow = false } = options + try { - const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoUUID) + const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId) if (!fullVideo) return const live = await VideoLiveModel.loadByVideoId(fullVideo.id) - if (!live.permanentLive) { - JobQueue.Instance.createJob({ - type: 'video-live-ending', - payload: { - videoId: fullVideo.id - } - }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY }) - - fullVideo.state = VideoState.LIVE_ENDED - } else { - fullVideo.state = VideoState.WAITING_FOR_LIVE - } + JobQueue.Instance.createJob({ + type: 'video-live-ending', + payload: { + videoId: fullVideo.id, + replayDirectory: live.saveReplay + ? await this.findReplayDirectory(fullVideo) + : undefined, + publishedAt: fullVideo.publishedAt.toISOString() + } + }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY }) + + fullVideo.state = live.permanentLive + ? VideoState.WAITING_FOR_LIVE + : VideoState.LIVE_ENDED await fullVideo.save() @@ -390,7 +402,7 @@ class LiveManager { await federateVideoIfNeeded(fullVideo, false) } catch (err) { - logger.error('Cannot save/federate new video state of live streaming of video %d.', videoUUID, { err, ...lTags(videoUUID) }) + logger.error('Cannot save/federate new video state of live streaming of video %d.', videoId, { err, ...lTags(videoId + '') }) } } @@ -398,10 +410,19 @@ class LiveManager { const videoUUIDs = await VideoModel.listPublishedLiveUUIDs() for (const uuid of videoUUIDs) { - await this.onAfterMuxingCleanup(uuid, true) + await this.onAfterMuxingCleanup({ videoId: uuid, cleanupNow: true }) } } + private async findReplayDirectory (video: MVideo) { + const directory = getLiveReplayBaseDirectory(video) + const files = await readdir(directory) + + if (files.length === 0) return undefined + + return join(directory, files.sort().reverse()[0]) + } + private buildAllResolutionsToTranscode (originResolution: number) { const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED ? computeLowerResolutionsToTranscode(originResolution, 'live') diff --git a/server/lib/live/live-utils.ts b/server/lib/live/live-utils.ts index 3bf723b98..46c7fd2f8 100644 --- a/server/lib/live/live-utils.ts +++ b/server/lib/live/live-utils.ts @@ -9,12 +9,12 @@ function buildConcatenatedName (segmentOrPlaylistPath: string) { return 'concat-' + num[1] + '.ts' } -async function cleanupLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) { +async function cleanupLive (video: MVideo, streamingPlaylist?: MStreamingPlaylist) { const hlsDirectory = getLiveDirectory(video) await remove(hlsDirectory) - await streamingPlaylist.destroy() + if (streamingPlaylist) await streamingPlaylist.destroy() } export { diff --git a/server/lib/live/shared/muxing-session.ts b/server/lib/live/shared/muxing-session.ts index a703f5b5f..588ee8749 100644 --- a/server/lib/live/shared/muxing-session.ts +++ b/server/lib/live/shared/muxing-session.ts @@ -11,7 +11,7 @@ import { CONFIG } from '@server/initializers/config' import { MEMOIZE_TTL, VIDEO_LIVE } from '@server/initializers/constants' import { VideoFileModel } from '@server/models/video/video-file' import { MStreamingPlaylistVideo, MUserId, MVideoLiveVideo } from '@server/types/models' -import { getLiveDirectory } from '../../paths' +import { getLiveDirectory, getLiveReplayBaseDirectory } from '../../paths' import { VideoTranscodingProfilesManager } from '../../transcoding/default-transcoding-profiles' import { isAbleToUploadVideo } from '../../user' import { LiveQuotaStore } from '../live-quota-store' @@ -63,6 +63,9 @@ class MuxingSession extends EventEmitter { private readonly videoUUID: string private readonly saveReplay: boolean + private readonly outDirectory: string + private readonly replayDirectory: string + private readonly lTags: LoggerTagsFn private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {} @@ -110,19 +113,22 @@ class MuxingSession extends EventEmitter { this.saveReplay = this.videoLive.saveReplay + this.outDirectory = getLiveDirectory(this.videoLive.Video) + this.replayDirectory = join(getLiveReplayBaseDirectory(this.videoLive.Video), new Date().toISOString()) + this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID) } async runMuxing () { this.createFiles() - const outPath = await this.prepareDirectories() + await this.prepareDirectories() this.ffmpegCommand = CONFIG.LIVE.TRANSCODING.ENABLED ? await getLiveTranscodingCommand({ inputUrl: this.inputUrl, - outPath, + outPath: this.outDirectory, masterPlaylistName: this.streamingPlaylist.playlistFilename, latencyMode: this.videoLive.latencyMode, @@ -137,15 +143,15 @@ class MuxingSession extends EventEmitter { }) : getLiveMuxingCommand({ inputUrl: this.inputUrl, - outPath, + outPath: this.outDirectory, masterPlaylistName: this.streamingPlaylist.playlistFilename, latencyMode: this.videoLive.latencyMode }) logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags()) - this.watchTSFiles(outPath) - this.watchMasterFile(outPath) + this.watchTSFiles(this.outDirectory) + this.watchMasterFile(this.outDirectory) let ffmpegShellCommand: string this.ffmpegCommand.on('start', cmdline => { @@ -155,10 +161,10 @@ class MuxingSession extends EventEmitter { }) this.ffmpegCommand.on('error', (err, stdout, stderr) => { - this.onFFmpegError({ err, stdout, stderr, outPath, ffmpegShellCommand }) + this.onFFmpegError({ err, stdout, stderr, outPath: this.outDirectory, ffmpegShellCommand }) }) - this.ffmpegCommand.on('end', () => this.onFFmpegEnded(outPath)) + this.ffmpegCommand.on('end', () => this.onFFmpegEnded(this.outDirectory)) this.ffmpegCommand.run() } @@ -304,16 +310,11 @@ class MuxingSession extends EventEmitter { } private async prepareDirectories () { - const outPath = getLiveDirectory(this.videoLive.Video) - await ensureDir(outPath) - - const replayDirectory = join(outPath, VIDEO_LIVE.REPLAY_DIRECTORY) + await ensureDir(this.outDirectory) if (this.videoLive.saveReplay === true) { - await ensureDir(replayDirectory) + await ensureDir(this.replayDirectory) } - - return outPath } private isDurationConstraintValid (streamingStartTime: number) { @@ -364,7 +365,7 @@ class MuxingSession extends EventEmitter { private async addSegmentToReplay (hlsVideoPath: string, segmentPath: string) { const segmentName = basename(segmentPath) - const dest = join(hlsVideoPath, VIDEO_LIVE.REPLAY_DIRECTORY, buildConcatenatedName(segmentName)) + const dest = join(this.replayDirectory, buildConcatenatedName(segmentName)) try { const data = await readFile(segmentPath) diff --git a/server/lib/paths.ts b/server/lib/paths.ts index 5a85bea42..b29854700 100644 --- a/server/lib/paths.ts +++ b/server/lib/paths.ts @@ -1,6 +1,6 @@ import { join } from 'path' import { CONFIG } from '@server/initializers/config' -import { HLS_REDUNDANCY_DIRECTORY, HLS_STREAMING_PLAYLIST_DIRECTORY } from '@server/initializers/constants' +import { HLS_REDUNDANCY_DIRECTORY, HLS_STREAMING_PLAYLIST_DIRECTORY, VIDEO_LIVE } from '@server/initializers/constants' import { isStreamingPlaylist, MStreamingPlaylistVideo, MVideo, MVideoFile, MVideoUUID } from '@server/types/models' import { removeFragmentedMP4Ext } from '@shared/core-utils' import { buildUUID } from '@shared/extra-utils' @@ -21,6 +21,10 @@ function getLiveDirectory (video: MVideoUUID) { return getHLSDirectory(video) } +function getLiveReplayBaseDirectory (video: MVideoUUID) { + return join(getLiveDirectory(video), VIDEO_LIVE.REPLAY_DIRECTORY) +} + function getHLSDirectory (video: MVideoUUID) { return join(HLS_STREAMING_PLAYLIST_DIRECTORY, video.uuid) } @@ -74,6 +78,7 @@ export { getHLSDirectory, getLiveDirectory, + getLiveReplayBaseDirectory, getHLSRedundancyDirectory, generateHLSMasterPlaylistFilename, diff --git a/server/lib/transcoding/transcoding.ts b/server/lib/transcoding/transcoding.ts index d55364e25..9a15f8613 100644 --- a/server/lib/transcoding/transcoding.ts +++ b/server/lib/transcoding/transcoding.ts @@ -3,13 +3,13 @@ import { copyFile, ensureDir, move, remove, stat } from 'fs-extra' import { basename, extname as extnameUtil, join } from 'path' import { toEven } from '@server/helpers/core-utils' import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent' -import { MStreamingPlaylistFilesVideo, MVideoFile, MVideoFullLight } from '@server/types/models' +import { MStreamingPlaylistFilesVideo, MVideo, MVideoFile, MVideoFullLight } from '@server/types/models' import { VideoResolution, VideoStorage } from '../../../shared/models/videos' import { VideoStreamingPlaylistType } from '../../../shared/models/videos/video-streaming-playlist.type' import { + buildFileMetadata, canDoQuickTranscode, getVideoStreamDuration, - buildFileMetadata, getVideoStreamFPS, transcodeVOD, TranscodeVODOptions, @@ -191,7 +191,7 @@ function mergeAudioVideofile (video: MVideoFullLight, resolution: VideoResolutio // Concat TS segments from a live video to a fragmented mp4 HLS playlist async function generateHlsPlaylistResolutionFromTS (options: { - video: MVideoFullLight + video: MVideo concatenatedTsFilePath: string resolution: VideoResolution isPortraitMode: boolean @@ -209,7 +209,7 @@ async function generateHlsPlaylistResolutionFromTS (options: { // Generate an HLS playlist from an input file, and update the master playlist function generateHlsPlaylistResolution (options: { - video: MVideoFullLight + video: MVideo videoInputPath: string resolution: VideoResolution copyCodecs: boolean @@ -265,7 +265,7 @@ async function onWebTorrentVideoFileTranscoding ( async function generateHlsPlaylistCommon (options: { type: 'hls' | 'hls-from-ts' - video: MVideoFullLight + video: MVideo inputPath: string resolution: VideoResolution copyCodecs?: boolean diff --git a/server/lib/video-blacklist.ts b/server/lib/video-blacklist.ts index 0984c0d7a..91f44cb11 100644 --- a/server/lib/video-blacklist.ts +++ b/server/lib/video-blacklist.ts @@ -73,8 +73,7 @@ async function blacklistVideo (videoInstance: MVideoAccountLight, options: Video unfederated: options.unfederate === true, reason: options.reason, type: VideoBlacklistType.MANUAL - } - ) + }) blacklist.Video = videoInstance if (options.unfederate === true) { -- cgit v1.2.3