feat: ✨ Split off queue workers into a separate worker process

classes/database/note.ts

@@ -41,7 +41,7 @@ import {
     parseTextMentions,
 } from "~/classes/functions/status";
 import { config } from "~/packages/config-manager";
-import { DeliveryJobType, deliveryQueue } from "~/worker.ts";
+import { DeliveryJobType, deliveryQueue } from "../queues/delivery.ts";
 import { Application } from "./application.ts";
 import { Attachment } from "./attachment.ts";
 import { BaseInterface } from "./base.ts";

classes/database/user.ts

@@ -53,7 +53,7 @@ import { findManyUsers } from "~/classes/functions/user";
 import { searchManager } from "~/classes/search/search-manager";
 import { type Config, config } from "~/packages/config-manager";
 import type { KnownEntity } from "~/types/api.ts";
-import { DeliveryJobType, deliveryQueue } from "~/worker.ts";
+import { DeliveryJobType, deliveryQueue } from "../queues/delivery.ts";
 import { BaseInterface } from "./base.ts";
 import { Emoji } from "./emoji.ts";
 import { Instance } from "./instance.ts";

classes/queues/delivery.ts

@@ -0,0 +1,20 @@
+import { Queue } from "bullmq";
+import type { KnownEntity } from "~/types/api";
+import { connection } from "~/utils/redis.ts";
+
+export enum DeliveryJobType {
+    FederateEntity = "federateEntity",
+}
+
+export type DeliveryJobData = {
+    entity: KnownEntity;
+    recipientId: string;
+    senderId: string;
+};
+
+export const deliveryQueue = new Queue<DeliveryJobData, void, DeliveryJobType>(
+    "delivery",
+    {
+        connection,
+    },
+);

classes/queues/fetch.ts

@@ -0,0 +1,17 @@
+import { Queue } from "bullmq";
+import { connection } from "~/utils/redis.ts";
+
+export enum FetchJobType {
+    Instance = "instance",
+    User = "user",
+    Note = "user",
+}
+
+export type FetchJobData = {
+    uri: string;
+    refetcher?: string;
+};
+
+export const fetchQueue = new Queue<FetchJobData, void, FetchJobType>("fetch", {
+    connection,
+});

classes/queues/inbox.ts

@@ -0,0 +1,31 @@
+import type { Entity } from "@versia/federation/types";
+import { Queue } from "bullmq";
+import type { SocketAddress } from "bun";
+import { connection } from "~/utils/redis.ts";
+
+export enum InboxJobType {
+    ProcessEntity = "processEntity",
+}
+
+export type InboxJobData = {
+    data: Entity;
+    headers: {
+        "x-signature"?: string;
+        "x-nonce"?: string;
+        "x-signed-by"?: string;
+        authorization?: string;
+    };
+    request: {
+        url: string;
+        method: string;
+        body: string;
+    };
+    ip: SocketAddress | null;
+};
+
+export const inboxQueue = new Queue<InboxJobData, Response, InboxJobType>(
+    "inbox",
+    {
+        connection,
+    },
+);

classes/workers/delivery.ts

@@ -0,0 +1,66 @@
+import { getLogger } from "@logtape/logtape";
+import { User } from "@versia/kit/db";
+import { Worker } from "bullmq";
+import chalk from "chalk";
+import { config } from "~/packages/config-manager";
+import { connection } from "~/utils/redis.ts";
+import {
+    type DeliveryJobData,
+    DeliveryJobType,
+    deliveryQueue,
+} from "../queues/delivery.ts";
+
+export const getDeliveryWorker = (): Worker<
+    DeliveryJobData,
+    void,
+    DeliveryJobType
+> =>
+    new Worker<DeliveryJobData, void, DeliveryJobType>(
+        deliveryQueue.name,
+        async (job) => {
+            switch (job.name) {
+                case DeliveryJobType.FederateEntity: {
+                    const { entity, recipientId, senderId } = job.data;
+
+                    const logger = getLogger(["federation", "delivery"]);
+
+                    const sender = await User.fromId(senderId);
+
+                    if (!sender) {
+                        throw new Error(
+                            `Could not resolve sender ID ${chalk.gray(senderId)}`,
+                        );
+                    }
+
+                    const recipient = await User.fromId(recipientId);
+
+                    if (!recipient) {
+                        throw new Error(
+                            `Could not resolve recipient ID ${chalk.gray(recipientId)}`,
+                        );
+                    }
+
+                    logger.debug`Federating entity ${chalk.gray(
+                        entity.id,
+                    )} from ${chalk.gray(`@${sender.getAcct()}`)} to ${chalk.gray(
+                        recipient.getAcct(),
+                    )}`;
+
+                    await sender.federateToUser(entity, recipient);
+
+                    logger.debug`${chalk.green(
+                        "✔",
+                    )} Finished federating entity ${chalk.gray(entity.id)}`;
+                }
+            }
+        },
+        {
+            connection,
+            removeOnComplete: {
+                age: config.queues.delivery.remove_on_complete,
+            },
+            removeOnFail: {
+                age: config.queues.delivery.remove_on_failure,
+            },
+        },
+    );

classes/workers/fetch.ts

