Implement EventQueueService abstracted from product syncs along with its database changes

apps/api/src/database/migrations/2026-06-03.ts

@@ -2,16 +2,13 @@ import { Kysely, sql } from 'kysely'
 import { addDefaultColumns } from '../db';
 
 export async function up(db: Kysely<any>): Promise<void> {
-    // TABLE: PRODUCT_SYNCS
-    await db.schema.createTable("product_syncs")
+    // TABLE: EVENTS
+    await db.schema.createTable("events")
         .$call(addDefaultColumns)
-        .addColumn("session_id", "uuid", (cb) => cb.notNull())
-        .addColumn("session_name", "text", (cb) => cb.notNull())
-        .addColumn("p_product_id_filter", sql`integer[]`)
-        .addColumn("syncing_p_product_ids", sql`integer[]`, (cb) => cb
-            .notNull()
-            .defaultTo(sql`'{}'::integer[]`)
-        )
+        .addColumn("type", "text", (cb) => cb.notNull())
+        .addColumn("source", "text", (cb) => cb.notNull())
+        .addColumn("payload", "jsonb", (cb) => cb.notNull())
+        .addColumn("state", "jsonb")
         .addColumn("started_at", "timestamptz")
         .addColumn("ended_at", "timestamptz")
         .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false))
@@ -55,8 +52,8 @@ export async function up(db: Kysely<any>): Promise<void> {
 }
 
 export async function down(db: Kysely<any>): Promise<void> {
-    // TABLE: PRODUCT_SYNCS
-    await db.schema.dropTable("product_syncs").ifExists().execute()
+    // TABLE: EVENTS
+    await db.schema.dropTable("events").ifExists().execute()
 
     // TABLE: CALCULATORS
     await db.schema.dropTable("calculators").ifExists().execute()

apps/api/src/services/commerce.ts

@@ -2,7 +2,7 @@ import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions
 import { env, log } from "../util";
 import { db } from "../database/db";
 import type { Selectable, Transaction } from "kysely";
-import type { DB, ProductSyncs } from "../database/out/db";
+import type { DB } from "../database/out/db";
 import { sleep } from "bun";
 
 type EventStatus = "queued" | "active" | "failed" | "fulfilled";

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

@@ -0,0 +1,188 @@
+import type { Selectable, Transaction } from "kysely";
+import type { DB, JsonValue } from "../database/out/db";
+import { log } from "../util";
+import { sleep } from "bun";
+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... */;
+type Source = "portal" | "printful" | "webflow";
+
+type EventHandler<Payload extends JsonValue, State extends JsonValue> =
+    (id: string, payload: Payload, setState: (state: State) => Promise<void>) => Promise<void>;
+type EventStatus = "queued" | "active" | "failed" | "fulfilled";
+
+class AlreadyClaimedEventError extends Error { }
+class ConcurrencyLimitReachedError extends Error { }
+
+type EventQueueServiceOptions = {
+    retry: {
+        limit: number,
+        delayMs: number,
+    },
+    concurrency: {
+        limit: number,
+        timeoutMs: number,
+    }
+}
+
+export class EventQueueService<Payload extends JsonValue, State extends JsonValue> {
+    constructor(
+        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 {
+        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");
+        let consecutiveFailures = 0;
+        while (true) {
+            try {
+                const event = await this.dequeue();
+                if (!event) break;
+                consecutiveFailures = 0;
+            }
+            catch (err) {
+                if (err instanceof AlreadyClaimedEventError) {
+                    log.warn("event handling already claimed by another worker, skipping");
+                    continue;
+                }
+                if (err instanceof ConcurrencyLimitReachedError) {
+                    await sleep(this.opt.retry.delayMs);
+                    continue;
+                }
+
+                if (++consecutiveFailures <= this.opt.retry.limit) {
+                    log.warn({ err, consecutiveFailures }, "Dequeue failed, retrying")
+                    await sleep(this.opt.retry.delayMs);
+                    continue;
+                }
+
+                log.error({ err }, "failed to drain queue");
+                throw err;
+            }
+        }
+    }
+
+    async dequeue() {
+        log.trace("dequeueing");
+
+        const front = await this.front();
+        if (!front) return;
+
+        const result = await db.updateTable("events")
+            .set({ started_at: new Date() })
+            .where("id", "=", front.id)
+            .where("started_at", "is", null)
+            .where((eb) =>
+                eb(
+                    eb.selectFrom("events")
+                        .select(eb.fn.countAll().as("count"))
+                        .where("type", "=", this.type)
+                        .where("started_at", "is not", null)
+                        .where("ended_at", "is", null),
+                    "<",
+                    this.opt.concurrency.limit
+                )
+            )
+            .executeTakeFirstOrThrow();
+
+        // Guard against another worker/machine having already claimed this event
+        if (result.numUpdatedRows === 0n) {
+            const freshFront = await db.selectFrom("events")
+                .select("started_at")
+                .where("id", "=", front.id)
+                .executeTakeFirst();
+
+            if (!freshFront) throw new AlreadyClaimedEventError();
+            if (freshFront?.started_at) throw new AlreadyClaimedEventError();
+            throw new ConcurrencyLimitReachedError();
+        };
+
+        log.trace({ name: front.id }, "dequeuing event");
+        await this.handleEvent(front.id, front.payload);
+        return front;
+    }
+
+    async enqueue(source: Source, payload: Payload): Promise<string> {
+        log.info({ payload }, "enqueuing");
+        const result = await db.insertInto("events")
+            .values({ type: this.type, source, payload })
+            .returning(["id"])
+            .executeTakeFirstOrThrow();
+        return result.id;
+    }
+
+    async fail(id: string, trx?: Transaction<DB>): Promise<void> {
+        await (trx ?? db).updateTable("events")
+            .set({ state: null, ended_at: new Date(), has_failed: true })
+            .where("id", "=", id)
+            .execute();
+    }
+
+    async fulfill(id: string, trx?: Transaction<DB>): Promise<void> {
+        await (trx ?? db).updateTable("events")
+            .set({ state: null, ended_at: new Date() })
+            .where("id", "=", id)
+            .execute();
+    }
+
+    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.concurrency.timeoutMs);
+            if (activeEventTimedOut) {
+                log.info({ name: activeEvent.id }, "active event has timed out. failing it and continuing...");
+                await this.fail(activeEvent.id);
+            }
+        }
+    }
+
+    async front(trx?: Transaction<DB>) {
+        const result = await (trx ?? db).selectFrom("events")
+            .selectAll()
+            .where("type", "=", this.type)
+            .where("started_at", "is", null)
+            .orderBy("created_at", "asc")
+            .limit(1)
+            .executeTakeFirst();
+        return result;
+    }
+
+    async activeEvents(trx?: Transaction<DB>) {
+        return await (trx ?? db).selectFrom("events")
+            .select(["started_at", "id"])
+            .where("type", "=", this.type)
+            .where("started_at", "is not", null)
+            .where("ended_at", "is", null)
+            .execute();
+    }
+
+    private async handleEvent(id: string, payload: JsonValue) {
+        log.info({ id, payload }, "handling event");
+        try {
+            Value.Assert(this.payloadSchema, payload);
+            await this.handler(id, payload, async (state) => {
+                await db.updateTable("events")
+                    .set({ state })
+                    .where("id", "=", id)
+                    .execute();
+            });
+        }
+        catch (err) {
+            await this.fail(id);
+            throw err;
+        }
+        await this.fulfill(id);
+    }
+}