From 58290216588d75c7ac4539083b00ecb5aec6a169 Mon Sep 17 00:00:00 2001
From: Dominic Ferrando <dominicf2001@gmail.com>
Date: Fri, 17 Jul 2026 10:24:37 -0400
Subject: [PATCH] Move queue management onto all workers and make clean single
 atomic operation

---
 apps/api/src/server.ts          | 38 ++++++++++++++++-----------------
 apps/api/src/services/events.ts | 25 +++++++++++++++-------
 2 files changed, 36 insertions(+), 27 deletions(-)

diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts
index a72513e..9676349 100644
--- a/apps/api/src/server.ts
+++ b/apps/api/src/server.ts
@@ -24,6 +24,7 @@ import { EventsService, EventStatusSchema } from "./services/events";
 // CONSTANTS
 // -----------------------
 const EVENT_QUEUE_MANAGE_DELAY_MS = 1000   // 1  sec
+const JWT_EXP = "7d";
 
 // SERVICES
 // -----------------------
@@ -124,7 +125,7 @@ export const app = new Elysia()
         if (body.password !== env.ADMIN_PASSWORD) throw status(401, "Invalid credentials");
 
         auth.set({
-            value: await jwt.sign({ sessionId: randomUUIDv7(), exp: "7d" }),
+            value: await jwt.sign({ sessionId: randomUUIDv7(), exp: JWT_EXP }),
             path: "/",
             maxAge: 60 * 60 * 24 * 7,
             sameSite: "lax",
@@ -345,27 +346,26 @@ export const app = new Elysia()
     });
 
 app.listen(3000, async () => {
-    if (cluster.worker?.id === 1) {
+    if (cluster.worker?.id === 1)
         log.info({ port: 3000 }, "server started")
 
-        // MANAGE QUEUES
-        for (const queue of queues) {
-            (async () => {
-                while (true) {
-                    try {
-                        // clean, if ready
-                        if ((Date.now() - queue.lastCleanDate.getTime()) >= EventsService.CONCURRENCY_TIMEOUT_MS)
-                            await queue.clean().catch((err) => log.error({ name: queue.group, err }));
-                        // drain
-                        await queue.drain();
-                    }
-                    catch (err) {
-                        log.error({ name: queue.group, err }, "error occurred during queue management");
-                    }
-                    await sleep(EVENT_QUEUE_MANAGE_DELAY_MS);
+    // MANAGE QUEUES
+    for (const queue of queues) {
+        (async () => {
+            while (true) {
+                try {
+                    // clean, if ready
+                    if ((Date.now() - queue.lastCleanDate.getTime()) >= EventsService.CONCURRENCY_TIMEOUT_MS)
+                        await queue.clean().catch((err) => log.error({ name: queue.group, err }));
+                    // drain
+                    await queue.drain();
                 }
-            })();
-        }
+                catch (err) {
+                    log.error({ name: queue.group, err }, "error occurred during queue management");
+                }
+                await sleep(EVENT_QUEUE_MANAGE_DELAY_MS);
+            }
+        })();
     }
 });
 
diff --git a/apps/api/src/services/events.ts b/apps/api/src/services/events.ts
index 52a1f20..80214ea 100644
--- a/apps/api/src/services/events.ts
+++ b/apps/api/src/services/events.ts
@@ -138,14 +138,23 @@ export class EventsService {
 
     async clean(opt: { filter?: { type?: string, group?: string } } = {}) {
         try {
-            // Find and fail any timed out processing events
-            const processingEvents = await this.list({ filter: { status: "processing", ...opt.filter } });
-            for (const processingEvent of processingEvents) {
-                const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > EventsService.CONCURRENCY_TIMEOUT_MS);
-                if (processingEventTimedOut) {
-                    log.warn({ name: processingEvent.id }, "processing event has timed out. failing it and continuing...");
-                    await this.fail(processingEvent.id);
-                }
+            const timedOutEvents = await db.updateTable("events")
+                .set((eb) => ({
+                    state: null,
+                    ended_at: new Date(),
+                    has_failed: true,
+                    errors: eb("errors", "+", 1),
+                }))
+                .where("started_at", "is not", null)
+                .where("ended_at", "is", null)
+                .where("started_at", "<", new Date(Date.now() - EventsService.CONCURRENCY_TIMEOUT_MS))
+                .$if(opt.filter?.type !== undefined, (qb) => qb.where("type", "=", opt.filter!.type!))
+                .$if(opt.filter?.group !== undefined, (qb) => qb.where("group", "=", opt.filter!.group!))
+                .returning(["id"])
+                .execute();
+
+            for (const event of timedOutEvents) {
+                log.warn({ name: event.id }, "processing event timed out, failed it");
             }
         }
         finally {