Move queue management onto all workers and make clean single atomic operation

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 {