Thumbnails generation

packages/api/src/repositories/jobs.ts

@@ -2,6 +2,7 @@ import {
   ffmpegQueue,
   ffprobeQueue,
   flowProducer,
+  imageQueue,
   outcomeQueue,
   packageQueue,
   pipelineQueue,
@@ -19,6 +20,7 @@ const allQueus = [
   ffmpegQueue,
   ffprobeQueue,
   outcomeQueue,
+  imageQueue,
 ];
 
 function findQueueByName(name: string): Queue {

packages/api/src/routes/jobs.ts

@@ -3,6 +3,7 @@ import {
   addToQueue,
   DEFAULT_PACKAGE_NAME,
   DEFAULT_SEGMENT_SIZE,
+  imageQueue,
   packageQueue,
   pipelineQueue,
   transcodeQueue,
@@ -168,6 +169,31 @@ export const jobs = new Elysia()
       },
     },
   )
+  .post(
+    "/image",
+    async ({ body }) => {
+      const jobId = await addToQueue(imageQueue, body, {
+        id: body.assetId,
+      });
+      return { jobId };
+    },
+    {
+      detail: {
+        summary: "Create image job",
+        tags: ["Jobs"],
+      },
+      body: t.Object({
+        assetId: t.String({
+          format: "uuid",
+        }),
+      }),
+      response: {
+        200: t.Object({
+          jobId: t.String(),
+        }),
+      },
+    },
+  )
   .get(
     "/jobs",
     async ({ query }) => {

packages/api/src/routes/storage.ts

@@ -45,6 +45,13 @@ export const storage = new Elysia()
             url: await getStorageFileUrl(query.path),
             type: "video",
           };
+        case "png":
+        case "webp":
+          return {
+            mode: "url",
+            url: await getStorageFileUrl(query.path),
+            type: "image",
+          };
         case "m4a":
         case "mp3":
           return {

packages/api/src/types.ts

@@ -58,7 +58,7 @@ export type StorageFolder = Static<typeof StorageFolderSchema>;
 export const StorageFileSchema = t.Union([
   t.Object({
     mode: t.Literal("url"),
-    type: t.Union([t.Literal("video"), t.Literal("audio")]),
+    type: t.Union([t.Literal("video"), t.Literal("image"), t.Literal("audio")]),
     url: t.String(),
   }),
   t.Object({

packages/api/src/workers/outcome.ts

@@ -26,5 +26,9 @@ export const outcomeCallback: WorkerCallback<OutcomeData> = async ({ job }) => {
       });
       break;
     }
+    case "image": {
+      // TODO: Store thumbnail in database.
+      break;
+    }
   }
 };

packages/app/src/components/FilePreview.tsx

@@ -63,6 +63,9 @@ function Preview({ file }: { file: StorageFile }) {
         <audio src={file.url} controls className="w-full max-w-lg mx-auto" />
       );
     }
+    if (file.type === "image") {
+      return <img src={file.url} className="w-full" />;
+    }
   }
   return null;
 }

packages/app/src/components/Image.tsx

@@ -0,0 +1,7 @@
+interface ImageProps {
+  src: string;
+}
+
+export function Image({ src }: ImageProps) {
+  return <img src={src} />;
+}

packages/artisan/src/lib/file-helpers.ts

