Add permanent live support

15 files changed, 326 insertions, 24 deletions

diff --git a/server/controllers/api/videos/index.ts b/server/controllers/api/videos/index.ts
index 0dcd38ad2..71a0f97e2 100644
--- a/server/controllers/api/videos/index.ts
+++ b/server/controllers/api/videos/index.ts
@@ -174,7 +174,7 @@ function listVideoPrivacies (req: express.Request, res: express.Response) {
 }
 
 async function addVideo (req: express.Request, res: express.Response) {
-  // Transferring the video could be long
+  // Uploading the video could be long
   // Set timeout to 10 minutes, as Express's default is 2 minutes
   req.setTimeout(1000 * 60 * 10, () => {
     logger.error('Upload video has timed out.')
diff --git a/server/controllers/api/videos/live.ts b/server/controllers/api/videos/live.ts
index a6f00c1bd..e67d89612 100644
--- a/server/controllers/api/videos/live.ts
+++ b/server/controllers/api/videos/live.ts
@@ -67,7 +67,9 @@ async function updateLiveVideo (req: express.Request, res: express.Response) {
 
   const video = res.locals.videoAll
   const videoLive = res.locals.videoLive
+
   videoLive.saveReplay = body.saveReplay || false
+  videoLive.permanentLive = body.permanentLive || false
 
   video.VideoLive = await videoLive.save()
 
@@ -90,6 +92,7 @@ async function addLiveVideo (req: express.Request, res: express.Response) {
 
   const videoLive = new VideoLiveModel()
   videoLive.saveReplay = videoInfo.saveReplay || false
+  videoLive.permanentLive = videoInfo.permanentLive || false
   videoLive.streamKey = uuidv4()
 
   const [ thumbnailModel, previewModel ] = await buildVideoThumbnailsFromReq({
diff --git a/server/helpers/activitypub.ts b/server/helpers/activitypub.ts
index d28453d79..1188d6cf9 100644
--- a/server/helpers/activitypub.ts
+++ b/server/helpers/activitypub.ts
@@ -39,6 +39,16 @@ function getContextData (type: ContextType) {
         sensitive: 'as:sensitive',
         language: 'sc:inLanguage',
 
+        isLiveBroadcast: 'sc:isLiveBroadcast',
+        liveSaveReplay: {
+          '@type': 'sc:Boolean',
+          '@id': 'pt:liveSaveReplay'
+        },
+        permanentLive: {
+          '@type': 'sc:Boolean',
+          '@id': 'pt:permanentLive'
+        },
+
         Infohash: 'pt:Infohash',
         Playlist: 'pt:Playlist',
         PlaylistElement: 'pt:PlaylistElement',
diff --git a/server/helpers/custom-validators/activitypub/videos.ts b/server/helpers/custom-validators/activitypub/videos.ts
index cb385b07d..a01429c83 100644
--- a/server/helpers/custom-validators/activitypub/videos.ts
+++ b/server/helpers/custom-validators/activitypub/videos.ts
@@ -64,6 +64,7 @@ function sanitizeAndCheckVideoTorrentObject (video: any) {
   if (!isBooleanValid(video.commentsEnabled)) video.commentsEnabled = false
   if (!isBooleanValid(video.isLiveBroadcast)) video.isLiveBroadcast = false
   if (!isBooleanValid(video.liveSaveReplay)) video.liveSaveReplay = false
+  if (!isBooleanValid(video.permanentLive)) video.permanentLive = false
 
   return isActivityPubUrlValid(video.id) &&
     isVideoNameValid(video.name) &&
@@ -74,8 +75,6 @@ function sanitizeAndCheckVideoTorrentObject (video: any) {
     (!video.language || isRemoteStringIdentifierValid(video.language)) &&
     isVideoViewsValid(video.views) &&
     isBooleanValid(video.sensitive) &&
-    isBooleanValid(video.commentsEnabled) &&
-    isBooleanValid(video.downloadEnabled) &&
     isDateValid(video.published) &&
     isDateValid(video.updated) &&
     (!video.originallyPublishedAt || isDateValid(video.originallyPublishedAt)) &&
diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts
index 2c7acd757..9e642af95 100644
--- a/server/initializers/constants.ts
+++ b/server/initializers/constants.ts
@@ -24,7 +24,7 @@ import { CONFIG, registerConfigChangedHandler } from './config'
 
 // ---------------------------------------------------------------------------
 
-const LAST_MIGRATION_VERSION = 565
+const LAST_MIGRATION_VERSION = 570
 
 // ---------------------------------------------------------------------------
 
diff --git a/server/initializers/migrations/0570-permanent-live.ts b/server/initializers/migrations/0570-permanent-live.ts
new file mode 100644
index 000000000..9572a9b2d
--- /dev/null
+++ b/server/initializers/migrations/0570-permanent-live.ts
@@ -0,0 +1,27 @@
+import * as Sequelize from 'sequelize'
+
+async function up (utils: {
+  transaction: Sequelize.Transaction
+  queryInterface: Sequelize.QueryInterface
+  sequelize: Sequelize.Sequelize
+  db: any
+}): Promise<void> {
+  {
+    const data = {
+      type: Sequelize.BOOLEAN,
+      defaultValue: false,
+      allowNull: false
+    }
+
+    await utils.queryInterface.addColumn('videoLive', 'permanentLive', data)
+  }
+}
+
+function down (options) {
+  throw new Error('Not implemented.')
+}
+
+export {
+  up,
+  down
+}
diff --git a/server/lib/activitypub/videos.ts b/server/lib/activitypub/videos.ts
index 4053f487c..04f0bfc23 100644
--- a/server/lib/activitypub/videos.ts
+++ b/server/lib/activitypub/videos.ts
@@ -429,6 +429,7 @@ async function updateVideoFromAP (options: {
         if (video.isLive) {
           const [ videoLive ] = await VideoLiveModel.upsert({
             saveReplay: videoObject.liveSaveReplay,
+            permanentLive: videoObject.permanentLive,
             videoId: video.id
           }, { transaction: t, returning: true })
 
@@ -631,6 +632,7 @@ async function createVideo (videoObject: VideoObject, channel: MChannelAccountLi
       const videoLive = new VideoLiveModel({
         streamKey: null,
         saveReplay: videoObject.liveSaveReplay,
+        permanentLive: videoObject.permanentLive,
         videoId: videoCreated.id
       })
 
diff --git a/server/lib/job-queue/handlers/video-live-ending.ts b/server/lib/job-queue/handlers/video-live-ending.ts
index d3c84ce75..e3c11caa2 100644
--- a/server/lib/job-queue/handlers/video-live-ending.ts
+++ b/server/lib/job-queue/handlers/video-live-ending.ts
@@ -1,5 +1,5 @@
 import * as Bull from 'bull'
-import { copy, readdir, remove } from 'fs-extra'
+import { copy, pathExists, readdir, remove } from 'fs-extra'
 import { join } from 'path'
 import { getDurationFromVideoFile, getVideoFileResolution } from '@server/helpers/ffprobe-utils'
 import { VIDEO_LIVE } from '@server/initializers/constants'
@@ -14,6 +14,7 @@ import { VideoStreamingPlaylistModel } from '@server/models/video/video-streamin
 import { MStreamingPlaylist, MVideo, MVideoLive } from '@server/types/models'
 import { ThumbnailType, VideoLiveEndingPayload, VideoState } from '@shared/models'
 import { logger } from '../../../helpers/logger'
+import { LiveManager } from '@server/lib/live-manager'
 
 async function processVideoLiveEnding (job: Bull.Job) {
   const payload = job.data as VideoLiveEndingPayload
@@ -36,6 +37,8 @@ async function processVideoLiveEnding (job: Bull.Job) {
     return
   }
 
+  LiveManager.Instance.cleanupShaSegments(video.uuid)
+
   if (live.saveReplay !== true) {
     return cleanupLive(video, streamingPlaylist)
   }
@@ -43,10 +46,19 @@ async function processVideoLiveEnding (job: Bull.Job) {
   return saveLive(video, live)
 }
 
+async function cleanupLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) {
+  const hlsDirectory = getHLSDirectory(video)
+
+  await remove(hlsDirectory)
+
+  await streamingPlaylist.destroy()
+}
+
 // ---------------------------------------------------------------------------
 
 export {
-  processVideoLiveEnding
+  processVideoLiveEnding,
+  cleanupLive
 }
 
 // ---------------------------------------------------------------------------
@@ -131,16 +143,9 @@ async function saveLive (video: MVideo, live: MVideoLive) {
   await publishAndFederateIfNeeded(videoWithFiles, true)
 }
 
-async function cleanupLive (video: MVideo, streamingPlaylist: MStreamingPlaylist) {
-  const hlsDirectory = getHLSDirectory(video)
-
-  await remove(hlsDirectory)
-
-  streamingPlaylist.destroy()
-    .catch(err => logger.error('Cannot remove live streaming playlist.', { err }))
-}
-
 async function cleanupLiveFiles (hlsDirectory: string) {
+  if (!await pathExists(hlsDirectory)) return
+
   const files = await readdir(hlsDirectory)
 
   for (const filename of files) {
diff --git a/server/lib/live-manager.ts b/server/lib/live-manager.ts
index 4f45ce530..dcf016169 100644
--- a/server/lib/live-manager.ts
+++ b/server/lib/live-manager.ts
@@ -19,6 +19,7 @@ import { VideoState, VideoStreamingPlaylistType } from '@shared/models'
 import { federateVideoIfNeeded } from './activitypub/videos'
 import { buildSha256Segment } from './hls'
 import { JobQueue } from './job-queue'
+import { cleanupLive } from './job-queue/handlers/video-live-ending'
 import { PeerTubeSocket } from './peertube-socket'
 import { isAbleToUploadVideo } from './user'
 import { getHLSDirectory } from './video-paths'
@@ -153,6 +154,10 @@ class LiveManager {
     watchers.push(new Date().getTime())
   }
 
+  cleanupShaSegments (videoUUID: string) {
+    this.segmentsSha256.delete(videoUUID)
+  }
+
   private getContext () {
     return context
   }
@@ -184,6 +189,14 @@ class LiveManager {
       return this.abortSession(sessionId)
     }
 
+    // Cleanup old potential live files (could happen with a permanent live)
+    this.cleanupShaSegments(video.uuid)
+
+    const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
+    if (oldStreamingPlaylist) {
+      await cleanupLive(video, oldStreamingPlaylist)
+    }
+
     this.videoSessions.set(video.id, sessionId)
 
     const playlistUrl = WEBSERVER.URL + VideoStreamingPlaylistModel.getHlsMasterPlaylistStaticPath(video.uuid)
@@ -372,7 +385,13 @@ class LiveManager {
       logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', rtmpUrl)
 
       this.transSessions.delete(sessionId)
+
       this.watchersPerVideo.delete(videoLive.videoId)
+      this.videoSessions.delete(videoLive.videoId)
+
+      const newLivesPerUser = this.livesPerUser.get(user.id)
+                                               .filter(o => o.liveId !== videoLive.id)
+      this.livesPerUser.set(user.id, newLivesPerUser)
 
       setTimeout(() => {
         // Wait latest segments generation, and close watchers
@@ -412,14 +431,21 @@ class LiveManager {
       const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
       if (!fullVideo) return
 
-      JobQueue.Instance.createJob({
-        type: 'video-live-ending',
-        payload: {
-          videoId: fullVideo.id
-        }
-      }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY })
+      const live = await VideoLiveModel.loadByVideoId(videoId)
+
+      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
+      }
 
-      fullVideo.state = VideoState.LIVE_ENDED
       await fullVideo.save()
 
       PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
diff --git a/server/middlewares/validators/videos/video-live.ts b/server/middlewares/validators/videos/video-live.ts
index ff92db910..69a14ccb1 100644
--- a/server/middlewares/validators/videos/video-live.ts
+++ b/server/middlewares/validators/videos/video-live.ts
@@ -49,9 +49,16 @@ const videoLiveAddValidator = getCommonVideoEditAttributes().concat([
     .customSanitizer(toBooleanOrNull)
     .custom(isBooleanValid).withMessage('Should have a valid saveReplay attribute'),
 
+  body('permanentLive')
+    .optional()
+    .customSanitizer(toBooleanOrNull)
+    .custom(isBooleanValid).withMessage('Should have a valid permanentLive attribute'),
+
   async (req: express.Request, res: express.Response, next: express.NextFunction) => {
     logger.debug('Checking videoLiveAddValidator parameters', { parameters: req.body })
 
+    if (areValidationErrors(req, res)) return cleanUpReqFiles(req)
+
     if (CONFIG.LIVE.ENABLED !== true) {
       cleanUpReqFiles(req)
 
@@ -66,7 +73,12 @@ const videoLiveAddValidator = getCommonVideoEditAttributes().concat([
         .json({ error: 'Saving live replay is not allowed instance' })
     }
 
-    if (areValidationErrors(req, res)) return cleanUpReqFiles(req)
+    if (req.body.permanentLive && req.body.saveReplay) {
+      cleanUpReqFiles(req)
+
+      return res.status(400)
+        .json({ error: 'Cannot set this live as permanent while saving its replay' })
+    }
 
     const user = res.locals.oauth.token.User
     if (!await doesVideoChannelOfAccountExist(req.body.channelId, user, res)) return cleanUpReqFiles(req)
@@ -116,6 +128,11 @@ const videoLiveUpdateValidator = [
 
     if (areValidationErrors(req, res)) return
 
+    if (req.body.permanentLive && req.body.saveReplay) {
+      return res.status(400)
+        .json({ error: 'Cannot set this live as permanent while saving its replay' })
+    }
+
     if (CONFIG.LIVE.ALLOW_REPLAY !== true && req.body.saveReplay === true) {
       return res.status(403)
         .json({ error: 'Saving live replay is not allowed instance' })
diff --git a/server/models/video/video-format-utils.ts b/server/models/video/video-format-utils.ts
index b1adbcb86..a1f022fb4 100644
--- a/server/models/video/video-format-utils.ts
+++ b/server/models/video/video-format-utils.ts
@@ -360,6 +360,10 @@ function videoModelToActivityPubObject (video: MVideoAP): VideoObject {
       ? video.VideoLive.saveReplay
       : null,
 
+    permanentLive: video.isLive
+      ? video.VideoLive.permanentLive
+      : null,
+
     state: video.state,
     commentsEnabled: video.commentsEnabled,
     downloadEnabled: video.downloadEnabled,
diff --git a/server/models/video/video-live.ts b/server/models/video/video-live.ts
index f3bff74ea..875ba9b31 100644
--- a/server/models/video/video-live.ts
+++ b/server/models/video/video-live.ts
@@ -38,6 +38,10 @@ export class VideoLiveModel extends Model<VideoLiveModel> {
   @Column
   saveReplay: boolean
 
+  @AllowNull(false)
+  @Column
+  permanentLive: boolean
+
   @CreatedAt
   createdAt: Date
 
@@ -99,6 +103,7 @@ export class VideoLiveModel extends Model<VideoLiveModel> {
         : null,
 
       streamKey: this.streamKey,
+      permanentLive: this.permanentLive,
       saveReplay: this.saveReplay
     }
   }
diff --git a/server/tests/api/check-params/live.ts b/server/tests/api/check-params/live.ts
index 2b2d1beec..055f2f295 100644
--- a/server/tests/api/check-params/live.ts
+++ b/server/tests/api/check-params/live.ts
@@ -84,7 +84,8 @@ describe('Test video lives API validator', function () {
         tags: [ 'tag1', 'tag2' ],
         privacy: VideoPrivacy.PUBLIC,
         channelId,
-        saveReplay: false
+        saveReplay: false,
+        permanentLive: false
       }
     })
 
@@ -211,6 +212,12 @@ describe('Test video lives API validator', function () {
       await makeUploadRequest({ url: server.url, path, token: server.accessToken, fields, attaches })
     })
 
+    it('Should fail with save replay and permanent live set to true', async function () {
+      const fields = immutableAssign(baseCorrectParams, { saveReplay: true, permanentLive: true })
+
+      await makePostBodyRequest({ url: server.url, path, token: server.accessToken, fields })
+    })
+
     it('Should succeed with the correct parameters', async function () {
       this.timeout(30000)
 
@@ -372,6 +379,12 @@ describe('Test video lives API validator', function () {
       await updateLive(server.url, server.accessToken, videoIdNotLive, {}, 404)
     })
 
+    it('Should fail with save replay and permanent live set to true', async function () {
+      const fields = { saveReplay: true, permanentLive: true }
+
+      await updateLive(server.url, server.accessToken, videoId, fields, 400)
+    })
+
     it('Should succeed with the correct params', async function () {
       await updateLive(server.url, server.accessToken, videoId, { saveReplay: false })
     })
diff --git a/server/tests/api/live/index.ts b/server/tests/api/live/index.ts
index 32219969a..c733f564e 100644
--- a/server/tests/api/live/index.ts
+++ b/server/tests/api/live/index.ts
@@ -1,3 +1,4 @@
 import './live-constraints'
+import './live-permanent'
 import './live-save-replay'
 import './live'
diff --git a/server/tests/api/live/live-permanent.ts b/server/tests/api/live/live-permanent.ts
new file mode 100644
index 000000000..a64588ed7
--- /dev/null
+++ b/server/tests/api/live/live-permanent.ts
@@ -0,0 +1,190 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import 'mocha'
+import * as chai from 'chai'
+import { LiveVideoCreate, VideoDetails, VideoPrivacy, VideoState } from '@shared/models'
+import {
+  checkLiveCleanup,
+  cleanupTests,
+  createLive,
+  doubleFollow,
+  flushAndRunMultipleServers,
+  getLive,
+  getPlaylistsCount,
+  getVideo,
+  sendRTMPStreamInVideo,
+  ServerInfo,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  stopFfmpeg,
+  updateCustomSubConfig,
+  updateLive,
+  wait,
+  waitJobs,
+  waitUntilLiveStarts
+} from '../../../../shared/extra-utils'
+
+const expect = chai.expect
+
+describe('Permenant live', function () {
+  let servers: ServerInfo[] = []
+  let videoUUID: string
+
+  async function createLiveWrapper (permanentLive: boolean) {
+    const attributes: LiveVideoCreate = {
+      channelId: servers[0].videoChannel.id,
+      privacy: VideoPrivacy.PUBLIC,
+      name: 'my super live',
+      saveReplay: false,
+      permanentLive
+    }
+
+    const res = await createLive(servers[0].url, servers[0].accessToken, attributes)
+    return res.body.video.uuid
+  }
+
+  async function checkVideoState (videoId: string, state: VideoState) {
+    for (const server of servers) {
+      const res = await getVideo(server.url, videoId)
+      expect((res.body as VideoDetails).state.id).to.equal(state)
+    }
+  }
+
+  before(async function () {
+    this.timeout(120000)
+
+    servers = await flushAndRunMultipleServers(2)
+
+    // Get the access tokens
+    await setAccessTokensToServers(servers)
+    await setDefaultVideoChannel(servers)
+
+    // Server 1 and server 2 follow each other
+    await doubleFollow(servers[0], servers[1])
+
+    await updateCustomSubConfig(servers[0].url, servers[0].accessToken, {
+      live: {
+        enabled: true,
+        allowReplay: true,
+        maxDuration: null,
+        transcoding: {
+          enabled: true,
+          resolutions: {
+            '240p': true,
+            '360p': true,
+            '480p': true,
+            '720p': true,
+            '1080p': true,
+            '2160p': true
+          }
+        }
+      }
+    })
+  })
+
+  it('Should create a non permanent live and update it to be a permanent live', async function () {
+    this.timeout(20000)
+
+    const videoUUID = await createLiveWrapper(false)
+
+    {
+      const res = await getLive(servers[0].url, servers[0].accessToken, videoUUID)
+      expect(res.body.permanentLive).to.be.false
+    }
+
+    await updateLive(servers[0].url, servers[0].accessToken, videoUUID, { permanentLive: true })
+
+    {
+      const res = await getLive(servers[0].url, servers[0].accessToken, videoUUID)
+      expect(res.body.permanentLive).to.be.true
+    }
+  })
+
+  it('Should create a permanent live', async function () {
+    this.timeout(20000)
+
+    videoUUID = await createLiveWrapper(true)
+
+    const res = await getLive(servers[0].url, servers[0].accessToken, videoUUID)
+    expect(res.body.permanentLive).to.be.true
+
+    await waitJobs(servers)
+  })
+
+  it('Should stream into this permanent live', async function () {
+    this.timeout(40000)
+
+    const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, videoUUID)
+
+    for (const server of servers) {
+      await waitUntilLiveStarts(server.url, server.accessToken, videoUUID)
+    }
+
+    await checkVideoState(videoUUID, VideoState.PUBLISHED)
+
+    await stopFfmpeg(command)
+
+    await waitJobs(servers)
+  })
+
+  it('Should not have cleaned up this live', async function () {
+    this.timeout(40000)
+
+    await wait(5000)
+    await waitJobs(servers)
+
+    for (const server of servers) {
+      const res = await getVideo(server.url, videoUUID)
+
+      const videoDetails = res.body as VideoDetails
+      expect(videoDetails.streamingPlaylists).to.have.lengthOf(1)
+    }
+  })
+
+  it('Should have set this live to waiting for live state', async function () {
+    this.timeout(20000)
+
+    await checkVideoState(videoUUID, VideoState.WAITING_FOR_LIVE)
+  })
+
+  it('Should be able to stream again in the permanent live', async function () {
+    this.timeout(20000)
+
+    await updateCustomSubConfig(servers[0].url, servers[0].accessToken, {
+      live: {
+        enabled: true,
+        allowReplay: true,
+        maxDuration: null,
+        transcoding: {
+          enabled: true,
+          resolutions: {
+            '240p': false,
+            '360p': false,
+            '480p': false,
+            '720p': false,
+            '1080p': false,
+            '2160p': false
+          }
+        }
+      }
+    })
+
+    const command = await sendRTMPStreamInVideo(servers[0].url, servers[0].accessToken, videoUUID)
+
+    for (const server of servers) {
+      await waitUntilLiveStarts(server.url, server.accessToken, videoUUID)
+    }
+
+    await checkVideoState(videoUUID, VideoState.PUBLISHED)
+
+    const count = await getPlaylistsCount(servers[0], videoUUID)
+    // master playlist and 720p playlist
+    expect(count).to.equal(2)
+
+    await stopFfmpeg(command)
+  })
+
+  after(async function () {
+    await cleanupTests(servers)
+  })
+})