diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts index 72ce880..9f5701e 100644 --- a/apps/api/src/database/migrations/2026-06-03.ts +++ b/apps/api/src/database/migrations/2026-06-03.ts @@ -17,16 +17,6 @@ export async function up(db: Kysely): Promise { .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false)) .execute() - // INDEX: ONE_ACTIVE_PRODUCT_SYNC - await db.schema.createIndex("idx_one_active_product_sync") - .on("product_syncs") - .column("ended_at") - .unique() - .where(sql.ref("started_at"), "is not", null) - .where("ended_at", "is", null) - .nullsNotDistinct() - .execute(); - // TABLE: STANDARDS_DATASETS await db.schema.createTable("standards_datasets") .$call(addDefaultColumns) @@ -68,9 +58,6 @@ export async function down(db: Kysely): Promise { // TABLE: PRODUCT_SYNCS await db.schema.dropTable("product_syncs").ifExists().execute() - // INDEX: ONE_ACTIVE_PRODUCT_SYNC - await db.schema.dropIndex("idx_one_active_product_sync").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 d9c100c..fe8d5c3 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -189,21 +189,20 @@ export const app = new Elysia() .group("/sync", (app) => app // Sync status .get("/", async ({ sessionId }) => { - const latestSync = await s.Commerce.Sync.getLatest(sessionId); - if (!latestSync) throw new NotFoundError("No sync found"); - return { - startDate: latestSync.started_at, - status: s.Commerce.Sync.deriveStatus(latestSync), - syncingPProductIds: latestSync.syncing_p_product_ids - } + const syncState = await s.Commerce.ProductSync.getSessionSyncState(sessionId); + if (!syncState) throw new NotFoundError("No product sync found for the provided session"); + return syncState; }) // Run sync .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => { - await s.Commerce.Sync.start({ + await s.Commerce.ProductSync.Queue.enqueue({ session: { id: sessionId, name: "Portal" }, filter: { pProductIds: pProductId ? [pProductId] : null } + }).then(async (id) => { + await s.Commerce.ProductSync.Queue.drain({ untilId: id }); + void s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err })); }); }, { params: t.Object({ pProductId: t.Optional(t.Numeric()) }) }), ) @@ -232,11 +231,17 @@ export const app = new Elysia() const pProduct = payload.data.sync_product; log.info({ productId: pProduct.id }, "printful webhook: product updated"); - await s.Commerce.Sync.start({ + // Controller layer handles draining the queue, potentionally across multiple workers + // TODO: maybe find better way of handling queue draining + // can cause a worker to hang for long time while draining + // this can cause an issue with webhook sender timing out and sending duplicate payload + // while we wait to process unrelated events + await s.Commerce.ProductSync.Queue.enqueue({ session: { id: randomUUIDv7(), name: "Printful" }, - filter: { - pProductIds: [pProduct.id] - } + filter: { pProductIds: [pProduct.id] } + }).then(async (id) => { + await s.Commerce.ProductSync.Queue.drain({ untilId: id }); + void s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err })); }); break; } @@ -311,13 +316,7 @@ app.listen(3000, async () => { if (cluster.worker?.id === 1) { log.info({ port: 3000 }, "server started") - try { - const frontQueuedSync = await s.Commerce.Sync.Queue.front(); - if (frontQueuedSync) await s.Commerce.Sync.start(frontQueuedSync); - } - catch (err) { - log.error({ err }, "failed to resume product sync queue") - } + await s.Commerce.ProductSync.Queue.drain().catch((err) => log.error({ err })); } }); diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index 2109c72..27744a4 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -1,14 +1,13 @@ import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions } from "@blade-and-brawn/commerce"; import { env, log } from "../util"; import { db } from "../database/db"; -import { DatabaseError } from "pg"; -import type { Insertable, Selectable } from "kysely"; -import type { ProductSyncs } from "../database/out/db"; +import type { Selectable, Transaction } from "kysely"; +import type { DB, ProductSyncs } from "../database/out/db"; +import { sleep } from "bun"; -type SyncStatus = "queued" | "active" | "failed" | "fulfilled"; +type EventStatus = "queued" | "active" | "failed" | "fulfilled"; -type ProductSyncPayload = { - id?: string, +export type ProductSyncPayload = { session: { name: string, id: string @@ -16,7 +15,18 @@ type ProductSyncPayload = { filter: ProductSyncerOptions["filter"] } -class AlreadyClaimedError extends Error { } +export type ProductSyncState = { + syncingPProductIds: number[] +} + +type QueueDrainOptions = { + untilId?: string +} + +type ProductSyncHandler = (id: string, payload: ProductSyncPayload, setState: (state: ProductSyncState) => Promise) => Promise + +class AlreadyClaimedEventError extends Error { } +class ActiveEventLimitReachedError extends Error { } export class CommerceService { public readonly Printful = new PrintfulClient({ @@ -29,173 +39,192 @@ export class CommerceService { token: env.WEBFLOW_AUTH, webhookSecret: env.WEBFLOW_WEBHOOK_SECRET, }); - readonly Sync = new ProductSyncService(this.Printful, this.Webflow); + readonly ProductSync = new ProductSyncService(this.Printful, this.Webflow); } class ProductSyncService { - readonly Queue = new ProductSyncQueueService(); + // TODO: make service have their own logger, so we can enforce .child({component: }) + readonly Queue: ProductSyncQueueService; private readonly ProductSyncer: ProductSyncer; constructor(private readonly Printful: PrintfulClient, private readonly Webflow: WebflowClient) { this.ProductSyncer = new ProductSyncer(this.Printful, this.Webflow, log); - } - - async start(payload: ProductSyncPayload) { - let id = payload.id ?? null; - let freedSlot = false; - try { - log.info({ session: payload.session }, "starting sync run"); + this.Queue = new ProductSyncQueueService(async (_id, payload, setState) => { await this.ProductSyncer.run({ filter: payload.filter, - beforeStart: async (startDate) => { - await this.Queue.enqueue(payload, startDate); - }, - beforeStep: async (pProductIds: number[]) => { - if (!id) throw new Error("Expected sync run ID"); - await db.updateTable("product_syncs") - .set({ syncing_p_product_ids: pProductIds }) - .where("id", "=", id) - .execute(); - }, - afterCompletion: async (endDate, duration) => { - if (!id) throw new Error("Expected sync run ID"); - await this.Queue.fulfill(id, endDate) - freedSlot = true; + beforeStep: async (syncingPProductIds: number[]) => { + await setState({ syncingPProductIds }); } }); - } - catch (err) { - if (err instanceof AlreadyClaimedError) { - log.info("sync already claimed by another worker, skipping"); - return; - } - - const activeSync = await this.getActive(); - - const activeSyncConflict = activeSync && err instanceof DatabaseError && err.code === "23505"; - if (activeSyncConflict) { - // If active sync has been syncing for 10 min (600000ms), clear it and retry this sync - if (Date.now() - (activeSync.started_at?.getTime() ?? 0) > 600000) { - log.info("already active sync exceeded timeout"); - await this.Queue.fail(activeSync.id, new Date()); - await this.start(payload); - } - else { - log.info("sync already active, enqueuing next sync"); - await this.Queue.enqueue(payload); - } - } - else { - if (id) { - await this.Queue.fail(id, new Date()); - freedSlot = true; - } - throw err; - } - } - finally { - if (freedSlot) { - try { - const frontQueuedSync = await this.Queue.front(); - if (frontQueuedSync) await this.start(frontQueuedSync); - } - catch (err) { - log.error({ err }, "failed to process next queued sync") - } - } - } + }); } - async getActive(sessionId?: string) { - return await db.selectFrom("product_syncs") + async getSessionSyncState(sessionId: string) { + const latestSync = await db.selectFrom("product_syncs") .selectAll() - .$if(sessionId !== undefined, (qb) => - qb.where("session_id", "=", sessionId!) - ) - .where("started_at", "is not", null) - .where("ended_at", "is", null) - .limit(1) - .executeTakeFirst(); - } - - async getLatest(sessionId?: string) { - return await db.selectFrom("product_syncs") - .selectAll() - .$if(sessionId !== undefined, (qb) => - qb.where("session_id", "=", sessionId!) - ) + .where("session_id", "=", sessionId) .orderBy("created_at", "desc") .limit(1) .executeTakeFirst(); - } + if (!latestSync) return; - deriveStatus(sync: Selectable): SyncStatus { - if (!sync.started_at) return "queued"; - if (!sync.ended_at) return "active"; - return sync.has_failed ? "failed" : "fulfilled"; + return { + startDate: latestSync.started_at, + status: ProductSyncQueueService.status(latestSync), + syncingPProductIds: latestSync.syncing_p_product_ids + } } } class ProductSyncQueueService { - async enqueue(payload: ProductSyncPayload, startDate?: Date): Promise { - const values: Insertable = { - session_id: payload.session.id, - session_name: payload.session.name, - p_product_id_filter: payload.filter?.pProductIds ?? null - }; - if (startDate) values.started_at = startDate; - if (payload.id) { - const result = await db.updateTable("product_syncs") - .set(values) - .where("id", "=", payload.id) - .where("started_at", "is", null) - .executeTakeFirstOrThrow(); - // Guard against another worker/machine having already claimed - // this queued row between `queue.front()` and now. - if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError(); - return payload.id; - } - else { - const result = await db.insertInto("product_syncs") - .values(values) - .returning("id") - .executeTakeFirstOrThrow(); - return result.id; + private static DRAIN_SLEEP_TIME = 10000; // 10 seconds + private static EVENT_TIMEOUT = 600000; // 10 min + private static MAX_ACTIVE_EVENTS = 1; + + constructor(private handler: ProductSyncHandler) { } + + 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(opt: QueueDrainOptions = {}): Promise { + log.info("draining queue"); + while (true) { + try { + const dequeued = await this.dequeue(); + + if (!dequeued) break; + if (dequeued.id === opt.untilId) break; + } + catch (err) { + if (err instanceof AlreadyClaimedEventError) { + log.warn("event handling already claimed by another worker, skipping"); + continue; + } + if (err instanceof ActiveEventLimitReachedError) { + await sleep(ProductSyncQueueService.DRAIN_SLEEP_TIME); + continue; + } + else { + log.error("failed to drain queue"); + throw err; + }; + } } } - async front(): Promise { - const result = await db.selectFrom("product_syncs") + async dequeue() { + log.info("dequeuing"); + const eventToHandle = await db.transaction().execute(async (trx) => { + const activeEvents = await this.activeEvents(trx); + let activeEventCount = activeEvents.length; + for (const activeEvent of activeEvents) { + // If event has been active past max active time, fail it + const activeEventTimedOut = (Date.now() - (activeEvent.started_at?.getTime() ?? 0) > ProductSyncQueueService.EVENT_TIMEOUT); + if (activeEventTimedOut) { + log.info({ name: activeEvent.id }, "active event has timed out. failing it and continuing..."); + await this.fail(activeEvent.id, trx); + --activeEventCount; + } + } + + if (activeEventCount >= ProductSyncQueueService.MAX_ACTIVE_EVENTS) + throw new ActiveEventLimitReachedError(); + + const front = await this.front(trx); + if (!front) return log.info("nothing to dequeue"); + + const result = await trx.updateTable("product_syncs") + .set({ started_at: new Date() }) + .where("id", "=", front.id) + .where("started_at", "is", null) + .executeTakeFirstOrThrow(); + + // Guard against another worker/machine having already claimed this event + if (result.numUpdatedRows === 0n) throw new AlreadyClaimedEventError(); + + return front; + }); + + if (eventToHandle) { + await this.handleEvent(eventToHandle.id, { + session: { + id: eventToHandle.session_id, + name: eventToHandle.session_name + }, + filter: { pProductIds: eventToHandle.p_product_id_filter } + }); + return eventToHandle; + } + } + + async enqueue(payload: ProductSyncPayload): Promise { + const result = await db.insertInto("product_syncs") + .values({ + session_id: payload.session.id, + session_name: payload.session.name, + p_product_id_filter: payload.filter?.pProductIds ?? null, + started_at: null, + }) + .returning(["id"]) + .executeTakeFirstOrThrow(); + return result.id; + } + + async fail(id: string, trx?: Transaction): Promise { + await (trx ?? db).updateTable("product_syncs") + .set({ syncing_p_product_ids: [], ended_at: new Date(), has_failed: true }) + .where("id", "=", id) + .execute(); + } + + async fulfill(id: string, trx?: Transaction): Promise { + await (trx ?? db).updateTable("product_syncs") + .set({ syncing_p_product_ids: [], ended_at: new Date() }) + .where("id", "=", id) + .execute(); + } + + async clean() { + // TODO + } + + async front(trx?: Transaction) { + const result = await (trx ?? db).selectFrom("product_syncs") .selectAll() .where("started_at", "is", null) .orderBy("created_at", "asc") .limit(1) .executeTakeFirst(); - if (!result) return; + return result; + } - return { - id: result.id, - session: { - id: result.session_id, - name: result.session_name - }, - filter: { - pProductIds: result.p_product_id_filter - } + async activeEvents(trx?: Transaction) { + return await (trx ?? db).selectFrom("product_syncs") + .select(["started_at", "id"]) + .where("started_at", "is not", null) + .where("ended_at", "is", null) + .execute(); + } + + private async handleEvent(id: string, payload: ProductSyncPayload) { + try { + log.info({ id, payload }, "handling event"); + await this.handler(id, payload, async (state) => { + // set event state + await db.updateTable("product_syncs") + .set({ syncing_p_product_ids: state.syncingPProductIds }) + .where("id", "=", id) + .execute(); + }); + } + catch (err) { + await this.fail(id); + throw err; } - } - async fail(id: string, endDate: Date) { - await db.updateTable("product_syncs") - .set({ syncing_p_product_ids: [], ended_at: endDate, has_failed: true }) - .where("id", "=", id) - .execute(); - } - - async fulfill(id: string, endDate: Date) { - await db.updateTable("product_syncs") - .set({ syncing_p_product_ids: [], ended_at: endDate }) - .where("id", "=", id) - .execute(); + await this.fulfill(id); } }