diff --git a/apps/api/src/database/migrations/2026-06-03.ts b/apps/api/src/database/migrations/2026-06-03.ts index 3c7b131..618d62e 100644 --- a/apps/api/src/database/migrations/2026-06-03.ts +++ b/apps/api/src/database/migrations/2026-06-03.ts @@ -16,6 +16,7 @@ export async function up(db: Kysely): Promise { .addColumn("errors", "integer", (cb) => cb.notNull().defaultTo(0)) .addColumn("rate_limits", "integer", (cb) => cb.notNull().defaultTo(0)) .addColumn("has_failed", "boolean", (cb) => cb.notNull().defaultTo(false)) + .addColumn("last_error", "jsonb") .execute() // TABLE: STANDARDS_DATASETS diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index 255fde9..233a609 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -19,6 +19,7 @@ import { randomUUIDv7, sleep } from "bun"; import { CalculatorService, CalculatorUnavailableError } from "./services/calculator"; import { StandardsParamsSchema } from "@blade-and-brawn/calculator"; import { StandardsService } from "./services/standards"; +import { EventsService, EventStatusSchema } from "./services/events"; // CONSTANTS // ----------------------- @@ -30,9 +31,17 @@ const s = (() => { const Standards = new StandardsService(); const Calculator = new CalculatorService(DEFAULT_NAME); const Commerce = new CommerceService(); - return { Standards, Calculator, Commerce }; + const Events = new EventsService(); + return { Standards, Calculator, Commerce, Events }; })(); +// QUEUES +// ----------------------- +const queues = [ + s.Commerce.Apparel.Syncs.Queue, + s.Commerce.Apparel.Orders.Queue, +] as const; + // PLUGINS // ----------------------- const authPlugin = new Elysia({ name: "auth" }) @@ -224,8 +233,38 @@ export const app = new Elysia() return { pProduct, wProduct }; }, { params: t.Object({ pProductId: t.Numeric() }) + }))) + + // EVENTS + .group("/events", { auth: true }, (app) => app + .get("/", async ({ query }) => { + const events = await s.Events.list({ + filter: { status: query.status, type: query.type, group: query.group }, + limit: query.limit, + offset: query.offset, + }); + return events.map((event) => ({ ...event, status: EventsService.status(event) })); + }, { + query: t.Object({ + status: t.Optional(EventStatusSchema), + type: t.Optional(t.String()), + group: t.Optional(t.String()), + limit: t.Optional(t.Numeric()), + offset: t.Optional(t.Numeric()), }) - )) + }) + .get("/:id", async ({ params: { id } }) => { + const event = await s.Events.get(id); + if (!event) throw new NotFoundError("Event not found"); + return { ...event, status: EventsService.status(event) }; + }, { + params: t.Object({ id: t.String() }) + }) + .post("/:id/replay", async ({ params: { id } }) => { + const replayed = await s.Events.replay(id); + if (!replayed) throw new NotFoundError("Event not found or not in a failed state"); + }, { params: t.Object({ id: t.String() }) }) + ) // WEBHOOKS .post("/webhooks/printful", async ({ body }) => { @@ -307,13 +346,12 @@ app.listen(3000, async () => { log.info({ port: 3000 }, "server started") // MANAGE QUEUES - const queues = [s.Commerce.Apparel.Syncs.Queue, s.Commerce.Apparel.Orders.Queue]; for (const queue of queues) { (async () => { while (true) { try { // clean, if ready - if ((Date.now() - queue.lastCleanDate.getTime()) >= queue.cfg.concurrency.timeoutMs) + if ((Date.now() - queue.lastCleanDate.getTime()) >= EventsService.CONCURRENCY_TIMEOUT_MS) await queue.clean().catch((err) => log.error({ name: queue.group, err })); // drain await queue.drain(); diff --git a/apps/api/src/services/calculator.ts b/apps/api/src/services/calculator.ts index d2cdbec..7d31db4 100644 --- a/apps/api/src/services/calculator.ts +++ b/apps/api/src/services/calculator.ts @@ -1,4 +1,4 @@ -import { LevelCalculator, Standards, type StandardsParams, StandardsParamsSchema, StandardsDataSchema, type StandardsData, type StandardsConfig, type LevelCalculatorOutput } from "@blade-and-brawn/calculator"; +import { LevelCalculator, Standards, StandardsParamsSchema, StandardsDataSchema, type StandardsConfig, type LevelCalculatorOutput } from "@blade-and-brawn/calculator"; import type { ActivityPerformance, Player } from "@blade-and-brawn/domain"; import { log } from "../util"; import { db } from "../database/db"; diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index f3b1a6c..8a5abc6 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -6,7 +6,8 @@ import { Type as t } from "@sinclair/typebox"; import { Value } from "@sinclair/typebox/value"; import { sql } from "kysely"; import zipcodesUs from "zipcodes-us"; -import { minToMs, secToMs } from "@blade-and-brawn/domain"; +import { secToMs } from "@blade-and-brawn/domain"; +import { EventsService } from "./events"; export class CommerceService { readonly Printful = new PrintfulClient({ @@ -39,9 +40,9 @@ class ApparelOrdersService { group: "apparel_order", cfg: { error: { limit: 3, initialRetryDelayMs: secToMs(1) }, - concurrency: { limit: 1, timeoutMs: minToMs(10) } + concurrency: { limit: 1 } }, - events: { + types: { "apparel_order_create": defineEventType({ schemas: { payload: t.Object({ wOrder: t.Object({}) }), state: t.Null() }, processor: async (payload) => { @@ -107,9 +108,9 @@ class ApparelSyncService { group: "apparel_sync", cfg: { error: { limit: 3, initialRetryDelayMs: secToMs(1) }, - concurrency: { limit: 1, timeoutMs: minToMs(10) } + concurrency: { limit: 1 } }, - events: { + types: { "apparel_sync_update": defineEventType({ schemas: { payload: t.Object({ @@ -164,11 +165,11 @@ class ApparelSyncService { .executeTakeFirst(); if (!latestSync) return; - Value.Assert(this.Queue.events["apparel_sync_update"].schemas.state, latestSync.state); + Value.Assert(this.Queue.types["apparel_sync_update"].schemas.state, latestSync.state); return { startDate: latestSync.started_at, - status: EventQueue.status(latestSync), + status: EventsService.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 3e5a132..6cc440f 100644 --- a/apps/api/src/services/event-queue.ts +++ b/apps/api/src/services/event-queue.ts @@ -1,14 +1,12 @@ -import { sql, type Selectable, type Transaction } from "kysely"; +import { type Selectable, type 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 { AssertError, Value } from "@sinclair/typebox/value"; import type { TSchema, Static } from "@sinclair/typebox"; -import { minToMs, RateLimitError, secToMs } from "@blade-and-brawn/domain"; - -export type EventSource = "portal" | "printful" | "webflow"; -export type EventStatus = "waiting" | "available" | "processing" | "failed" | "fulfilled"; +import { RateLimitError, secToMs } from "@blade-and-brawn/domain"; +import { EventsService, type EventSource } from "./events"; type RetryType = "error" | "ratelimit"; @@ -27,7 +25,6 @@ type EventQueueCfg = { }, concurrency: { limit: number, - timeoutMs: number, }, }; @@ -47,32 +44,33 @@ function exponentialDelay(initialDelay: number, tries: number, cap: number = Inf return Math.min(initialDelay * (2 ** tries), cap); } +function serializeError(err: unknown): JsonValue { + if (err instanceof Error) { + const payload = "payload" in err ? { payload: err.payload as JsonValue } : {}; + return { name: err.name, message: err.message, ...payload }; + } + return String(err); +} + export class EventQueue>> { private static readonly DEQUEUE_RETRY_DELAY_MS = secToMs(10); private static readonly DEQUEUE_RETRY_LIMIT = 3; private static readonly RATE_LIMIT_RETRY_LIMIT = 15; private static readonly JITTER_FACTOR = 0.2; - private _lastCleanDate: Date = new Date(0); readonly cfg: EventQueueCfg; readonly group: G; - readonly events: T; + readonly types: T; - constructor(opt: { group: G, cfg: EventQueueCfg, events: T }) { - this.cfg = opt.cfg; - this.group = opt.group; - this.events = opt.events; - } + private readonly Events = new EventsService(); + private _lastCleanDate: Date = new Date(0); get lastCleanDate() { return this._lastCleanDate; }; - static status(event: Pick, "available_at" | "started_at" | "ended_at" | "has_failed">): EventStatus { - if (!event.started_at) { - if (Date.now() < event.available_at.getTime()) return "waiting"; - return "available"; - }; - if (!event.ended_at) return "processing"; - return event.has_failed ? "failed" : "fulfilled"; + constructor(opt: { group: G, cfg: EventQueueCfg, types: T }) { + this.cfg = opt.cfg; + this.group = opt.group; + this.types = opt.types; } async drain(): Promise { @@ -160,37 +158,6 @@ export class EventQueue): 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() { - try { - // Find and fail any timed out processing events - const processingEvents = await this.list({ filter: { status: "processing" } }); - for (const processingEvent of processingEvents) { - const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > this.cfg.concurrency.timeoutMs); - if (processingEventTimedOut) { - log.warn({ name: processingEvent.id }, "processing event has timed out. failing it and continuing..."); - await this.fail(processingEvent.id); - } - } - } - finally { - this._lastCleanDate = new Date(); - } - } - async front(trx?: Transaction) { const result = await (trx ?? db).selectFrom("events") .select(["id", "type", "payload", "available_at", "errors", "rate_limits"]) @@ -203,48 +170,28 @@ export class EventQueue) { - return await (trx ?? db).selectFrom("events") - .select(["started_at", "id"]) - .where("group", "=", this.group) - .$if(opt.filter?.status === "waiting", (qb) => qb - .where("started_at", "is", null) - .where("available_at", ">", new Date()) - ) - .$if(opt.filter?.status === "available", (qb) => qb - .where("started_at", "is", null) - .where("available_at", "<=", new Date()) - ) - .$if(opt.filter?.status === "processing", (qb) => qb - .where("started_at", "is not", null) - .where("ended_at", "is", null) - ) - .$if(opt.filter?.status === "failed", (qb) => qb - .where("started_at", "is not", null) - .where("ended_at", "is not", null) - .where("has_failed", "is", true) - ) - .$if(opt.filter?.status === "fulfilled", (qb) => qb - .where("started_at", "is not", null) - .where("ended_at", "is not", null) - .where("has_failed", "is", false) - ) - .orderBy(sql`(started_at is not null and ended_at is null)`, "desc") - .orderBy("created_at", "desc") - .execute(); + async clean(opt: { filter?: { type?: string } } = {}) { + try { + await this.Events.clean({ filter: { group: this.group, ...opt.filter } }); + } + finally { + this._lastCleanDate = new Date(); + } } private async processEvent(event: Pick, "id" | "type" | "errors" | "payload" | "rate_limits">) { log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event"); - if (!(event.type in this.events)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`); - const cfg = this.events[event.type as keyof T]!; + 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]!; try { Value.Assert(cfg.schemas.payload, event.payload); await cfg.processor(event.payload, (state) => this.setEventState(event.id, state), event.id); - await this.fulfill(event.id); + await this.Events.fulfill(event.id); } catch (err) { + await this.setLastError(event.id, err); + if (err instanceof RateLimitError && event.rate_limits < EventQueue.RATE_LIMIT_RETRY_LIMIT) { log.warn( { rateLimits: event.rate_limits, delay: err.retryDelayMs, source: err.source, payload: err.payload }, @@ -256,7 +203,7 @@ export class EventQueue ({ diff --git a/apps/api/src/services/events.ts b/apps/api/src/services/events.ts new file mode 100644 index 0000000..ba0e4e3 --- /dev/null +++ b/apps/api/src/services/events.ts @@ -0,0 +1,134 @@ +import type { Static } from "@sinclair/typebox"; +import { t } from "elysia"; +import { sql, type Selectable, type Transaction } from "kysely"; +import type { DB } from "../database/out/db"; +import { db } from "../database/db"; +import { log } from "../util"; +import { minToMs } from "@blade-and-brawn/domain"; + +export const EventStatusSchema = t.Union([ + t.Literal("waiting"), + t.Literal("available"), + t.Literal("processing"), + t.Literal("failed"), + t.Literal("fulfilled"), +]); +export type EventStatus = Static; + +export type EventSource = "portal" | "printful" | "webflow"; + +export class EventsService { + static readonly CONCURRENCY_TIMEOUT_MS = minToMs(10); + + private _lastCleanDate: Date = new Date(0); + + get lastCleanDate() { return this._lastCleanDate; }; + + async get(id: string, trx?: Transaction) { + return await (trx ?? db).selectFrom("events") + .selectAll() + .where("id", "=", id) + .executeTakeFirst(); + } + + async list(opt: { + filter?: { status?: EventStatus, type?: string, group?: string }, + limit?: number, + offset?: number, + } = {}, trx?: Transaction) { + return await (trx ?? db).selectFrom("events") + .select([ + "id", "type", "source", "started_at", "ended_at", "available_at", + "errors", "rate_limits", "has_failed", "last_error", "created_at", + ]) + .$if(opt.filter?.group !== undefined, (qb) => qb.where("group", "=", opt.filter!.group!)) + .$if(opt.filter?.type !== undefined, (qb) => qb.where("type", "=", opt.filter!.type!)) + .$if(opt.filter?.status === "waiting", (qb) => qb + .where("started_at", "is", null) + .where("available_at", ">", new Date()) + ) + .$if(opt.filter?.status === "available", (qb) => qb + .where("started_at", "is", null) + .where("available_at", "<=", new Date()) + ) + .$if(opt.filter?.status === "processing", (qb) => qb + .where("started_at", "is not", null) + .where("ended_at", "is", null) + ) + .$if(opt.filter?.status === "failed", (qb) => qb + .where("started_at", "is not", null) + .where("ended_at", "is not", null) + .where("has_failed", "is", true) + ) + .$if(opt.filter?.status === "fulfilled", (qb) => qb + .where("started_at", "is not", null) + .where("ended_at", "is not", null) + .where("has_failed", "is", false) + ) + .orderBy(sql`(started_at is not null and ended_at is null)`, "desc") + .orderBy("created_at", "desc") + .orderBy("id", "desc") + .$if(opt.limit !== undefined, (qb) => qb.limit(opt.limit!)) + .$if(opt.offset !== undefined, (qb) => qb.offset(opt.offset!)) + .execute(); + } + + static status(event: Pick, "available_at" | "started_at" | "ended_at" | "has_failed">): EventStatus { + if (!event.started_at) { + if (Date.now() < event.available_at.getTime()) return "waiting"; + return "available"; + }; + if (!event.ended_at) return "processing"; + return event.has_failed ? "failed" : "fulfilled"; + } + + async fail(id: string, trx?: Transaction): Promise { + await (trx ?? db).updateTable("events") + .set((eb) => ({ + state: null, + ended_at: new Date(), + has_failed: true, + errors: eb("errors", "+", 1) + })) + .where("id", "=", id) + .execute(); + } + + async fulfill(id: string, trx?: Transaction): Promise { + await (trx ?? db).updateTable("events") + .set({ state: null, ended_at: new Date(), last_error: null }) + .where("id", "=", id) + .execute(); + } + + async replay(id: string): Promise { + const result = await db.updateTable("events") + .set({ + started_at: null, + ended_at: null, + has_failed: false, + available_at: new Date(), + }) + .where("id", "=", id) + .where("has_failed", "=", true) + .executeTakeFirst(); + return result.numUpdatedRows > 0n; + } + + async clean(opt: { filter?: { type?: string, group?: string } } = {}) { + try { + // Find and fail any timed out processing events + const processingEvents = await this.list({ filter: { status: "processing", ...opt.filter } }); + for (const processingEvent of processingEvents) { + const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > EventsService.CONCURRENCY_TIMEOUT_MS); + if (processingEventTimedOut) { + log.warn({ name: processingEvent.id }, "processing event has timed out. failing it and continuing..."); + await this.fail(processingEvent.id); + } + } + } + finally { + this._lastCleanDate = new Date(); + } + } +}