From 3ef361f521631b7329a1d2050dfb4fb7906b3b22 Mon Sep 17 00:00:00 2001
From: Jesse Wierzbinski <contact@cpluspatch.com>
Date: Mon, 25 Nov 2024 08:59:48 +0100
Subject: [PATCH] fix(federation): :bug: Correctly remove listeners after job
 processing finishes

---
 api/users/:uuid/inbox/index.ts | 29 ++++++++++++++++++++++++-----
 worker.ts                      |  2 +-
 2 files changed, 25 insertions(+), 6 deletions(-)

diff --git a/api/users/:uuid/inbox/index.ts b/api/users/:uuid/inbox/index.ts
index 6d9091ac..4942e8ec 100644
--- a/api/users/:uuid/inbox/index.ts
+++ b/api/users/:uuid/inbox/index.ts
@@ -1,9 +1,15 @@
 import { apiRoute, applyConfig } from "@/api";
 import { createRoute } from "@hono/zod-openapi";
 import type { Entity } from "@versia/federation/types";
+import type { Job } from "bullmq";
 import { z } from "zod";
 import { ErrorSchema } from "~/types/api";
-import { InboxJobType, inboxQueue, inboxWorker } from "~/worker";
+import {
+    type InboxJobData,
+    InboxJobType,
+    inboxQueue,
+    inboxWorker,
+} from "~/worker";
 
 export const meta = applyConfig({
     auth: {
@@ -117,17 +123,30 @@ export default apiRoute((app) =>
         });
 
         return new Promise<Response>((resolve, reject) => {
-            inboxWorker.on("completed", (job) => {
+            const successCallback = (
+                job: Job<InboxJobData, Response, InboxJobType>,
+            ): void => {
                 if (job.id === result.id) {
+                    inboxWorker.off("completed", successCallback);
+                    inboxWorker.off("failed", failureCallback);
                     resolve(job.returnvalue);
                 }
-            });
+            };
 
-            inboxWorker.on("failed", (job, error) => {
+            const failureCallback = (
+                job: Job<InboxJobData, Response, InboxJobType> | undefined,
+                error: Error,
+            ): void => {
                 if (job && job.id === result.id) {
+                    inboxWorker.off("completed", successCallback);
+                    inboxWorker.off("failed", failureCallback);
                     reject(error);
                 }
-            });
+            };
+
+            inboxWorker.on("completed", successCallback);
+
+            inboxWorker.on("failed", failureCallback);
         });
     }),
 );
diff --git a/worker.ts b/worker.ts
index db4d2ccc..92e75a92 100644
--- a/worker.ts
+++ b/worker.ts
@@ -24,7 +24,7 @@ export enum InboxJobType {
     ProcessEntity = "processEntity",
 }
 
-type InboxJobData = {
+export type InboxJobData = {
     data: Entity;
     headers: {
         "x-signature"?: string;