feat: NoteStore event-emitter

packages/app/src/Hooks/useRefreshFeedcache.tsx

@@ -28,7 +28,7 @@ export function useRefreshFeedCache<T>(c: RefreshFeedCache<T>, leaveOpen = false
       const q = system.Query(NoopStore, sub);
       let t: ReturnType<typeof setTimeout> | undefined;
       let tBuf: Array<TaggedNostrEvent> = [];
-      const releaseOnEvent = q.feed.onEvent(evs => {
+      q.feed.on("event", evs => {
         if (!t) {
           tBuf = [...evs];
           t = setTimeout(() => {
@@ -41,9 +41,9 @@ export function useRefreshFeedCache<T>(c: RefreshFeedCache<T>, leaveOpen = false
       });
       q.uncancel();
       return () => {
+        q.feed.off("event");
         q.cancel();
         q.sendClose();
-        releaseOnEvent();
       };
     }
   }, [sub]);

packages/system-react/src/useRequestBuilder.tsx

@@ -14,11 +14,11 @@ const useRequestBuilder = <TStore extends NoteStore, TSnapshot = ReturnType<TSto
   const subscribe = (onChanged: () => void) => {
     if (rb) {
       const q = system.Query<TStore>(type, rb);
-      const release = q.feed.hook(onChanged);
+      q.feed.on("event", onChanged);
       q.uncancel();
       return () => {
+        q.feed.off("event", onChanged);
         q.cancel();
-        release();
       };
     }
     return () => {

packages/system/src/index.ts

@@ -43,7 +43,7 @@ export * from "./cache/user-relays";
 export * from "./cache/user-metadata";
 export * from "./cache/relay-metric";
 
-export * from "./worker/system-worker";
+export * from "./worker";
 
 export interface SystemInterface {
   /**
@@ -79,7 +79,7 @@ export interface SystemInterface {
    * @param req Request to send to relays
    * @param cb A callback which will fire every 100ms when new data is received
    */
-  Fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void): Promise<Array<TaggedNostrEvent>>;
+  Fetch(req: RequestBuilder, cb?: (evs: ReadonlyArray<TaggedNostrEvent>) => void): Promise<Array<TaggedNostrEvent>>;
 
   /**
    * Create a new permanent connection to a relay

packages/system/src/nostr-connection-pool.ts

@@ -17,10 +17,19 @@ export interface NostrConnectionPoolEvents {
   notice: (address: string, msg: string) => void;
 }
 
+export type ConnectionPool = {
+  getState(): ConnectionStateSnapshot[];
+  getConnection(id: string): Connection | undefined;
+  connect(address: string, options: RelaySettings, ephemeral: boolean): Promise<Connection | undefined>;
+  disconnect(address: string): void;
+  broadcast(system: SystemInterface, ev: NostrEvent, cb?: (rsp: OkResponse) => void): Promise<OkResponse[]>;
+  broadcastTo(address: string, ev: NostrEvent): Promise<OkResponse>;
+} & EventEmitter<NostrConnectionPoolEvents>;
+
 /**
  * Simple connection pool containing connections to multiple nostr relays
  */
-export class NostrConnectionPool extends EventEmitter<NostrConnectionPoolEvents> {
+export class NostrConnectionPool extends EventEmitter<NostrConnectionPoolEvents> implements ConnectionPool {
   #log = debug("NostrConnectionPool");
 
   /**

packages/system/src/nostr-query-manager.ts

@@ -73,28 +73,17 @@ export class NostrQueryManager extends EventEmitter<NostrQueryManagerEvents> {
   /**
    * Async fetch results
    */
-  fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void) {
+  fetch(req: RequestBuilder, cb?: (evs: ReadonlyArray<TaggedNostrEvent>) => void) {
     const q = this.query(NoteCollection, req);
     return new Promise<Array<TaggedNostrEvent>>(resolve => {
       let t: ReturnType<typeof setTimeout> | undefined;
       let tBuf: Array<TaggedNostrEvent> = [];
-      const releaseOnEvent = cb
+      if (cb) {
-        ? q.feed.onEvent(evs => {
+        q.feed.on("event", cb);
-            if (!t) {
+      }
-              tBuf = [...evs];
+      q.feed.on("progress", loading => {
-              t = setTimeout(() => {
+        if (!loading) {
-                t = undefined;
+          q.feed.off("event");
-                cb(tBuf);
-              }, 100);
-            } else {
-              tBuf.push(...evs);
-            }
-          })
-        : undefined;
-      const releaseFeedHook = q.feed.hook(() => {
-        if (q.progress === 1) {
-          releaseOnEvent?.();
-          releaseFeedHook();
           q.cancel();
           resolve(unwrap((q.feed as NoteCollection).snapshot.data));
         }

packages/system/src/nostr-system.ts

@@ -228,7 +228,7 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
     return this.#queryManager.get(id);
   }
 
-  Fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void) {
+  Fetch(req: RequestBuilder, cb?: (evs: ReadonlyArray<TaggedNostrEvent>) => void) {
     return this.#queryManager.fetch(req, cb);
   }
 

packages/system/src/note-collection.ts

@@ -1,6 +1,7 @@
 import { appendDedupe, SortedMap } from "@snort/shared";
 import { EventExt, EventType, TaggedNostrEvent, u256 } from ".";
 import { findTag } from "./utils";
+import EventEmitter from "eventemitter3";
 
 export interface StoreSnapshot<TSnapshot> {
   data: TSnapshot | undefined;
@@ -28,28 +29,25 @@ export type OnEventCallbackRelease = () => void;
 export type OnEoseCallback = (c: string) => void;
 export type OnEoseCallbackRelease = () => void;
 
+export interface NostrStoreEvents {
+  progress: (loading: boolean) => void;
+  event: (evs: Readonly<Array<TaggedNostrEvent>>) => void;
+}
+
 /**
  * Generic note store interface
  */
-export abstract class NoteStore {
+export abstract class NoteStore extends EventEmitter<NostrStoreEvents> {
   abstract add(ev: Readonly<TaggedNostrEvent> | Readonly<Array<TaggedNostrEvent>>): void;
   abstract clear(): void;
-
-  // react hooks
-  abstract hook(cb: NoteStoreHook): NoteStoreHookRelease;
   abstract getSnapshotData(): NoteStoreSnapshotData | undefined;
 
-  // events
-  abstract onEvent(cb: OnEventCallback): OnEventCallbackRelease;
-
   abstract get snapshot(): StoreSnapshot<NoteStoreSnapshotData>;
   abstract get loading(): boolean;
   abstract set loading(v: boolean);
 }
 
-export abstract class HookedNoteStore<TSnapshot extends NoteStoreSnapshotData> implements NoteStore {
+export abstract class HookedNoteStore<TSnapshot extends NoteStoreSnapshotData> extends NoteStore {
-  #hooks: Array<NoteStoreHook> = [];
-  #eventHooks: Array<OnEventCallback> = [];
   #loading = true;
   #storeSnapshot: StoreSnapshot<TSnapshot> = {
     clear: () => this.clear(),
@@ -59,6 +57,7 @@ export abstract class HookedNoteStore<TSnapshot extends NoteStoreSnapshotData> i
   };
   #needsSnapshot = true;
   #nextNotifyTimer?: ReturnType<typeof setTimeout>;
+  #bufEmit: Array<TaggedNostrEvent> = [];
 
   get snapshot() {
     this.#updateSnapshot();
@@ -71,56 +70,29 @@ export abstract class HookedNoteStore<TSnapshot extends NoteStoreSnapshotData> i
 
   set loading(v: boolean) {
     this.#loading = v;
-    this.onChange([]);
+    this.emit("progress", v);
   }
 
-  abstract add(ev: Readonly<TaggedNostrEvent> | Readonly<Array<TaggedNostrEvent>>): void;
+  abstract override add(ev: Readonly<TaggedNostrEvent> | Readonly<Array<TaggedNostrEvent>>): void;
-  abstract clear(): void;
+  abstract override clear(): void;
-
-  hook(cb: NoteStoreHook): NoteStoreHookRelease {
-    this.#hooks.push(cb);
-    return () => {
-      const idx = this.#hooks.findIndex(a => a === cb);
-      this.#hooks.splice(idx, 1);
-    };
-  }
 
   getSnapshotData() {
     this.#updateSnapshot();
     return this.#storeSnapshot.data;
   }
 
-  onEvent(cb: OnEventCallback): OnEventCallbackRelease {
-    const existing = this.#eventHooks.find(a => a === cb);
-    if (!existing) {
-      this.#eventHooks.push(cb);
-      return () => {
-        const idx = this.#eventHooks.findIndex(a => a === cb);
-        this.#eventHooks.splice(idx, 1);
-      };
-    }
-    return () => {
-      //noop
-    };
-  }
-
   protected abstract takeSnapshot(): TSnapshot | undefined;
 
   protected onChange(changes: Readonly<Array<TaggedNostrEvent>>): void {
     this.#needsSnapshot = true;
+    this.#bufEmit.push(...changes);
     if (!this.#nextNotifyTimer) {
       this.#nextNotifyTimer = setTimeout(() => {
         this.#nextNotifyTimer = undefined;
-        for (const hk of this.#hooks) {
+        this.emit("event", this.#bufEmit);
-          hk();
+        this.#bufEmit = [];
-        }
       }, 500);
     }
-    if (changes.length > 0) {
-      for (const hkE of this.#eventHooks) {
-        hkE(changes);
-      }
-    }
   }
 
   #updateSnapshot() {

packages/system/src/worker/index.ts

@@ -1,4 +1,4 @@
-export const enum NostrSystemCommand {
+export const enum WorkerCommand {
   OkResponse,
   ErrorResponse,
   Init,
@@ -6,8 +6,8 @@ export const enum NostrSystemCommand {
   DisconnectRelay,
 }
 
-export interface NostrSystemMessage<T> {
+export interface WorkerMessage<T> {
   id: string;
-  type: NostrSystemCommand;
+  type: WorkerCommand;
   data: T;
 }

packages/system/src/worker/system-worker.ts

@@ -15,7 +15,7 @@ import {
 } from "..";
 import { NostrSystemEvents, NostrsystemProps } from "../nostr-system";
 import { Query } from "../query";
-import { NostrSystemCommand, NostrSystemMessage } from ".";
+import { WorkerCommand, WorkerMessage } from ".";
 
 export class SystemWorker extends EventEmitter<NostrSystemEvents> implements SystemInterface {
   #worker: Worker;
@@ -36,7 +36,7 @@ export class SystemWorker extends EventEmitter<NostrSystemEvents> implements Sys
   }
 
   async Init() {
-    await this.#workerRpc<void, string>(NostrSystemCommand.Init, undefined);
+    await this.#workerRpc<void, string>(WorkerCommand.Init, undefined);
   }
 
   GetQuery(id: string): Query | undefined {
@@ -83,19 +83,19 @@ export class SystemWorker extends EventEmitter<NostrSystemEvents> implements Sys
     throw new Error("Method not implemented.");
   }
 
-  #workerRpc<T, R>(type: NostrSystemCommand, data: T, timeout = 5_000) {
+  #workerRpc<T, R>(type: WorkerCommand, data: T, timeout = 5_000) {
     const id = uuid();
     this.#worker.postMessage({
       id,
       type,
       data,
-    } as NostrSystemMessage<T>);
+    } as WorkerMessage<T>);
     return new Promise<R>((resolve, reject) => {
       let t: ReturnType<typeof setTimeout>;
       this.#commandQueue.set(id, v => {
         clearTimeout(t);
-        const cmdReply = v as NostrSystemMessage<R>;
+        const cmdReply = v as WorkerMessage<R>;
-        if (cmdReply.type === NostrSystemCommand.OkResponse) {
+        if (cmdReply.type === WorkerCommand.OkResponse) {
           resolve(cmdReply.data);
         } else {
           reject(cmdReply.data);