feat(client): add method

Author: hasundue

client.ts

@@ -1,3 +1,4 @@
+import type { PromiseCallbacks } from "./core/types.ts";
 import type {
   ClientToRelayMessage,
   EoseMessage,
@@ -16,6 +17,8 @@ import { NonExclusiveWritableStream } from "./core/streams.ts";
 import { LazyWebSocket } from "./core/websockets.ts";
 import { Lock } from "./core/x/async.ts";
 
+export class EventRejectionError extends Error {}
+
 /**
  * A class that represents a remote Nostr Relay.
  */
@@ -28,7 +31,8 @@ export class Relay extends NostrNode<ClientToRelayMessage> {
     Lock<WritableStreamDefaultWriter<EventMessage | EoseMessage>>
   >();
 
-  readonly #published = new Map<EventId, PromiseResolveCallback<OkMessage>>();
+  readonly #publisher = this.getWriter();
+  readonly #published = new Map<EventId, PromiseCallbacks<OkMessage>>();
 
   constructor(
     init: RelayUrl | RelayInit,
@@ -58,21 +62,40 @@ export class Relay extends NostrNode<ClientToRelayMessage> {
       }
       if (type === "OK") {
         const [, eid] = msg;
-        const resolve = this.#published.get(eid);
-        if (!resolve) {
-          opts?.logger?.warn?.(type, this.config.name, "Unknown event id");
+        const callbacks = this.#published.get(eid);
+        if (!callbacks) {
+          opts?.logger?.warn?.(type, this.config.name, "Unknown event id", eid);
           return;
         }
-        return resolve(msg);
+        return callbacks.resolve(msg);
       }
       if (type === "EVENT" || type === "EOSE") {
         const [, sid] = msg;
-        return this._notify(sid, msg);
+        return this.#notify(sid, msg);
       }
       opts?.logger?.warn?.(type, this.config.name, "Unknown message type");
     });
   }
 
+  async #notify(
+    sid: SubscriptionId,
+    msg: EventMessage | EoseMessage,
+  ) {
+    const messenger = new Lock(this.getWriter());
+    const sub = this.#subs.get(sid);
+    const promise = sub
+      ? sub.lock((writer) => writer.write(msg))
+      // Subscription is already closed. TODO: should we throw an error?
+      : messenger.lock((writer) => writer.write(["CLOSE", sid]));
+    await promise.catch((err) => {
+      if (err instanceof TypeError) {
+        // Stream is already closing or closed.
+        return;
+      }
+      throw err;
+    });
+  }
+
   subscribe<K extends EventKind>(
     filter: SubscriptionFilter<K> | SubscriptionFilter<K>[],
     opts: Partial<SubscriptionOptions> = {},
@@ -140,15 +163,17 @@ export class Relay extends NostrNode<ClientToRelayMessage> {
   }
 
   async publish(event: NostrEvent): Promise<void> {
-    const writer = this.getWriter();
-    await writer.ready;
-    await writer.write(["EVENT", event]);
-    writer.releaseLock();
-    const [, , accepted, body] = await new Promise<OkMessage>((resolve) => {
-      this.#published.set(event.id, resolve);
-    });
+    await this.#publisher.ready;
+    // We found this blocks for a long time, so we don't await it.
+    this.#publisher.write(["EVENT", event]);
+    const [, , accepted, body] = await new Promise<OkMessage>(
+      (resolve, reject) => {
+        this.#published.set(event.id, { resolve, reject });
+      },
+    );
+    await this.#publisher.ready;
     if (!accepted) {
-      throw new Error(`Event rejected: ${body}`, { cause: event });
+      throw new EventRejectionError(body, { cause: event });
     }
     this.config.logger?.debug?.("OK", this.config.name, body);
   }
@@ -159,27 +184,13 @@ export class Relay extends NostrNode<ClientToRelayMessage> {
         sub.lock((writer) => writer.close())
       ),
     );
+    await Promise.all(
+      Array.from(this.#published.values()).map(({ reject }) =>
+        reject("Relay closed")
+      ),
+    );
     await super.close();
   }
