diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts index e21a14d..a0013a0 100644 --- a/apps/api/src/server.ts +++ b/apps/api/src/server.ts @@ -20,7 +20,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 type { EventQueue, EventSpec } from "./services/event-queue"; +import type { EventQueue, EventTypeOptions } from "./services/event-queue"; // CONSTANTS // ----------------------- @@ -326,7 +326,7 @@ app.listen(3000, async () => { log.info({ port: 3000 }, "server started") // MANAGE QUEUES - const queues: EventQueue>>[] = [s.Commerce.ProductSync.Queue]; + const queues: EventQueue>>[] = [s.Commerce.ProductSync.Queue]; for (const queue of queues) { (async () => { while (true) { diff --git a/apps/api/src/services/commerce.ts b/apps/api/src/services/commerce.ts index f75f9a6..87b6be1 100644 --- a/apps/api/src/services/commerce.ts +++ b/apps/api/src/services/commerce.ts @@ -1,44 +1,11 @@ import { PrintfulClient, ProductSyncer, WebflowClient } from "@blade-and-brawn/commerce"; import { env, log } from "../util"; import { db } from "../database/db"; -import { EventQueue, type EventSpec } from "./event-queue"; +import { defineEventType, EventQueue } 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"; -// product_sync_update -const ProductSyncUpdatePayloadSchema = t.Object({ - session: t.Object({ - name: t.String(), - id: t.String() - }), - filter: t.Optional( - t.Object({ - pProductIds: t.Optional(t.Array(t.Number())) - }) - ) -}); -type ProductSyncUpdatePayload = Static; -const ProductSyncUpdateStateSchema = t.Union([ - t.Object({ - pProductIds: t.Optional(t.Array(t.Number())) - }), - t.Null() -]); -type ProductSyncUpdateState = Static; - -// product_sync_delete -const ProductSyncDeletePayloadSchema = t.Object({ wProductId: t.String() }); -type ProductSyncDeletePayload = Static; -const ProductSyncDeleteStateSchema = t.Union([t.Object({}), t.Null()]); -type ProductSyncDeleteState = Static; - -export type ProductSyncEvents = { - product_sync_update: EventSpec; - product_sync_delete: EventSpec; -}; - export class CommerceService { readonly Printful = new PrintfulClient({ token: env.PRINTFUL_AUTH, @@ -55,22 +22,40 @@ export class CommerceService { class ProductSyncService { static readonly EVENT_GROUP = "product_sync"; - readonly Queue: EventQueue<"product_sync", ProductSyncEvents>; + readonly Queue; 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 EventQueue<"product_sync", ProductSyncEvents>({ + this.Queue = new EventQueue({ group: ProductSyncService.EVENT_GROUP, cfg: { retry: { limit: 3, delayMs: 10000 }, concurrency: { limit: 1, timeoutMs: 600000 } }, events: { - "product_sync_update": { - schemas: { payload: ProductSyncUpdatePayloadSchema, state: ProductSyncUpdateStateSchema }, + "product_sync_update": defineEventType({ + schemas: { + payload: t.Object({ + session: t.Object({ + name: t.String(), + id: t.String() + }), + filter: t.Optional( + t.Object({ + pProductIds: t.Optional(t.Array(t.Number())) + }) + ) + }), + state: t.Union([ + t.Object({ + pProductIds: t.Optional(t.Array(t.Number())) + }), + t.Null() + ]) + }, processor: async (id, payload, setState) => { await this.ProductSyncer.sync({ filter: payload.filter, @@ -79,13 +64,16 @@ class ProductSyncService { } }); } - }, - "product_sync_delete": { - schemas: { payload: ProductSyncDeletePayloadSchema, state: ProductSyncDeleteStateSchema }, + }), + "product_sync_delete": defineEventType({ + schemas: { + payload: t.Object({ wProductId: t.String() }), + state: t.Union([t.Object({}), t.Null()]) + }, processor: async (id, payload, setState) => { await this.Webflow.Products.remove(payload.wProductId); } - } + }) } }); } @@ -102,7 +90,7 @@ class ProductSyncService { .executeTakeFirst(); if (!latestSync) return; - Value.Assert(ProductSyncUpdateStateSchema, latestSync.state); + Value.Assert(this.Queue.events["product_sync_update"].schemas.state, latestSync.state); return { startDate: latestSync.started_at, diff --git a/apps/api/src/services/event-queue.ts b/apps/api/src/services/event-queue.ts index 67c104d..559a177 100644 --- a/apps/api/src/services/event-queue.ts +++ b/apps/api/src/services/event-queue.ts @@ -4,60 +4,47 @@ import { log } from "../util"; import { sleep } from "bun"; import { db } from "../database/db"; import { Value } from "@sinclair/typebox/value"; -import type { TSchema } from "@sinclair/typebox"; +import type { TSchema, Static } from "@sinclair/typebox"; -export type EventSpec

= { - payload: P, - state: S -} export type EventSource = "portal" | "printful" | "webflow"; +export type EventStatus = "pending" | "processing" | "failed" | "fulfilled"; -type EventProcessor> = - (id: string, payload: S["payload"], setState: (state: S["state"]) => Promise) => Promise; -type EventStatus = "pending" | "processing" | "failed" | "fulfilled"; +type EventProcessor

= + (id: string, payload: Static

, setState: (state: Static) => Promise) => Promise; -type EventTypeOptions> = { - schemas: { - payload: TSchema & { static: S["payload"] }, - state: TSchema & { static: S["state"] }, - }, - processor: EventProcessor, +export type EventTypeOptions

= { + schemas: { payload: P, state: S }, + processor: EventProcessor, }; -type EventQueueOptions>> = { - group: G, - cfg: { - retry: { - limit: number, - delayMs: number, - }, - concurrency: { - limit: number, - timeoutMs: number, - }, +type EventQueueCfg = { + retry: { + limit: number, + delayMs: number, }, - events: { - [T in keyof M]: EventTypeOptions + concurrency: { + limit: number, + timeoutMs: number, }, -} - -type EnqueueOptions>, T extends keyof M & string> = { - source: EventSource, - type: T, - payload: M[T]["payload"] -} +}; class AlreadyClaimedEventError extends Error { } class ConcurrencyLimitReachedError extends Error { } -export class EventQueue>> { - private _lastCleanDate: Date = new Date(0); - readonly cfg: EventQueueOptions["cfg"]; - readonly group: G; +export function defineEventType

(opt: EventTypeOptions): EventTypeOptions { + return opt; +} - constructor(private readonly opt: EventQueueOptions) { +export class EventQueue>> { + private _lastCleanDate: Date = new Date(0); + readonly cfg: EventQueueCfg; + readonly group: G; + readonly events: T; + + constructor(opt: { group: G, cfg: EventQueueCfg, events: T }) { this.cfg = opt.cfg; this.group = opt.group; + this.events = opt.events; } get lastCleanDate() { return this._lastCleanDate; }; @@ -139,10 +126,14 @@ export class EventQueue({ source, type, payload }: EnqueueOptions): Promise { - log.info({ payload }, "enqueuing"); + async enqueue(opt: { + source: EventSource, + type: K, + payload: Static + }): Promise { + log.info({ payload: opt.payload }, "enqueuing"); const result = await db.insertInto("events") - .values({ group: this.group, type, source, payload }) + .values({ group: this.group, type: opt.type, source: opt.source, payload: opt.payload as JsonValue }) .returning(["id"]) .executeTakeFirstOrThrow(); return result.id; @@ -201,8 +192,8 @@ export class EventQueue {