Merges pull request #48

Introduce WebSocketPool for Connection Management

.cursor/rules/alexandria.mdc

@@ -19,6 +19,12 @@ In reader mode, Alexandria loads a document tree from a root publication index e
 
 Svelte components in Alexandria use TypeScript exclusively over plain JavaScript.  Styles are defined via Tailwind 4 utility classes, and some custom utility classes are defined in [app.css](mdc:src/app.css).  The app runs on Deno, but maintains compatibility with Node.js.
 
+### Utilities
+
+The project contains a number of modules that define utility classes and functions to support the app's operations.  Make use of these utilities when they are relevant to reduce code duplication and maintain common patterns:
+
+- Use the `WebSocketPool` class defined in [websocket_pool.ts](../src/lib/data_structures/websocket_pool.ts) when handling raw WebSockets to efficiently manage connections.
+
 ## General Guidelines
 
 When responding to prompts, adhere to the following rules:
@@ -49,7 +55,7 @@ Observe the following style guidelines when writing code:
 - Use blocks enclosed by curly brackets when writing control flow expressions such as `for` and `while` loops, and `if` and `switch` statements.
 - Begin `case` expressions in a `switch` statement at the same indentation level as the `switch` itself.  Indent code within a `case` block.
 - Limit line length to 100 characters; break statements across lines if necessary.
-- Default to single quotes.
+- Default to double quotes.
 
 ### HTML
 
@@ -57,5 +63,3 @@ Observe the following style guidelines when writing code:
 - Break long tags across multiple lines.
 - Use Tailwind 4 utility classes for styling.
 - Default to single quotes.
-
-

deno.lock

