Generalize queue logic

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

@@ -17,16 +17,6 @@ export async function up(db: Kysely<any>): Promise<void> {
         .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false))
         .execute()
 
-    // INDEX: ONE_ACTIVE_PRODUCT_SYNC
-    await db.schema.createIndex("idx_one_active_product_sync")
-        .on("product_syncs")
-        .column("ended_at")
-        .unique()
-        .where(sql.ref("started_at"), "is not", null)
-        .where("ended_at", "is", null)
-        .nullsNotDistinct()
-        .execute();
-
     // TABLE: STANDARDS_DATASETS
     await db.schema.createTable("standards_datasets")
         .$call(addDefaultColumns)
@@ -68,9 +58,6 @@ export async function down(db: Kysely<any>): Promise<void> {
     // TABLE: PRODUCT_SYNCS
     await db.schema.dropTable("product_syncs").ifExists().execute()
 
-    // INDEX: ONE_ACTIVE_PRODUCT_SYNC
-    await db.schema.dropIndex("idx_one_active_product_sync").ifExists().execute();
-
     // TABLE: CALCULATORS
     await db.schema.dropTable("calculators").ifExists().execute()
 

apps/api/src/server.ts

@@ -189,21 +189,20 @@ export const app = new Elysia()
                 .group("/sync", (app) => app
                     // Sync status
                     .get("/", async ({ sessionId }) => {
-                        const latestSync = await s.Commerce.Sync.getLatest(sessionId);
+                        const syncState = await s.Commerce.ProductSync.getSessionSyncState(sessionId);
-                        if (!latestSync) throw new NotFoundError("No sync found");
+                        if (!syncState) throw new NotFoundError("No product sync found for the provided session");
-                        return {
+                        return syncState;
-                            startDate: latestSync.started_at,
-                            status: s.Commerce.Sync.deriveStatus(latestSync),
-                            syncingPProductIds: latestSync.syncing_p_product_ids
-                        }
                     })
                     // Run sync
                     .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => {
-                        await s.Commerce.Sync.start({
+                        await s.Commerce.ProductSync.Queue.enqueue({
                             session: { id: sessionId, name: "Portal" },
                             filter: {
                                 pProductIds: pProductId ? [pProductId] : null
                             }
+                        }).then(async (id) => {
+                            await s.Commerce.ProductSync.Queue.drain({ untilId: id });
+                            void s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err }));
                         });
                     }, { params: t.Object({ pProductId: t.Optional(t.Numeric()) }) }),
                 )
@@ -232,11 +231,17 @@ export const app = new Elysia()
                 const pProduct = payload.data.sync_product;
                 log.info({ productId: pProduct.id }, "printful webhook: product updated");
 
-                await s.Commerce.Sync.start({
+                // Controller layer handles draining the queue, potentionally across multiple workers
+                // TODO: maybe find better way of handling queue draining
+                // can cause a worker to hang for long time while draining
+                // this can cause an issue with webhook sender timing out and sending duplicate payload
+                // while we wait to process unrelated events
+                await s.Commerce.ProductSync.Queue.enqueue({
                     session: { id: randomUUIDv7(), name: "Printful" },
-                    filter: {
+                    filter: { pProductIds: [pProduct.id] }
-                        pProductIds: [pProduct.id]
+                }).then(async (id) => {
-                    }
+                    await s.Commerce.ProductSync.Queue.drain({ untilId: id });
+                    void s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err }));
                 });
                 break;
             }
@@ -311,13 +316,7 @@ app.listen(3000, async () => {
     if (cluster.worker?.id === 1) {
         log.info({ port: 3000 }, "server started")
 
-        try {
+        await s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err }));
-            const frontQueuedSync = await s.Commerce.Sync.Queue.front();
-            if (frontQueuedSync) await s.Commerce.Sync.start(frontQueuedSync);
-        }
-        catch (err) {
-            log.error({ err }, "failed to resume product sync queue")
-        }
     }
 });
 

apps/api/src/services/commerce.ts

@@ -1,14 +1,13 @@
 import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions } from "@blade-and-brawn/commerce";
 import { env, log } from "../util";
 import { db } from "../database/db";
