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