diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 56a2c8c..00bee3c 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -21,8 +21,13 @@ import { CalculatorService, CalculatorUnavailableError } from "./services/calcul import { StandardsParamsSchema } from "@blade-and-brawn/calculator"; import { StandardsService } from "./services/standards"; +// CONSTANTS +// ----------------------- +const EVENT_QUEUE_MANAGE_DELAY_MS = 1000 // 1 sec +const EVENT_QUEUE_CLEAN_DELAY_MS = 600000 // 10 min + // SERVICES -// ----------- +// ----------------------- const s = (() => { const Standards = new StandardsService(); const Calculator = new CalculatorService(DEFAULT_NAME); @@ -31,7 +36,7 @@ const s = (() => { })(); // ELYSIA -// ----------- +// ----------------------- export const app = new Elysia() .use(serverTiming()) .use( @@ -196,9 +201,12 @@ export const app = new Elysia() // Run sync .post("/:pProductId?", async ({ params: { pProductId }, sessionId }) => { await s.Commerce.ProductSync.Queue.enqueue({ - session: { id: sessionId, name: "Portal" }, - filter: { - pProductIds: pProductId ? [pProductId] : null + source: "portal", + payload: { + session: { id: sessionId, name: "Portal" }, + filter: { + pProductIds: pProductId ? [pProductId] : undefined + } } }); }, { params: t.Object({ pProductId: t.Optional(t.Numeric()) }) }), @@ -229,8 +237,11 @@ export const app = new Elysia() log.info({ productId: pProduct.id }, "printful webhook: product updated"); await s.Commerce.ProductSync.Queue.enqueue({ - session: { id: randomUUIDv7(), name: "Printful" }, - filter: { pProductIds: [pProduct.id] } + source: "printful", + payload: { + session: { id: randomUUIDv7(), name: "Printful" }, + filter: { pProductIds: [pProduct.id] } + } }); break; } @@ -305,9 +316,17 @@ app.listen(3000, async () => { if (cluster.worker?.id === 1) { log.info({ port: 3000 }, "server started") + // MANAGE QUEUES + const queues = [s.Commerce.ProductSync.Queue]; while (true) { - await s.Commerce.ProductSync.Queue.drain().catch((err) => log.error(err)); - await sleep(1000); // every 1 second; + 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); } } }); diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index b238a49..c547f2d 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -1,28 +1,32 @@ -import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions } from "@blade-and-brawn/commerce"; +import { PrintfulClient, ProductSyncer, WebflowClient } from "@blade-and-brawn/commerce"; import { env, log } from "../util"; import { db } from "../database/db"; -import type { Selectable, Transaction } from "kysely"; -import type { DB } from "../database/out/db"; -import { sleep } from "bun"; +import { EventQueueService, type EventType } from "./event-queue"; +import { Type as t } from "@sinclair/typebox"; +import type { Static } from "@sinclair/typebox"; +import { Value } from "@sinclair/typebox/value"; +import { sql } from "kysely"; -type EventStatus = "queued" | "active" | "failed" | "fulfilled"; +const ProductSyncPayloadSchema = t.Object({ + session: t.Object({ + name: t.String(), + id: t.String() + }), + filter: t.Optional( + t.Object({ + pProductIds: t.Optional(t.Array(t.Number())) + }) + ) +}); +type ProductSyncPayload = Static; -export type ProductSyncPayload = { - session: { - name: string, - id: string - }, - filter: ProductSyncerOptions["filter"] -} - -export type ProductSyncState = { - syncingPProductIds: number[] -} - -type ProductSyncHandler = (id: string, payload: ProductSyncPayload, setState: (state: ProductSyncState) => Promise) => Promise - -class AlreadyClaimedEventError extends Error { } -class ActiveEventLimitReachedError extends Error { } +const ProductSyncStateSchema = t.Union([ + t.Object({ + pProductIds: t.Optional(t.Array(t.Number())) + }), + t.Null() +]); +type ProductSyncState = Static; export class CommerceService { public readonly Printful = new PrintfulClient({ @@ -39,188 +43,49 @@ export class CommerceService { } class ProductSyncService { - // TODO: make service have their own logger, so we can enforce .child({component: }) - readonly Queue: ProductSyncQueueService; + static readonly EVENT_TYPE: EventType = "product_sync"; + + readonly Queue: EventQueueService; private readonly ProductSyncer: ProductSyncer; + // TODO: make each service have their own logger, so we can enforce .child({component: }) + constructor(private readonly Printful: PrintfulClient, private readonly Webflow: WebflowClient) { this.ProductSyncer = new ProductSyncer(this.Printful, this.Webflow, log); - this.Queue = new ProductSyncQueueService(async (_id, payload, setState) => { - await this.ProductSyncer.run({ - filter: payload.filter, - beforeStep: async (syncingPProductIds: number[]) => { - await setState({ syncingPProductIds }); - } - }); + this.Queue = new EventQueueService({ + type: ProductSyncService.EVENT_TYPE, + schemas: { payload: ProductSyncPayloadSchema, state: ProductSyncStateSchema }, + cfg: { + retry: { limit: 3, delayMs: 10000 }, + concurrency: { limit: 1, timeoutMs: 600000 } + }, + handler: async (id, payload, setState) => { + await this.ProductSyncer.run({ + filter: payload.filter, + beforeStep: async (pProductIds: number[]) => { + await setState({ pProductIds }); + } + }); + } }); } async getSessionSyncState(sessionId: string) { - const latestSync = await db.selectFrom("product_syncs") - .selectAll() - .where("session_id", "=", sessionId) + 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("created_at", "desc") .limit(1) .executeTakeFirst(); if (!latestSync) return; + Value.Assert(ProductSyncStateSchema, latestSync.state); + return { startDate: latestSync.started_at, - status: ProductSyncQueueService.status(latestSync), - syncingPProductIds: latestSync.syncing_p_product_ids - } - } -} - -class ProductSyncQueueService { - 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(): Promise { - log.trace("draining queue"); - while (true) { - try { - const nextActiveEvent = await this.dequeue(); - if (!nextActiveEvent) 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 dequeue() { - log.trace("attempting to dequeue"); - 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; - - 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) { - log.trace({ name: eventToHandle.id }, "dequeuing event"); - 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 { - log.info({ payload }, "enqueuing"); - 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(); - return result; - } - - 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) { - log.info({ id, payload }, "handling event"); - try { - 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; - } - - await this.fulfill(id); + status: EventQueueService.status(latestSync), + syncingPProductIds: latestSync.state?.pProductIds ?? [] + }; } } diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index 39f0c2f..0d19def 100644 --- a/apps/api/src/services/event-queue.ts +++ b/apps/api/src/services/event-queue.ts @@ -6,8 +6,8 @@ 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"; +export type EventType = "product_sync" | "product_order_created" /* TODO: more events... */; +export type EventSource = "portal" | "printful" | "webflow"; type EventHandler = (id: string, payload: Payload, setState: (state: State) => Promise) => Promise; @@ -16,27 +16,33 @@ type EventStatus = "queued" | "active" | "failed" | "fulfilled"; class AlreadyClaimedEventError extends Error { } class ConcurrencyLimitReachedError extends Error { } -type EventQueueServiceOptions = { - retry: { - limit: number, - delayMs: number, +type EventQueueServiceOptions = { + type: EventType, + schemas: { + payload: TSchema & { static: Payload }, + state: TSchema & { static: State }, }, - concurrency: { - limit: number, - timeoutMs: number, - } + cfg: { + retry: { + limit: number, + delayMs: number, + }, + concurrency: { + limit: number, + timeoutMs: number, + }, + }, + handler: EventHandler, } export class EventQueueService { - constructor( - readonly type: EventType, - private payloadSchema: TSchema & { static: Payload }, - private stateSchema: TSchema & { static: State }, - private handler: EventHandler, - readonly opt: EventQueueServiceOptions - ) { } + private _lastCleanDate: Date = new Date(0); - static status(event: Selectable): EventStatus { + constructor(private opt: EventQueueServiceOptions) { } + + 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"; return event.has_failed ? "failed" : "fulfilled"; @@ -57,17 +63,17 @@ export class EventQueueService { + async enqueue({ source, payload }: { source: EventSource, payload: Payload }): Promise { log.info({ payload }, "enqueuing"); const result = await db.insertInto("events") - .values({ type: this.type, source, payload }) + .values({ type: this.opt.type, source, payload }) .returning(["id"]) .executeTakeFirstOrThrow(); return result.id; @@ -140,18 +146,20 @@ export class EventQueueService this.opt.concurrency.timeoutMs); + const activeEventTimedOut = (Date.now() - (activeEvent.started_at?.getTime() ?? 0) > this.opt.cfg.concurrency.timeoutMs); if (activeEventTimedOut) { - log.info({ name: activeEvent.id }, "active event has timed out. failing it and continuing..."); + log.warn({ name: activeEvent.id }, "active event has timed out. failing it and continuing..."); await this.fail(activeEvent.id); } } + + this._lastCleanDate = new Date(); } async front(trx?: Transaction) { const result = await (trx ?? db).selectFrom("events") .selectAll() - .where("type", "=", this.type) + .where("type", "=", this.opt.type) .where("started_at", "is", null) .orderBy("created_at", "asc") .limit(1) @@ -162,7 +170,7 @@ export class EventQueueService) { return await (trx ?? db).selectFrom("events") .select(["started_at", "id"]) - .where("type", "=", this.type) + .where("type", "=", this.opt.type) .where("started_at", "is not", null) .where("ended_at", "is", null) .execute(); @@ -171,8 +179,8 @@ export class EventQueueService { + Value.Assert(this.opt.schemas.payload, payload); + await this.opt.handler(id, payload, async (state) => { await db.updateTable("events") .set({ state }) .where("id", "=", id) diff --git a/packages/commerce/src/sync.ts b/packages/commerce/src/sync.ts index ba391c6..17def48 100644 --- a/packages/commerce/src/sync.ts +++ b/packages/commerce/src/sync.ts @@ -19,7 +19,7 @@ type PMetaProduct = { export type ProductSyncerOptions = { filter?: { - pProductIds: number[] | null; + pProductIds?: number[]; }, beforeStart?: (startDate: Date) => Promise afterCompletion?: (completionDate: Date, duration: number) => Promise