From 29a7a05a08c46755cba5289c261012b4c710b90e Mon Sep 17 00:00:00 2001 From: Dominic Ferrando Date: Sat, 11 Jul 2026 23:15:06 -0400 Subject: [PATCH] Implement EventQueueService abstracted from product syncs along with its database changes --- .../api/src/database/migrations/2026-06-03.ts | 19 +- apps/api/src/services/commerce.ts | 2 +- apps/api/src/services/event-queue.ts | 188 ++++++++++++++++++ 3 files changed, 197 insertions(+), 12 deletions(-) create mode 100644 apps/api/src/services/event-queue.ts diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts index 9f5701e..af5a188 100644 --- a/apps/api/src/database/migrations/2026-06-03.ts +++ b/apps/api/src/database/migrations/2026-06-03.ts @@ -2,16 +2,13 @@ import { Kysely, sql } from 'kysely' import { addDefaultColumns } from '../db'; export async function up(db: Kysely): Promise { - // TABLE: PRODUCT_SYNCS - await db.schema.createTable("product_syncs") + // TABLE: EVENTS + await db.schema.createTable("events") .$call(addDefaultColumns) - .addColumn("session_id", "uuid", (cb) => cb.notNull()) - .addColumn("session_name", "text", (cb) => cb.notNull()) - .addColumn("p_product_id_filter", sql`integer[]`) - .addColumn("syncing_p_product_ids", sql`integer[]`, (cb) => cb - .notNull() - .defaultTo(sql`'{}'::integer[]`) - ) + .addColumn("type", "text", (cb) => cb.notNull()) + .addColumn("source", "text", (cb) => cb.notNull()) + .addColumn("payload", "jsonb", (cb) => cb.notNull()) + .addColumn("state", "jsonb") .addColumn("started_at", "timestamptz") .addColumn("ended_at", "timestamptz") .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false)) @@ -55,8 +52,8 @@ export async function up(db: Kysely): Promise { } export async function down(db: Kysely): Promise { - // TABLE: PRODUCT_SYNCS - await db.schema.dropTable("product_syncs").ifExists().execute() + // TABLE: EVENTS + await db.schema.dropTable("events").ifExists().execute() // TABLE: CALCULATORS await db.schema.dropTable("calculators").ifExists().execute() diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 5326cdf..b238a49 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -2,7 +2,7 @@ import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions import { env, log } from "../util"; import { db } from "../database/db"; import type { Selectable, Transaction } from "kysely"; -import type { DB, ProductSyncs } from "../database/out/db"; +import type { DB } from "../database/out/db"; import { sleep } from "bun"; type EventStatus = "queued" | "active" | "failed" | "fulfilled"; diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts new file mode 100644 index 0000000..39f0c2f --- /dev/null +++ b/apps/api/src/services/event-queue.ts @@ -0,0 +1,188 @@ +import type { Selectable, 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 type { TSchema } from "@sinclair/typebox"; + +type EventType = "product_sync" | "product_order_created" /* TODO: more events... */; +type Source = "portal" | "printful" | "webflow"; + +type EventHandler = + (id: string, payload: Payload, setState: (state: State) => Promise) => Promise; +type EventStatus = "queued" | "active" | "failed" | "fulfilled"; + +class AlreadyClaimedEventError extends Error { } +class ConcurrencyLimitReachedError extends Error { } + +type EventQueueServiceOptions = { + retry: { + limit: number, + delayMs: number, + }, + concurrency: { + limit: number, + timeoutMs: number, + } +} + +export class EventQueueService { + constructor( + readonly type: EventType, + private payloadSchema: TSchema & { static: Payload }, + private stateSchema: TSchema & { static: State }, + private handler: EventHandler, + readonly opt: EventQueueServiceOptions + ) { } + + static status(event: Selectable): EventStatus { + if (!event.started_at) return "queued"; + if (!event.ended_at) return "active"; + return event.has_failed ? "failed" : "fulfilled"; + } + + async drain(): Promise { + log.trace("draining queue"); + let consecutiveFailures = 0; + while (true) { + try { + const event = await this.dequeue(); + if (!event) break; + consecutiveFailures = 0; + } + catch (err) { + if (err instanceof AlreadyClaimedEventError) { + log.warn("event handling already claimed by another worker, skipping"); + continue; + } + if (err instanceof ConcurrencyLimitReachedError) { + await sleep(this.opt.retry.delayMs); + continue; + } + + if (++consecutiveFailures <= this.opt.retry.limit) { + log.warn({ err, consecutiveFailures }, "Dequeue failed, retrying") + await sleep(this.opt.retry.delayMs); + continue; + } + + log.error({ err }, "failed to drain queue"); + throw err; + } + } + } + + async dequeue() { + log.trace("dequeueing"); + + const front = await this.front(); + if (!front) return; + + const result = await db.updateTable("events") + .set({ started_at: new Date() }) + .where("id", "=", front.id) + .where("started_at", "is", null) + .where((eb) => + eb( + eb.selectFrom("events") + .select(eb.fn.countAll().as("count")) + .where("type", "=", this.type) + .where("started_at", "is not", null) + .where("ended_at", "is", null), + "<", + this.opt.concurrency.limit + ) + ) + .executeTakeFirstOrThrow(); + + // Guard against another worker/machine having already claimed this event + if (result.numUpdatedRows === 0n) { + const freshFront = await db.selectFrom("events") + .select("started_at") + .where("id", "=", front.id) + .executeTakeFirst(); + + if (!freshFront) throw new AlreadyClaimedEventError(); + if (freshFront?.started_at) throw new AlreadyClaimedEventError(); + throw new ConcurrencyLimitReachedError(); + }; + + log.trace({ name: front.id }, "dequeuing event"); + await this.handleEvent(front.id, front.payload); + return front; + } + + async enqueue(source: Source, payload: Payload): Promise { + log.info({ payload }, "enqueuing"); + const result = await db.insertInto("events") + .values({ type: this.type, source, payload }) + .returning(["id"]) + .executeTakeFirstOrThrow(); + return result.id; + } + + async fail(id: string, trx?: Transaction): Promise { + await (trx ?? db).updateTable("events") + .set({ state: null, ended_at: new Date(), has_failed: true }) + .where("id", "=", id) + .execute(); + } + + async fulfill(id: string, trx?: Transaction): Promise { + await (trx ?? db).updateTable("events") + .set({ state: null, ended_at: new Date() }) + .where("id", "=", id) + .execute(); + } + + async clean() { + // Find and fail any timed out active events + const activeEvents = await this.activeEvents(); + for (const activeEvent of activeEvents) { + const activeEventTimedOut = (Date.now() - (activeEvent.started_at?.getTime() ?? 0) > this.opt.concurrency.timeoutMs); + if (activeEventTimedOut) { + log.info({ name: activeEvent.id }, "active event has timed out. failing it and continuing..."); + await this.fail(activeEvent.id); + } + } + } + + async front(trx?: Transaction) { + const result = await (trx ?? db).selectFrom("events") + .selectAll() + .where("type", "=", this.type) + .where("started_at", "is", null) + .orderBy("created_at", "asc") + .limit(1) + .executeTakeFirst(); + return result; + } + + async activeEvents(trx?: Transaction) { + return await (trx ?? db).selectFrom("events") + .select(["started_at", "id"]) + .where("type", "=", this.type) + .where("started_at", "is not", null) + .where("ended_at", "is", null) + .execute(); + } + + private async handleEvent(id: string, payload: JsonValue) { + log.info({ id, payload }, "handling event"); + try { + Value.Assert(this.payloadSchema, payload); + await this.handler(id, payload, async (state) => { + await db.updateTable("events") + .set({ state }) + .where("id", "=", id) + .execute(); + }); + } + catch (err) { + await this.fail(id); + throw err; + } + await this.fulfill(id); + } +}