Add retries to processEvent handler with DB integration

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

@@ -12,6 +12,8 @@ export async function up(db: Kysely<any>): Promise<void> {
         .addColumn("state", "jsonb")
         .addColumn("started_at", "timestamptz")
         .addColumn("ended_at", "timestamptz")
+        .addColumn("available_at", "timestamptz", (cb) => cb.notNull().defaultTo(sql`now()`))
+        .addColumn("tries", "integer", (cb) => cb.notNull().defaultTo(0))
         .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false))
         .execute()
 

apps/api/src/server.ts

@@ -228,7 +228,6 @@ export const app = new Elysia()
         ))
 
     // WEBHOOKS
-    // TODO: account for automatic webhook retries from the webhook sender end (especially timeouts)
     .post("/webhooks/printful", async ({ body }) => {
         // https://webflow.com/integrations/printful
         const payload = body as Printful.Webhook.EventPayload;

apps/api/src/services/commerce.ts

@@ -152,7 +152,7 @@ class ApparelSyncService {
 
     async getLatestSyncState(sessionId: string) {
         const latestSync = await db.selectFrom("events")
-            .select(["started_at", "ended_at", "has_failed", "state"])
+            .select(["started_at", "available_at", "ended_at", "has_failed", "state"])
             .where("group", "=", this.Queue.group)
             .where("type", "=", "apparel_sync_update")
             .where(sql<string>`payload->'session'->>'id'`, "=", sessionId)

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

@@ -1,13 +1,13 @@
-import type { Selectable, Transaction } from "kysely";
+import { sql, type Selectable, type 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 { AssertError, Value } from "@sinclair/typebox/value";
 import type { TSchema, Static } from "@sinclair/typebox";
 
 export type EventSource = "portal" | "printful" | "webflow";
-export type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
+export type EventStatus = "waiting" | "available" | "processing" | "failed" | "fulfilled";
 
 type EventProcessor<P extends TSchema, S extends TSchema> =
     (payload: Static<P>, setState: (state: Static<S>) => Promise<void>, id: string) => Promise<void>;
@@ -36,6 +36,9 @@ export function defineEventType<P extends TSchema, S extends TSchema>(opt: Event
 }
 
 export class EventQueue<G extends string, T extends Record<string, EventTypeOptions<any, any>>> {
+    private static readonly DEQUEUE_RETRY_DELAY_MS = 10000; // 10 seconds
+    private static readonly DEQUEUE_RETRY_LIMIT = 3;
+
     private _lastCleanDate: Date = new Date(0);
     readonly cfg: EventQueueCfg;
     readonly group: G;
@@ -49,8 +52,11 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
 
     get lastCleanDate() { return this._lastCleanDate; };
 
-    static status(event: Pick<Selectable<DB["events"]>, "started_at" | "ended_at" | "has_failed">): EventStatus {
+    static status(event: Pick<Selectable<DB["events"]>, "available_at" | "started_at" | "ended_at" | "has_failed">): EventStatus {
-        if (!event.started_at) return "pending";
+        if (!event.started_at) {
+            if (Date.now() < event.available_at.getTime()) return "waiting";
+            return "available";
+        };
         if (!event.ended_at) return "processing";
         return event.has_failed ? "failed" : "fulfilled";
     }
@@ -66,24 +72,24 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
             }
             catch (err) {
                 if (err instanceof AlreadyClaimedEventError) {
-                    log.warn("event handling already claimed by another worker, skipping");
+                    log.warn("event processing already claimed by another worker, skipping");
                     continue;
                 }
                 if (err instanceof ConcurrencyLimitReachedError) {
-                    await sleep(this.cfg.retry.delayMs);
+                    await sleep(EventQueue.DEQUEUE_RETRY_DELAY_MS);
                     continue;
                 }
 
-                if (++consecutiveFailures <= this.cfg.retry.limit) {
+                if (++consecutiveFailures <= EventQueue.DEQUEUE_RETRY_LIMIT) {
-                    log.warn({ err, consecutiveFailures }, "Dequeue failed, retrying")
+                    log.warn({ err, consecutiveFailures }, "dequeue failed, retrying")
-                    await sleep(this.cfg.retry.delayMs);
+                    await sleep(EventQueue.DEQUEUE_RETRY_DELAY_MS);
                     continue;
                 }
 
                 log.error({ err, consecutiveFailures }, "failed to drain queue");
                 throw err;
             }
-        }
+        };
     }
 
     async dequeue() {
@@ -109,20 +115,21 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
             )
             .executeTakeFirstOrThrow();
 
-        // Guard against another worker/machine having already claimed this event
+        // Guard against another worker/machine having already claimed this event and guard against max concurrency
         if (result.numUpdatedRows === 0n) {
-            const freshFront = await db.selectFrom("events")
+            const frontFresh = await db.selectFrom("events")
                 .select("started_at")
                 .where("id", "=", front.id)
                 .executeTakeFirst();
 
-            if (!freshFront) throw new AlreadyClaimedEventError();
+            if (!frontFresh) throw new AlreadyClaimedEventError();
-            if (freshFront?.started_at) throw new AlreadyClaimedEventError();
+            if (frontFresh?.started_at) throw new AlreadyClaimedEventError();
             throw new ConcurrencyLimitReachedError();
         };
 
         log.trace({ name: front.id }, "dequeuing event");
-        await this.processEvent(front.id, front.type, front.payload);
+        await this.processEvent(front);
+
         return front;
     }
 
@@ -156,7 +163,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
     async clean() {
         try {
             // Find and fail any timed out processing events
-            const processingEvents = await this.processingEvents();
+            const processingEvents = await this.list({ filter: { status: "processing" } });
             for (const processingEvent of processingEvents) {
                 const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > this.cfg.concurrency.timeoutMs);
                 if (processingEventTimedOut) {
@@ -172,41 +179,92 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
 
     async front(trx?: Transaction<DB>) {
         const result = await (trx ?? db).selectFrom("events")
-            .selectAll()
+            .select(["id", "type", "payload", "available_at", "tries"])
             .where("group", "=", this.group)
             .where("started_at", "is", null)
+            .where("available_at", "<=", new Date())
             .orderBy("created_at", "asc")
             .limit(1)
             .executeTakeFirst();
         return result;
     }
 
-    async processingEvents(trx?: Transaction<DB>) {
+    async list(opt: { filter?: { status?: EventStatus } } = {}, trx?: Transaction<DB>) {
         return await (trx ?? db).selectFrom("events")
             .select(["started_at", "id"])
             .where("group", "=", this.group)
+            .$if(opt.filter?.status === "waiting", (qb) => qb
+                .where("started_at", "is", null)
+                .where("available_at", ">", new Date())
+            )
+            .$if(opt.filter?.status === "available", (qb) => qb
+                .where("started_at", "is", null)
+                .where("available_at", "<=", new Date())
+            )
+            .$if(opt.filter?.status === "processing", (qb) => qb
                 .where("started_at", "is not", null)
                 .where("ended_at", "is", null)
+            )
+            .$if(opt.filter?.status === "failed", (qb) => qb
+                .where("started_at", "is not", null)
+                .where("ended_at", "is not", null)
+                .where("has_failed", "is", true)
+            )
+            .$if(opt.filter?.status === "fulfilled", (qb) => qb
+                .where("started_at", "is not", null)
+                .where("ended_at", "is not", null)
+                .where("has_failed", "is", false)
+            )
+            .orderBy(sql`(started_at is not null and ended_at is null)`, "desc")
+            .orderBy("created_at", "desc")
             .execute();
     }
 
-    private async processEvent(id: string, type: string, payload: JsonValue) {
+    private async processEvent(event: Pick<Selectable<DB["events"]>, "id" | "type" | "tries" | "payload">) {
-        log.info({ id, payload }, "processing event");
+        log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event");
-        if (!(type in this.events)) throw new Error(`event type not registerd in this queue: ${type}`);
+        if (!(event.type in this.events)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`);
-        const cfg = this.events[type as keyof T]!;
+        const cfg = this.events[event.type as keyof T]!;
+
         try {
-            Value.Assert(cfg.schemas.payload, payload);
+            Value.Assert(cfg.schemas.payload, event.payload);
-            await cfg.processor(payload, async (state) => {
+            await cfg.processor(event.payload, (state) => this.setEventState(event.id, state), event.id);
+            await this.fulfill(event.id);
+        }
+        catch (err) {
+            if (err instanceof AssertError) {
+                log.error({ tries: event.tries }, "could not process event: malformed payload");
+                await this.fail(event.id);
+                return;
+            }
+
+            if (event.tries < this.cfg.retry.limit) {
+                log.warn({ err, tries: event.tries, delay: this.cfg.retry.delayMs }, "encountered an error during event processing, retrying")
+                await this.retry(event.id);
+                return;
+            }
+
+            log.error({ err, tries: event.tries }, "event failed");
+            await this.fail(event.id);
+        }
+    }
+
+    private async setEventState(id: string, state: JsonValue) {
         await db.updateTable("events")
             .set({ state })
             .where("id", "=", id)
             .execute();
-            }, id);
     }
-        catch (err) {
+
-            await this.fail(id);
+    private async retry(id: string) {
-            throw err;
+        await db.updateTable("events")
-        }
+            .set((eb) => ({
-        await this.fulfill(id);
+                tries: eb("tries", "+", 1),
+                started_at: null,
+                ended_at: null,
+                has_failed: false,
+                available_at: new Date(Date.now() + this.cfg.retry.delayMs)
+            }))
+            .where("id", "=", id)
+            .execute();
     }
 }

apps/portal/src/routes/(app)/commerce/products/+page.svelte

@@ -3,7 +3,12 @@
     import { onMount } from "svelte";
     import { api } from "$lib/api.js";
 
-    type SyncStatus = "pending" | "processing" | "failed" | "fulfilled";
+    type SyncStatus =
+        | "waiting"
+        | "available"
+        | "processing"
+        | "failed"
+        | "fulfilled";
 
     const { data } = $props();
 
@@ -11,7 +16,11 @@
         status: "none" as SyncStatus | "none",
         syncingPProductIds: [] as number[],
         get isInProgress() {
-            return this.status === "processing" || this.status === "pending";
+            return (
+                this.status !== "failed" &&
+                this.status !== "fulfilled" &&
+                this.status !== "none"
+            );
         },
         poll: async function () {
             const res = await api.commerce.products.sync.get();
@@ -20,8 +29,7 @@
                 synchronizer.syncingPProductIds = [];
             } else {
                 synchronizer.status = res.data.status;
-                synchronizer.syncingPProductIds =
+                synchronizer.syncingPProductIds = res.data.syncingPProductIds;
-                    res.data.syncingPProductIds;
             }
         },
         startPolling: () => {
@@ -47,9 +55,7 @@
         }
     });
 
-    const isProductSynced = async (
+    const isProductSynced = async (pProduct: Printful.Products.SyncProduct) => {
-        pProduct: Printful.Products.SyncProduct,
-    ) => {
         const wProducts = await data.products.webflow;
         return wProducts.some(
             (p) => p.product.id === pProduct.external_id.split("-")[0],
@@ -88,8 +94,7 @@
                                 onclick={async () => {
                                     const res = await api.commerce
                                         .products({
-                                            pProductId:
+                                            pProductId: pProduct.id,
-                                                pProduct.id,
                                         })
                                         .get();
                                     console.log(res.data);
@@ -119,8 +124,7 @@
                         <td>
                             <button
                                 disabled={synchronizer.status === "processing"}
-                                onclick={() =>
+                                onclick={() => synchronizer.sync(pProduct.id)}
-                                    synchronizer.sync(pProduct.id)}
                                 class="btn"
                             >
                                 Sync