Majorly improve and simplify event queue

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<string, Record<string, EventSpec<any, any>>>[] = [s.Commerce.ProductSync.Queue];
+        const queues: EventQueue<string, Record<string, EventTypeOptions<any, any>>>[] = [s.Commerce.ProductSync.Queue];
         for (const queue of queues) {
             (async () => {
                 while (true) {

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<typeof ProductSyncUpdatePayloadSchema>;
-const ProductSyncUpdateStateSchema = t.Union([
-    t.Object({
-        pProductIds: t.Optional(t.Array(t.Number()))
-    }),
-    t.Null()
-]);
-type ProductSyncUpdateState = Static<typeof ProductSyncUpdateStateSchema>;
-
-// product_sync_delete
-const ProductSyncDeletePayloadSchema = t.Object({ wProductId: t.String() });
-type ProductSyncDeletePayload = Static<typeof ProductSyncDeletePayloadSchema>;
-const ProductSyncDeleteStateSchema = t.Union([t.Object({}), t.Null()]);
-type ProductSyncDeleteState = Static<typeof ProductSyncDeleteStateSchema>;
-
-export type ProductSyncEvents = {
-    product_sync_update: EventSpec<ProductSyncUpdatePayload, ProductSyncUpdateState>;
-    product_sync_delete: EventSpec<ProductSyncDeletePayload, ProductSyncDeleteState>;
-};
-
 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: <ServiceName>})
 
     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": {
+                "product_sync_update": defineEventType({
-                    schemas: { payload: ProductSyncUpdatePayloadSchema, state: ProductSyncUpdateStateSchema },
+                    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": {
+                "product_sync_delete": defineEventType({
-                    schemas: { payload: ProductSyncDeletePayloadSchema, state: ProductSyncDeleteStateSchema },
+                    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,

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<P extends JsonValue, S extends JsonValue> = {
-    payload: P,
-    state: S
-}
 export type EventSource = "portal" | "printful" | "webflow";
+export type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
 
-type EventProcessor<S extends EventSpec<JsonValue, JsonValue>> =
+type EventProcessor<P extends TSchema, S extends TSchema> =
-    (id: string, payload: S["payload"], setState: (state: S["state"]) => Promise<void>) => Promise<void>;
+    (id: string, payload: Static<P>, setState: (state: Static<S>) => Promise<void>) => Promise<void>;
-type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
 
-type EventTypeOptions<S extends EventSpec<JsonValue, JsonValue>> = {
+export type EventTypeOptions<P extends TSchema, S extends TSchema> = {
-    schemas: {
+    schemas: { payload: P, state: S },
-        payload: TSchema & { static: S["payload"] },
+    processor: EventProcessor<P, S>,
-        state: TSchema & { static: S["state"] },
-    },
-    processor: EventProcessor<S>,
 };
 
-type EventQueueOptions<G extends string, M extends Record<string, EventSpec<JsonValue, JsonValue>>> = {
+type EventQueueCfg = {
-    group: G,
+    retry: {
-    cfg: {
+        limit: number,
-        retry: {
+        delayMs: number,
-            limit: number,
-            delayMs: number,
-        },
-        concurrency: {
-            limit: number,
-            timeoutMs: number,
-        },
     },
-    events: {
+    concurrency: {
-        [T in keyof M]: EventTypeOptions<M[T]>
+        limit: number,
+        timeoutMs: number,
     },
-}
+};
-
-type EnqueueOptions<M extends Record<string, EventSpec<JsonValue, JsonValue>>, T extends keyof M & string> = {
-    source: EventSource,
-    type: T,
-    payload: M[T]["payload"]
-}
 
 class AlreadyClaimedEventError extends Error { }
 class ConcurrencyLimitReachedError extends Error { }
 
-export class EventQueue<G extends string, M extends Record<string, EventSpec<JsonValue, JsonValue>>> {
+export function defineEventType<P extends TSchema, S extends TSchema>(opt: EventTypeOptions<P, S>): EventTypeOptions<P, S> {
-    private _lastCleanDate: Date = new Date(0);
+    return opt;
-    readonly cfg: EventQueueOptions<G, M>["cfg"];
+}
-    readonly group: G;
 
-    constructor(private readonly opt: EventQueueOptions<G, M>) {
+export class EventQueue<G extends string, T extends Record<string, EventTypeOptions<any, any>>> {
+    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<G extends string, M extends Record<string, EventSpec<Jso
         return front;
     }
 
-    async enqueue<T extends keyof M & string>({ source, type, payload }: EnqueueOptions<M, T>): Promise<string> {
+    async enqueue<K extends keyof T & string>(opt: {
-        log.info({ payload }, "enqueuing");
+        source: EventSource,
+        type: K,
+        payload: Static<T[K]["schemas"]["payload"]>
+    }): Promise<string> {
+        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<G extends string, M extends Record<string, EventSpec<Jso
 
     private async processEvent(id: string, type: string, payload: JsonValue) {
         log.info({ id, payload }, "processing event");
-        if (!(type in this.opt.events)) throw new Error(`Unknown event type: ${type}`);
+        if (!(type in this.events)) throw new Error(`event type not registerd in this queue: ${type}`);
-        const cfg = this.opt.events[type as keyof M];
+        const cfg = this.events[type as keyof T]!;
         try {
             Value.Assert(cfg.schemas.payload, payload);
             await cfg.processor(id, payload, async (state) => {