Cleanup terminology

apps/api/src/server.ts

@@ -20,11 +20,11 @@ import { randomUUIDv7, sleep } from "bun";
 import { CalculatorService, CalculatorUnavailableError } from "./services/calculator";
 import { StandardsParamsSchema } from "@blade-and-brawn/calculator";
 import { StandardsService } from "./services/standards";
+import type { EventQueueService } from "./services/event-queue";
 
 // CONSTANTS
 // -----------------------
 const EVENT_QUEUE_MANAGE_DELAY_MS = 1000   // 1  sec
-const EVENT_QUEUE_CLEAN_DELAY_MS = 600000 // 10 min
 
 // SERVICES
 // -----------------------
@@ -194,9 +194,9 @@ export const app = new Elysia()
                 .group("/sync", (app) => app
                     // Sync status
                     .get("/", async ({ sessionId }) => {
-                        const syncState = await s.Commerce.ProductSync.getSessionSyncState(sessionId);
-                        if (!syncState) throw new NotFoundError("No product sync found for the provided session");
-                        return syncState;
+                        const latestSyncState = await s.Commerce.ProductSync.getLatestSyncState(sessionId);
+                        if (!latestSyncState) throw new NotFoundError("No product sync found for the provided session");
+                        return latestSyncState;
                     })
                     // Run sync
                     .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => {
@@ -317,16 +317,18 @@ app.listen(3000, async () => {
         log.info({ port: 3000 }, "server started")
 
         // MANAGE QUEUES
-        const queues = [s.Commerce.ProductSync.Queue];
-        while (true) {
-            for (const queue of queues) {
-                // clean, if ready
-                if ((Date.now() - queue.lastCleanDate.getTime()) >= EVENT_QUEUE_CLEAN_DELAY_MS)
-                    await queue.clean();
-                // drain
-                await queue.drain().catch((err) => log.error(err));
-            }
-            await sleep(EVENT_QUEUE_MANAGE_DELAY_MS);
+        const queues: EventQueueService<any, any>[] = [s.Commerce.ProductSync.Queue];
+        for (const queue of queues) {
+            (async () => {
+                while (true) {
+                    // clean, if ready
+                    if ((Date.now() - queue.lastCleanDate.getTime()) >= queue.cfg.concurrency.timeoutMs)
+                        await queue.clean().catch((err) => log.error(err));
+                    // drain
+                    await queue.drain().catch((err) => log.error(err));
+                    await sleep(EVENT_QUEUE_MANAGE_DELAY_MS);
+                }
+            })();
         }
     }
 });

apps/api/src/services/commerce.ts

@@ -59,7 +59,7 @@ class ProductSyncService {
                 retry: { limit: 3, delayMs: 10000 },
                 concurrency: { limit: 1, timeoutMs: 600000 }
             },
-            handler: async (id, payload, setState) => {
+            processor: async (id, payload, setState) => {
                 await this.ProductSyncer.run({
                     filter: payload.filter,
                     beforeStep: async (pProductIds: number[]) => {
@@ -70,11 +70,12 @@ class ProductSyncService {
         });
     }
 
-    async getSessionSyncState(sessionId: string) {
+    async getLatestSyncState(sessionId: string) {
         const latestSync = await db.selectFrom("events")
             .select(["started_at", "ended_at", "has_failed", "state"])
             .where("type", "=", ProductSyncService.EVENT_TYPE)
             .where(sql<string>`payload->'session'->>'id'`, "=", sessionId)
+            .orderBy(sql`(started_at is not null and ended_at is null)`, "desc")
             .orderBy("created_at", "desc")
             .limit(1)
             .executeTakeFirst();

apps/api/src/services/event-queue.ts

@@ -9,9 +9,9 @@ import type { TSchema } from "@sinclair/typebox";
 export type EventType = "product_sync" | "product_order_created" /* TODO: more events... */;
 export type EventSource = "portal" | "printful" | "webflow";
 
-type EventHandler<Payload extends JsonValue, State extends JsonValue> =
+type EventProcessor<Payload extends JsonValue, State extends JsonValue> =
     (id: string, payload: Payload, setState: (state: State) => Promise<void>) => Promise<void>;
-type EventStatus = "queued" | "active" | "failed" | "fulfilled";
+type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
 
 class AlreadyClaimedEventError extends Error { }
 class ConcurrencyLimitReachedError extends Error { }
@@ -32,19 +32,22 @@ type EventQueueServiceOptions<Payload extends JsonValue, State extends JsonValue
             timeoutMs: number,
         },
     },
-    handler: EventHandler<Payload, State>,
+    processor: EventProcessor<Payload, State>,
 }
 
 export class EventQueueService<Payload extends JsonValue, State extends JsonValue> {
     private _lastCleanDate: Date = new Date(0);
+    readonly cfg: EventQueueServiceOptions<Payload, State>["cfg"];
 
-    constructor(private opt: EventQueueServiceOptions<Payload, State>) { }
+    constructor(private readonly opt: EventQueueServiceOptions<Payload, State>) {
+        this.cfg = opt.cfg;
+    }
 
     get lastCleanDate() { return this._lastCleanDate; };
 
     static status(event: Pick<Selectable<DB["events"]>, "started_at" | "ended_at" | "has_failed">): EventStatus {
-        if (!event.started_at) return "queued";
-        if (!event.ended_at) return "active";
+        if (!event.started_at) return "pending";
+        if (!event.ended_at) return "processing";
         return event.has_failed ? "failed" : "fulfilled";
     }
 
@@ -143,17 +146,20 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
     }
 
     async clean() {
-        // Find and fail any timed out active events
-        const activeEvents = await this.activeEvents();
-        for (const activeEvent of activeEvents) {
-            const activeEventTimedOut = (Date.now() - (activeEvent.started_at?.getTime() ?? 0) > this.opt.cfg.concurrency.timeoutMs);
-            if (activeEventTimedOut) {
-                log.warn({ name: activeEvent.id }, "active event has timed out. failing it and continuing...");
-                await this.fail(activeEvent.id);
+        try {
+            // Find and fail any timed out processing events
+            const processingEvents = await this.processingEvents();
+            for (const processingEvent of processingEvents) {
+                const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > this.opt.cfg.concurrency.timeoutMs);
+                if (processingEventTimedOut) {
+                    log.warn({ name: processingEvent.id }, "processing event has timed out. failing it and continuing...");
+                    await this.fail(processingEvent.id);
+                }
             }
         }
-
-        this._lastCleanDate = new Date();
+        finally {
+            this._lastCleanDate = new Date();
+        }
     }
 
     async front(trx?: Transaction<DB>) {
@@ -167,7 +173,7 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
         return result;
     }
 
-    async activeEvents(trx?: Transaction<DB>) {
+    async processingEvents(trx?: Transaction<DB>) {
         return await (trx ?? db).selectFrom("events")
             .select(["started_at", "id"])
             .where("type", "=", this.opt.type)
@@ -180,7 +186,7 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
         log.info({ id, payload }, "handling event");
         try {
             Value.Assert(this.opt.schemas.payload, payload);
-            await this.opt.handler(id, payload, async (state) => {
+            await this.opt.processor(id, payload, async (state) => {
                 await db.updateTable("events")
                     .set({ state })
                     .where("id", "=", id)