fp/services/factory/src/tasks/combine_video_segments.ts

import { Helpers, type Task } from 'graphile-worker'
import { basename, join } from 'node:path';
import { S3Client } from "@aws-sdk/client-s3"
import { Upload } from "@aws-sdk/lib-storage"
import { createId } from '@paralleldrive/cuid2';
// import { downloadS3File, uploadToS3, createStrapiB2File, createStrapiVod, createStrapiStream } from '@futureporn/storage'
// import { concatenateVideoSegments } from '@futureporn/video'
import { execFile } from 'node:child_process';
import { configs } from '../config';
import { pipeline, PassThrough, Readable } from 'node:stream';
import { createReadStream, createWriteStream, write } from 'node:fs';
import { writeFile, readFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { promisify } from 'node:util';
import patchVodInDatabase from '@futureporn/fetchers/patchVodInDatabase.ts'
import { downloadFile } from '@futureporn/storage/s3.ts';
import { S3FileRecord, VodRecord } from '@futureporn/types';
const pipelinePromise = promisify(pipeline)

interface s3ManifestEntry {
  key: string;
  bytes: number;
}

interface Payload {
  s3_manifest: s3ManifestEntry[];
  vod_id?: string;
}

interface S3Target {
  Bucket: string;
  Key: string;
  Body: Readable;
}

interface S3UploadParameters {
  bucket: string;
  keyName: string;
  uploadStream: PassThrough;
  client: S3Client;
  onProgress?: Function;
}


function assertPayload(payload: any): asserts payload is Payload {
  if (typeof payload !== "object" || !payload) throw new Error("invalid payload");
  if (typeof payload.s3_manifest !== "object") throw new Error("invalid s3_manifest");
}



/**
 * 
 * generate a txt file on disk which ffmpeg's concat filter will use to concatenate files.
 * example: ffmpeg -f concat -safe 0 -i ./files.txt -c copy ./projektmelody-chaturbate-2023-07-18.mp4
 * the text file is written to os.tmpdir()
 * the text file contents looks like the following. 
 * 
 *     file './cb-recording-part-1.mp4'
 *     file './cb-recording-part-2.mp4'
 *     file './cb-recording-part-3.mp4'
 */
const getFFmpegConcatSpecFile = async function (inputVideoFilePaths: string[]): Promise<string> {
  if (!inputVideoFilePaths) throw new Error('getFFmpegConcatSpec file requires an array of filepaths as argument, but it was undefined');
  if (inputVideoFilePaths.length < 1) throw new Error('getFFmpegConcatSpec arg0, inputVideoFilePaths was length 0 which is unsupported.');
  let lines = []
  for (const filePath of inputVideoFilePaths) {
    lines.push(`file '${filePath}'`)
  }
  const specFilePath = join(tmpdir(), `concat-spec-${createId()}`)
  await writeFile(specFilePath, lines.join('\n'), { encoding: 'utf-8' })
  return specFilePath
}

const getFFmpegConcatenation = async function (specFilePath: string, outputPath: string) {
  if (!specFilePath) throw new Error('getFFmpegStream requires specFilePath as arg');
  const execFileP = promisify(execFile)
  const { stdout, stderr } = await execFileP('ffmpeg', [
    '-f', 'concat',
    '-safe', '0',
    '-i', specFilePath,
    '-c', 'copy',
    outputPath
  ])
  console.log(stdout)
  console.log(stderr)
  return outputPath
}

const concatVideos = async function (videoFilePaths: string[]): Promise<string> {
  console.log(`concatVideos with ${JSON.stringify(videoFilePaths)}`)
  if (!videoFilePaths || videoFilePaths.length < 1 || typeof videoFilePaths[0] !== 'string') throw new Error('concatVideos requires videoFilePaths as arg, but it was undefined or not an array of strings');
  if (videoFilePaths.length === 1) {
    // if there is only one video, we don't need to do anything.
    return videoFilePaths[0]
  }
  const concatSpec = await getFFmpegConcatSpecFile(videoFilePaths)
  const outputVideoPath = join(tmpdir(), `${basename(videoFilePaths[0], '.mp4')}-${createId()}.mp4`)
  try {
    await getFFmpegConcatenation(concatSpec, outputVideoPath)
  } catch (err) {
    console.error(`error encountered while concatenating video files together`)
    console.error(err)
    throw err
  }
  return outputVideoPath
}




const getS3ParallelUpload = async function ({ 
  filePath, 
  client,
  s3KeyName
}: { 
  s3KeyName: string,
  filePath: string, 
  client: S3Client,
}): Promise<{upload: Upload, uploadStream: PassThrough}> {
  if (!filePath) throw new Error("first argument passed to uploadToS3, 'filePath' is undefined");

  console.log(`uploading ${s3KeyName} to S3`)

  const uploadStream = new PassThrough()
  const target: S3Target = {
    Bucket: configs.s3UscBucket,
    Key: s3KeyName,
    Body: uploadStream
  }

  const upload = new Upload({
    client,
    partSize: 1024 * 1024 * 5,
    queueSize: 1,
    leavePartsOnError: false,
    params: target,
  });


  return { upload, uploadStream }
}




export const combine_video_segments: Task = async function (payload: unknown, helpers: Helpers) {
  // helpers.logger.info('the following is the raw Task payload')
  // helpers.logger.info(payload)
  // helpers.logger.info(JSON.stringify(payload?.s3_manifest))
  assertPayload(payload)
  const { s3_manifest, vod_id } = payload
  if (!vod_id) throw new Error('combine_video_segments was called without a vod_id.');
  helpers.logger.info(`🏗️ combine_video_segments started with s3_manifest=${JSON.stringify(s3_manifest)}, vod_id=${vod_id}`)

  /**
   * Here we take a manifest of S3 files and we download each of them.
   * Then we combine them all, preserving original order using `ffmpeg -f concat`
   * Then we upload the resulting video to S3
   * Then we create records in Postgrest
   *   * s3_file
   *   * vod
   * 
   * After the records are created, 
   *  if we were told about a stream record that this recording belongs to,
   *  we edit the stream record, adding a relation to the vod we just created.
   */


  try {
    
    const client = new S3Client({
      endpoint: configs.s3Endpoint,
      region: configs.s3Region,
      credentials: {
        accessKeyId: configs.s3AccessKeyId,
        secretAccessKey: configs.s3SecretAccessKey
      }
    });
    const s3Manifest = s3_manifest
    const inputVideoFilePaths = await Promise.all(s3Manifest.filter((m) => (m.bytes !== 0)).map((m) => downloadFile(client, configs.s3UscBucket, m.key)))
    const concatenatedVideoFile = await concatVideos(inputVideoFilePaths)
    const s3KeyName = basename(concatenatedVideoFile)
    const inputStream = createReadStream(concatenatedVideoFile)
    const filePath = concatenatedVideoFile
    const { uploadStream, upload } = await getS3ParallelUpload({ client, s3KeyName, filePath })
    pipelinePromise(inputStream, uploadStream)
    await upload.done()

    if (!vod_id) throw new Error('vod_id was missing from payload');
    // update the vod with the s3_file of the combined video
    const payload = { 
      s3_file: s3KeyName,
      vod_id
    }
    await patchVodInDatabase(vod_id, payload)
    

  } catch (e: any) {
    helpers.logger.error('combined_video_segments failed')
    if (e instanceof Error) {
      helpers.logger.error(e.message)
    } else {
      helpers.logger.error(e)
    }
    throw e
  }


}


export default combine_video_segments