Live supports object storage

 * Sync live files (segments, master playlist, resolution playlist,
   segment sha file) into object storage
 * Automatically delete them when the live ends
 * Segment sha file is now a file on disk, and not stored in memory
   anymore

9 files changed, 263 insertions, 98 deletions

diff --git a/server/lib/hls.ts b/server/lib/hls.ts
index a0a5afc0f..a41f1ae48 100644
--- a/server/lib/hls.ts
+++ b/server/lib/hls.ts
@@ -15,7 +15,7 @@ import { P2P_MEDIA_LOADER_PEER_VERSION, REQUEST_TIMEOUTS } from '../initializers
 import { sequelizeTypescript } from '../initializers/database'
 import { VideoFileModel } from '../models/video/video-file'
 import { VideoStreamingPlaylistModel } from '../models/video/video-streaming-playlist'
-import { storeHLSFile } from './object-storage'
+import { storeHLSFileFromFilename } from './object-storage'
 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getHlsResolutionPlaylistFilename } from './paths'
 import { VideoPathManager } from './video-path-manager'
 
@@ -95,7 +95,7 @@ function updateMasterHLSPlaylist (video: MVideo, playlistArg: MStreamingPlaylist
     await writeFile(masterPlaylistPath, masterPlaylists.join('\n') + '\n')
 
     if (playlist.storage === VideoStorage.OBJECT_STORAGE) {
-      playlist.playlistUrl = await storeHLSFile(playlist, playlist.playlistFilename)
+      playlist.playlistUrl = await storeHLSFileFromFilename(playlist, playlist.playlistFilename)
       await remove(masterPlaylistPath)
     }
 
@@ -146,7 +146,7 @@ function updateSha256VODSegments (video: MVideo, playlistArg: MStreamingPlaylist
     await outputJSON(outputPath, json)
 
     if (playlist.storage === VideoStorage.OBJECT_STORAGE) {
-      playlist.segmentsSha256Url = await storeHLSFile(playlist, playlist.segmentsSha256Filename)
+      playlist.segmentsSha256Url = await storeHLSFileFromFilename(playlist, playlist.segmentsSha256Filename)
       await remove(outputPath)
     }
 
diff --git a/server/lib/job-queue/handlers/move-to-object-storage.ts b/server/lib/job-queue/handlers/move-to-object-storage.ts
index 25bdebeea..28c3d325d 100644
--- a/server/lib/job-queue/handlers/move-to-object-storage.ts
+++ b/server/lib/job-queue/handlers/move-to-object-storage.ts
@@ -5,7 +5,7 @@ import { logger, loggerTagsFactory } from '@server/helpers/logger'
 import { updateTorrentMetadata } from '@server/helpers/webtorrent'
 import { CONFIG } from '@server/initializers/config'
 import { P2P_MEDIA_LOADER_PEER_VERSION } from '@server/initializers/constants'
-import { storeHLSFile, storeWebTorrentFile } from '@server/lib/object-storage'
+import { storeHLSFileFromFilename, storeWebTorrentFile } from '@server/lib/object-storage'
 import { getHLSDirectory, getHlsResolutionPlaylistFilename } from '@server/lib/paths'
 import { moveToFailedMoveToObjectStorageState, moveToNextState } from '@server/lib/video-state'
 import { VideoModel } from '@server/models/video/video'
@@ -88,10 +88,10 @@ async function moveHLSFiles (video: MVideoWithAllFiles) {
 
       // Resolution playlist
       const playlistFilename = getHlsResolutionPlaylistFilename(file.filename)
-      await storeHLSFile(playlistWithVideo, playlistFilename)
+      await storeHLSFileFromFilename(playlistWithVideo, playlistFilename)
 
       // Resolution fragmented file
-      const fileUrl = await storeHLSFile(playlistWithVideo, file.filename)
+      const fileUrl = await storeHLSFileFromFilename(playlistWithVideo, file.filename)
 
       const oldPath = join(getHLSDirectory(video), file.filename)
 
@@ -113,9 +113,9 @@ async function doAfterLastJob (options: {
     const playlistWithVideo = playlist.withVideo(video)
 
     // Master playlist
-    playlist.playlistUrl = await storeHLSFile(playlistWithVideo, playlist.playlistFilename)
+    playlist.playlistUrl = await storeHLSFileFromFilename(playlistWithVideo, playlist.playlistFilename)
     // Sha256 segments file
-    playlist.segmentsSha256Url = await storeHLSFile(playlistWithVideo, playlist.segmentsSha256Filename)
+    playlist.segmentsSha256Url = await storeHLSFileFromFilename(playlistWithVideo, playlist.segmentsSha256Filename)
 
     playlist.storage = VideoStorage.OBJECT_STORAGE
 
diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts
index 8a3ee09a2..abfaf1cd7 100644
--- a/server/lib/job-queue/handlers/video-live-ending.ts
+++ b/server/lib/job-queue/handlers/video-live-ending.ts
@@ -4,7 +4,7 @@ import { join } from 'path'
 import { ffprobePromise, getAudioStream, getVideoStreamDimensionsInfo } from '@server/helpers/ffmpeg'
 import { getLocalVideoActivityPubUrl } from '@server/lib/activitypub/url'
 import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
-import { cleanupPermanentLive, cleanupTMPLiveFiles, cleanupUnsavedNormalLive } from '@server/lib/live'
+import { cleanupAndDestroyPermanentLive, cleanupTMPLiveFiles, cleanupUnsavedNormalLive } from '@server/lib/live'
 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getLiveReplayBaseDirectory } from '@server/lib/paths'
 import { generateVideoMiniature } from '@server/lib/thumbnail'
 import { generateHlsPlaylistResolutionFromTS } from '@server/lib/transcoding/transcoding'
@@ -141,23 +141,22 @@ async function replaceLiveByReplay (options: {
 }) {
   const { video, liveSession, live, permanentLive, replayDirectory } = options
 
-  await cleanupTMPLiveFiles(video)
+  const videoWithFiles = await VideoModel.loadFull(video.id)
+  const hlsPlaylist = videoWithFiles.getHLSPlaylist()
+
+  await cleanupTMPLiveFiles(videoWithFiles, hlsPlaylist)
 
   await live.destroy()
 
-  video.isLive = false
-  video.waitTranscoding = true
-  video.state = VideoState.TO_TRANSCODE
+  videoWithFiles.isLive = false
+  videoWithFiles.waitTranscoding = true
+  videoWithFiles.state = VideoState.TO_TRANSCODE
 
-  await video.save()
+  await videoWithFiles.save()
 
-  liveSession.replayVideoId = video.id
+  liveSession.replayVideoId = videoWithFiles.id
   await liveSession.save()
 
-  // Remove old HLS playlist video files
-  const videoWithFiles = await VideoModel.loadFull(video.id)
-
-  const hlsPlaylist = videoWithFiles.getHLSPlaylist()
   await VideoFileModel.removeHLSFilesOfVideoId(hlsPlaylist.id)
 
   // Reset playlist
@@ -234,7 +233,7 @@ async function cleanupLiveAndFederate (options: {
 
   if (streamingPlaylist) {
     if (permanentLive) {
-      await cleanupPermanentLive(video, streamingPlaylist)
+      await cleanupAndDestroyPermanentLive(video, streamingPlaylist)
     } else {
       await cleanupUnsavedNormalLive(video, streamingPlaylist)
     }
diff --git a/server/lib/live/live-manager.ts b/server/lib/live/live-manager.ts
index 16715862b..9470b530b 100644
--- a/server/lib/live/live-manager.ts
+++ b/server/lib/live/live-manager.ts
@@ -21,14 +21,14 @@ import { VideoLiveSessionModel } from '@server/models/video/video-live-session'
 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
 import { MStreamingPlaylistVideo, MVideo, MVideoLiveSession, MVideoLiveVideo } from '@server/types/models'
 import { pick, wait } from '@shared/core-utils'
-import { LiveVideoError, VideoState, VideoStreamingPlaylistType } from '@shared/models'
+import { LiveVideoError, VideoState, VideoStorage, VideoStreamingPlaylistType } from '@shared/models'
 import { federateVideoIfNeeded } from '../activitypub/videos'
 import { JobQueue } from '../job-queue'
 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getLiveReplayBaseDirectory } from '../paths'
 import { PeerTubeSocket } from '../peertube-socket'
 import { Hooks } from '../plugins/hooks'
 import { LiveQuotaStore } from './live-quota-store'
-import { cleanupPermanentLive } from './live-utils'
+import { cleanupAndDestroyPermanentLive } from './live-utils'
 import { MuxingSession } from './shared'
 
 const NodeRtmpSession = require('node-media-server/src/node_rtmp_session')
@@ -224,7 +224,7 @@ class LiveManager {
     if (oldStreamingPlaylist) {
       if (!videoLive.permanentLive) throw new Error('Found previous session in a non permanent live: ' + video.uuid)
 
-      await cleanupPermanentLive(video, oldStreamingPlaylist)
+      await cleanupAndDestroyPermanentLive(video, oldStreamingPlaylist)
     }
 
     this.videoSessions.set(video.id, sessionId)
@@ -301,7 +301,7 @@ class LiveManager {
       ...pick(options, [ 'streamingPlaylist', 'inputUrl', 'bitrate', 'ratio', 'fps', 'allResolutions', 'hasAudio' ])
     })
 
-    muxingSession.on('master-playlist-created', () => this.publishAndFederateLive(videoLive, localLTags))
+    muxingSession.on('live-ready', () => this.publishAndFederateLive(videoLive, localLTags))
 
     muxingSession.on('bad-socket-health', ({ videoId }) => {
       logger.error(
@@ -485,6 +485,10 @@ class LiveManager {
 
     playlist.assignP2PMediaLoaderInfoHashes(video, allResolutions)
 
+    playlist.storage = CONFIG.OBJECT_STORAGE.ENABLED
+      ? VideoStorage.OBJECT_STORAGE
+      : VideoStorage.FILE_SYSTEM
+
     return playlist.save()
   }
 
diff --git a/server/lib/live/live-segment-sha-store.ts b/server/lib/live/live-segment-sha-store.ts
index 4af6f3ebf..faf03dccf 100644
--- a/server/lib/live/live-segment-sha-store.ts
+++ b/server/lib/live/live-segment-sha-store.ts
@@ -1,62 +1,73 @@
+import { writeJson } from 'fs-extra'
 import { basename } from 'path'
+import { mapToJSON } from '@server/helpers/core-utils'
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
+import { MStreamingPlaylistVideo } from '@server/types/models'
 import { buildSha256Segment } from '../hls'
+import { storeHLSFileFromPath } from '../object-storage'
 
 const lTags = loggerTagsFactory('live')
 
 class LiveSegmentShaStore {
 
-  private static instance: LiveSegmentShaStore
-
-  private readonly segmentsSha256 = new Map<string, Map<string, string>>()
-
-  private constructor () {
+  private readonly segmentsSha256 = new Map<string, string>()
+
+  private readonly videoUUID: string
+  private readonly sha256Path: string
+  private readonly streamingPlaylist: MStreamingPlaylistVideo
+  private readonly sendToObjectStorage: boolean
+
+  constructor (options: {
+    videoUUID: string
+    sha256Path: string
+    streamingPlaylist: MStreamingPlaylistVideo
+    sendToObjectStorage: boolean
+  }) {
+    this.videoUUID = options.videoUUID
+    this.sha256Path = options.sha256Path
+    this.streamingPlaylist = options.streamingPlaylist
+    this.sendToObjectStorage = options.sendToObjectStorage
   }
 
-  getSegmentsSha256 (videoUUID: string) {
-    return this.segmentsSha256.get(videoUUID)
-  }
-
-  async addSegmentSha (videoUUID: string, segmentPath: string) {
-    const segmentName = basename(segmentPath)
-    logger.debug('Adding live sha segment %s.', segmentPath, lTags(videoUUID))
+  async addSegmentSha (segmentPath: string) {
+    logger.debug('Adding live sha segment %s.', segmentPath, lTags(this.videoUUID))
 
     const shaResult = await buildSha256Segment(segmentPath)
 
-    if (!this.segmentsSha256.has(videoUUID)) {
-      this.segmentsSha256.set(videoUUID, new Map())
-    }
+    const segmentName = basename(segmentPath)
+    this.segmentsSha256.set(segmentName, shaResult)
 
-    const filesMap = this.segmentsSha256.get(videoUUID)
-    filesMap.set(segmentName, shaResult)
+    await this.writeToDisk()
   }
 
-  removeSegmentSha (videoUUID: string, segmentPath: string) {
+  async removeSegmentSha (segmentPath: string) {
     const segmentName = basename(segmentPath)
 
-    logger.debug('Removing live sha segment %s.', segmentPath, lTags(videoUUID))
+    logger.debug('Removing live sha segment %s.', segmentPath, lTags(this.videoUUID))
 
-    const filesMap = this.segmentsSha256.get(videoUUID)
-    if (!filesMap) {
-      logger.warn('Unknown files map to remove sha for %s.', videoUUID, lTags(videoUUID))
+    if (!this.segmentsSha256.has(segmentName)) {
+      logger.warn('Unknown segment in files map for video %s and segment %s.', this.videoUUID, segmentPath, lTags(this.videoUUID))
       return
     }
 
-    if (!filesMap.has(segmentName)) {
-      logger.warn('Unknown segment in files map for video %s and segment %s.', videoUUID, segmentPath, lTags(videoUUID))
-      return
-    }
+    this.segmentsSha256.delete(segmentName)
 
-    filesMap.delete(segmentName)
+    await this.writeToDisk()
   }
 
-  cleanupShaSegments (videoUUID: string) {
-    this.segmentsSha256.delete(videoUUID)
-  }
+  private async writeToDisk () {
+    await writeJson(this.sha256Path, mapToJSON(this.segmentsSha256))
 
-  static get Instance () {
-    return this.instance || (this.instance = new this())
+    if (this.sendToObjectStorage) {
+      const url = await storeHLSFileFromPath(this.streamingPlaylist, this.sha256Path)
+
+      if (this.streamingPlaylist.segmentsSha256Url !== url) {
+        this.streamingPlaylist.segmentsSha256Url = url
+        await this.streamingPlaylist.save()
+      }
+    }
   }
+
 }
 
 export {
diff --git a/server/lib/live/live-utils.ts b/server/lib/live/live-utils.ts
index bba876642..d2b8e3a55 100644
--- a/server/lib/live/live-utils.ts
+++ b/server/lib/live/live-utils.ts
@@ -1,9 +1,10 @@
 import { pathExists, readdir, remove } from 'fs-extra'
 import { basename, join } from 'path'
 import { logger } from '@server/helpers/logger'
-import { MStreamingPlaylist, MVideo } from '@server/types/models'
+import { MStreamingPlaylist, MStreamingPlaylistVideo, MVideo } from '@server/types/models'
+import { VideoStorage } from '@shared/models'
+import { listHLSFileKeysOf, removeHLSFileObjectStorage, removeHLSObjectStorage } from '../object-storage'
 import { getLiveDirectory } from '../paths'
-import { LiveSegmentShaStore } from './live-segment-sha-store'
 
 function buildConcatenatedName (segmentOrPlaylistPath: string) {
   const num = basename(segmentOrPlaylistPath).match(/^(\d+)(-|\.)/)
@@ -11,8 +12,8 @@ function buildConcatenatedName (segmentOrPlaylistPath: string) {
   return 'concat-' + num[1] + '.ts'
 }
 
-async function cleanupPermanentLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) {
-  await cleanupTMPLiveFiles(video)
+async function cleanupAndDestroyPermanentLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) {
+  await cleanupTMPLiveFiles(video, streamingPlaylist)
 
   await streamingPlaylist.destroy()
 }
@@ -20,32 +21,51 @@ async function cleanupPermanentLive (video: MVideo, streamingPlaylist: MStreamin
 async function cleanupUnsavedNormalLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) {
   const hlsDirectory = getLiveDirectory(video)
 
+  // We uploaded files to object storage too, remove them
+  if (streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
+    await removeHLSObjectStorage(streamingPlaylist.withVideo(video))
+  }
+
   await remove(hlsDirectory)
 
   await streamingPlaylist.destroy()
+}
 
-  LiveSegmentShaStore.Instance.cleanupShaSegments(video.uuid)
+async function cleanupTMPLiveFiles (video: MVideo, streamingPlaylist: MStreamingPlaylist) {
+  await cleanupTMPLiveFilesFromObjectStorage(streamingPlaylist.withVideo(video))
+
+  await cleanupTMPLiveFilesFromFilesystem(video)
 }
 
-async function cleanupTMPLiveFiles (video: MVideo) {
-  const hlsDirectory = getLiveDirectory(video)
+export {
+  cleanupAndDestroyPermanentLive,
+  cleanupUnsavedNormalLive,
+  cleanupTMPLiveFiles,
+  buildConcatenatedName
+}
+
+// ---------------------------------------------------------------------------
 
-  LiveSegmentShaStore.Instance.cleanupShaSegments(video.uuid)
+function isTMPLiveFile (name: string) {
+  return name.endsWith('.ts') ||
+    name.endsWith('.m3u8') ||
+    name.endsWith('.json') ||
+    name.endsWith('.mpd') ||
+    name.endsWith('.m4s') ||
+    name.endsWith('.tmp')
+}
+
+async function cleanupTMPLiveFilesFromFilesystem (video: MVideo) {
+  const hlsDirectory = getLiveDirectory(video)
 
   if (!await pathExists(hlsDirectory)) return
 
-  logger.info('Cleanup TMP live files of %s.', hlsDirectory)
+  logger.info('Cleanup TMP live files from filesystem of %s.', hlsDirectory)
 
   const files = await readdir(hlsDirectory)
 
   for (const filename of files) {
-    if (
-      filename.endsWith('.ts') ||
-      filename.endsWith('.m3u8') ||
-      filename.endsWith('.mpd') ||
-      filename.endsWith('.m4s') ||
-      filename.endsWith('.tmp')
-    ) {
+    if (isTMPLiveFile(filename)) {
       const p = join(hlsDirectory, filename)
 
       remove(p)
@@ -54,9 +74,14 @@ async function cleanupTMPLiveFiles (video: MVideo) {
   }
 }
 
-export {
-  cleanupPermanentLive,
-  cleanupUnsavedNormalLive,
-  cleanupTMPLiveFiles,
-  buildConcatenatedName
+async function cleanupTMPLiveFilesFromObjectStorage (streamingPlaylist: MStreamingPlaylistVideo) {
+  if (streamingPlaylist.storage !== VideoStorage.OBJECT_STORAGE) return
+
+  const keys = await listHLSFileKeysOf(streamingPlaylist)
+
+  for (const key of keys) {
+    if (isTMPLiveFile(key)) {
+      await removeHLSFileObjectStorage(streamingPlaylist, key)
+    }
+  }
 }
diff --git a/server/lib/live/shared/muxing-session.ts b/server/lib/live/shared/muxing-session.ts
index 505717dce..4c27d5dd8 100644
--- a/server/lib/live/shared/muxing-session.ts
+++ b/server/lib/live/shared/muxing-session.ts
@@ -9,8 +9,10 @@ import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers
 import { logger, loggerTagsFactory, LoggerTagsFn } from '@server/helpers/logger'
 import { CONFIG } from '@server/initializers/config'
 import { MEMOIZE_TTL, VIDEO_LIVE } from '@server/initializers/constants'
+import { removeHLSFileObjectStorage, storeHLSFileFromFilename, storeHLSFileFromPath } from '@server/lib/object-storage'
 import { VideoFileModel } from '@server/models/video/video-file'
 import { MStreamingPlaylistVideo, MUserId, MVideoLiveVideo } from '@server/types/models'
+import { VideoStorage } from '@shared/models'
 import { getLiveDirectory, getLiveReplayBaseDirectory } from '../../paths'
 import { VideoTranscodingProfilesManager } from '../../transcoding/default-transcoding-profiles'
 import { isAbleToUploadVideo } from '../../user'
@@ -21,7 +23,7 @@ import { buildConcatenatedName } from '../live-utils'
 import memoizee = require('memoizee')
 
 interface MuxingSessionEvents {
-  'master-playlist-created': (options: { videoId: number }) => void
+  'live-ready': (options: { videoId: number }) => void
 
   'bad-socket-health': (options: { videoId: number }) => void
   'duration-exceeded': (options: { videoId: number }) => void
@@ -68,12 +70,18 @@ class MuxingSession extends EventEmitter {
   private readonly outDirectory: string
   private readonly replayDirectory: string
 
+  private readonly liveSegmentShaStore: LiveSegmentShaStore
+
   private readonly lTags: LoggerTagsFn
 
   private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {}
 
   private tsWatcher: FSWatcher
   private masterWatcher: FSWatcher
+  private m3u8Watcher: FSWatcher
+
+  private masterPlaylistCreated = false
+  private liveReady = false
 
   private aborted = false
 
@@ -123,6 +131,13 @@ class MuxingSession extends EventEmitter {
     this.outDirectory = getLiveDirectory(this.videoLive.Video)
     this.replayDirectory = join(getLiveReplayBaseDirectory(this.videoLive.Video), new Date().toISOString())
 
+    this.liveSegmentShaStore = new LiveSegmentShaStore({
+      videoUUID: this.videoLive.Video.uuid,
+      sha256Path: join(this.outDirectory, this.streamingPlaylist.segmentsSha256Filename),
+      streamingPlaylist: this.streamingPlaylist,
+      sendToObjectStorage: CONFIG.OBJECT_STORAGE.ENABLED
+    })
+
     this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID)
   }
 
@@ -159,8 +174,9 @@ class MuxingSession extends EventEmitter {
 
     logger.info('Running live muxing/transcoding for %s.', this.videoUUID, this.lTags())
 
-    this.watchTSFiles()
     this.watchMasterFile()
+    this.watchTSFiles()
+    this.watchM3U8File()
 
     let ffmpegShellCommand: string
     this.ffmpegCommand.on('start', cmdline => {
@@ -219,7 +235,7 @@ class MuxingSession extends EventEmitter {
     setTimeout(() => {
       // Wait latest segments generation, and close watchers
 
-      Promise.all([ this.tsWatcher.close(), this.masterWatcher.close() ])
+      Promise.all([ this.tsWatcher.close(), this.masterWatcher.close(), this.m3u8Watcher.close() ])
         .then(() => {
           // Process remaining segments hash
           for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) {
@@ -240,14 +256,41 @@ class MuxingSession extends EventEmitter {
   private watchMasterFile () {
     this.masterWatcher = watch(this.outDirectory + '/' + this.streamingPlaylist.playlistFilename)
 
-    this.masterWatcher.on('add', () => {
-      this.emit('master-playlist-created', { videoId: this.videoId })
+    this.masterWatcher.on('add', async () => {
+      if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
+        try {
+          const url = await storeHLSFileFromFilename(this.streamingPlaylist, this.streamingPlaylist.playlistFilename)
+
+          this.streamingPlaylist.playlistUrl = url
+          await this.streamingPlaylist.save()
+        } catch (err) {
+          logger.error('Cannot upload live master file to object storage.', { err, ...this.lTags() })
+        }
+      }
+
+      this.masterPlaylistCreated = true
 
       this.masterWatcher.close()
         .catch(err => logger.error('Cannot close master watcher of %s.', this.outDirectory, { err, ...this.lTags() }))
     })
   }
 
+  private watchM3U8File () {
+    this.m3u8Watcher = watch(this.outDirectory + '/*.m3u8')
+
+    const onChangeOrAdd = async (m3u8Path: string) => {
+      if (this.streamingPlaylist.storage !== VideoStorage.OBJECT_STORAGE) return
+
+      try {
+        await storeHLSFileFromPath(this.streamingPlaylist, m3u8Path)
+      } catch (err) {
+        logger.error('Cannot store in object storage m3u8 file %s', m3u8Path, { err, ...this.lTags() })
+      }
+    }
+
+    this.m3u8Watcher.on('change', onChangeOrAdd)
+  }
+
   private watchTSFiles () {
     const startStreamDateTime = new Date().getTime()
 
@@ -282,7 +325,21 @@ class MuxingSession extends EventEmitter {
       }
     }
 
-    const deleteHandler = (segmentPath: string) => LiveSegmentShaStore.Instance.removeSegmentSha(this.videoUUID, segmentPath)
+    const deleteHandler = async (segmentPath: string) => {
+      try {
+        await this.liveSegmentShaStore.removeSegmentSha(segmentPath)
+      } catch (err) {
+        logger.warn('Cannot remove segment sha %s from sha store', segmentPath, { err, ...this.lTags() })
+      }
+
+      if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
+        try {
+          await removeHLSFileObjectStorage(this.streamingPlaylist, segmentPath)
+        } catch (err) {
+          logger.error('Cannot remove segment %s from object storage', segmentPath, { err, ...this.lTags() })
+        }
+      }
+    }
 
     this.tsWatcher.on('add', p => addHandler(p))
     this.tsWatcher.on('unlink', p => deleteHandler(p))
@@ -315,6 +372,7 @@ class MuxingSession extends EventEmitter {
         extname: '.ts',
         infoHash: null,
         fps: this.fps,
+        storage: this.streamingPlaylist.storage,
         videoStreamingPlaylistId: this.streamingPlaylist.id
       })
 
@@ -343,18 +401,36 @@ class MuxingSession extends EventEmitter {
   }
 
   private processSegments (segmentPaths: string[]) {
-    mapSeries(segmentPaths, async previousSegment => {
-      // Add sha hash of previous segments, because ffmpeg should have finished generating them
-      await LiveSegmentShaStore.Instance.addSegmentSha(this.videoUUID, previousSegment)
+    mapSeries(segmentPaths, previousSegment => this.processSegment(previousSegment))
+      .catch(err => {
+        if (this.aborted) return
+
+        logger.error('Cannot process segments', { err, ...this.lTags() })
+      })
+  }
 
-      if (this.saveReplay) {
-        await this.addSegmentToReplay(previousSegment)
+  private async processSegment (segmentPath: string) {
+    // Add sha hash of previous segments, because ffmpeg should have finished generating them
+    await this.liveSegmentShaStore.addSegmentSha(segmentPath)
+
+    if (this.saveReplay) {
+      await this.addSegmentToReplay(segmentPath)
+    }
+
+    if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
+      try {
+        await storeHLSFileFromPath(this.streamingPlaylist, segmentPath)
+      } catch (err) {
+        logger.error('Cannot store TS segment %s in object storage', segmentPath, { err, ...this.lTags() })
       }
-    }).catch(err => {
-      if (this.aborted) return
+    }
 
-      logger.error('Cannot process segments', { err, ...this.lTags() })
-    })
+    // Master playlist and segment JSON file are created, live is ready
+    if (this.masterPlaylistCreated && !this.liveReady) {
+      this.liveReady = true
+
+      this.emit('live-ready', { videoId: this.videoId })
+    }
   }
 
   private hasClientSocketInBadHealth (sessionId: string) {
diff --git a/server/lib/object-storage/shared/object-storage-helpers.ts b/server/lib/object-storage/shared/object-storage-helpers.ts
index 16161362c..c131977e8 100644
--- a/server/lib/object-storage/shared/object-storage-helpers.ts
+++ b/server/lib/object-storage/shared/object-storage-helpers.ts
@@ -22,6 +22,24 @@ type BucketInfo = {
   PREFIX?: string
 }
 
+async function listKeysOfPrefix (prefix: string, bucketInfo: BucketInfo) {
+  const s3Client = getClient()
+
+  const commandPrefix = bucketInfo.PREFIX + prefix
+  const listCommand = new ListObjectsV2Command({
+    Bucket: bucketInfo.BUCKET_NAME,
+    Prefix: commandPrefix
+  })
+
+  const listedObjects = await s3Client.send(listCommand)
+
+  if (isArray(listedObjects.Contents) !== true) return []
+
+  return listedObjects.Contents.map(c => c.Key)
+}
+
+// ---------------------------------------------------------------------------
+
 async function storeObject (options: {
   inputPath: string
   objectStorageKey: string
@@ -36,6 +54,8 @@ async function storeObject (options: {
   return uploadToStorage({ objectStorageKey, content: fileStream, bucketInfo })
 }
 
+// ---------------------------------------------------------------------------
+
 async function removeObject (filename: string, bucketInfo: BucketInfo) {
   const command = new DeleteObjectCommand({
     Bucket: bucketInfo.BUCKET_NAME,
@@ -89,6 +109,8 @@ async function removePrefix (prefix: string, bucketInfo: BucketInfo) {
   if (listedObjects.IsTruncated) await removePrefix(prefix, bucketInfo)
 }
 
+// ---------------------------------------------------------------------------
+
 async function makeAvailable (options: {
   key: string
   destination: string
@@ -122,7 +144,8 @@ export {
   storeObject,
   removeObject,
   removePrefix,
-  makeAvailable
+  makeAvailable,
+  listKeysOfPrefix
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/object-storage/videos.ts b/server/lib/object-storage/videos.ts
index 66e738200..62aae248b 100644
--- a/server/lib/object-storage/videos.ts
+++ b/server/lib/object-storage/videos.ts
@@ -1,19 +1,35 @@
-import { join } from 'path'
+import { basename, join } from 'path'
 import { logger } from '@server/helpers/logger'
 import { CONFIG } from '@server/initializers/config'
 import { MStreamingPlaylistVideo, MVideoFile } from '@server/types/models'
 import { getHLSDirectory } from '../paths'
 import { generateHLSObjectBaseStorageKey, generateHLSObjectStorageKey, generateWebTorrentObjectStorageKey } from './keys'
-import { lTags, makeAvailable, removeObject, removePrefix, storeObject } from './shared'
+import { listKeysOfPrefix, lTags, makeAvailable, removeObject, removePrefix, storeObject } from './shared'
 
-function storeHLSFile (playlist: MStreamingPlaylistVideo, filename: string, path?: string) {
+function listHLSFileKeysOf (playlist: MStreamingPlaylistVideo) {
+  return listKeysOfPrefix(generateHLSObjectBaseStorageKey(playlist), CONFIG.OBJECT_STORAGE.STREAMING_PLAYLISTS)
+}
+
+// ---------------------------------------------------------------------------
+
+function storeHLSFileFromFilename (playlist: MStreamingPlaylistVideo, filename: string) {
   return storeObject({
-    inputPath: path ?? join(getHLSDirectory(playlist.Video), filename),
+    inputPath: join(getHLSDirectory(playlist.Video), filename),
     objectStorageKey: generateHLSObjectStorageKey(playlist, filename),
     bucketInfo: CONFIG.OBJECT_STORAGE.STREAMING_PLAYLISTS
   })
 }
 
+function storeHLSFileFromPath (playlist: MStreamingPlaylistVideo, path: string) {
+  return storeObject({
+    inputPath: path,
+    objectStorageKey: generateHLSObjectStorageKey(playlist, basename(path)),
+    bucketInfo: CONFIG.OBJECT_STORAGE.STREAMING_PLAYLISTS
+  })
+}
+
+// ---------------------------------------------------------------------------
+
 function storeWebTorrentFile (filename: string) {
   return storeObject({
     inputPath: join(CONFIG.STORAGE.VIDEOS_DIR, filename),
@@ -22,6 +38,8 @@ function storeWebTorrentFile (filename: string) {
   })
 }
 
+// ---------------------------------------------------------------------------
+
 function removeHLSObjectStorage (playlist: MStreamingPlaylistVideo) {
   return removePrefix(generateHLSObjectBaseStorageKey(playlist), CONFIG.OBJECT_STORAGE.STREAMING_PLAYLISTS)
 }
@@ -30,10 +48,14 @@ function removeHLSFileObjectStorage (playlist: MStreamingPlaylistVideo, filename
   return removeObject(generateHLSObjectStorageKey(playlist, filename), CONFIG.OBJECT_STORAGE.STREAMING_PLAYLISTS)
 }
 
+// ---------------------------------------------------------------------------
+
 function removeWebTorrentObjectStorage (videoFile: MVideoFile) {
   return removeObject(generateWebTorrentObjectStorageKey(videoFile.filename), CONFIG.OBJECT_STORAGE.VIDEOS)
 }
 
+// ---------------------------------------------------------------------------
+
 async function makeHLSFileAvailable (playlist: MStreamingPlaylistVideo, filename: string, destination: string) {
   const key = generateHLSObjectStorageKey(playlist, filename)
 
@@ -62,9 +84,14 @@ async function makeWebTorrentFileAvailable (filename: string, destination: strin
   return destination
 }
 
+// ---------------------------------------------------------------------------
+
 export {
+  listHLSFileKeysOf,
+
   storeWebTorrentFile,
-  storeHLSFile,
+  storeHLSFileFromFilename,
+  storeHLSFileFromPath,
 
   removeHLSObjectStorage,
   removeHLSFileObjectStorage,