diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts index 618d62e..aee1d5e 100644 --- a/apps/api/src/database/migrations/2026-06-03.ts +++ b/apps/api/src/database/migrations/2026-06-03.ts @@ -17,6 +17,15 @@ export async function up(db: Kysely): Promise { .addColumn("rate_limits", "integer", (cb) => cb.notNull().defaultTo(0)) .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false)) .addColumn("last_error", "jsonb") + .addColumn("dedupe_key", "text") + .execute() + + await db.schema.createIndex("events_idx_dedupe_key") + .on("events") + .columns(["group", "type", "dedupe_key"]) + .unique() + .where("dedupe_key", "is not", null) + .where(sql.ref("ended_at"), "is", null) .execute() // TABLE: STANDARDS_DATASETS @@ -60,6 +69,8 @@ export async function down(db: Kysely): Promise { // TABLE: EVENTS await db.schema.dropTable("events").ifExists().execute() + await db.schema.dropIndex("events_idx_dedupe_key").ifExists().execute(); + // TABLE: CALCULATORS await db.schema.dropTable("calculators").ifExists().execute() diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 9676349..006b63d 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -24,7 +24,7 @@ import { EventsService, EventStatusSchema } from "./services/events"; // CONSTANTS // ----------------------- const EVENT_QUEUE_MANAGE_DELAY_MS = 1000 // 1 sec -const JWT_EXP = "7d"; +const JWT_EXP = "1d"; // SERVICES // ----------------------- diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 8a5abc6..ff44e1f 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -45,6 +45,7 @@ class ApparelOrdersService { types: { "apparel_order_create": defineEventType({ schemas: { payload: t.Object({ wOrder: t.Object({}) }), state: t.Null() }, + dedupeKey: (p) => (p.wOrder as Webflow.Orders.Order).orderId, processor: async (payload) => { const wOrder = payload.wOrder as Webflow.Orders.Order; @@ -76,6 +77,7 @@ class ApparelOrdersService { }), "apparel_order_fulfill": defineEventType({ schemas: { payload: t.Object({ wOrderId: t.String(), shipment: t.Object({}) }), state: t.Null() }, + dedupeKey: (p) => `${p.wOrderId}:${(p.shipment as Printful.Webhook.PackageShipped["data"]["shipment"]).tracking_number}`, processor: async (payload) => { const wOrderId = payload.wOrderId; const shipment = payload.shipment as Printful.Webhook.PackageShipped["data"]["shipment"]; @@ -131,6 +133,7 @@ class ApparelSyncService { t.Null() ]) }, + dedupeKey: (p) => (p.filter?.pProductIds ?? []).join(","), processor: async (payload, setState) => { await this.ProductSyncer.syncApparel({ filter: payload.filter, @@ -145,6 +148,7 @@ class ApparelSyncService { payload: t.Object({ wProductId: t.String() }), state: t.Union([t.Object({}), t.Null()]) }, + dedupeKey: (p) => p.wProductId, processor: async (payload) => { await this.Webflow.Products.remove(payload.wProductId); } diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index f90c287..9863f88 100644 --- a/apps/api/src/services/event-queue.ts +++ b/apps/api/src/services/event-queue.ts @@ -16,6 +16,7 @@ type EventProcessor

= export type EventTypeOptions

= { schemas: { payload: P, state: S }, processor: EventProcessor, + dedupeKey: (payload: Static

) => string | undefined }; type EventQueueCfg = { @@ -137,17 +138,33 @@ export class EventQueue(opt: { + async enqueue({ source, type, payload }: { source: EventSource, type: K, payload: Static - }): Promise { - log.info({ payload: opt.payload }, "enqueuing"); - const result = await db.insertInto("events") - .values({ group: this.group, type: opt.type, source: opt.source, payload: opt.payload as JsonValue }) + }): Promise { + if (!(type in this.types)) throw new Error(`event type not registered in this queue (${this.group}): ${type}`); + const cfg = this.types[type as keyof T]!; + + const dedupeKey = cfg.dedupeKey(payload); + log.info({ payload, dedupeKey }, "enqueuing"); + + await db.insertInto("events") + .values({ + group: this.group, + type, + source, + payload, + dedupe_key: dedupeKey ?? null, + }) + .onConflict((oc) => oc + .columns(["group", "type", "dedupe_key"]) + .where("dedupe_key", "is not", null) + .where("ended_at", "is", null) + .doNothing() + ) .returning(["id"]) - .executeTakeFirstOrThrow(); - return result.id; + .executeTakeFirst(); } async front(trx?: Transaction) { @@ -174,11 +191,11 @@ export class EventQueue, "id" | "type" | "errors" | "payload" | "rate_limits">) { log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event"); if (!(event.type in this.types)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`); - const cfg = this.types[event.type as keyof T]!; + const typeCfg = this.types[event.type as keyof T]!; try { - Value.Assert(cfg.schemas.payload, event.payload); - await cfg.processor(event.payload, (state) => this.Events.setState(event.id, state), event.id); + Value.Assert(typeCfg.schemas.payload, event.payload); + await typeCfg.processor(event.payload, (state) => this.Events.setState(event.id, state), event.id); await this.Events.resolve(event.id); } catch (err) {