Merge pull request #384 from v0l/nostr-package-event-emitter-api

`nostr` package: use `EventEmitter`

packages/nostr/package.json

@@ -10,6 +10,7 @@
     "lint": "eslint ."
   },
   "devDependencies": {
+    "@types/events": "^3.0.0",
     "@types/expect": "^24.3.0",
     "@types/mocha": "^10.0.1",
     "@typescript-eslint/eslint-plugin": "^5.53.0",
@@ -25,6 +26,7 @@
   "dependencies": {
     "@noble/secp256k1": "^1.7.1",
     "bech32": "^2.0.0",
+    "events": "^3.3.0",
     "isomorphic-ws": "^5.0.0",
     "ws": "^8.12.1"
   }

packages/nostr/src/client/conn.ts

@@ -47,14 +47,7 @@ export class Conn {
         const msg = await parseIncomingMessage(value)
         this.#msgCallback?.(msg)
       } catch (err) {
-        if (err instanceof ProtocolError) {
+        this.#errorCallback?.(err)
-          this.#errorCallback?.(err)
-        } else {
-          // TODO Not sure if this is the best idea.
-          // Investigate what WebSocket does if the callback throws?
-          // Either way it seems like the best idea is to have `onError` called on all types of errors
-          throw err
-        }
       }
     })
 
@@ -65,6 +58,10 @@ export class Conn {
       }
       this.#pending = []
     })
+
+    this.#socket.addEventListener("error", (err) => {
+      this.#errorCallback?.(err)
+    })
   }
 
   on(on: "message", cb: IncomingMessageCallback): void
@@ -84,11 +81,23 @@ export class Conn {
       this.#pending.push(msg)
       return
     }
-    this.#socket.send(serializeOutgoingMessage(msg))
+    try {
+      this.#socket.send(serializeOutgoingMessage(msg), (err) => {
+        if (err !== undefined && err !== null) {
+          this.#errorCallback?.(err)
+        }
+      })
+    } catch (err) {
+      this.#errorCallback?.(err)
+    }
   }
 
   close(): void {
-    this.#socket.close()
+    try {
+      this.#socket.close()
+    } catch (err) {
+      this.#errorCallback?.(err)
+    }
   }
 }
 
@@ -160,7 +169,7 @@ export interface OutgoingCloseSubscription {
 }
 
 type IncomingMessageCallback = (message: IncomingMessage) => unknown
