From e4cb2da9cc1636c1dfbcb8b8a2e7953d73dd3368 Mon Sep 17 00:00:00 2001
From: Dominic Ferrando <dominicf2001@gmail.com>
Date: Sat, 18 Jul 2026 01:10:03 -0400
Subject: [PATCH] Improve event timeout with heartbeat_at

---
 apps/api/src/database/migrations/2026-06-03.ts |  1 +
 apps/api/src/scripts/register-webhooks.ts      |  5 ++++-
 apps/api/src/services/event-queue.ts           |  5 +++--
 apps/api/src/services/events.ts                | 10 +++++++---
 4 files changed, 15 insertions(+), 6 deletions(-)

diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts
index aee1d5e..7cbe15b 100644
--- a/apps/api/src/database/migrations/2026-06-03.ts
+++ b/apps/api/src/database/migrations/2026-06-03.ts
@@ -10,6 +10,7 @@ export async function up(db: Kysely<any>): Promise<void> {
         .addColumn("source", "text", (cb) => cb.notNull())
         .addColumn("payload", "jsonb", (cb) => cb.notNull())
         .addColumn("state", "jsonb")
+        .addColumn("heartbeat_at", "timestamptz")
         .addColumn("started_at", "timestamptz")
         .addColumn("ended_at", "timestamptz")
         .addColumn("available_at", "timestamptz", (cb) => cb.notNull().defaultTo(sql`now()`))
diff --git a/apps/api/src/scripts/register-webhooks.ts b/apps/api/src/scripts/register-webhooks.ts
index cd14127..ced5611 100644
--- a/apps/api/src/scripts/register-webhooks.ts
+++ b/apps/api/src/scripts/register-webhooks.ts
@@ -3,7 +3,10 @@ import { env } from "../util";
 
 const DOMAIN = "dev.api.bladeandbrawn.com";
 
-const PRINTFUL_WEBHOOK_URL = `http://${DOMAIN}/webhooks/printful?secret=${env.PRINTFUL_WEBHOOK_SECRET}`;
+const PRINTFUL_WEBHOOK_URL = env.NODE_ENV === "development" ?
+    `http://${DOMAIN}/webhooks/printful?secret=${env.PRINTFUL_WEBHOOK_SECRET}` :
+    `https://${DOMAIN}/webhooks/printful?secret=${env.PRINTFUL_WEBHOOK_SECRET}`;
+// http dosent work for webflow, so cant register properly for development env
 const WEBFLOW_WEBHOOK_URL = `https://${DOMAIN}/webhooks/webflow`;
 
 function printError(err: unknown): never {
diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts
index 3114fbd..5b71e3f 100644
--- a/apps/api/src/services/event-queue.ts
+++ b/apps/api/src/services/event-queue.ts
@@ -104,7 +104,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
         if (!front) return;
 
         const result = await db.updateTable("events")
-            .set({ started_at: new Date() })
+            .set({ started_at: new Date(), heartbeat_at: new Date() })
             .where("id", "=", front.id)
             .where("started_at", "is", null)
             .where((eb) =>
@@ -195,7 +195,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
 
         try {
             Value.Assert(typeCfg.schemas.payload, event.payload);
-            await typeCfg.processor(event.payload, (state) => this.Events.setState(event.id, state), event.id);
+            await typeCfg.processor(event.payload, (s) => this.Events.setState(event.id, s), event.id);
             await this.Events.resolve(event.id);
         }
         catch (err) {
@@ -236,6 +236,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
             .set((eb) => ({
                 errors: type === "error" ? eb("errors", "+", 1) : eb.ref("errors"),
                 rate_limits: type === "ratelimit" ? eb("rate_limits", "+", 1) : eb.ref("rate_limits"),
+                heartbeat_at: null,
                 started_at: null,
                 ended_at: null,
                 has_failed: false,
diff --git a/apps/api/src/services/events.ts b/apps/api/src/services/events.ts
index 9a91886..7ed1283 100644
--- a/apps/api/src/services/events.ts
+++ b/apps/api/src/services/events.ts
@@ -89,9 +89,12 @@ export class EventsService {
             .execute();
     }
 
-    async setState(id: string, state: JsonValue) {
+    async setState(id: string, state: JsonValue, shouldHeartbeat: boolean = true) {
         await db.updateTable("events")
-            .set({ state })
+            .set((eb) => ({
+                state,
+                heartbeat_at: shouldHeartbeat ? new Date() : eb.ref("heartbeat_at")
+            }))
             .where("id", "=", id)
             .execute();
     }
@@ -125,6 +128,7 @@ export class EventsService {
     async retry(id: string): Promise<boolean> {
         const result = await db.updateTable("events")
             .set({
+                heartbeat_at: null,
                 started_at: null,
                 ended_at: null,
                 has_failed: false,
@@ -147,7 +151,7 @@ export class EventsService {
                 }))
                 .where("started_at", "is not", null)
                 .where("ended_at", "is", null)
-                .where("started_at", "<", new Date(Date.now() - EventsService.CONCURRENCY_TIMEOUT_MS))
+                .where("heartbeat_at", "<", new Date(Date.now() - EventsService.CONCURRENCY_TIMEOUT_MS))
                 .$if(opt.filter?.type !== undefined, (qb) => qb.where("type", "=", opt.filter!.type!))
                 .$if(opt.filter?.group !== undefined, (qb) => qb.where("group", "=", opt.filter!.group!))
                 .returning(["id"])