@@ -3,7 +3,6 @@
   "specifiers": {
     "npm:@noble/curves@^1.9.4": "1.9.4",
     "npm:@noble/hashes@^1.8.0": "1.8.0",
-    "npm:@noble/secp256k1@^2.3.0": "2.3.0",
     "npm:@nostr-dev-kit/ndk-cache-dexie@2.6": "2.6.33_nostr-tools@2.15.1__typescript@5.8.3_typescript@5.8.3",
     "npm:@nostr-dev-kit/ndk-cache-dexie@^2.6.33": "2.6.33_nostr-tools@2.15.1__typescript@5.8.3_typescript@5.8.3",
     "npm:@nostr-dev-kit/ndk@^2.14.32": "2.14.32_nostr-tools@2.15.1__typescript@5.8.3_typescript@5.8.3",

src/lib/components/EventInput.svelte

@@ -3,7 +3,6 @@
     getTitleTagForEvent,
     getDTagForEvent,
     requiresDTag,
-    hasDTag,
     validateNotAsciidoc,
     validateAsciiDoc,
     build30040EventSet,
@@ -22,8 +21,8 @@
   import { prefixNostrAddresses } from "$lib/utils/nostrUtils";
   import { activeInboxRelays, activeOutboxRelays } from "$lib/ndk";
   import { Button } from "flowbite-svelte";
-  import { nip19 } from "nostr-tools";
   import { goto } from "$app/navigation";
+  import { WebSocketPool } from "$lib/data_structures/websocket_pool";
 
   let kind = $state<number>(30023);
   let tags = $state<[string, string][]>([]);
@@ -308,17 +307,14 @@
 
           for (const relayUrl of relays) {
             try {
-              const ws = new WebSocket(relayUrl);
+              const ws = await WebSocketPool.instance.acquire(relayUrl);
+
               await new Promise<void>((resolve, reject) => {
                 const timeout = setTimeout(() => {
-                  ws.close();
+                  WebSocketPool.instance.release(ws);
                   reject(new Error("Timeout"));
                 }, 5000);
 
-                ws.onopen = () => {
-                  ws.send(JSON.stringify(["EVENT", signedEvent]));
-                };
-
                 ws.onmessage = (e) => {
                   const [type, id, ok, message] = JSON.parse(e.data);
                   if (type === "OK" && id === signedEvent.id) {
@@ -326,20 +322,14 @@
                     if (ok) {
                       published = true;
                       relaysPublished.push(relayUrl);
-                      ws.close();
+                      WebSocketPool.instance.release(ws);
                       resolve();
                     } else {
-                      ws.close();
+                      WebSocketPool.instance.release(ws);
                       reject(new Error(message));
                     }
                   }
                 };
-
-                ws.onerror = () => {
-                  clearTimeout(timeout);
-                  ws.close();
-                  reject(new Error("WebSocket error"));
-                };
               });
               if (published) break;
             } catch (e) {

src/lib/data_structures/websocket_pool.ts

@@ -0,0 +1,338 @@
+/**
+ * A representation of a WebSocket connection used internally by the WebSocketPool. Attaches
+ * information about the connection's state within the resource pool to the connection itself.
+ */
+interface WebSocketHandle {
+  ws: WebSocket;
+  refCount: number;
+  idleTimer?: ReturnType<typeof setTimeout>;
+}
+
+interface WebSocketPoolWaitingQueueItem {
+  url: string;
+  resolve: (handle: WebSocketHandle) => void;
+  reject: (reason?: any) => void;
+}
+
+/**
+ * A resource pool for WebSocket connections. Its purpose is to allow multiple requestors to share
+ * an open WebSocket connection for greater resource efficiency.
+ *
+ * The pool maintains a single connection per URL. Requestors may acquire a reference to a
+ * connection, and are expected to release it when it is no longer needed. If the number of
+ * requestors using a connection drops to zero and no new requestors acquire it for a set period,
+ * the connection is closed.
+ */
+export class WebSocketPool {
+  static #shared: WebSocketPool;
+
+  /**
+   * A map of WebSocket URLs to the handles containing the connection and its state.
+   */
+  #pool: Map<string, WebSocketHandle> = new Map();
+
+  #idleTimeoutMs: number;
+  #maxConnections: number;
+  #waitingQueue: WebSocketPoolWaitingQueueItem[] = [];
+
+  /**
+   * Private constructor invoked when the singleton instance is first created.
+   * @param idleTimeoutMs - The timeout in milliseconds after which idle connections will be
+   * closed. Defaults to 60 seconds.
+   * @param maxConnections - The maximum number of simultaneous WebSocket connections. Defaults to
+   * 16.
+   */
+  private constructor(idleTimeoutMs: number = 60000, maxConnections: number = 16) {
+    this.#idleTimeoutMs = idleTimeoutMs;
+    this.#maxConnections = maxConnections;
+  }
+
+  /**
+   * Returns the singleton instance of the WebsocketPool.
+   * @returns The singleton instance.
+   */
+  public static get instance(): WebSocketPool {
+    if (!WebSocketPool.#shared) {
+      WebSocketPool.#shared = new WebSocketPool();
+    }
+    return WebSocketPool.#shared;
+  }
+
+  // #region Resource Management Interface
+
+  /**
+   * Sets the maximum number of simultaneous WebSocket connections.
+   *
+   * @param limit - The new connection limit.
+   */
+  public set maxConnections(limit: number) {
+    if (limit === this.#maxConnections) {
+      return;
+    }
+
+    if (limit == null || isNaN(limit)) {
+      throw new Error('[WebSocketPool] Connection limit must be a number.');
+    }
+
+    if (limit <= 0) {
+      throw new Error('[WebSocketPool] Connection limit must be greater than 0.');
+    }
+
+    if (!Number.isInteger(limit)) {
+      throw new Error('[WebSocketPool] Connection limit must be an integer.');
+    }
+
+    this.#maxConnections = limit;
+    this.#processWaitingQueue();
+  }
+
+  /**
+   * Gets the current maximum number of simultaneous WebSocket connections.
+   *
+   * @returns The current connection limit.
+   */
+  public get maxConnections(): number {
+    return this.#maxConnections;
+  }
+
+  /**
+   * Sets the idle timeout for WebSocket connections.
+   *
+   * @param timeoutMs - The timeout in milliseconds after which idle connections will be closed.
+   */
+  public set idleTimeoutMs(timeoutMs: number) {
+    if (timeoutMs === this.#idleTimeoutMs) {
+      return;
+    }
+
+    if (timeoutMs == null || isNaN(timeoutMs)) {
+      throw new Error('[WebSocketPool] Idle timeout must be a number.');
+    }
+
+    if (timeoutMs <= 0) {
+      throw new Error('[WebSocketPool] Idle timeout must be greater than 0.');
+    }
+
+    if (!Number.isInteger(timeoutMs)) {
+      throw new Error('[WebSocketPool] Idle timeout must be an integer.');
+    }
+
+    this.#idleTimeoutMs = timeoutMs;
+  }
+
+  /**
+   * Gets the current idle timeout setting.
+   *
+   * @returns The current idle timeout in milliseconds.
+   */
+  public get idleTimeoutMs(): number {
+    return this.#idleTimeoutMs;
+  }
+
+  /**
+   * Acquires a WebSocket connection for the specified URL. If a connection is available in the
+   * pool, that connection is returned. If no connection is available but the pool is not full, a
+   * new connection is created and returned. If the pool is full, the request is queued until a
+   * connection is released, at which point the newly-available connection is returned to the
+   * caller.
+   *
+   * @param url - The URL to connect to.
+   * @returns A promise that resolves with a WebSocket connection.
+   */
+  public async acquire(url: string): Promise<WebSocket> {
+    const normalizedUrl = this.#normalizeUrl(url);
+    const handle = this.#pool.get(normalizedUrl);
+
+    try {
+      if (handle && handle.ws.readyState === WebSocket.OPEN) {
+        this.#checkOut(handle);
+        return handle.ws;
+      }
+
+      if (this.#pool.size >= this.#maxConnections) {
+        return new Promise((resolve, reject) => {
+          this.#waitingQueue.push({ 
+            url: normalizedUrl, 
+            resolve: (handle) => resolve(handle.ws), 
+            reject,
+          });
+        });
+      }
+
+      const newHandle = await this.#createSocket(normalizedUrl);
+      return newHandle.ws;
+    } catch (error) {
+      throw new Error(
+        `[WebSocketPool] Failed to acquire connection for ${normalizedUrl}: ${error}`
+      );
+    }
+  }
+
+  /**
+   * Releases a WebSocket connection back to the pool. If there are pending requests for the same
+   * URL, the connection is passed to the requestor in the queue. Otherwise, the connection is
+   * marked as available.
+   *
+   * @param handle - The WebSocketHandle to release.
+   */
+  public release(ws: WebSocket): void {
+    const normalizedUrl = this.#normalizeUrl(ws.url);
+    const handle = this.#pool.get(normalizedUrl);
+    if (!handle) {
+      throw new Error('[WebSocketPool] Attempted to release an unmanaged WebSocket connection.');
+    }
+
+    if (--handle.refCount === 0) {
+      this.#startIdleTimer(handle);
+    }
+  }
+
+  /**
+   * Closes all WebSocket connections and "drains" the pool.
+   */
+  public drain(): void {
+    // Clear all idle timers first
+    for (const handle of this.#pool.values()) {
+      this.#clearIdleTimer(handle);
+    }
+
+    for (const { reject } of this.#waitingQueue) {
+      reject(new Error('[WebSocketPool] Draining pool.'));
+    }
+    this.#waitingQueue = [];
+
+    for (const handle of this.#pool.values()) {
+      handle.ws.close();
+    }
+    this.#pool.clear();
+  }
+
+  // #endregion
+
+  // #region Private Helper Methods
+
+  #createSocket(url: string): Promise<WebSocketHandle> {
+    return new Promise((resolve, reject) => {
+      try {
+        const handle: WebSocketHandle = {
+          ws: new WebSocket(url),
+          refCount: 1,
+        };
+        handle.ws.onopen = () => {
+          this.#pool.set(url, handle);
+
+          // Remove the socket from the pool when it is closed. The socket may be closed by
+          // either the client or the server.
+          handle.ws.onclose = () => this.#removeSocket(handle);
+          resolve(handle);
+        };
+
+        handle.ws.onerror = (event) => {
+          this.#removeSocket(handle);
+          this.#processWaitingQueue();
+          reject(
+            new Error(`[WebSocketPool] WebSocket connection failed for ${url}: ${event.type}`)
+          );
+        };
+      } catch (error) {
+        this.#processWaitingQueue();
+        reject(error);
+      }
+    });
+  }
+
+  #removeSocket(handle: WebSocketHandle): void {
+    this.#clearIdleTimer(handle);
+    handle.ws.onopen = handle.ws.onerror = handle.ws.onclose = null;
+    this.#pool.delete(this.#normalizeUrl(handle.ws.url));
+    this.#processWaitingQueue();
+  }
+
+  /**
+   * Starts an idle timer for the specified WebSocket. The connection will be automatically
+   * closed after the idle timeout period if it remains unused.
+   *
+   * @param ws - The WebSocket for which to start the idle timer.
+   */
+  #startIdleTimer(handle: WebSocketHandle): void {
+    // Clear any existing timer first
+    this.#clearIdleTimer(handle);
+
+    handle.idleTimer = setTimeout(() => {
+      const refCount = handle.refCount;
+      if (refCount === 0 && handle.ws.readyState === WebSocket.OPEN) {
+        handle.ws.close();
+        this.#removeSocket(handle);
+      }
+    }, this.#idleTimeoutMs);
+  }
+
+  /**
+   * Clears the idle timer for the specified WebSocket.
+   *
+   * @param handle - The WebSocketHandle for which to clear the idle timer.
+   */
+  #clearIdleTimer(handle: WebSocketHandle): void {
+    const timer = handle.idleTimer;
+    if (timer) {
+      clearTimeout(timer);
+      handle.idleTimer = undefined;
+    }
+  }
+
+  /**
+   * Processes pending requests to acquire a connection. Reuses existing connections when possible.
+   */
+  #processWaitingQueue(): void {
+    while (
+      this.#waitingQueue.length > 0 &&
+      this.#pool.size < this.#maxConnections
+    ) {
+      const nextInQueue = this.#waitingQueue.shift();
+      if (!nextInQueue) {
+        continue;
+      }
+
+      const { url, resolve, reject } = nextInQueue;
+
+      const existingHandle = this.#pool.get(url);
+      if (existingHandle && existingHandle.ws.readyState === WebSocket.OPEN) {
+        this.#checkOut(existingHandle);
+        resolve(existingHandle);
+        return;
+      }
+
+      this.#createSocket(url).then(resolve, reject);
+    }
+  }
+
+  #checkOut(handle: WebSocketHandle): void {
+    if (handle.refCount == null) {
+      throw new Error('[WebSocketPool] Handle refCount unexpectedly null.');
+    }
+
+    ++handle.refCount;
+    this.#clearIdleTimer(handle);
+  }
+
+  #normalizeUrl(url: string): string {
+    try {
+      const urlObj = new URL(url);
+      // The URL constructor correctly normalizes scheme and hostname casing.
+      let normalized = urlObj.toString();
+
+      // The logic to remove a trailing slash for connection coalescing can be kept,
+      // but should be done on the normalized string.
+      if (urlObj.pathname !== '/' && normalized.endsWith('/')) {
+        normalized = normalized.slice(0, -1);
+      }
+      
+      return normalized;
+    } catch {
+      // If URL is invalid, return it as-is and let WebSocket constructor handle the error.
+      return url;
+    }
+  }
+
+  // #endregion
+}

