worker.ts

import { Job, Worker } from "bullmq";
import IORedis from "ioredis";
import { models } from "./models";
import { AcBackgroundJob } from "./models/acBackgroundJob";

const fs = require("fs");

const redisUrl = process.env.REDIS_URL || undefined;

const redisOptions = {
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
};

const redisConnection = redisUrl
  ? new IORedis(redisUrl, redisOptions)
  : new IORedis(redisOptions);

const uploadAllToS3 = require('./s3').uploadAllToS3;
const uploadToS3 = require('./s3').uploadToS3;
const downloadFromS3 = require('./s3').downloadFromS3;

const encodeVideo = require('./video').encodeVideo;
const createScreenshots = require('./video').createScreenshots;
const encodeAudio = require('./audio').encodeAudio;
const encodeFlac = require('./audio').encodeFlac;

const handleVideoJob = async (job: Job<JobDataAttributes>) => {
  let acBackgroundJob: AcBackgroundJob | null = null;

  console.log(`Got job ${job.id} ${JSON.stringify(job.data)}`);

  try {
    const jobData = job.data as JobDataAttributes;

    console.log(`AcBackgroundJob Id: ${jobData.acBackgroundJobId}`);

    acBackgroundJob = await models.AcBackgroundJob.findOne({
      where: {
        id: jobData.acBackgroundJobId,
      },
    });

    if (!acBackgroundJob) {
      throw new Error("Can't find acBackgroundJob");
    }

    const tempInDir = `/tmp/${acBackgroundJob.id}/in`;
    const tempOutVideoDir = `/tmp/${acBackgroundJob.id}/out`;
    const tempOutVThumbnailDir = `/tmp/${acBackgroundJob.id}/outThumbnails`;
    await fs.promises.mkdir(tempInDir, { recursive: true });
    await fs.promises.mkdir(tempOutVideoDir, { recursive: true });
    await fs.promises.mkdir(tempOutVThumbnailDir, { recursive: true });
    const videoInFilename = `${tempInDir}/${jobData.fileKey}`;
    const videoOutFilename = `${tempOutVideoDir}/${jobData.fileKey}`;
    const videoOutFlacFilename = `${tempOutVideoDir}/${jobData.flacFilename}`;

    console.log(`Video Download from S3`);

    await downloadFromS3(
      process.env.S3_VIDEO_UPLOAD_BUCKET!,
      jobData.fileKey,
      videoInFilename
    );

    console.log(`Video Encoding`);

    const videoDuration = await encodeVideo(
      videoInFilename,
      videoOutFilename,
      jobData,
      acBackgroundJob
    );

    console.log(`Video duration: ${videoDuration}`);

    console.log(`Video Upload to S3`);

    await uploadToS3(
      process.env.S3_VIDEO_PUBLIC_BUCKET!,
      jobData.fileKey,
      videoOutFilename
    );

    console.log(`Screenshots created`);

    await createScreenshots(
      videoOutFilename,
      tempOutVThumbnailDir,
      jobData,
      videoDuration as number
    );

    console.log(`Screenshots Uploaded to S3`);

    await uploadAllToS3(
      tempOutVThumbnailDir,
      process.env.S3_VIDEO_THUMBNAIL_BUCKET!
    );

    if (process.env.GOOGLE_TRANSCODING_FLAC_BUCKET) {
      console.log("Encode flac");

      const hasFlac = (await encodeFlac(
        videoOutFilename,
        videoOutFlacFilename
      )) as boolean;

      if (hasFlac) {
        console.log("Uploading flac");

        await uploadToS3(
          process.env.S3_VIDEO_PUBLIC_BUCKET!,
          jobData.flacFilename,
          videoOutFlacFilename
        );
      } else {
        console.warn("No flac produced");
      }
    }

    console.log(`Video Completed saving...`);

    acBackgroundJob.progress = 100;
    acBackgroundJob.data.finalDuration = videoDuration;
    //@ts-ignore
    acBackgroundJob.set("data.status", "Complete");
    await acBackgroundJob.save();
    console.log(`Video Job ${job.id} Completed`);

    return {
      duration: videoDuration,
    };
  } catch (error) {
    if (acBackgroundJob) {
      try {
        console.log(`${JSON.stringify(acBackgroundJob)}`);
        acBackgroundJob.progress = 0;
        //@ts-ignore
        acBackgroundJob.set("data.status", "Error");
        await acBackgroundJob.save();
      } catch (innerError) {
        console.error(innerError);
      }
    }
    console.error(error);
    throw error;
  }
};

