Finish up event queue abstraction

apps/api/src/server.ts

@@ -21,8 +21,13 @@ import { CalculatorService, CalculatorUnavailableError } from "./services/calcul
 import { StandardsParamsSchema } from "@blade-and-brawn/calculator";
 import { StandardsService } from "./services/standards";
 
+// CONSTANTS
+// -----------------------
+const EVENT_QUEUE_MANAGE_DELAY_MS = 1000   // 1  sec
+const EVENT_QUEUE_CLEAN_DELAY_MS = 600000 // 10 min
+
 // SERVICES
-// -----------
+// -----------------------
 const s = (() => {
     const Standards = new StandardsService();
     const Calculator = new CalculatorService(DEFAULT_NAME);
@@ -31,7 +36,7 @@ const s = (() => {
 })();
 
 // ELYSIA
-// -----------
+// -----------------------
 export const app = new Elysia()
     .use(serverTiming())
     .use(
@@ -196,9 +201,12 @@ export const app = new Elysia()
                     // Run sync
                     .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => {
                         await s.Commerce.ProductSync.Queue.enqueue({
+                            source: "portal",
+                            payload: {
                                 session: { id: sessionId, name: "Portal" },
                                 filter: {
-                                pProductIds: pProductId ? [pProductId] : null
+                                    pProductIds: pProductId ? [pProductId] : undefined
+                                }
                             }
                         });
                     }, { params: t.Object({ pProductId: t.Optional(t.Numeric()) }) }),
@@ -229,8 +237,11 @@ export const app = new Elysia()
                 log.info({ productId: pProduct.id }, "printful webhook: product updated");
 
                 await s.Commerce.ProductSync.Queue.enqueue({
+                    source: "printful",
+                    payload: {
                         session: { id: randomUUIDv7(), name: "Printful" },
                         filter: { pProductIds: [pProduct.id] }
+                    }
                 });
                 break;
             }
@@ -305,9 +316,17 @@ app.listen(3000, async () => {
     if (cluster.worker?.id === 1) {
         log.info({ port: 3000 }, "server started")
 
+        // MANAGE QUEUES
+        const queues = [s.Commerce.ProductSync.Queue];
         while (true) {
-            await s.Commerce.ProductSync.Queue.drain().catch((err) => log.error(err));
+            for (const queue of queues) {
-            await sleep(1000); // every 1 second;
+                // 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);
         }
     }
 });

apps/api/src/services/commerce.ts

@@ -1,28 +1,32 @@
-import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions } from "@blade-and-brawn/commerce";
+import { PrintfulClient, ProductSyncer, WebflowClient } from "@blade-and-brawn/commerce";
 import { env, log } from "../util";
 import { db } from "../database/db";
-import type { Selectable, Transaction } from "kysely";
+import { EventQueueService, type EventType } from "./event-queue";
-import type { DB } from "../database/out/db";
+import { Type as t } from "@sinclair/typebox";
-import { sleep } from "bun";
+import type { Static } from "@sinclair/typebox";
+import { Value } from "@sinclair/typebox/value";
+import { sql } from "kysely";
 
-type EventStatus = "queued" | "active" | "failed" | "fulfilled";
+const ProductSyncPayloadSchema = t.Object({
+    session: t.Object({
+        name: t.String(),
+        id: t.String()
+    }),
+    filter: t.Optional(
+        t.Object({
+            pProductIds: t.Optional(t.Array(t.Number()))
+        })
+    )
+});
+type ProductSyncPayload = Static<typeof ProductSyncPayloadSchema>;
 
-export type ProductSyncPayload = {
+const ProductSyncStateSchema = t.Union([
-    session: {
+    t.Object({
-        name: string,
+        pProductIds: t.Optional(t.Array(t.Number()))
-        id: string
+    }),
-    },
+    t.Null()
-    filter: ProductSyncerOptions["filter"]
+]);
-}
+type ProductSyncState = Static<typeof ProductSyncStateSchema>;
-
-export type ProductSyncState = {
-    syncingPProductIds: number[]
-}
-
-type ProductSyncHandler = (id: string, payload: ProductSyncPayload, setState: (state: ProductSyncState) => Promise<void>) => Promise<void>
-
-class AlreadyClaimedEventError extends Error { }
-class ActiveEventLimitReachedError extends Error { }
 
 export class CommerceService {
     public readonly Printful = new PrintfulClient({
@@ -39,188 +43,49 @@ export class CommerceService {
 }
 
 class ProductSyncService {
-    // TODO: make service have their own logger, so we can enforce .child({component: <ServiceName>})
+    static readonly EVENT_TYPE: EventType = "product_sync";
-    readonly Queue: ProductSyncQueueService;
+
+    readonly Queue: EventQueueService<ProductSyncPayload, ProductSyncState>;
     private readonly ProductSyncer: ProductSyncer;
 
+    // TODO: make each service have their own logger, so we can enforce .child({component: <ServiceName>})
+
     constructor(private readonly Printful: PrintfulClient, private readonly Webflow: WebflowClient) {
         this.ProductSyncer = new ProductSyncer(this.Printful, this.Webflow, log);
-        this.Queue = new ProductSyncQueueService(async (_id, payload, setState) => {
+        this.Queue = new EventQueueService({
+            type: ProductSyncService.EVENT_TYPE,
+            schemas: { payload: ProductSyncPayloadSchema, state: ProductSyncStateSchema },
+            cfg: {
+                retry: { limit: 3, delayMs: 10000 },
+                concurrency: { limit: 1, timeoutMs: 600000 }
+            },
+            handler: async (id, payload, setState) => {
                 await this.ProductSyncer.run({
                     filter: payload.filter,
-                beforeStep: async (syncingPProductIds: number[]) => {
+                    beforeStep: async (pProductIds: number[]) => {
-                    await setState({ syncingPProductIds });
+                        await setState({ pProductIds });
                     }
                 });
+            }
         });
     }
 
     async getSessionSyncState(sessionId: string) {
-        const latestSync = await db.selectFrom("product_syncs")
+        const latestSync = await db.selectFrom("events")
-            .selectAll()
+            .select(["started_at", "ended_at", "has_failed", "state"])
-            .where("session_id", "=", sessionId)
+            .where("type", "=", ProductSyncService.EVENT_TYPE)
+            .where(sql<string>`payload->'session'->>'id'`, "=", sessionId)
             .orderBy("created_at", "desc")
             .limit(1)
             .executeTakeFirst();
         if (!latestSync) return;
 
+        Value.Assert(ProductSyncStateSchema, latestSync.state);
+
         return {
             startDate: latestSync.started_at,
-            status: ProductSyncQueueService.status(latestSync),
+            status: EventQueueService.status(latestSync),
-            syncingPProductIds: latestSync.syncing_p_product_ids
+            syncingPProductIds: latestSync.state?.pProductIds ?? []
-        }
-    }
-}
-
-class ProductSyncQueueService {
-    private static DRAIN_SLEEP_TIME = 10000; // 10 seconds
-    private static EVENT_TIMEOUT = 600000; // 10 min
-    private static MAX_ACTIVE_EVENTS = 1;
-
-    constructor(private handler: ProductSyncHandler) { }
-
-    static status(event: Selectable<ProductSyncs>): EventStatus {
-        if (!event.started_at) return "queued";
-        if (!event.ended_at) return "active";
-        return event.has_failed ? "failed" : "fulfilled";
-    }
-
-    async drain(): Promise<void> {
-        log.trace("draining queue");
-        while (true) {
-            try {
-                const nextActiveEvent = await this.dequeue();
-                if (!nextActiveEvent) break;
-            }
-            catch (err) {
-                if (err instanceof AlreadyClaimedEventError) {
-                    log.warn("event handling already claimed by another worker, skipping");
-                    continue;
-                }
-                if (err instanceof ActiveEventLimitReachedError) {
-                    await sleep(ProductSyncQueueService.DRAIN_SLEEP_TIME);
-                    continue;
-                }
-                else {
-                    log.error("failed to drain queue");
-                    throw err;
         };
     }
-        }
-    }
-
-    async dequeue() {
-        log.trace("attempting to dequeue");
-        const eventToHandle = await db.transaction().execute(async (trx) => {
-            const activeEvents = await this.activeEvents(trx);
-            let activeEventCount = activeEvents.length;
-            for (const activeEvent of activeEvents) {
-                // If event has been active past max active time, fail it
-                const activeEventTimedOut = (Date.now() - (activeEvent.started_at?.getTime() ?? 0) > ProductSyncQueueService.EVENT_TIMEOUT);
-                if (activeEventTimedOut) {
-                    log.info({ name: activeEvent.id }, "active event has timed out. failing it and continuing...");
-                    await this.fail(activeEvent.id, trx);
-                    --activeEventCount;
-                }
-            }
-
-            if (activeEventCount >= ProductSyncQueueService.MAX_ACTIVE_EVENTS)
-                throw new ActiveEventLimitReachedError();
-
-            const front = await this.front(trx);
-            if (!front) return;
-
-            const result = await trx.updateTable("product_syncs")
-                .set({ started_at: new Date() })
-                .where("id", "=", front.id)
-                .where("started_at", "is", null)
-                .executeTakeFirstOrThrow();
-
-            // Guard against another worker/machine having already claimed this event
-            if (result.numUpdatedRows === 0n) throw new AlreadyClaimedEventError();
-
-            return front;
-        });
-
-        if (eventToHandle) {
-            log.trace({ name: eventToHandle.id }, "dequeuing event");
-            await this.handleEvent(eventToHandle.id, {
-                session: {
-                    id: eventToHandle.session_id,
-                    name: eventToHandle.session_name
-                },
-                filter: { pProductIds: eventToHandle.p_product_id_filter }
-            });
-            return eventToHandle;
-        }
-    }
-
-    async enqueue(payload: ProductSyncPayload): Promise<string> {
-        log.info({ payload }, "enqueuing");
-        const result = await db.insertInto("product_syncs")
-            .values({
-                session_id: payload.session.id,
-                session_name: payload.session.name,
-                p_product_id_filter: payload.filter?.pProductIds ?? null,
-                started_at: null,
-            })
-            .returning(["id"])
-            .executeTakeFirstOrThrow();
-        return result.id;
-    }
-
-    async fail(id: string, trx?: Transaction<DB>): Promise<void> {
-        await (trx ?? db).updateTable("product_syncs")
-            .set({ syncing_p_product_ids: [], ended_at: new Date(), has_failed: true })
-            .where("id", "=", id)
-            .execute();
-    }
-
-    async fulfill(id: string, trx?: Transaction<DB>): Promise<void> {
-        await (trx ?? db).updateTable("product_syncs")
-            .set({ syncing_p_product_ids: [], ended_at: new Date() })
-            .where("id", "=", id)
-            .execute();
-    }
-
-    async clean() {
-        // TODO
-    }
-
-    async front(trx?: Transaction<DB>) {
-        const result = await (trx ?? db).selectFrom("product_syncs")
-            .selectAll()
-            .where("started_at", "is", null)
-            .orderBy("created_at", "asc")
-            .limit(1)
-            .executeTakeFirst();
-        return result;
-    }
-
-    async activeEvents(trx?: Transaction<DB>) {
-        return await (trx ?? db).selectFrom("product_syncs")
-            .select(["started_at", "id"])
-            .where("started_at", "is not", null)
-            .where("ended_at", "is", null)
-            .execute();
-    }
-
-    private async handleEvent(id: string, payload: ProductSyncPayload) {
-        log.info({ id, payload }, "handling event");
-        try {
-            await this.handler(id, payload, async (state) => {
-                // set event state
-                await db.updateTable("product_syncs")
-                    .set({ syncing_p_product_ids: state.syncingPProductIds })
-                    .where("id", "=", id)
-                    .execute();
-            });
-        }
-        catch (err) {
-            await this.fail(id);
-            throw err;
-        }
-
-        await this.fulfill(id);
-    }
 }

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

@@ -6,8 +6,8 @@ import { db } from "../database/db";
 import { Value } from "@sinclair/typebox/value";
 import type { TSchema } from "@sinclair/typebox";
 
-type EventType = "product_sync" | "product_order_created" /* TODO: more events... */;
+export type EventType = "product_sync" | "product_order_created" /* TODO: more events... */;
-type Source = "portal" | "printful" | "webflow";
+export type EventSource = "portal" | "printful" | "webflow";
 
 type EventHandler<Payload extends JsonValue, State extends JsonValue> =
     (id: string, payload: Payload, setState: (state: State) => Promise<void>) => Promise<void>;
@@ -16,7 +16,13 @@ type EventStatus = "queued" | "active" | "failed" | "fulfilled";
 class AlreadyClaimedEventError extends Error { }
 class ConcurrencyLimitReachedError extends Error { }
 
-type EventQueueServiceOptions = {
+type EventQueueServiceOptions<Payload extends JsonValue, State extends JsonValue> = {
+    type: EventType,
+    schemas: {
+        payload: TSchema & { static: Payload },
+        state: TSchema & { static: State },
+    },
+    cfg: {
         retry: {
             limit: number,
             delayMs: number,
@@ -24,19 +30,19 @@ type EventQueueServiceOptions = {
         concurrency: {
             limit: number,
             timeoutMs: number,
-    }
+        },
+    },
+    handler: EventHandler<Payload, State>,
 }
 
 export class EventQueueService<Payload extends JsonValue, State extends JsonValue> {
-    constructor(
+    private _lastCleanDate: Date = new Date(0);
-        readonly type: EventType,
-        private payloadSchema: TSchema & { static: Payload },
-        private stateSchema: TSchema & { static: State },
-        private handler: EventHandler<Payload, State>,
-        readonly opt: EventQueueServiceOptions
-    ) { }
 
-    static status(event: Selectable<DB["events"]>): EventStatus {
+    constructor(private opt: EventQueueServiceOptions<Payload, State>) { }
+
+    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";
         return event.has_failed ? "failed" : "fulfilled";
@@ -57,17 +63,17 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
                     continue;
                 }
                 if (err instanceof ConcurrencyLimitReachedError) {
-                    await sleep(this.opt.retry.delayMs);
+                    await sleep(this.opt.cfg.retry.delayMs);
                     continue;
                 }
 
-                if (++consecutiveFailures <= this.opt.retry.limit) {
+                if (++consecutiveFailures <= this.opt.cfg.retry.limit) {
                     log.warn({ err, consecutiveFailures }, "Dequeue failed, retrying")
-                    await sleep(this.opt.retry.delayMs);
+                    await sleep(this.opt.cfg.retry.delayMs);
                     continue;
                 }
 
-                log.error({ err }, "failed to drain queue");
+                log.error({ err, consecutiveFailures }, "failed to drain queue");
                 throw err;
             }
         }
@@ -87,11 +93,11 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
                 eb(
                     eb.selectFrom("events")
                         .select(eb.fn.countAll().as("count"))
-                        .where("type", "=", this.type)
+                        .where("type", "=", this.opt.type)
                         .where("started_at", "is not", null)
                         .where("ended_at", "is", null),
                     "<",
-                    this.opt.concurrency.limit
+                    this.opt.cfg.concurrency.limit
                 )
             )
             .executeTakeFirstOrThrow();
@@ -113,10 +119,10 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
         return front;
     }
 
-    async enqueue(source: Source, payload: Payload): Promise<string> {
+    async enqueue({ source, payload }: { source: EventSource, payload: Payload }): Promise<string> {
         log.info({ payload }, "enqueuing");
         const result = await db.insertInto("events")
-            .values({ type: this.type, source, payload })
+            .values({ type: this.opt.type, source, payload })
             .returning(["id"])
             .executeTakeFirstOrThrow();
         return result.id;
@@ -140,18 +146,20 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
         // 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.concurrency.timeoutMs);
+            const activeEventTimedOut = (Date.now() - (activeEvent.started_at?.getTime() ?? 0) > this.opt.cfg.concurrency.timeoutMs);
             if (activeEventTimedOut) {
-                log.info({ name: activeEvent.id }, "active event has timed out. failing it and continuing...");
+                log.warn({ name: activeEvent.id }, "active event has timed out. failing it and continuing...");
                 await this.fail(activeEvent.id);
             }
         }
+
+        this._lastCleanDate = new Date();
     }
 
     async front(trx?: Transaction<DB>) {
         const result = await (trx ?? db).selectFrom("events")
             .selectAll()
-            .where("type", "=", this.type)
+            .where("type", "=", this.opt.type)
             .where("started_at", "is", null)
             .orderBy("created_at", "asc")
             .limit(1)
@@ -162,7 +170,7 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
     async activeEvents(trx?: Transaction<DB>) {
         return await (trx ?? db).selectFrom("events")
             .select(["started_at", "id"])
-            .where("type", "=", this.type)
+            .where("type", "=", this.opt.type)
             .where("started_at", "is not", null)
             .where("ended_at", "is", null)
             .execute();
@@ -171,8 +179,8 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
     private async handleEvent(id: string, payload: JsonValue) {
         log.info({ id, payload }, "handling event");
         try {
-            Value.Assert(this.payloadSchema, payload);
+            Value.Assert(this.opt.schemas.payload, payload);
-            await this.handler(id, payload, async (state) => {
+            await this.opt.handler(id, payload, async (state) => {
                 await db.updateTable("events")
                     .set({ state })
                     .where("id", "=", id)

packages/commerce/src/sync.ts

@@ -19,7 +19,7 @@ type PMetaProduct = {
 
 export type ProductSyncerOptions = {
     filter?: {
-        pProductIds: number[] | null;
+        pProductIds?: number[];
     },
     beforeStart?: (startDate: Date) => Promise<void>
     afterCompletion?: (completionDate: Date, duration: number) => Promise<void>