@@ -1,5 +1,4 @@
-import * as fs from "node:fs/promises";
-import { getS3SignedUrl } from "./s3";
+import { getS3SignedUrl, getTextFromS3 } from "./s3";
 import type { PartialInput, Stream } from "bolt";
 
 export async function getBinaryPath(name: string) {
@@ -41,11 +40,11 @@ export interface MetaStruct {
 }
 
 /**
- * Will fetch meta file when meta.json is found in path.
- * @param path S3 dir
+ * Will fetch meta file when meta.json
+ * @param assetId
  * @returns
  */
-export async function getMetaStruct(path: string): Promise<MetaStruct> {
-  const text = await fs.readFile(`${path}/meta.json`, "utf8");
-  return JSON.parse(text.toString());
+export async function getMetaStruct(assetId: string): Promise<MetaStruct> {
+  const text = await getTextFromS3(`transcode/${assetId}/meta.json`);
+  return JSON.parse(text);
 }

packages/artisan/src/lib/s3.ts

@@ -5,6 +5,7 @@ import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
 import { ConfiguredRetryStrategy } from "@smithy/util-retry";
 import { lookup } from "mime-types";
 import { S3SyncClient } from "s3-sync-client";
+import { assert } from "shared/assert";
 import { env } from "../env";
 import type { PutObjectCommandInput } from "@aws-sdk/client-s3";
 import type { CommandInput } from "s3-sync-client";
@@ -116,3 +117,16 @@ export async function getS3SignedUrl(
   });
   return url;
 }
+
+export async function getTextFromS3(remoteFilePath: string) {
+  const command = new GetObjectCommand({
+    Bucket: env.S3_BUCKET,
+    Key: remoteFilePath,
+  });
+  const response = await client.send(command);
+
+  const text = response.Body?.transformToString("utf-8");
+  assert(text, `Failed to get text from S3 "${remoteFilePath}"`);
+
+  return text;
+}

packages/artisan/src/workers/image.ts

@@ -0,0 +1,79 @@
+import { addToQueue, outcomeQueue } from "bolt";
+import { ffmpeg } from "../lib/ffmpeg";
+import { getMetaStruct } from "../lib/file-helpers";
+import { getS3SignedUrl, syncToS3 } from "../lib/s3";
+import type { MetaStruct } from "../lib/file-helpers";
+import type { ImageData, ImageResult, WorkerCallback } from "bolt";
+
+export const imageCallback: WorkerCallback<ImageData, ImageResult> = async ({
+  job,
+  dir,
+  progressTracker,
+}) => {
+  const metaStruct = await getMetaStruct(job.data.assetId);
+  const name = findStreamInputName(metaStruct);
+
+  if (!name) {
+    throw new Error("Failed to find suitable stream.");
+  }
+
+  const publicUrl = await getS3SignedUrl(
+    `transcode/${job.data.assetId}/${name}`,
+    60 * 30,
+  );
+
+  const outDir = await dir.createTempDir();
+
+  const outputOptions = ["-ss 00:00:01.000", "-vframes 1"];
+
+  await ffmpeg(
+    publicUrl,
+    `${outDir}/thumbnail.png`,
+    outputOptions,
+    (command) => {
+      job.log(command);
+    },
+    (value) => {
+      progressTracker.set("screenshot", value);
+    },
+  );
+
+  const s3Dir = `image/${job.data.assetId}`;
+  job.log(`Uploading to ${s3Dir}`);
+
+  await syncToS3(outDir, s3Dir, {
+    public: true,
+  });
+
+  await addToQueue(
+    outcomeQueue,
+    {
+      type: "image",
+      data: job.data,
+    },
+    {
+      options: {
+        removeOnComplete: true,
+      },
+    },
+  );
+
+  return {
+    assetId: job.data.assetId,
+  };
+};
+
+function findStreamInputName(metaStruct: MetaStruct): string | null {
+  let name: string | null = null;
+  let lastHeight = 0;
+
+  const entries = Object.entries(metaStruct.streams);
+  for (const [key, value] of entries) {
+    if (value.type === "video" && value.height > lastHeight) {
+      name = key;
+      lastHeight = value.height;
+    }
+  }
+
+  return name;
+}

packages/artisan/src/workers/index.ts

@@ -1,6 +1,7 @@
 import { runWorkers } from "bolt";
 import { ffmpegCallback } from "./ffmpeg";
 import { ffprobeCallback } from "./ffprobe";
