diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 00bee3c..6b319f7 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -20,11 +20,11 @@ import { randomUUIDv7, sleep } from "bun"; import { CalculatorService, CalculatorUnavailableError } from "./services/calculator"; import { StandardsParamsSchema } from "@blade-and-brawn/calculator"; import { StandardsService } from "./services/standards"; +import type { EventQueueService } from "./services/event-queue"; // CONSTANTS // ----------------------- const EVENT_QUEUE_MANAGE_DELAY_MS = 1000 // 1 sec -const EVENT_QUEUE_CLEAN_DELAY_MS = 600000 // 10 min // SERVICES // ----------------------- @@ -194,9 +194,9 @@ export const app = new Elysia() .group("/sync", (app) => app // Sync status .get("/", async ({ sessionId }) => { - const syncState = await s.Commerce.ProductSync.getSessionSyncState(sessionId); - if (!syncState) throw new NotFoundError("No product sync found for the provided session"); - return syncState; + const latestSyncState = await s.Commerce.ProductSync.getLatestSyncState(sessionId); + if (!latestSyncState) throw new NotFoundError("No product sync found for the provided session"); + return latestSyncState; }) // Run sync .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => { @@ -317,16 +317,18 @@ app.listen(3000, async () => { log.info({ port: 3000 }, "server started") // MANAGE QUEUES - const queues = [s.Commerce.ProductSync.Queue]; - while (true) { - for (const queue of queues) { - // clean, if ready - if ((Date.now() - queue.lastCleanDate.getTime()) >= EVENT_QUEUE_CLEAN_DELAY_MS) - await queue.clean(); - // drain - await queue.drain().catch((err) => log.error(err)); - } - await sleep(EVENT_QUEUE_MANAGE_DELAY_MS); + const queues: EventQueueService[] = [s.Commerce.ProductSync.Queue]; + for (const queue of queues) { + (async () => { + while (true) { + // clean, if ready + if ((Date.now() - queue.lastCleanDate.getTime()) >= queue.cfg.concurrency.timeoutMs) + await queue.clean().catch((err) => log.error(err)); + // drain + await queue.drain().catch((err) => log.error(err)); + await sleep(EVENT_QUEUE_MANAGE_DELAY_MS); + } + })(); } } }); diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index c547f2d..a194a86 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -59,7 +59,7 @@ class ProductSyncService { retry: { limit: 3, delayMs: 10000 }, concurrency: { limit: 1, timeoutMs: 600000 } }, - handler: async (id, payload, setState) => { + processor: async (id, payload, setState) => { await this.ProductSyncer.run({ filter: payload.filter, beforeStep: async (pProductIds: number[]) => { @@ -70,11 +70,12 @@ class ProductSyncService { }); } - async getSessionSyncState(sessionId: string) { + async getLatestSyncState(sessionId: string) { const latestSync = await db.selectFrom("events") .select(["started_at", "ended_at", "has_failed", "state"]) .where("type", "=", ProductSyncService.EVENT_TYPE) .where(sql`payload->'session'->>'id'`, "=", sessionId) + .orderBy(sql`(started_at is not null and ended_at is null)`, "desc") .orderBy("created_at", "desc") .limit(1) .executeTakeFirst(); diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index 0d19def..f45759f 100644 --- a/apps/api/src/services/event-queue.ts +++ b/apps/api/src/services/event-queue.ts @@ -9,9 +9,9 @@ import type { TSchema } from "@sinclair/typebox"; export type EventType = "product_sync" | "product_order_created" /* TODO: more events... */; export type EventSource = "portal" | "printful" | "webflow"; -type EventHandler = +type EventProcessor = (id: string, payload: Payload, setState: (state: State) => Promise) => Promise; -type EventStatus = "queued" | "active" | "failed" | "fulfilled"; +type EventStatus = "pending" | "processing" | "failed" | "fulfilled"; class AlreadyClaimedEventError extends Error { } class ConcurrencyLimitReachedError extends Error { } @@ -32,19 +32,22 @@ type EventQueueServiceOptions, + processor: EventProcessor, } export class EventQueueService { private _lastCleanDate: Date = new Date(0); + readonly cfg: EventQueueServiceOptions["cfg"]; - constructor(private opt: EventQueueServiceOptions) { } + constructor(private readonly opt: EventQueueServiceOptions) { + this.cfg = opt.cfg; + } get lastCleanDate() { return this._lastCleanDate; }; static status(event: Pick, "started_at" | "ended_at" | "has_failed">): EventStatus { - if (!event.started_at) return "queued"; - if (!event.ended_at) return "active"; + if (!event.started_at) return "pending"; + if (!event.ended_at) return "processing"; return event.has_failed ? "failed" : "fulfilled"; } @@ -143,17 +146,20 @@ export class EventQueueService this.opt.cfg.concurrency.timeoutMs); - if (activeEventTimedOut) { - log.warn({ name: activeEvent.id }, "active event has timed out. failing it and continuing..."); - await this.fail(activeEvent.id); + try { + // Find and fail any timed out processing events + const processingEvents = await this.processingEvents(); + for (const processingEvent of processingEvents) { + const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > this.opt.cfg.concurrency.timeoutMs); + if (processingEventTimedOut) { + log.warn({ name: processingEvent.id }, "processing event has timed out. failing it and continuing..."); + await this.fail(processingEvent.id); + } } } - - this._lastCleanDate = new Date(); + finally { + this._lastCleanDate = new Date(); + } } async front(trx?: Transaction) { @@ -167,7 +173,7 @@ export class EventQueueService) { + async processingEvents(trx?: Transaction) { return await (trx ?? db).selectFrom("events") .select(["started_at", "id"]) .where("type", "=", this.opt.type) @@ -180,7 +186,7 @@ export class EventQueueService { + await this.opt.processor(id, payload, async (state) => { await db.updateTable("events") .set({ state }) .where("id", "=", id) diff --git a/apps/portal/src/routes/(app)/commerce/products/+page.svelte b/apps/portal/src/routes/(app)/commerce/products/+page.svelte index 9d72923..562bc71 100644 --- a/apps/portal/src/routes/(app)/commerce/products/+page.svelte +++ b/apps/portal/src/routes/(app)/commerce/products/+page.svelte @@ -3,15 +3,15 @@ import { onMount } from "svelte"; import { api } from "$lib/api.js"; - type SyncStatus = "queued" | "active" | "failed" | "fulfilled" | "none"; + type SyncStatus = "pending" | "processing" | "failed" | "fulfilled"; const { data } = $props(); const synchronizer = $state({ - status: "none" as SyncStatus, + status: "none" as SyncStatus | "none", syncingPProductIds: [] as number[], get isInProgress() { - return this.status === "active" || this.status === "queued"; + return this.status === "processing" || this.status === "pending"; }, poll: async function () { const res = await api.commerce.products.sync.get(); @@ -118,7 +118,7 @@