Implement basic event deduplication

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

@@ -16,6 +16,7 @@ type EventProcessor<P extends TSchema, S extends TSchema> =
 export type EventTypeOptions<P extends TSchema, S extends TSchema> = {
     schemas: { payload: P, state: S },
     processor: EventProcessor<P, S>,
+    dedupeKey: (payload: Static<P>) => string | undefined
 };
 
 type EventQueueCfg = {
@@ -137,17 +138,33 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
         return front;
     }
 
-    async enqueue<K extends keyof T & string>(opt: {
+    async enqueue<K extends keyof T & string>({ source, type, payload }: {
         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: opt.type, source: opt.source, payload: opt.payload as JsonValue })
+    }): Promise<void> {
+        if (!(type in this.types)) throw new Error(`event type not registered in this queue (${this.group}): ${type}`);
+        const cfg = this.types[type as keyof T]!;
+
+        const dedupeKey = cfg.dedupeKey(payload);
+        log.info({ payload, dedupeKey }, "enqueuing");
+
+        await db.insertInto("events")
+            .values({
+                group: this.group,
+                type,
+                source,
+                payload,
+                dedupe_key: dedupeKey ?? null,
+            })
+            .onConflict((oc) => oc
+                .columns(["group", "type", "dedupe_key"])
+                .where("dedupe_key", "is not", null)
+                .where("ended_at", "is", null)
+                .doNothing()
+            )
             .returning(["id"])
-            .executeTakeFirstOrThrow();
-        return result.id;
+            .executeTakeFirst();
     }
 
     async front(trx?: Transaction<DB>) {
@@ -174,11 +191,11 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
     private async process(event: Pick<Selectable<DB["events"]>, "id" | "type" | "errors" | "payload" | "rate_limits">) {
         log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event");
         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]!;
+        const typeCfg = this.types[event.type as keyof T]!;
 
         try {
-            Value.Assert(cfg.schemas.payload, event.payload);
-            await cfg.processor(event.payload, (state) => this.Events.setState(event.id, state), event.id);
+            Value.Assert(typeCfg.schemas.payload, event.payload);
+            await typeCfg.processor(event.payload, (state) => this.Events.setState(event.id, state), event.id);
             await this.Events.resolve(event.id);
         }
         catch (err) {