refactor(worker): ⚡ Move blurhash processing to worker

.vscode/settings.json

@@ -6,7 +6,8 @@
         "cli",
         "federation",
         "config",
-        "plugin"
+        "plugin",
+        "worker"
     ],
     "languageToolLinter.languageTool.ignoredWordsInWorkspace": ["versia"]
 }

classes/database/attachment.ts

@@ -189,7 +189,7 @@ export class Attachment extends BaseInterface<typeof Attachments> {
 
         const mediaManager = new MediaManager(config);
 
-        const { path, blurhash } = await mediaManager.addFile(file);
+        const { path } = await mediaManager.addFile(file);
 
         const url = Attachment.getUrl(path);
 
@@ -208,7 +208,6 @@ export class Attachment extends BaseInterface<typeof Attachments> {
             mimeType: file.type,
             description: options?.description ?? "",
             size: file.size,
-            blurhash: blurhash ?? undefined,
             width: metadata?.width ?? undefined,
             height: metadata?.height ?? undefined,
         });

classes/media/media-manager.test.ts

@@ -86,7 +86,6 @@ describe("MediaManager", () => {
             uploadedFile: new File(["hey"], "test.webp"),
             path: "test/test.webp",
             hash: "testhash",
-            blurhash: null,
         });
     });
 

classes/media/media-manager.ts

@@ -7,7 +7,6 @@ import type { Config } from "~/packages/config-manager/config.type";
 import { DiskMediaDriver } from "./drivers/disk.ts";
 import type { MediaDriver } from "./drivers/media-driver.ts";
 import { S3MediaDriver } from "./drivers/s3.ts";