+import { imageCallback } from "./image";
 import { packageCallback } from "./package";
 import { pipelineCallback } from "./pipeline";
 import { transcodeCallback } from "./transcode";
@@ -26,4 +27,8 @@ runWorkers([
     name: "pipeline",
     callback: pipelineCallback,
   },
+  {
+    name: "image",
+    callback: imageCallback,
+  },
 ]);

packages/artisan/src/workers/package.ts

@@ -3,68 +3,24 @@ import { execa } from "execa";
 import parseFilePath from "parse-filepath";
 import { getBinaryPath, getMetaStruct } from "../lib/file-helpers";
 import { syncFromS3, syncToS3 } from "../lib/s3";
-import type {
-  PackageData,
-  PackageResult,
-  Stream,
-  WorkerCallback,
-  WorkerDir,
-} from "bolt";
-import type { Job } from "bullmq";
+import type { PackageData, PackageResult, Stream, WorkerCallback } from "bolt";
 
 const packagerBin = await getBinaryPath("packager");
 
-enum Step {
-  Initial,
-  Outcome,
-  Finish,
-}
-
 export const packageCallback: WorkerCallback<
-  PackageData & { step?: Step },
+  PackageData,
   PackageResult
 > = async ({ job, dir }) => {
-  let step = job.data.step ?? Step.Initial;
-  while (step !== Step.Finish) {
-    switch (step) {
-      case Step.Initial: {
-        await handleStepInitial(job, dir);
-        await job.updateData({
-          ...job.data,
-          step: Step.Outcome,
-        });
-        step = Step.Outcome;
-        break;
-      }
-
-      case Step.Outcome: {
-        await handleJobOutcome(job);
-        await job.updateData({
-          ...job.data,
-          step: Step.Finish,
-        });
-        step = Step.Finish;
-        break;
-      }
-    }
-  }
+  const inDir = await dir.createTempDir();
 
-  return {
-    assetId: job.data.assetId,
-  };
-};
+  const meta = await getMetaStruct(job.data.assetId);
 
-async function handleStepInitial(job: Job<PackageData>, dir: WorkerDir) {
-  const inDir = await dir.createTempDir();
+  job.log(`Got meta: "${JSON.stringify(meta)}"`);
 
   await syncFromS3(`transcode/${job.data.assetId}`, inDir);
 
   job.log(`Synced folder in ${inDir}`);
 
-  const meta = await getMetaStruct(inDir);
-
-  job.log(`Got meta: "${JSON.stringify(meta)}"`);
-
   // If we do not specify the segmentSize, grab it from the meta file.
   const segmentSize = job.data.segmentSize ?? meta.segmentSize;
 
@@ -143,9 +99,7 @@ async function handleStepInitial(job: Job<PackageData>, dir: WorkerDir) {
     del: true,
     public: true,
   });
-}
 
-async function handleJobOutcome(job: Job<PackageData>) {
   await addToQueue(
     outcomeQueue,
     {
@@ -158,7 +112,11 @@ async function handleJobOutcome(job: Job<PackageData>) {
       },
     },
   );
-}
+
+  return {
+    assetId: job.data.assetId,
+  };
+};
 
 function getGroupId(
   stream:

packages/artisan/src/workers/pipeline.ts

@@ -1,5 +1,6 @@
 import {
   addToQueue,
+  imageQueue,
   packageQueue,
   transcodeQueue,
   waitForChildren,
@@ -88,4 +89,14 @@ async function handleStepContinue(job: Job<PipelineData>, token?: string) {
       parent: job,
     },
   );
+
+  await addToQueue(
+    imageQueue,
+    {
+      assetId: job.data.assetId,
+    },
+    {
+      parent: job,
+    },
+  );
 }

packages/bolt/src/queue-result.ts

@@ -33,3 +33,7 @@ export interface PackageResult {
 export interface PipelineResult {
   assetId: string;
 }
+
+export interface ImageResult {
+  assetId: string;
+}