-type ErrorCallback = (error: ProtocolError) => unknown
+type ErrorCallback = (error: unknown) => unknown
 
 interface RawFilters {
   ids?: string[]

packages/nostr/src/client/emitter.ts

@@ -0,0 +1,145 @@
+import Base from "events"
+import { EventParams, Nostr } from "."
+
+/**
+ * Overrides providing better types for EventEmitter methods.
+ */
+export class EventEmitter extends Base {
+  override addListener(eventName: "newListener", listener: NewListener): this
+  override addListener(
+    eventName: "removeListener",
+    listener: RemoveListener
+  ): this
+  override addListener(eventName: "notice", listener: NoticeListener): this
+  override addListener(eventName: "error", listener: ErrorListener): this
+  override addListener(eventName: "newListener", listener: ErrorListener): this
+  override addListener(eventName: EventName, listener: Listener): this {
+    return super.addListener(eventName, listener)
+  }
+
+  override emit(eventName: "newListener", listener: NewListener): boolean
+  override emit(eventName: "removeListener", listener: RemoveListener): boolean
+  override emit(eventName: "event", params: EventParams, nostr: Nostr): boolean
+  override emit(eventName: "notice", notice: string, nostr: Nostr): boolean
+  override emit(eventName: "error", err: unknown, nostr: Nostr): boolean
+  override emit(eventName: EventName, ...args: unknown[]): boolean {
+    return super.emit(eventName, ...args)
+  }
+
+  override eventNames(): EventName[] {
+    return super.eventNames() as EventName[]
+  }
+
+  override listeners(eventName: "newListener"): EventListener[]
+  override listeners(eventName: "removeListener"): EventListener[]
+  override listeners(eventName: "event"): EventListener[]
+  override listeners(eventName: "notice"): NoticeListener[]
+  override listeners(eventName: "error"): ErrorListener[]
+  override listeners(eventName: EventName): Listener[] {
+    return super.listeners(eventName) as Listener[]
+  }
+
+  override off(eventName: "newListener", listener: NewListener): this
+  override off(eventName: "removeListener", listener: RemoveListener): this
+  override off(eventName: "event", listener: EventListener): this
+  override off(eventName: "notice", listener: NoticeListener): this
+  override off(eventName: "error", listener: ErrorListener): this
+  override off(eventName: string, listener: Listener): this {
+    return super.off(eventName, listener)
+  }
+
+  override on(eventName: "newListener", listener: NewListener): this
+  override on(eventName: "removeListener", listener: RemoveListener): this
+  override on(eventName: "event", listener: EventListener): this
+  override on(eventName: "notice", listener: NoticeListener): this
+  override on(eventName: "error", listener: ErrorListener): this
+  override on(eventName: EventName, listener: Listener): this {
+    return super.on(eventName, listener)
+  }
+
+  override once(eventName: "newListener", listener: NewListener): this
+  override once(eventName: "removeListener", listener: RemoveListener): this
+  override once(eventName: "event", listener: EventListener): this
+  override once(eventName: "notice", listener: NoticeListener): this
+  override once(eventName: "error", listener: ErrorListener): this
+  override once(eventName: EventName, listener: Listener): this {
+    return super.once(eventName, listener)
+  }
+
+  override prependListener(
+    eventName: "newListener",
+    listener: NewListener
+  ): this
+  override prependListener(
+    eventName: "removeListener",
+    listener: RemoveListener
+  ): this
+  override prependListener(eventName: "event", listener: EventListener): this
+  override prependListener(eventName: "notice", listener: NoticeListener): this
+  override prependListener(eventName: "error", listener: ErrorListener): this
+  override prependListener(eventName: EventName, listener: Listener): this {
+    return super.prependListener(eventName, listener)
+  }
+
+  override prependOnceListener(
+    eventName: "newListener",
+    listener: NewListener
+  ): this
+  override prependOnceListener(
+    eventName: "removeListener",
+    listener: RemoveListener
+  ): this
+  override prependOnceListener(
+    eventName: "event",
+    listener: EventListener
+  ): this
+  override prependOnceListener(
+    eventName: "notice",
+    listener: NoticeListener
+  ): this
+  override prependOnceListener(
+    eventName: "error",
+    listener: ErrorListener
+  ): this
+  override prependOnceListener(eventName: EventName, listener: Listener): this {
+    return super.prependOnceListener(eventName, listener)
+  }
+
+  override removeAllListeners(event?: EventName): this {
+    return super.removeAllListeners(event)
+  }
+
+  override removeListener(eventName: "newListener", listener: NewListener): this
+  override removeListener(
+    eventName: "removeListener",
+    listener: RemoveListener
+  ): this
+  override removeListener(eventName: "event", listener: EventListener): this
+  override removeListener(eventName: "notice", listener: NoticeListener): this
+  override removeListener(eventName: "error", listener: ErrorListener): this
+  override removeListener(eventName: EventName, listener: Listener): this {
+    return super.removeListener(eventName, listener)
+  }
+
+  override rawListeners(eventName: EventName): Listener[] {
+    return super.rawListeners(eventName) as Listener[]
+  }
+
+  // TODO
+  // emitter[Symbol.for('nodejs.rejection')](err, eventName[, ...args]) shenanigans?
+}
+
+// TODO Also add on: ("subscribed", subscriptionId) which checks "OK"/"NOTICE" and makes a callback?
+// TODO Also add on: ("ok", boolean, eventId) which checks "OK"/"NOTICE" and makes a callback?
+type EventName = "newListener" | "removeListener" | "event" | "notice" | "error"
+type NewListener = (eventName: EventName, listener: Listener) => void
+type RemoveListener = (eventName: EventName, listener: Listener) => void
+type EventListener = (params: EventParams, nostr: Nostr) => void
+type NoticeListener = (notice: string, nostr: Nostr) => void
+type ErrorListener = (error: unknown, nostr: Nostr) => void
+type Listener =
+  | NewListener
+  | RemoveListener
+  | EventListener
+  | NoticeListener
+  | ErrorListener

packages/nostr/src/client/index.ts

@@ -3,11 +3,14 @@ import { EventId, Event, EventKind, SignedEvent, RawEvent } from "../event"
 import { PrivateKey, PublicKey } from "../keypair"
 import { Conn, IncomingKind, OutgoingKind } from "./conn"
 import * as secp from "@noble/secp256k1"
+import { EventEmitter } from "./emitter"
 
 /**
  * A nostr client.
+ *
+ * TODO Document the events here
  */
-export class Nostr {
+export class Nostr extends EventEmitter {
   // TODO NIP-44 AUTH, leave this for later
   /**
    * Open connections to relays.
@@ -19,39 +22,6 @@ export class Nostr {
    */
   readonly #subscriptions: Map<string, Filters[]> = new Map()
 
-  #eventCallback?: EventCallback
-  #noticeCallback?: NoticeCallback
-  #errorCallback?: ErrorCallback
-
-  /**
-   * Add a new callback for received events.
-   */
-  on(on: "event", cb: EventCallback | undefined | null): void
-  /**
-   * Add a new callback for received notices.
-   */
-  on(on: "notice", cb: NoticeCallback | undefined | null): void
-  /**
-   * Add a new callback for errors.
-   */
-  on(on: "error", cb: ErrorCallback | undefined | null): void
-  // TODO Also add on: ("subscribed", subscriptionId) which checks "OK"/"NOTICE" and makes a callback?
-  // TODO Also add on: ("sent", eventId) which checks "OK"/"NOTICE" and makes a callback?
-  on(
-    on: "event" | "notice" | "error",
-    cb: EventCallback | NoticeCallback | ErrorCallback | undefined | null
-  ) {
-    if (on === "event") {
-      this.#eventCallback = (cb as EventCallback) ?? undefined
-    } else if (on === "notice") {
-      this.#noticeCallback = (cb as NoticeCallback) ?? undefined
-    } else if (on === "error") {
-      this.#errorCallback = (cb as ErrorCallback) ?? undefined
-    } else {
-      throw new Error(`unexpected input: ${on}`)
-    }
-  }
-
   /**
    * Open a connection and start communicating with a relay. This method recreates all existing
    * subscriptions on the new relay as well. If there is already an existing connection,
@@ -81,26 +51,30 @@ export class Nostr {
 
     // Handle messages on this connection.
     conn.on("message", async (msg) => {
-      if (msg.kind === IncomingKind.Event) {
+      try {
-        this.#eventCallback?.(
+        if (msg.kind === IncomingKind.Event) {
-          {
+          this.emit(
-            signed: msg.signed,
+            "event",
-            subscriptionId: msg.subscriptionId,
+            {
-            raw: msg.raw,
+              signed: msg.signed,
-          },
+              subscriptionId: msg.subscriptionId,
-          this
+              raw: msg.raw,
-        )
+            },
-      } else if (msg.kind === IncomingKind.Notice) {
+            this
-        this.#noticeCallback?.(msg.notice, this)
+          )
-      } else {
+        } else if (msg.kind === IncomingKind.Notice) {
-        const err = new ProtocolError(`invalid message ${msg}`)
+          this.emit("notice", msg.notice, this)
-        this.#errorCallback?.(err, this)
+        } else {
+          throw new ProtocolError(`invalid message ${msg}`)
+        }
+      } catch (err) {
+        this.emit("error", err, this)
       }
     })
 
     // Forward connection errors to the error callbacks.
     conn.on("error", (err) => {
-      this.#errorCallback?.(err, this)
+      this.emit("error", err, this)
     })
 
     // Resend existing subscriptions to this connection.
@@ -278,6 +252,7 @@ export class SubscriptionId {
   }
 }
 
+// TODO Rethink this type. Maybe it's not very idiomatic.
 /**
  * A prefix filter. These filters match events which have the appropriate prefix.
  * This also means that exact matches pass the filters. No special syntax is required.
@@ -315,10 +290,7 @@ export interface Filters {
   limit?: number
 }
 
-export type EventCallback = (params: EventParams, nostr: Nostr) => unknown
+// TODO Document this
-export type NoticeCallback = (notice: string, nostr: Nostr) => unknown
-export type ErrorCallback = (error: ProtocolError, nostr: Nostr) => unknown
-
 export interface EventParams {
   signed: SignedEvent
   subscriptionId: SubscriptionId

packages/nostr/test/simple-communication.ts

@@ -1,4 +1,4 @@
-import { Nostr } from "../src/client"
+import { EventParams, Nostr } from "../src/client"
 import { EventKind } from "../src/event"
 import { PrivateKey } from "../src/keypair"
 import assert from "assert"
@@ -20,7 +20,7 @@ describe("single event communication", function () {
     const subscriber = new Nostr()
     subscriber.open("ws://localhost:12648")
 
-    subscriber.on("event", ({ signed: { event } }) => {
+    function listener({ signed: { event } }: EventParams) {
       assert.equal(event.kind, EventKind.TextNote)
       assert.equal(event.pubkey.toString(), pubkey.toString())
       assert.equal(event.createdAt.toString(), timestamp.toString())
@@ -32,13 +32,15 @@ describe("single event communication", function () {
       // subscribe happen at the same time, the same event might end up being broadcast twice.
       // To prevent reacting to the same event and calling done() twice, remove the callback
       // for future events.
-      subscriber.on("event", null)
+      subscriber.off("event", listener)
 
       publisher.close()
       subscriber.close()
 
       done()
-    })
+    }
+
+    subscriber.on("event", listener)
 
     subscriber.subscribe([])
 

packages/nostr/tsconfig.json

@@ -8,7 +8,8 @@
     "outDir": "dist",
     "moduleResolution": "node",
     "esModuleInterop": true,
-    "skipLibCheck": true
+    "skipLibCheck": true,
+    "noImplicitOverride": true
   },
   "include": ["src"]
 }

yarn.lock

@@ -2059,6 +2059,11 @@
   resolved "https://registry.yarnpkg.com/@types/estree/-/estree-0.0.51.tgz#cfd70924a25a3fd32b218e5e420e6897e1ac4f40"
   integrity sha512-CuPgU6f3eT/XgKKPqKd/gLZV1Xmvf1a2R5POBOGQa6uv82xpls89HU5zKeVoyR8XzHd1RGNOlQlvUe3CFkjWNQ==
 
+"@types/events@^3.0.0":
+  version "3.0.0"
+  resolved "https://registry.yarnpkg.com/@types/events/-/events-3.0.0.tgz#2862f3f58a9a7f7c3e78d79f130dd4d71c25c2a7"
+  integrity sha512-EaObqwIvayI5a8dCzhFrjKzVwKLxjoG9T6Ppd5CEo07LRKfQ8Yokw54r5+Wq7FaBQ+yXRvQAYPrHwya1/UFt9g==
+
 "@types/expect@^24.3.0":
   version "24.3.0"
   resolved "https://registry.yarnpkg.com/@types/expect/-/expect-24.3.0.tgz#d7cab8b3c10c2d92a0cbb31981feceb81d3486f1"
@@ -4707,7 +4712,7 @@ eventemitter3@^4.0.0:
   resolved "https://registry.yarnpkg.com/eventemitter3/-/eventemitter3-4.0.7.tgz#2de9b68f6528d5644ef5c59526a1b4a07306169f"
   integrity sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==
 
-events@^3.2.0:
+events@^3.2.0, events@^3.3.0:
   version "3.3.0"
   resolved "https://registry.yarnpkg.com/events/-/events-3.3.0.tgz#31a95ad0a924e2d2c419a813aeb2c4e878ea7400"
   integrity sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==