@@ -0,0 +1,64 @@
+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";
+import {
+    type FetchJobData,
+    FetchJobType,
+    fetchQueue,
+} from "../queues/fetch.ts";
+
+export const getFetchWorker = (): Worker<FetchJobData, void, FetchJobType> =>
+    new Worker<FetchJobData, void, FetchJobType>(
+        fetchQueue.name,
+        async (job) => {
+            switch (job.name) {
+                case FetchJobType.Instance: {
+                    const { uri } = job.data;
+
+                    await job.log(`Fetching instance metadata from [${uri}]`);
+
+                    // Check if exists
+                    const host = new URL(uri).host;
+
+                    const existingInstance = await Instance.fromSql(
+                        eq(Instances.baseUrl, host),
+                    );
+
+                    if (existingInstance) {
+                        await job.log(
+                            "Instance is known, refetching remote data.",
+                        );
+
+                        await existingInstance.updateFromRemote();
+
+                        await job.log(
+                            `Instance [${uri}] successfully refetched`,
+                        );
+
+                        return;
+                    }
+
+                    await Instance.resolve(uri);
+
+                    await job.log(
+                        `${chalk.green(
+                            "✔",
+                        )} Finished fetching instance metadata from [${uri}]`,
+                    );
+                }
+            }
+        },
+        {
+            connection,
+            removeOnComplete: {
+                age: config.queues.fetch.remove_on_complete,
+            },
+            removeOnFail: {
+                age: config.queues.fetch.remove_on_failure,
+            },
+        },
+    );

classes/workers/inbox.ts

@@ -0,0 +1,167 @@
+import { getLogger } from "@logtape/logtape";
+import { Instance, User } from "@versia/kit/db";
+import { Worker } from "bullmq";
+import chalk from "chalk";
+import { config } from "~/packages/config-manager/index.ts";
+import { connection } from "~/utils/redis.ts";
+import { InboxProcessor } from "../inbox/processor.ts";
+import {
+    type InboxJobData,
+    InboxJobType,
+    inboxQueue,
+} from "../queues/inbox.ts";
+
+export const getInboxWorker = (): Worker<
+    InboxJobData,
+    Response,
+    InboxJobType
+> =>
+    new Worker<InboxJobData, Response, InboxJobType>(
+        inboxQueue.name,
+        async (job) => {
+            switch (job.name) {
+                case InboxJobType.ProcessEntity: {
+                    const {
+                        data,
+                        headers: {
+                            "x-signature": signature,
+                            "x-nonce": nonce,
+                            "x-signed-by": signedBy,
+                            authorization,
+                        },
+                        request,
+                        ip,
+                    } = job.data;
+
+                    const logger = getLogger(["federation", "inbox"]);
+
+                    logger.debug`Processing entity ${chalk.gray(
+                        data.id,
+                    )} from ${chalk.gray(signedBy)}`;
+
+                    if (authorization) {
+                        const processor = new InboxProcessor(
+                            request,
+                            data,
+                            null,
+                            {
+                                signature,
+                                nonce,
+                                authorization,
+                            },
+                            logger,
+                            ip,
+                        );
+
+                        logger.debug`Entity ${chalk.gray(
+                            data.id,
+                        )} is potentially from a bridge`;
+
+                        return await processor.process();
+                    }
+
+                    // If not potentially from bridge, check for required headers
+                    if (!(signature && nonce && signedBy)) {
+                        return Response.json(
+                            {
+                                error: "Missing required headers: x-signature, x-nonce, or x-signed-by",
+                            },
+                            {
+                                status: 400,
+                            },
+                        );
+                    }
+
+                    const sender = await User.resolve(signedBy);
+
+                    if (!(sender || signedBy.startsWith("instance "))) {
+                        return Response.json(
+                            {
+                                error: `Couldn't resolve sender URI ${signedBy}`,
+                            },
+                            {
+                                status: 404,
+                            },
+                        );
+                    }
+
+                    if (sender?.isLocal()) {
+                        return Response.json(
+                            {
+                                error: "Cannot process federation requests from local users",
+                            },
+                            {
+                                status: 400,
+                            },
+                        );
+                    }
+
+                    const remoteInstance = sender
+                        ? await Instance.fromUser(sender)
+                        : await Instance.resolveFromHost(
+                              signedBy.split(" ")[1],
+                          );
+
+                    if (!remoteInstance) {
+                        return Response.json(
+                            { error: "Could not resolve the remote instance." },
+                            {
+                                status: 500,
+                            },
+                        );
+                    }
+
+                    logger.debug`Entity ${chalk.gray(
+                        data.id,
+                    )} is from remote instance ${chalk.gray(
+                        remoteInstance.data.baseUrl,
+                    )}`;
+
+                    if (!remoteInstance.data.publicKey?.key) {
+                        throw new Error(
+                            `Instance ${remoteInstance.data.baseUrl} has no public key stored in database`,
+                        );
+                    }
+
+                    const processor = new InboxProcessor(
+                        request,
+                        data,
+                        {
+                            instance: remoteInstance,
+                            key:
+                                sender?.data.publicKey ??
+                                remoteInstance.data.publicKey.key,
+                        },
+                        {
+                            signature,
+                            nonce,
+                            authorization,
+                        },
+                        logger,
+                        ip,
+                    );
+
+                    const output = await processor.process();
+
+                    logger.debug`${chalk.green(
+                        "✔",
+                    )} Finished processing entity ${chalk.gray(data.id)}`;
+
+                    return output;
+                }
+
+                default: {
+                    throw new Error(`Unknown job type: ${job.name}`);
+                }
+            }
+        },
+        {
+            connection,
+            removeOnComplete: {
+                age: config.queues.inbox.remove_on_complete,
+            },
+            removeOnFail: {
+                age: config.queues.inbox.remove_on_failure,
+            },
+        },
+    );