server/lib/hls.ts

   1 import { close, ensureDir, move, open, outputJSON, read, readFile, remove, stat, writeFile } from 'fs-extra'
   2 import { flatten } from 'lodash'
   3 import PQueue from 'p-queue'
   4 import { basename, dirname, join } from 'path'
   5 import { MStreamingPlaylist, MStreamingPlaylistFilesVideo, MVideo } from '@server/types/models'
   6 import { uniqify, uuidRegex } from '@shared/core-utils'
   7 import { sha256 } from '@shared/extra-utils'
   8 import { getVideoStreamDimensionsInfo } from '@shared/ffmpeg'
   9 import { VideoStorage } from '@shared/models'
  10 import { getAudioStreamCodec, getVideoStreamCodec } from '../helpers/ffmpeg'
  11 import { logger } from '../helpers/logger'
  12 import { doRequest, doRequestAndSaveToFile } from '../helpers/requests'
  13 import { generateRandomString } from '../helpers/utils'
  14 import { CONFIG } from '../initializers/config'
  15 import { P2P_MEDIA_LOADER_PEER_VERSION, REQUEST_TIMEOUTS } from '../initializers/constants'
  16 import { sequelizeTypescript } from '../initializers/database'
  17 import { VideoFileModel } from '../models/video/video-file'
  18 import { VideoStreamingPlaylistModel } from '../models/video/video-streaming-playlist'
  19 import { storeHLSFileFromFilename } from './object-storage'
  20 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getHlsResolutionPlaylistFilename } from './paths'
  21 import { VideoPathManager } from './video-path-manager'
  22
  23 async function updateStreamingPlaylistsInfohashesIfNeeded () {
  24   const playlistsToUpdate = await VideoStreamingPlaylistModel.listByIncorrectPeerVersion()
  25
  26   // Use separate SQL queries, because we could have many videos to update
  27   for (const playlist of playlistsToUpdate) {
  28     await sequelizeTypescript.transaction(async t => {
  29       const videoFiles = await VideoFileModel.listByStreamingPlaylist(playlist.id, t)
  30
  31       playlist.assignP2PMediaLoaderInfoHashes(playlist.Video, videoFiles)
  32       playlist.p2pMediaLoaderPeerVersion = P2P_MEDIA_LOADER_PEER_VERSION
  33
  34       await playlist.save({ transaction: t })
  35     })
  36   }
  37 }
  38
  39 async function updatePlaylistAfterFileChange (video: MVideo, playlist: MStreamingPlaylist) {
  40   try {
  41     let playlistWithFiles = await updateMasterHLSPlaylist(video, playlist)
  42     playlistWithFiles = await updateSha256VODSegments(video, playlist)
  43
  44     // Refresh playlist, operations can take some time
  45     playlistWithFiles = await VideoStreamingPlaylistModel.loadWithVideoAndFiles(playlist.id)
  46     playlistWithFiles.assignP2PMediaLoaderInfoHashes(video, playlistWithFiles.VideoFiles)
  47     await playlistWithFiles.save()
  48
  49     video.setHLSPlaylist(playlistWithFiles)
  50   } catch (err) {
  51     logger.info('Cannot update playlist after file change. Maybe due to concurrent transcoding', { err })
  52   }
  53 }
  54
  55 // ---------------------------------------------------------------------------
  56
  57 // Avoid concurrency issues when updating streaming playlist files
  58 const playlistFilesQueue = new PQueue({ concurrency: 1 })
  59
  60 function updateMasterHLSPlaylist (video: MVideo, playlistArg: MStreamingPlaylist): Promise<MStreamingPlaylistFilesVideo> {
  61   return playlistFilesQueue.add(async () => {
  62     const playlist = await VideoStreamingPlaylistModel.loadWithVideoAndFiles(playlistArg.id)
  63
  64     const masterPlaylists: string[] = [ '#EXTM3U', '#EXT-X-VERSION:3' ]
  65
  66     for (const file of playlist.VideoFiles) {
  67       const playlistFilename = getHlsResolutionPlaylistFilename(file.filename)
  68
  69       await VideoPathManager.Instance.makeAvailableVideoFile(file.withVideoOrPlaylist(playlist), async videoFilePath => {
  70         const size = await getVideoStreamDimensionsInfo(videoFilePath)
  71
  72         const bandwidth = 'BANDWIDTH=' + video.getBandwidthBits(file)
  73         const resolution = `RESOLUTION=${size?.width || 0}x${size?.height || 0}`
  74
  75         let line = `#EXT-X-STREAM-INF:${bandwidth},${resolution}`
  76         if (file.fps) line += ',FRAME-RATE=' + file.fps
  77
  78         const codecs = await Promise.all([
  79           getVideoStreamCodec(videoFilePath),
  80           getAudioStreamCodec(videoFilePath)
  81         ])
  82
  83         line += `,CODECS="${codecs.filter(c => !!c).join(',')}"`
  84
  85         masterPlaylists.push(line)
  86         masterPlaylists.push(playlistFilename)
  87       })
  88     }
  89
  90     if (playlist.playlistFilename) {
  91       await video.removeStreamingPlaylistFile(playlist, playlist.playlistFilename)
  92     }
  93     playlist.playlistFilename = generateHLSMasterPlaylistFilename(video.isLive)
  94
  95     const masterPlaylistPath = VideoPathManager.Instance.getFSHLSOutputPath(video, playlist.playlistFilename)
  96     await writeFile(masterPlaylistPath, masterPlaylists.join('\n') + '\n')
  97
  98     if (playlist.storage === VideoStorage.OBJECT_STORAGE) {
  99       playlist.playlistUrl = await storeHLSFileFromFilename(playlist, playlist.playlistFilename)
 100       await remove(masterPlaylistPath)
 101     }
 102
 103     return playlist.save()
 104   })
 105 }
 106
 107 // ---------------------------------------------------------------------------
 108
 109 function updateSha256VODSegments (video: MVideo, playlistArg: MStreamingPlaylist): Promise<MStreamingPlaylistFilesVideo> {
 110   return playlistFilesQueue.add(async () => {
 111     const json: { [filename: string]: { [range: string]: string } } = {}
 112
 113     const playlist = await VideoStreamingPlaylistModel.loadWithVideoAndFiles(playlistArg.id)
 114
 115     // For all the resolutions available for this video
 116     for (const file of playlist.VideoFiles) {
 117       const rangeHashes: { [range: string]: string } = {}
 118       const fileWithPlaylist = file.withVideoOrPlaylist(playlist)
 119
 120       await VideoPathManager.Instance.makeAvailableVideoFile(fileWithPlaylist, videoPath => {
 121
 122         return VideoPathManager.Instance.makeAvailableResolutionPlaylistFile(fileWithPlaylist, async resolutionPlaylistPath => {
 123           const playlistContent = await readFile(resolutionPlaylistPath)
 124           const ranges = getRangesFromPlaylist(playlistContent.toString())
 125
 126           const fd = await open(videoPath, 'r')
 127           for (const range of ranges) {
 128             const buf = Buffer.alloc(range.length)
 129             await read(fd, buf, 0, range.length, range.offset)
 130
 131             rangeHashes[`${range.offset}-${range.offset + range.length - 1}`] = sha256(buf)
 132           }
 133           await close(fd)
 134
 135           const videoFilename = file.filename
 136           json[videoFilename] = rangeHashes
 137         })
 138       })
 139     }
 140
 141     if (playlist.segmentsSha256Filename) {
 142       await video.removeStreamingPlaylistFile(playlist, playlist.segmentsSha256Filename)
 143     }
 144     playlist.segmentsSha256Filename = generateHlsSha256SegmentsFilename(video.isLive)
 145
 146     const outputPath = VideoPathManager.Instance.getFSHLSOutputPath(video, playlist.segmentsSha256Filename)
 147     await outputJSON(outputPath, json)
 148
 149     if (playlist.storage === VideoStorage.OBJECT_STORAGE) {
 150       playlist.segmentsSha256Url = await storeHLSFileFromFilename(playlist, playlist.segmentsSha256Filename)
 151       await remove(outputPath)
 152     }
 153
 154     return playlist.save()
 155   })
 156 }
 157
 158 // ---------------------------------------------------------------------------
 159
 160 async function buildSha256Segment (segmentPath: string) {
 161   const buf = await readFile(segmentPath)
 162   return sha256(buf)
 163 }
 164
 165 function downloadPlaylistSegments (playlistUrl: string, destinationDir: string, timeout: number, bodyKBLimit: number) {
 166   let timer
 167   let remainingBodyKBLimit = bodyKBLimit
 168
 169   logger.info('Importing HLS playlist %s', playlistUrl)
 170
 171   return new Promise<void>(async (res, rej) => {
 172     const tmpDirectory = join(CONFIG.STORAGE.TMP_DIR, await generateRandomString(10))
 173
 174     await ensureDir(tmpDirectory)
 175
 176     timer = setTimeout(() => {
 177       deleteTmpDirectory(tmpDirectory)
 178
 179       return rej(new Error('HLS download timeout.'))
 180     }, timeout)
 181
 182     try {
 183       // Fetch master playlist
 184       const subPlaylistUrls = await fetchUniqUrls(playlistUrl)
 185
 186       const subRequests = subPlaylistUrls.map(u => fetchUniqUrls(u))
 187       const fileUrls = uniqify(flatten(await Promise.all(subRequests)))
 188
 189       logger.debug('Will download %d HLS files.', fileUrls.length, { fileUrls })
 190
 191       for (const fileUrl of fileUrls) {
 192         const destPath = join(tmpDirectory, basename(fileUrl))
 193
 194         await doRequestAndSaveToFile(fileUrl, destPath, { bodyKBLimit: remainingBodyKBLimit, timeout: REQUEST_TIMEOUTS.REDUNDANCY })
 195
 196         const { size } = await stat(destPath)
 197         remainingBodyKBLimit -= (size / 1000)
 198
 199         logger.debug('Downloaded HLS playlist file %s with %d kB remained limit.', fileUrl, Math.floor(remainingBodyKBLimit))
 200       }
 201
 202       clearTimeout(timer)
 203
 204       await move(tmpDirectory, destinationDir, { overwrite: true })
 205
 206       return res()
 207     } catch (err) {
 208       deleteTmpDirectory(tmpDirectory)
 209
 210       return rej(err)
 211     }
 212   })
 213
 214   function deleteTmpDirectory (directory: string) {
 215     remove(directory)
 216       .catch(err => logger.error('Cannot delete path on HLS download error.', { err }))
 217   }
 218
 219   async function fetchUniqUrls (playlistUrl: string) {
 220     const { body } = await doRequest(playlistUrl)
 221
 222     if (!body) return []
 223
 224     const urls = body.split('\n')
 225       .filter(line => line.endsWith('.m3u8') || line.endsWith('.mp4'))
 226       .map(url => {
 227         if (url.startsWith('http://') || url.startsWith('https://')) return url
 228
 229         return `${dirname(playlistUrl)}/${url}`
 230       })
 231
 232     return uniqify(urls)
 233   }
 234 }
 235
 236 // ---------------------------------------------------------------------------
 237
 238 async function renameVideoFileInPlaylist (playlistPath: string, newVideoFilename: string) {
 239   const content = await readFile(playlistPath, 'utf8')
 240
 241   const newContent = content.replace(new RegExp(`${uuidRegex}-\\d+-fragmented.mp4`, 'g'), newVideoFilename)
 242
 243   await writeFile(playlistPath, newContent, 'utf8')
 244 }
 245
 246 // ---------------------------------------------------------------------------
 247
 248 function injectQueryToPlaylistUrls (content: string, queryString: string) {
 249   return content.replace(/\.(m3u8|ts|mp4)/gm, '.$1?' + queryString)
 250 }
 251
 252 // ---------------------------------------------------------------------------
 253
 254 export {
 255   updateMasterHLSPlaylist,
 256   updateSha256VODSegments,
 257   buildSha256Segment,
 258   downloadPlaylistSegments,
 259   updateStreamingPlaylistsInfohashesIfNeeded,
 260   updatePlaylistAfterFileChange,
 261   injectQueryToPlaylistUrls,
 262   renameVideoFileInPlaylist
 263 }
 264
 265 // ---------------------------------------------------------------------------
 266
 267 function getRangesFromPlaylist (playlistContent: string) {
 268   const ranges: { offset: number, length: number }[] = []
 269   const lines = playlistContent.split('\n')
 270   const regex = /^#EXT-X-BYTERANGE:(\d+)@(\d+)$/
 271
 272   for (const line of lines) {
 273     const captured = regex.exec(line)
 274
 275     if (captured) {
 276       ranges.push({ length: parseInt(captured[1], 10), offset: parseInt(captured[2], 10) })
 277     }
 278   }
 279
 280   return ranges
 281 }