-import { DatabaseError } from "pg";
+import type { Selectable, Transaction } from "kysely";
-import type { Insertable, Selectable } from "kysely";
+import type { DB, ProductSyncs } from "../database/out/db";
-import type { ProductSyncs } from "../database/out/db";
+import { sleep } from "bun";
 
-type SyncStatus = "queued" | "active" | "failed" | "fulfilled";
+type EventStatus = "queued" | "active" | "failed" | "fulfilled";
 
-type ProductSyncPayload = {
+export type ProductSyncPayload = {
-    id?: string,
     session: {
         name: string,
         id: string
@@ -16,7 +15,18 @@ type ProductSyncPayload = {
     filter: ProductSyncerOptions["filter"]
 }
 
-class AlreadyClaimedError extends Error { }
+export type ProductSyncState = {
+    syncingPProductIds: number[]
+}
+
+type QueueDrainOptions = {
+    untilId?: string
+}
+
+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({
@@ -29,173 +39,192 @@ export class CommerceService {
         token: env.WEBFLOW_AUTH,
         webhookSecret: env.WEBFLOW_WEBHOOK_SECRET,
     });
-    readonly Sync = new ProductSyncService(this.Printful, this.Webflow);
+    readonly ProductSync = new ProductSyncService(this.Printful, this.Webflow);
 }
 
 class ProductSyncService {
-    readonly Queue = new ProductSyncQueueService();
+    // TODO: make service have their own logger, so we can enforce .child({component: <ServiceName>})
+    readonly Queue: ProductSyncQueueService;
     private readonly ProductSyncer: ProductSyncer;
 
     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) => {
-
-    async start(payload: ProductSyncPayload) {
-        let id = payload.id ?? null;
-        let freedSlot = false;
-        try {
-            log.info({ session: payload.session }, "starting sync run");
             await this.ProductSyncer.run({
                 filter: payload.filter,
-                beforeStart: async (startDate) => {
+                beforeStep: async (syncingPProductIds: number[]) => {
-                    await this.Queue.enqueue(payload, startDate);
+                    await setState({ syncingPProductIds });
-                },
-                beforeStep: async (pProductIds: number[]) => {
-                    if (!id) throw new Error("Expected sync run ID");
-                    await db.updateTable("product_syncs")
-                        .set({ syncing_p_product_ids: pProductIds })
-                        .where("id", "=", id)
-                        .execute();
-                },
-                afterCompletion: async (endDate, duration) => {
-                    if (!id) throw new Error("Expected sync run ID");
-                    await this.Queue.fulfill(id, endDate)
-                    freedSlot = true;
                 }
             });
-        }
+        });
-        catch (err) {
-            if (err instanceof AlreadyClaimedError) {
-                log.info("sync already claimed by another worker, skipping");
-                return;
     }
 
-            const activeSync = await this.getActive();
+    async getSessionSyncState(sessionId: string) {
-
+        const latestSync = await db.selectFrom("product_syncs")
-            const activeSyncConflict = activeSync && err instanceof DatabaseError && err.code === "23505";
-            if (activeSyncConflict) {
-                // If active sync has been syncing for 10 min (600000ms), clear it and retry this sync
-                if (Date.now() - (activeSync.started_at?.getTime() ?? 0) > 600000) {
-                    log.info("already active sync exceeded timeout");
-                    await this.Queue.fail(activeSync.id, new Date());
-                    await this.start(payload);
-                }
-                else {
-                    log.info("sync already active, enqueuing next sync");
-                    await this.Queue.enqueue(payload);
-                }
-            }
-            else {
-                if (id) {
-                    await this.Queue.fail(id, new Date());
-                    freedSlot = true;
-                }
-                throw err;
-            }
-        }
-        finally {
-            if (freedSlot) {
-                try {
-                    const frontQueuedSync = await this.Queue.front();
-                    if (frontQueuedSync) await this.start(frontQueuedSync);
-                }
-                catch (err) {
-                    log.error({ err }, "failed to process next queued sync")
-                }
-            }
-        }
-    }
-
-    async getActive(sessionId?: string) {
-        return await db.selectFrom("product_syncs")
             .selectAll()
-            .$if(sessionId !== undefined, (qb) =>
+            .where("session_id", "=", sessionId)
-                qb.where("session_id", "=", sessionId!)
-            )
-            .where("started_at", "is not", null)
-            .where("ended_at", "is", null)
-            .limit(1)
-            .executeTakeFirst();
-    }
-
-    async getLatest(sessionId?: string) {
-        return await db.selectFrom("product_syncs")
-            .selectAll()
-            .$if(sessionId !== undefined, (qb) =>
-                qb.where("session_id", "=", sessionId!)
-            )
             .orderBy("created_at", "desc")
             .limit(1)
             .executeTakeFirst();
-    }
+        if (!latestSync) return;
 
-    deriveStatus(sync: Selectable<ProductSyncs>): SyncStatus {
+        return {
-        if (!sync.started_at) return "queued";
+            startDate: latestSync.started_at,
-        if (!sync.ended_at) return "active";
+            status: ProductSyncQueueService.status(latestSync),
-        return sync.has_failed ? "failed" : "fulfilled";
+            syncingPProductIds: latestSync.syncing_p_product_ids
+        }
     }
 }
 
 class ProductSyncQueueService {
-    async enqueue(payload: ProductSyncPayload, startDate?: Date): Promise<string> {
+    private static DRAIN_SLEEP_TIME = 10000; // 10 seconds
-        const values: Insertable<ProductSyncs> = {
+    private static EVENT_TIMEOUT = 600000; // 10 min
-            session_id: payload.session.id,
+    private static MAX_ACTIVE_EVENTS = 1;
-            session_name: payload.session.name,
+
-            p_product_id_filter: payload.filter?.pProductIds ?? null
+    constructor(private handler: ProductSyncHandler) { }
-        };
+
-        if (startDate) values.started_at = startDate;
+    static status(event: Selectable<ProductSyncs>): EventStatus {
-        if (payload.id) {
+        if (!event.started_at) return "queued";
-            const result = await db.updateTable("product_syncs")
+        if (!event.ended_at) return "active";
-                .set(values)
+        return event.has_failed ? "failed" : "fulfilled";
-                .where("id", "=", payload.id)
+    }
-                .where("started_at", "is", null)
+
-                .executeTakeFirstOrThrow();
+    async drain(opt: QueueDrainOptions = {}): Promise<void> {
-            // Guard against another worker/machine having already claimed
+        log.info("draining queue");
-            // this queued row between `queue.front()` and now.
+        while (true) {
-            if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError();
+            try {
-            return payload.id;
+                const dequeued = await this.dequeue();
+
+                if (!dequeued) break;
+                if (dequeued.id === opt.untilId) 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 {
-            const result = await db.insertInto("product_syncs")
+                    log.error("failed to drain queue");
-                .values(values)
+                    throw err;
-                .returning("id")
+                };
-                .executeTakeFirstOrThrow();
+            }
-            return result.id;
         }
     }
 
-    async front(): Promise<ProductSyncPayload | undefined> {
+    async dequeue() {
-        const result = await db.selectFrom("product_syncs")
+        log.info("dequeuing");
+        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 log.info("nothing to dequeue");
+
+            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) {
+            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> {
+        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();
-        if (!result) return;
+        return result;
-
-        return {
-            id: result.id,
-            session: {
-                id: result.session_id,
-                name: result.session_name
-            },
-            filter: {
-                pProductIds: result.p_product_id_filter
-            }
-        }
     }
 
-    async fail(id: string, endDate: Date) {
+    async activeEvents(trx?: Transaction<DB>) {
-        await db.updateTable("product_syncs")
+        return await (trx ?? db).selectFrom("product_syncs")
-            .set({ syncing_p_product_ids: [], ended_at: endDate, has_failed: true })
+            .select(["started_at", "id"])
-            .where("id", "=", id)
+            .where("started_at", "is not", null)
+            .where("ended_at", "is", null)
             .execute();
     }
 
-    async fulfill(id: string, endDate: Date) {
+    private async handleEvent(id: string, payload: ProductSyncPayload) {
+        try {
+            log.info({ id, payload }, "handling event");
+            await this.handler(id, payload, async (state) => {
+                // set event state
                 await db.updateTable("product_syncs")
-            .set({ syncing_p_product_ids: [], ended_at: endDate })
+                    .set({ syncing_p_product_ids: state.syncingPProductIds })
                     .where("id", "=", id)
                     .execute();
+            });
+        }
+        catch (err) {
+            await this.fail(id);
+            throw err;
+        }
+
+        await this.fulfill(id);
     }
 }