diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts index ae40c02..bcfe1d7 100644 --- a/apps/api/src/database/migrations/2026-06-03.ts +++ b/apps/api/src/database/migrations/2026-06-03.ts @@ -12,6 +12,8 @@ export async function up(db: Kysely): Promise { .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() diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index c65ed04..255fde9 100644 --- a/apps/api/src/server.ts +++ b/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; diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 46570ee..b5d5d3b 100644 --- a/apps/api/src/services/commerce.ts +++ b/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`payload->'session'->>'id'`, "=", sessionId) diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index 4fbea2c..d02bf45 100644 --- a/apps/api/src/services/event-queue.ts +++ b/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

= (payload: Static

, setState: (state: Static) => Promise, id: string) => Promise; @@ -36,6 +36,9 @@ export function defineEventType

(opt: Event } export class EventQueue>> { + 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, "started_at" | "ended_at" | "has_failed">): EventStatus { - if (!event.started_at) return "pending"; + static status(event: Pick, "available_at" | "started_at" | "ended_at" | "has_failed">): EventStatus { + 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 this.cfg.concurrency.timeoutMs); if (processingEventTimedOut) { @@ -172,41 +179,92 @@ export class EventQueue) { 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) { + async list(opt: { filter?: { status?: EventStatus } } = {}, trx?: Transaction) { return await (trx ?? db).selectFrom("events") .select(["started_at", "id"]) .where("group", "=", this.group) - .where("started_at", "is not", null) - .where("ended_at", "is", null) + .$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) { - log.info({ id, payload }, "processing event"); - if (!(type in this.events)) throw new Error(`event type not registerd in this queue: ${type}`); - const cfg = this.events[type as keyof T]!; + private async processEvent(event: Pick, "id" | "type" | "tries" | "payload">) { + log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event"); + if (!(event.type in this.events)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`); + const cfg = this.events[event.type as keyof T]!; + try { - Value.Assert(cfg.schemas.payload, payload); - await cfg.processor(payload, async (state) => { - await db.updateTable("events") - .set({ state }) - .where("id", "=", id) - .execute(); - }, id); + Value.Assert(cfg.schemas.payload, event.payload); + await cfg.processor(event.payload, (state) => this.setEventState(event.id, state), event.id); + await this.fulfill(event.id); } catch (err) { - await this.fail(id); - throw 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); } - await this.fulfill(id); + } + + private async setEventState(id: string, state: JsonValue) { + await db.updateTable("events") + .set({ state }) + .where("id", "=", id) + .execute(); + } + + private async retry(id: string) { + await db.updateTable("events") + .set((eb) => ({ + 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(); } } diff --git a/apps/portal/src/routes/(app)/commerce/products/+page.svelte b/apps/portal/src/routes/(app)/commerce/products/+page.svelte index 562bc71..417a132 100644 --- a/apps/portal/src/routes/(app)/commerce/products/+page.svelte +++ b/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 = - res.data.syncingPProductIds; + synchronizer.syncingPProductIds = res.data.syncingPProductIds; } }, startPolling: () => { @@ -47,9 +55,7 @@ } }); - const isProductSynced = async ( - pProduct: Printful.Products.SyncProduct, - ) => { + const isProductSynced = async (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: - pProduct.id, + pProductId: pProduct.id, }) .get(); console.log(res.data); @@ -119,8 +124,7 @@