Improve how events are drained

apps/api/src/server.ts

@@ -16,7 +16,7 @@ import serverTiming from "@elysia/server-timing";
 import jwt from "@elysia/jwt";
 import { CommerceService } from "./services/commerce";
 import cluster from "node:cluster";
-import { randomUUIDv7 } from "bun";
+import { randomUUIDv7, sleep } from "bun";
 import { CalculatorService, CalculatorUnavailableError } from "./services/calculator";
 import { StandardsParamsSchema } from "@blade-and-brawn/calculator";
 import { StandardsService } from "./services/standards";
@@ -200,9 +200,6 @@ export const app = new Elysia()
                             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()) }) }),
                 )
@@ -231,17 +228,9 @@ export const app = new Elysia()
                 const pProduct = payload.data.sync_product;
                 log.info({ productId: pProduct.id }, "printful webhook: product updated");
 
-                // 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: { 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;
             }
@@ -316,7 +305,10 @@ app.listen(3000, async () => {
     if (cluster.worker?.id === 1) {
         log.info({ port: 3000 }, "server started")
 
-        await s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err }));
+        while (true) {
+            await s.Commerce.ProductSync.Queue.drain().catch((err) => log.error(err));
+            await sleep(1000); // every 1 second;
+        }
     }
 });
 

apps/api/src/services/commerce.ts

@@ -19,10 +19,6 @@ 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 { }
@@ -89,14 +85,12 @@ class ProductSyncQueueService {
         return event.has_failed ? "failed" : "fulfilled";
     }
 
-    async drain(opt: QueueDrainOptions = {}): Promise<void> {
+    async drain(): Promise<void> {
-        log.info("draining queue");
+        log.trace("draining queue");
         while (true) {
             try {
-                const dequeued = await this.dequeue();
+                const nextActiveEvent = await this.dequeue();
-
+                if (!nextActiveEvent) break;
-                if (!dequeued) break;
-                if (dequeued.id === opt.untilId) break;
             }
             catch (err) {
                 if (err instanceof AlreadyClaimedEventError) {
@@ -116,7 +110,7 @@ class ProductSyncQueueService {
     }
 
     async dequeue() {
-        log.info("dequeuing");
+        log.trace("attempting to dequeue");
         const eventToHandle = await db.transaction().execute(async (trx) => {
             const activeEvents = await this.activeEvents(trx);
             let activeEventCount = activeEvents.length;
@@ -134,7 +128,7 @@ class ProductSyncQueueService {
                 throw new ActiveEventLimitReachedError();
 
             const front = await this.front(trx);
-            if (!front) return log.info("nothing to dequeue");
+            if (!front) return;
 
             const result = await trx.updateTable("product_syncs")
                 .set({ started_at: new Date() })
@@ -149,6 +143,7 @@ class ProductSyncQueueService {
         });
 
         if (eventToHandle) {
+            log.trace({ name: eventToHandle.id }, "dequeuing event");
             await this.handleEvent(eventToHandle.id, {
                 session: {
                     id: eventToHandle.session_id,
@@ -161,6 +156,7 @@ class ProductSyncQueueService {
     }
 
     async enqueue(payload: ProductSyncPayload): Promise<string> {
+        log.info({ payload }, "enqueuing");
         const result = await db.insertInto("product_syncs")
             .values({
                 session_id: payload.session.id,
@@ -210,8 +206,8 @@ class ProductSyncQueueService {
     }
 
     private async handleEvent(id: string, payload: ProductSyncPayload) {
+        log.info({ id, payload }, "handling event");
         try {
-            log.info({ id, payload }, "handling event");
             await this.handler(id, payload, async (state) => {
                 // set event state
                 await db.updateTable("product_syncs")