const handleAudioJob = async (job: Job<JobDataAttributes>) => {
  let acBackgroundJob: AcBackgroundJob | null = null;

  console.log(`Got job ${job.id}`);

  try {
    const jobData = job.data as JobDataAttributes;
    acBackgroundJob = await models.AcBackgroundJob.findOne({
      where: {
        id: jobData.acBackgroundJobId,
      },
    });

    if (!acBackgroundJob) {
      throw new Error("Can't find acBackgroundJob");
    }

    const tempInDir = `/tmp/${acBackgroundJob.id}/in`;
    const tempOutAudioDir = `/tmp/${acBackgroundJob.id}/outAudio`;
    await fs.promises.mkdir(tempInDir, { recursive: true });
    await fs.promises.mkdir(tempOutAudioDir, { recursive: true });
    const audioInFilename = `${tempInDir}/${jobData.fileKey}`;
    const audioOutFilename = `${tempOutAudioDir}/${jobData.fileKey}`;
    const audioOutFlacFilename = `${tempOutAudioDir}/${jobData.flacFilename}`;

    console.log(`Audio Download from S3`);

    await downloadFromS3(
      process.env.S3_AUDIO_UPLOAD_BUCKET!,
      jobData.fileKey,
      audioInFilename
    );

    console.log(`Audio Encoding`);

    const audioDuration = await encodeAudio(
      audioInFilename,
      audioOutFilename,
      jobData,
      acBackgroundJob
    );

    console.log(`Audio duration: ${audioDuration}`);

    console.log(`Audio Upload to S3`);

    await uploadToS3(
      process.env.S3_AUDIO_PUBLIC_BUCKET!,
      jobData.fileKey,
      audioOutFilename
    );

    if (process.env.GOOGLE_TRANSCODING_FLAC_BUCKET) {
      console.log("Encode flac");

      await encodeFlac(
        audioOutFilename,
        audioOutFlacFilename
      );

      console.log("Uploading flac");

      await uploadToS3(
        process.env.S3_AUDIO_PUBLIC_BUCKET!,
        jobData.flacFilename,
        audioOutFlacFilename
      );
    }

    acBackgroundJob.progress = 100;
    acBackgroundJob.data.finalDuration = audioDuration;
    //@ts-ignore
    acBackgroundJob.set("data.status", "Complete");
    await acBackgroundJob.save();

    console.log(`Audio Job ${job.id} Completed`);

    return {
      duration: audioDuration,
    };
  } catch (error) {
    if (acBackgroundJob) {
      try {
        acBackgroundJob.progress = 0;
        //@ts-ignore
        acBackgroundJob.set("data.status", "Error");
        await acBackgroundJob.save();
      } catch (innerError) {
        console.error(innerError);
      }
    }
    console.error(error);
    throw error;
  }
};

const workerOptions = { connection: redisConnection };

const videoWorker = new Worker<JobDataAttributes>("VideoEncoding", handleVideoJob, workerOptions);
const audioWorker = new Worker<JobDataAttributes>("AudioEncoding", handleAudioJob, workerOptions);

videoWorker.on("error", (error) => {
  console.error("Video worker error", error);
});

audioWorker.on("error", (error) => {
  console.error("Audio worker error", error);
});