feat: emit updates from relay-worker

packages/worker-relay/src/interface.ts

@@ -5,7 +5,7 @@ import { v4 as uuid } from "uuid";
 export class WorkerRelayInterface {
   #worker: Worker;
   #log = (msg: any) => console.debug(msg);
-  #commandQueue: Map<string, (v: unknown) => void> = new Map();
+  #commandQueue: Map<string, (v: unknown, ports: ReadonlyArray<MessagePort>) => void> = new Map();
 
   constructor(path: string) {
     this.#log(`Module path: ${path}`);
@@ -14,7 +14,7 @@ export class WorkerRelayInterface {
       const cmd = e.data as WorkerMessage<any>;
       if (cmd.cmd === "reply") {
         const q = this.#commandQueue.get(cmd.id);
-        q?.(cmd);
+        q?.(cmd, e.ports);
         this.#commandQueue.delete(cmd.id);
       }
     };
@@ -37,7 +37,7 @@ export class WorkerRelayInterface {
   }
 
   async req(req: ReqCommand) {
-    return await this.#workerRpc<ReqCommand, Array<NostrEvent>>("req", req);
+    return await this.#workerRpc<ReqCommand, { results: Array<NostrEvent>; port?: Readonly<MessagePort> }>("req", req);
   }
 
   async count(req: ReqCommand) {
@@ -48,6 +48,10 @@ export class WorkerRelayInterface {
     return await this.#workerRpc<void, Record<string, number>>("summary");
   }
 
+  async close(id: string) {
+    return await this.#workerRpc<string, boolean>("close", id);
+  }
+
   #workerRpc<T, R>(cmd: string, args?: T, timeout = 30_000) {
     const id = uuid();
     const msg = {
@@ -58,13 +62,14 @@ export class WorkerRelayInterface {
     this.#worker.postMessage(msg);
     return new Promise<R>((resolve, reject) => {
       let t: ReturnType<typeof setTimeout>;
-      this.#commandQueue.set(id, v => {
+      this.#commandQueue.set(id, (v, ports) => {
         clearTimeout(t);
         const cmdReply = v as WorkerMessage<R>;
-        resolve(cmdReply.args);
+        resolve({ ...cmdReply.args, port: ports.length > 0 ? ports[0] : undefined });
       });
       t = setTimeout(() => {
         reject("timeout");
+        this.#commandQueue.delete(id);
       }, timeout);
     });
   }

packages/worker-relay/src/relay.ts

@@ -1,8 +1,13 @@
 import sqlite3InitModule, { Database, Sqlite3Static } from "@sqlite.org/sqlite-wasm";
 import debug from "debug";
+import { EventEmitter } from "eventemitter3";
 import { NostrEvent, ReqFilter, unixNowMs } from "./types";
 
-export class WorkerRelay {
+export interface WorkerRelayEvents {
+  event: (evs: Array<NostrEvent>) => void;
+}
+
+export class WorkerRelay extends EventEmitter<WorkerRelayEvents> {
   #sqlite?: Sqlite3Static;
   #log = debug("WorkerRelay");
   #db?: Database;
@@ -72,6 +77,7 @@ export class WorkerRelay {
     });
     if (eventInserted) {
       this.#log(`Inserted: kind=${ev.kind},authors=${ev.pubkey},id=${ev.id}`);
+      this.emit("event", [ev]);
     }
     return eventInserted;
   }
@@ -90,6 +96,7 @@ export class WorkerRelay {
     });
     if (eventsInserted.length > 0) {
       this.#log(`Inserted Batch: ${eventsInserted.length}/${evs.length}`);
+      this.emit("event", eventsInserted);
     }
     return eventsInserted.length > 0;
   }

packages/worker-relay/src/types.ts

@@ -1,6 +1,6 @@
 export interface WorkerMessage<T> {
   id: string;
-  cmd: "reply" | "init" | "open" | "migrate" | "event" | "req" | "count" | "summary";
+  cmd: "reply" | "init" | "open" | "migrate" | "event" | "req" | "count" | "summary" | "close";
   args: T;
 }
 
@@ -14,7 +14,11 @@ export interface NostrEvent {
   sig: string;
 }
 
