server/lib/live/live-segment-sha-store.ts

   1 import { rename, writeJson } from 'fs-extra'
   2 import PQueue from 'p-queue'
   3 import { basename } from 'path'
   4 import { mapToJSON } from '@server/helpers/core-utils'
   5 import { logger, loggerTagsFactory } from '@server/helpers/logger'
   6 import { MStreamingPlaylistVideo } from '@server/types/models'
   7 import { buildSha256Segment } from '../hls'
   8 import { storeHLSFileFromPath } from '../object-storage'
   9
  10 const lTags = loggerTagsFactory('live')
  11
  12 class LiveSegmentShaStore {
  13
  14   private readonly segmentsSha256 = new Map<string, string>()
  15
  16   private readonly videoUUID: string
  17
  18   private readonly sha256Path: string
  19   private readonly sha256PathTMP: string
  20
  21   private readonly streamingPlaylist: MStreamingPlaylistVideo
  22   private readonly sendToObjectStorage: boolean
  23   private readonly writeQueue = new PQueue({ concurrency: 1 })
  24
  25   constructor (options: {
  26     videoUUID: string
  27     sha256Path: string
  28     streamingPlaylist: MStreamingPlaylistVideo
  29     sendToObjectStorage: boolean
  30   }) {
  31     this.videoUUID = options.videoUUID
  32
  33     this.sha256Path = options.sha256Path
  34     this.sha256PathTMP = options.sha256Path + '.tmp'
  35
  36     this.streamingPlaylist = options.streamingPlaylist
  37     this.sendToObjectStorage = options.sendToObjectStorage
  38   }
  39
  40   async addSegmentSha (segmentPath: string) {
  41     logger.debug('Adding live sha segment %s.', segmentPath, lTags(this.videoUUID))
  42
  43     const shaResult = await buildSha256Segment(segmentPath)
  44
  45     const segmentName = basename(segmentPath)
  46     this.segmentsSha256.set(segmentName, shaResult)
  47
  48     try {
  49       await this.writeToDisk()
  50     } catch (err) {
  51       logger.error('Cannot write sha segments to disk.', { err })
  52     }
  53   }
  54
  55   async removeSegmentSha (segmentPath: string) {
  56     const segmentName = basename(segmentPath)
  57
  58     logger.debug('Removing live sha segment %s.', segmentPath, lTags(this.videoUUID))
  59
  60     if (!this.segmentsSha256.has(segmentName)) {
  61       logger.warn(
  62         'Unknown segment in live segment hash store for video %s and segment %s.',
  63         this.videoUUID, segmentPath, lTags(this.videoUUID)
  64       )
  65       return
  66     }
  67
  68     this.segmentsSha256.delete(segmentName)
  69
  70     await this.writeToDisk()
  71   }
  72
  73   private writeToDisk () {
  74     return this.writeQueue.add(async () => {
  75       logger.debug(`Writing segment sha JSON ${this.sha256Path} of ${this.videoUUID} on disk.`, lTags(this.videoUUID))
  76
  77       // Atomic write: use rename instead of move that is not atomic
  78       await writeJson(this.sha256PathTMP, mapToJSON(this.segmentsSha256))
  79       await rename(this.sha256PathTMP, this.sha256Path)
  80
  81       if (this.sendToObjectStorage) {
  82         const url = await storeHLSFileFromPath(this.streamingPlaylist, this.sha256Path)
  83
  84         if (this.streamingPlaylist.segmentsSha256Url !== url) {
  85           this.streamingPlaylist.segmentsSha256Url = url
  86           await this.streamingPlaylist.save()
  87         }
  88       }
  89     })
  90   }
  91 }
  92
  93 export {
  94   LiveSegmentShaStore
  95 }