Refactor event queue to allow multiple types of events

apps/api/src/services/event-queue.ts

@@ -6,22 +6,26 @@ import { db } from "../database/db";
 import { Value } from "@sinclair/typebox/value";
 import type { TSchema } from "@sinclair/typebox";
 
-export type EventType = "product_sync" | "product_order_created" /* TODO: more events... */;
+export type EventSpec<P extends JsonValue = JsonValue, S extends JsonValue = JsonValue> = {
+    payload: P,
+    state: S
+}
 export type EventSource = "portal" | "printful" | "webflow";
 
-type EventProcessor<Payload extends JsonValue, State extends JsonValue> =
-    (id: string, payload: Payload, setState: (state: State) => Promise<void>) => Promise<void>;
+type EventProcessor<S extends EventSpec> =
+    (id: string, payload: S["payload"], setState: (state: S["state"]) => Promise<void>) => Promise<void>;
 type EventStatus = "pending" | "processing" | "failed" | "fulfilled";
 
-class AlreadyClaimedEventError extends Error { }
-class ConcurrencyLimitReachedError extends Error { }
-
-type EventQueueServiceOptions<Payload extends JsonValue, State extends JsonValue> = {
-    type: EventType,
+type EventTypeOptions<S extends EventSpec> = {
     schemas: {
-        payload: TSchema & { static: Payload },
-        state: TSchema & { static: State },
+        payload: TSchema & { static: S["payload"] },
+        state: TSchema & { static: S["state"] },
     },
+    processor: EventProcessor<S>,
+};
+
+type EventQueueOptions<G extends string, M extends Record<string, EventSpec>> = {
+    group: G,
     cfg: {
         retry: {
             limit: number,
@@ -32,17 +36,28 @@ type EventQueueServiceOptions<Payload extends JsonValue, State extends JsonValue
             timeoutMs: number,
         },
     },
-    processor: EventProcessor<Payload, State>,
+    types: {
+        [T in keyof M]: EventTypeOptions<M[T]>
+    },
 }
 
-export class EventQueueService<Payload extends JsonValue, State extends JsonValue> {
-    private _lastCleanDate: Date = new Date(0);
-    readonly cfg: EventQueueServiceOptions<Payload, State>["cfg"];
-    readonly type: EventType
+type EnqueueOptions<M extends Record<string, EventSpec>, T extends keyof M & string> = {
+    source: EventSource,
+    type: T,
+    payload: M[T]["payload"]
+}
 
-    constructor(private readonly opt: EventQueueServiceOptions<Payload, State>) {
+class AlreadyClaimedEventError extends Error { }
+class ConcurrencyLimitReachedError extends Error { }
+
+export class EventQueue<G extends string, M extends Record<string, EventSpec>> {
+    private _lastCleanDate: Date = new Date(0);
+    readonly cfg: EventQueueOptions<G, M>["cfg"];
+    readonly group: G;
+
+    constructor(private readonly opt: EventQueueOptions<G, M>) {
         this.cfg = opt.cfg;
-        this.type = opt.type;
+        this.group = opt.group;
     }
 
     get lastCleanDate() { return this._lastCleanDate; };
@@ -68,13 +83,13 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
                     continue;
                 }
                 if (err instanceof ConcurrencyLimitReachedError) {
-                    await sleep(this.opt.cfg.retry.delayMs);
+                    await sleep(this.cfg.retry.delayMs);
                     continue;
                 }
 
-                if (++consecutiveFailures <= this.opt.cfg.retry.limit) {
+                if (++consecutiveFailures <= this.cfg.retry.limit) {
                     log.warn({ err, consecutiveFailures }, "Dequeue failed, retrying")
-                    await sleep(this.opt.cfg.retry.delayMs);
+                    await sleep(this.cfg.retry.delayMs);
                     continue;
                 }
 
@@ -98,11 +113,11 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
                 eb(
                     eb.selectFrom("events")
                         .select(eb.fn.countAll().as("count"))
-                        .where("type", "=", this.opt.type)
+                        .where("group", "=", this.group)
                         .where("started_at", "is not", null)
                         .where("ended_at", "is", null),
                     "<",
-                    this.opt.cfg.concurrency.limit
+                    this.cfg.concurrency.limit
                 )
             )
             .executeTakeFirstOrThrow();
@@ -120,14 +135,14 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
         };
 
         log.trace({ name: front.id }, "dequeuing event");
-        await this.handleEvent(front.id, front.payload);
+        await this.handleEvent(front.id, front.type, front.payload);
         return front;
     }
 
-    async enqueue({ source, payload }: { source: EventSource, payload: Payload }): Promise<string> {
+    async enqueue<T extends keyof M & string>({ source, type, payload }: EnqueueOptions<M, T>): Promise<string> {
         log.info({ payload }, "enqueuing");
         const result = await db.insertInto("events")
-            .values({ type: this.opt.type, source, payload })
+            .values({ group: this.group, type, source, payload })
             .returning(["id"])
             .executeTakeFirstOrThrow();
         return result.id;
@@ -152,7 +167,7 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
             // 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);
+                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);
@@ -167,7 +182,7 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
     async front(trx?: Transaction<DB>) {
         const result = await (trx ?? db).selectFrom("events")
             .selectAll()
-            .where("type", "=", this.opt.type)
+            .where("group", "=", this.group)
             .where("started_at", "is", null)
             .orderBy("created_at", "asc")
             .limit(1)
@@ -178,17 +193,19 @@ export class EventQueueService<Payload extends JsonValue, State extends JsonValu
     async processingEvents(trx?: Transaction<DB>) {
         return await (trx ?? db).selectFrom("events")
             .select(["started_at", "id"])
-            .where("type", "=", this.opt.type)
+            .where("group", "=", this.group)
             .where("started_at", "is not", null)
             .where("ended_at", "is", null)
             .execute();
     }
 
-    private async handleEvent(id: string, payload: JsonValue) {
+    private async handleEvent(id: string, type: string, payload: JsonValue) {
         log.info({ id, payload }, "handling event");
+        if (!(type in this.opt.types)) throw new Error(`Unknown event type: ${type}`);
+        const cfg = this.opt.types[type as keyof M];
         try {
-            Value.Assert(this.opt.schemas.payload, payload);
-            await this.opt.processor(id, payload, async (state) => {
+            Value.Assert(cfg.schemas.payload, payload);
+            await cfg.processor(id, payload, async (state) => {
                 await db.updateTable("events")
                     .set({ state })
                     .where("id", "=", id)