-import { BlurhashPreprocessor } from "./preprocessors/blurhash.ts";
 import type { MediaPreprocessor } from "./preprocessors/media-preprocessor.ts";
 
 /**
@@ -33,7 +32,6 @@ export class MediaManager {
      */
     public constructor(private config: Config) {
         this.driver = this.initializeDriver();
-        this.initializePreprocessors();
     }
 
     /**
@@ -53,14 +51,6 @@ export class MediaManager {
         }
     }
 
-    /**
-     * Initializes the preprocessors based on the configuration.
-     */
-    private initializePreprocessors(): void {
-        this.preprocessors.push(new BlurhashPreprocessor());
-        // Add other preprocessors here as needed
-    }
-
     /**
      * Adds a file to the media storage.
      * @param file - The file to add.
@@ -68,21 +58,16 @@ export class MediaManager {
      */
     public async addFile(file: File): Promise<UploadedFileMetadata> {
         let processedFile = file;
-        let blurhash: string | null = null;
 
         for (const preprocessor of this.preprocessors) {
             const result = await preprocessor.process(processedFile);
 
-            if ("blurhash" in result) {
-                blurhash = result.blurhash as string;
-            }
-
             processedFile = result.file;
         }
 
         const uploadResult = await this.driver.addFile(processedFile);
 
-        return { ...uploadResult, blurhash };
+        return uploadResult;
     }
     /**
      * Retrieves a file from the media storage by its hash.
@@ -123,5 +108,4 @@ export interface UploadedFileMetadata {
     uploadedFile: File;
     path: string;
     hash: string;
-    blurhash: string | null;
 }

classes/workers/delivery.ts

@@ -1,4 +1,3 @@
-import { getLogger } from "@logtape/logtape";
 import { User } from "@versia/kit/db";
 import { Worker } from "bullmq";
 import chalk from "chalk";
@@ -22,8 +21,6 @@ export const getDeliveryWorker = (): Worker<
                 case DeliveryJobType.FederateEntity: {
                     const { entity, recipientId, senderId } = job.data;
 
-                    const logger = getLogger(["federation", "delivery"]);
-
                     const sender = await User.fromId(senderId);
 
                     if (!sender) {
@@ -40,17 +37,15 @@ export const getDeliveryWorker = (): Worker<
                         );
                     }
 
-                    logger.debug`Federating entity ${chalk.gray(
+                    await job.log(
-                        entity.id,
+                        `Federating entity [${entity.id}] from @${sender.getAcct()} to @${recipient.getAcct()}`,
-                    )} from ${chalk.gray(`@${sender.getAcct()}`)} to ${chalk.gray(
+                    );
-                        recipient.getAcct(),
-                    )}`;
 
                     await sender.federateToUser(entity, recipient);
 
-                    logger.debug`${chalk.green(
+                    await job.log(
-                        "✔",
+                        `✔ Finished federating entity [${entity.id}]`,
-                    )} Finished federating entity ${chalk.gray(entity.id)}`;
+                    );
                 }
             }
         },

classes/workers/fetch.ts

@@ -1,7 +1,6 @@
 import { Instance } from "@versia/kit/db";
 import { Instances } from "@versia/kit/tables";
 import { Worker } from "bullmq";
-import chalk from "chalk";
 import { eq } from "drizzle-orm";
 import { config } from "~/packages/config-manager";
 import { connection } from "~/utils/redis.ts";
@@ -45,9 +44,7 @@ export const getFetchWorker = (): Worker<FetchJobData, void, FetchJobType> =>
                     await Instance.resolve(uri);
 
                     await job.log(
-                        `${chalk.green(
+                        `✔ Finished fetching instance metadata from [${uri}]`,
-                            "✔",
-                        )} Finished fetching instance metadata from [${uri}]`,
                     );
                 }
             }

classes/workers/inbox.ts

@@ -51,7 +51,7 @@ export const getInboxWorker = (): Worker<InboxJobData, void, InboxJobType> =>
                         }
 
                         await job.log(
-                            `Finished processing entity [${data.id}]`,
+                            `✔ Finished processing entity [${data.id}]`,
                         );
 
                         return;

classes/workers/media.ts

@@ -3,6 +3,7 @@ import { Worker } from "bullmq";
 import { config } from "~/packages/config-manager";
 import { connection } from "~/utils/redis.ts";
 import { MediaManager } from "../media/media-manager.ts";
+import { BlurhashPreprocessor } from "../media/preprocessors/blurhash.ts";
 import { ImageConversionPreprocessor } from "../media/preprocessors/image-conversion.ts";
 import {
     type MediaJobData,
@@ -18,6 +19,8 @@ export const getMediaWorker = (): Worker<MediaJobData, void, MediaJobType> =>
                 case MediaJobType.ConvertMedia: {
                     const { attachmentId, filename } = job.data;
 
+                    await job.log(`Fetching attachment ID [${attachmentId}]`);
+
                     const attachment = await Attachment.fromId(attachmentId);
 
                     if (!attachment) {
@@ -27,6 +30,7 @@ export const getMediaWorker = (): Worker<MediaJobData, void, MediaJobType> =>
                     }
 
                     const processor = new ImageConversionPreprocessor(config);
+                    const blurhashProcessor = new BlurhashPreprocessor();
 
                     const hash = attachment?.data.sha256;
 
@@ -36,6 +40,11 @@ export const getMediaWorker = (): Worker<MediaJobData, void, MediaJobType> =>
                         );
                     }
 
+                    await job.log(`Processing attachment [${attachmentId}]`);
+                    await job.log(
+                        `Fetching file from [${attachment.data.url}]`,
+                    );
+
                     // Download the file and process it.
                     const blob = await (
                         await fetch(attachment.data.url)
@@ -43,11 +52,19 @@ export const getMediaWorker = (): Worker<MediaJobData, void, MediaJobType> =>
 
                     const file = new File([blob], filename);
 
+                    await job.log(`Converting attachment [${attachmentId}]`);
+
                     const { file: processedFile } =
                         await processor.process(file);
 
+                    await job.log(`Generating blurhash for [${attachmentId}]`);
+
+                    const { blurhash } = await blurhashProcessor.process(file);
+
                     const mediaManager = new MediaManager(config);
 
+                    await job.log(`Uploading attachment [${attachmentId}]`);
+
                     const { path, uploadedFile } =
                         await mediaManager.addFile(processedFile);
 
@@ -62,7 +79,12 @@ export const getMediaWorker = (): Worker<MediaJobData, void, MediaJobType> =>
                             .digest("hex"),
                         mimeType: uploadedFile.type,
                         size: uploadedFile.size,
+                        blurhash,
                     });
+
+                    await job.log(
+                        `✔ Finished processing attachment [${attachmentId}]`,
+                    );
                 }
             }
         },