src/lib/ndk.ts

@@ -21,6 +21,7 @@ export { testRelayConnection };
 import { userStore } from "./stores/userStore.ts";
 import { userPubkey } from "./stores/authStore.Svelte.ts";
 import { startNetworkStatusMonitoring, stopNetworkStatusMonitoring } from "./stores/networkStore.ts";
+import { WebSocketPool } from "./data_structures/websocket_pool.ts";
 
 export const ndkInstance: Writable<NDK> = writable();
 export const ndkSignedIn = writable(false);
@@ -199,16 +200,14 @@ export function checkWebSocketSupport(): void {
 
   // Test if secure WebSocket is supported
   try {
-    const testWs = new WebSocket("wss://echo.websocket.org");
-    testWs.onopen = () => {
+    WebSocketPool.instance.acquire("wss://echo.websocket.org").then((ws) => {
       console.debug("[NDK.ts] ✓ Secure WebSocket (wss://) is supported");
-      testWs.close();
-    };
-    testWs.onerror = () => {
+      WebSocketPool.instance.release(ws);
+    }).catch((_) => {
       console.warn("[NDK.ts] ✗ Secure WebSocket (wss://) may not be supported");
-    };
-  } catch (error) {
-    console.warn("[NDK.ts] ✗ WebSocket test failed:", error);
+    });
+  } catch {
+    console.warn("[NDK.ts] ✗ WebSocket test failed");
   }
 }
 