-export type ReqCommand = [cmd: "REQ", id: string, ...filters: Array<ReqFilter>];
+export interface ReqCommand {
+  id: string;
+  filters: Array<ReqFilter>;
+  leaveOpen?: boolean;
+}
 
 export interface ReqFilter {
   ids?: string[];

packages/worker-relay/src/worker.ts

@@ -4,14 +4,41 @@ import { WorkQueueItem, barrierQueue, processWorkQueue } from "./queue";
 import { WorkerRelay } from "./relay";
 import { NostrEvent, ReqCommand, ReqFilter, WorkerMessage } from "./types";
 
-const relay = new WorkerRelay();
+interface PortedFilter {
+  filters: Array<ReqFilter>;
+  port: MessagePort;
+}
 
-async function reply<T>(id: string, obj?: T) {
-  globalThis.postMessage({
-    id,
-    cmd: "reply",
-    args: obj,
-  } as WorkerMessage<T>);
+// Active open subscriptions awaiting new events
+const ActiveSubscriptions = new Map<string, PortedFilter>();
+
+const relay = new WorkerRelay();
+relay.on("event", evs => {
+  for (const pf of ActiveSubscriptions.values()) {
+    const pfSend = [];
+    for (const ev of evs) {
+      for (const fx of pf.filters) {
+        if (eventMatchesFilter(ev, fx)) {
+          pfSend.push(ev);
+          continue;
+        }
+      }
+    }
+    if (pfSend.length > 0) {
+      pf.port.postMessage(pfSend);
+    }
+  }
+});
+
+async function reply<T>(id: string, obj?: T, transferables?: Transferable[]) {
+  globalThis.postMessage(
+    {
+      id,
+      cmd: "reply",
+      args: obj,
+    } as WorkerMessage<T>,
+    transferables ?? [],
+  );
 }
 
 // Event inserter queue
@@ -25,7 +52,7 @@ async function insertBatch() {
 }
 setTimeout(() => insertBatch(), 100);
 
-let cmdQueue: Array<WorkQueueItem> = [];
+const cmdQueue: Array<WorkQueueItem> = [];
 processWorkQueue(cmdQueue, 50);
 
 globalThis.onclose = () => {
@@ -64,14 +91,26 @@ globalThis.onmessage = ev => {
         reply(msg.id, true);
         break;
       }
+      case "close": {
+        ActiveSubscriptions.delete(msg.args as string);
+        reply(msg.id, true);
+        break;
+      }
       case "req": {
         barrierQueue(cmdQueue, async () => {
           const req = msg.args as ReqCommand;
+          const chan = new MessageChannel();
+          if (req.leaveOpen) {
+            ActiveSubscriptions.set(req.id, {
+              filters: req.filters,
+              port: chan.port1,
+            });
+          }
           const results = [];
-          for (const r of req.slice(2)) {
+          for (const r of req.filters) {
             results.push(...relay.req(r as ReqFilter));
           }
-          reply(msg.id, results);
+          reply(msg.id, { results }, req.leaveOpen ? [chan.port2] : undefined);
         });
         break;
       }
@@ -79,7 +118,7 @@ globalThis.onmessage = ev => {
         barrierQueue(cmdQueue, async () => {
           const req = msg.args as ReqCommand;
           let results = 0;
-          for (const r of req.slice(2)) {
+          for (const r of req.filters) {
             const c = relay.count(r as ReqFilter);
             results += c;
           }
@@ -103,3 +142,22 @@ globalThis.onmessage = ev => {
     reply(msg.id, { error: e });
   }
 };
+
+export function eventMatchesFilter(ev: NostrEvent, filter: ReqFilter) {
+  if (filter.since && ev.created_at < filter.since) {
+    return false;
+  }
+  if (filter.until && ev.created_at > filter.until) {
+    return false;
+  }
+  if (!(filter.ids?.includes(ev.id) ?? true)) {
+    return false;
+  }
+  if (!(filter.authors?.includes(ev.pubkey) ?? true)) {
+    return false;
+  }
+  if (!(filter.kinds?.includes(ev.kind) ?? true)) {
+    return false;
+  }
+  return true;
+}