-
-  private async _notify(
-    sid: SubscriptionId,
-    msg: EventMessage | EoseMessage,
-  ) {
-    const messenger = new Lock(this.getWriter());
-    const sub = this.#subs.get(sid);
-    const promise = sub
-      ? sub.lock((writer) => writer.write(msg))
-      // Subscription is already closed. TODO: should we throw an error?
-      : messenger.lock((writer) => writer.write(["CLOSE", sid]));
-    await promise.catch((err) => {
-      if (err instanceof TypeError) {
-        // Stream is already closing or closed.
-        return;
-      }
-      throw err;
-    });
-  }
 }
 
 export type RelayConfig = NostrNodeConfig & {
@@ -205,5 +216,3 @@ export interface RelayLike
   extends NonExclusiveWritableStream<ClientToRelayMessage> {
   subscribe: Relay["subscribe"];
 }
-
-type PromiseResolveCallback<T> = (value: T | PromiseLike<T>) => void;

core/types.ts

@@ -18,6 +18,14 @@ export type Optional<T, K extends keyof T> = Expand<
   & Partial<Pick<T, K>>
 >;
 
+//
+// Promises
+//
+export type PromiseCallbacks<T> = {
+  resolve: (value: T | PromiseLike<T>) => void;
+  reject: (reason?: unknown) => void;
+};
+
 // Logger
 export interface Logger {
   debug?: (...args: unknown[]) => void;

deno.lock

@@ -3,3 +3,5 @@ export { assertArrayIncludes } from "https://deno.land/std@0.203.0/assert/assert
 export { assertEquals } from "https://deno.land/std@0.203.0/assert/assert_equals.ts";
 export { assertFalse } from "https://deno.land/std@0.203.0/assert/assert_false.ts";
 export { assertObjectMatch } from "https://deno.land/std@0.203.0/assert/assert_object_match.ts";
+export { assertThrows } from "https://deno.land/std@0.203.0/assert/assert_throws.ts";
+export { assertRejects } from "https://deno.land/std@0.203.0/assert/assert_rejects.ts";

lib/std/assert.ts

@@ -11,7 +11,7 @@ export class MockWebSocket extends EventTarget implements WebSocket {
     this.url = url?.toString() ?? "";
     this.protocol = protocols ? [...protocols].flat()[0] : "";
     MockWebSocket.#instances.push(this);
-    // Simulate a slow opening of a WebSocket as much as possible.
+    // Simulate async behavior of WebSocket as much as possible.
     queueMicrotask(() => {
       this.#readyState = 1;
       this.dispatchEvent(new Event("open"));
@@ -45,7 +45,7 @@ export class MockWebSocket extends EventTarget implements WebSocket {
 
   close(code?: number, reason?: string): void {
     this.#readyState = 2;
-    // Simulate a slow closing of a WebSocket as much as possible.
+    // Simulate async behavior of WebSocket as much as possible.
     queueMicrotask(() => {
       if (this.#remote) {
         this.#remote.#readyState = 3;
@@ -57,7 +57,10 @@ export class MockWebSocket extends EventTarget implements WebSocket {
   }
 
   send(data: MessageEventData): void {
-    this.#remote?.dispatchEvent(new MessageEvent("message", { data }));
+    // Simulate async behavior of WebSocket as much as possible.
+    queueMicrotask(() =>
+      this.#remote?.dispatchEvent(new MessageEvent("message", { data }))
+    );
   }
 
   onclose = null;

lib/testing.ts

@@ -1,7 +1,17 @@
-import { NostrEvent } from "../core/nips/01.ts";
+import {
+  EventId,
+  NostrEvent,
+  OkMessage,
+  PublishMessage,
+} from "../core/nips/01.ts";
 import { Relay } from "../client.ts";
 import { afterAll, beforeAll, describe, it } from "../lib/std/testing.ts";
-import { assert, assertEquals, assertObjectMatch } from "../lib/std/assert.ts";
+import {
+  assert,
+  assertEquals,
+  assertObjectMatch,
+  assertRejects,
+} from "../lib/std/assert.ts";
 import { MockWebSocket } from "../lib/testing.ts";
 
 const url = "wss://localhost:8080";
@@ -13,19 +23,15 @@ describe("Relay constructor", () => {
     beforeAll(() => {
       relay = new Relay(url);
     });
-
     it("should be constructable", () => {
       assert(relay instanceof Relay);
     });
-
     it("should have a url", () => {
       assertEquals(relay.config.url, url);
     });
-
     it("should have a name", () => {
       assertEquals(relay.config.name, "localhost:8080");
     });
-
     it("should have default options", () => {
       assertObjectMatch(relay.config, {
         nbuffer: 10,
@@ -37,7 +43,6 @@ describe("Relay constructor", () => {
 
   describe("called with url and options", () => {
     const logger = { info: () => {} };
-
     beforeAll(() => {
       relay = new Relay(url, {
         name: "test",
@@ -47,15 +52,12 @@ describe("Relay constructor", () => {
         logger,
       });
     });
-
-    afterAll(async () => {
-      await relay.close();
+    afterAll(() => {
+      relay.close();
     });
-
     it("should be constructable", () => {
       assert(relay instanceof Relay);
     });
-
     it("should have the given options", () => {
       assertObjectMatch(relay.config, {
         name: "test",
@@ -79,21 +81,18 @@ describe("Relay", () => {
     globalThis.WebSocket = MockWebSocket;
     relay = new Relay(url);
   });
-
   afterAll(() => {
     relay.close();
   });
 
   it("should not be connected initially", () => {
     assertEquals(relay.status, WebSocket.CLOSED);
   });
-
   it("should not connect when a subscription is created", () => {
     sub_1 = relay.subscribe({ kinds: [1] }, { id: "test-1" });
     assert(sub_1 instanceof ReadableStream);
     assertEquals(relay.status, WebSocket.CLOSED);
   });
-
   it("should receive text notes", async () => {
     const reader = sub_1.getReader();
     const read = reader.read();
@@ -108,15 +107,13 @@ describe("Relay", () => {
     assertEquals(value.kind, 1);
     reader.releaseLock();
   });
-
   it("should be able to open multiple subscriptions", () => {
     sub_0 = relay.subscribe({ kinds: [0], limit: 1 }, { id: "test-0" });
     assert(sub_0 instanceof ReadableStream);
   });
-
   it("should recieve metas and notes simultaneously", async () => {
-    const read_0 = sub_0.getReader().read();
-    const read_1 = sub_1.getReader().read();
+    const reader_0 = sub_0.getReader();
+    const reader_1 = sub_1.getReader();
     const ws = MockWebSocket.instances[0];
     ws.dispatchEvent(
       new MessageEvent("message", {
@@ -129,12 +126,58 @@ describe("Relay", () => {
       }),
     );
     const [{ value: value_0 }, { value: value_1 }] = await Promise.all([
-      read_0,
-      read_1,
+      reader_0.read(),
+      reader_1.read(),
     ]);
     assert(value_0);
     assertEquals(value_0.kind, 0);
     assert(value_1);
     assertEquals(value_1.kind, 1);
+    reader_0.releaseLock();
+    reader_1.releaseLock();
+  });
+  it("should publish an event and recieve an accepting OK message", async () => {
+    const eid = "test-true" as EventId;
+    const ok = ["OK", eid, true, ""] satisfies OkMessage<true>;
+    const ws = MockWebSocket.instances[0];
+    const arrived = new Promise<true>((resolve) => {
+      ws.remote.addEventListener(
+        "message",
+        (ev: MessageEvent<string>) => {
+          const [, event] = JSON.parse(ev.data) as PublishMessage<1>;
+          if (event.id === eid) {
+            assertObjectMatch(event, { kind: 1 });
+            resolve(true);
+            ws.remote.send(JSON.stringify(ok));
+          }
+        },
+      );
+    });
+    const event = { id: eid, kind: 1 };
+    // deno-lint-ignore no-explicit-any
+    await relay.publish(event as any);
+    assert(await arrived);
+  });
+  it("should receieve a rejecting OK message", async () => {
+    const eid = "test-false" as EventId;
+    const ok = ["OK", eid, false, "error: test"] satisfies OkMessage<false>;
+    const ws = MockWebSocket.instances[0];
+    const arrived = new Promise<true>((resolve) => {
+      ws.remote.addEventListener(
+        "message",
+        (ev: MessageEvent<string>) => {
+          const [, event] = JSON.parse(ev.data) as PublishMessage<1>;
+          if (event.id === eid) {
+            assertEquals(event.kind, 1);
+            resolve(true);
+            ws.remote.send(JSON.stringify(ok));
+          }
+        },
+      );
+    });
+    const event = { id: eid, kind: 1 };
+    // deno-lint-ignore no-explicit-any
+    assertRejects(() => relay.publish(event as any));
+    await arrived;
   });
 });