src/lib/utils/community_checker.ts

@@ -1,4 +1,5 @@
 import { communityRelays } from "$lib/consts";
+import { WebSocketPool } from "../data_structures/websocket_pool.ts";
 import { RELAY_CONSTANTS, SEARCH_LIMITS } from "./search_constants";
 
 // Cache for pubkeys with kind 1 events on communityRelay
@@ -16,37 +17,31 @@ export async function checkCommunity(pubkey: string): Promise<boolean> {
     // Try each community relay until we find one that works
     for (const relayUrl of communityRelays) {
       try {
-        const ws = new WebSocket(relayUrl);
+        const ws = await WebSocketPool.instance.acquire(relayUrl);
         const result = await new Promise<boolean>((resolve) => {
-          ws.onopen = () => {
-            ws.send(
-              JSON.stringify([
-                "REQ",
-                RELAY_CONSTANTS.COMMUNITY_REQUEST_ID,
-                {
-                  kinds: RELAY_CONSTANTS.COMMUNITY_REQUEST_KINDS,
-                  authors: [pubkey],
-                  limit: SEARCH_LIMITS.COMMUNITY_CHECK,
-                },
-              ]),
-            );
-          };
+          ws.send(
+            JSON.stringify([
+              "REQ",
+              RELAY_CONSTANTS.COMMUNITY_REQUEST_ID,
+              {
+                kinds: RELAY_CONSTANTS.COMMUNITY_REQUEST_KINDS,
+                authors: [pubkey],
+                limit: SEARCH_LIMITS.COMMUNITY_CHECK,
+              },
+            ]),
+          );
           ws.onmessage = (event) => {
             const data = JSON.parse(event.data);
             if (data[0] === "EVENT" && data[2]?.kind === 1) {
               communityCache.set(pubkey, true);
-              ws.close();
+              WebSocketPool.instance.release(ws);
               resolve(true);
             } else if (data[0] === "EOSE") {
               communityCache.set(pubkey, false);
-              ws.close();
+              WebSocketPool.instance.release(ws);
               resolve(false);
             }
           };
-          ws.onerror = () => {
-            ws.close();
-            resolve(false);
-          };
         });
         
         if (result) {

src/lib/utils/relayDiagnostics.ts

@@ -1,3 +1,4 @@
+import { WebSocketPool } from "../data_structures/websocket_pool.ts";
 import { activeInboxRelays, activeOutboxRelays } from "../ndk.ts";
 import { TIMEOUTS } from "./search_constants.ts";
 import { get } from "svelte/store";
@@ -13,72 +14,37 @@ export interface RelayDiagnostic {
 /**
  * Tests connection to a single relay
  */
-export function testRelay(url: string): Promise<RelayDiagnostic> {
+export async function testRelay(url: string): Promise<RelayDiagnostic> {
   const startTime = Date.now();
+  const ws = await WebSocketPool.instance.acquire(url);
 
   return new Promise((resolve) => {
-    const ws = new WebSocket(url);
-    let resolved = false;
-
     const timeout = setTimeout(() => {
-      if (!resolved) {
-        resolved = true;
-        ws.close();
-        resolve({
-          url,
-          connected: false,
-          requiresAuth: false,
-          error: "Connection timeout",
-          responseTime: Date.now() - startTime,
-        });
-      }
+      WebSocketPool.instance.release(ws);
+      resolve({
+        url,
+        connected: false,
+        requiresAuth: false,
+        error: "Connection timeout",
+        responseTime: Date.now() - startTime,
+      });
     }, TIMEOUTS.RELAY_DIAGNOSTICS);
 
-    ws.onopen = () => {
-      if (!resolved) {
-        resolved = true;
+    ws.onmessage = (event) => {
+      const data = JSON.parse(event.data);
+      if (data[0] === "NOTICE" && data[1]?.includes("auth-required")) {
         clearTimeout(timeout);
-        ws.close();
+        WebSocketPool.instance.release(ws);
         resolve({
           url,
           connected: true,
-          requiresAuth: false,
+          requiresAuth: true,
           responseTime: Date.now() - startTime,
         });
       }
-    };
-
-    ws.onerror = () => {
-      if (!resolved) {
-        resolved = true;
-        clearTimeout(timeout);
-        resolve({
-          url,
-          connected: false,
-          requiresAuth: false,
-          error: "WebSocket error",
-          responseTime: Date.now() - startTime,
-        });
-      }
-    };
-
-    ws.onmessage = (event) => {
-      const data = JSON.parse(event.data);
-      if (data[0] === "NOTICE" && data[1]?.includes("auth-required")) {
-        if (!resolved) {
-          resolved = true;
-          clearTimeout(timeout);
-          ws.close();
-          resolve({
-            url,
-            connected: true,
-            requiresAuth: true,
-            responseTime: Date.now() - startTime,
-          });
-        }
-      }
-    };
+    }
   });
+
 }
 
 /**