Merge branch 'outboxy-req-router' into 'main'

Add an outboxy-style REQ router

See merge request soapbox-pub/ditto!713

packages/ditto/storages/DittoPool.test.ts

@@ -0,0 +1,66 @@
+import { DittoConf } from '@ditto/conf';
+import { genEvent, MockRelay } from '@nostrify/nostrify/test';
+import { assertEquals } from '@std/assert';
+import { generateSecretKey, getPublicKey, nip19 } from 'nostr-tools';
+
+import { DittoPool } from './DittoPool.ts';
+
+Deno.test('DittoPool.reqRouter', async (t) => {
+  const conf = new DittoConf(new Map([['DITTO_NSEC', nip19.nsecEncode(generateSecretKey())]]));
+  const relay = new MockRelay();
+
+  const pool = new DittoPool({ conf, relay });
+
+  const [alex, mk] = [
+    generateKeypair(),
+    generateKeypair(),
+  ];
+
+  const [ditto, henhouse, gleasonator] = [
+    'wss://ditto.pub/relay',
+    'wss://henhouse.social/relay',
+    'wss://gleasonator.dev/relay',
+  ];
+
+  const events = [
+    genEvent({ kind: 10002, tags: [['r', gleasonator], ['r', ditto]] }, alex.sk),
+    genEvent({ kind: 10002, tags: [['r', henhouse], ['r', ditto]] }, mk.sk),
+  ];
+
+  for (const event of events) {
+    await relay.event(event);
+  }
+
+  await t.step('no authors', async () => {
+    const reqRoutes = await pool.reqRouter([{ kinds: [1] }]);
+    assertEquals(reqRoutes, new Map());
+  });
+
+  await t.step('single author', async () => {
+    const reqRoutes = await pool.reqRouter([{ kinds: [10002], authors: [alex.pk] }]);
+
+    const expected = new Map([
+      [ditto, [{ kinds: [10002], authors: [alex.pk] }]],
+      [gleasonator, [{ kinds: [10002], authors: [alex.pk] }]],
+    ]);
+
+    assertEquals(reqRoutes, expected);
+  });
+
+  await t.step('multiple authors', async () => {
+    const reqRoutes = await pool.reqRouter([{ kinds: [10002], authors: [alex.pk, mk.pk] }]);
+
+    const expected = new Map([
+      [ditto, [{ kinds: [10002], authors: [alex.pk, mk.pk] }]],
+      [henhouse, [{ kinds: [10002], authors: [mk.pk] }]],
+      [gleasonator, [{ kinds: [10002], authors: [alex.pk] }]],
+    ]);
+
+    assertEquals(reqRoutes, expected);
+  });
+});
+
+function generateKeypair(): { pk: string; sk: Uint8Array } {
+  const sk = generateSecretKey();
+  return { pk: getPublicKey(sk), sk };
+}

packages/ditto/storages/DittoPool.ts

@@ -6,6 +6,7 @@ import { logi } from '@soapbox/logi';
 interface DittoPoolOpts {
   conf: DittoConf;
   relay: NRelay;
+  maxReqRelays?: number;
   maxEventRelays?: number;
 }
 
@@ -32,18 +33,66 @@ export class DittoPool extends NPool<NRelay1> {
     this._opts = opts;
   }
 
-  private async reqRouter(filters: NostrFilter[]): Promise<Map<string, NostrFilter[]>> {
+  async reqRouter(filters: NostrFilter[]): Promise<Map<string, NostrFilter[]>> {
-    const routes = new Map<string, NostrFilter[]>();
+    const { conf, relay, maxReqRelays = 5 } = this._opts;
 
-    for (const relayUrl of await this.getRelayUrls({ marker: 'read' })) {
+    const routes = new Map<string, NostrFilter[]>();
-      routes.set(relayUrl, filters);
+    const authors = new Set<string>();
+
+    for (const filter of filters) {
+      if (filter.authors) {
+        for (const author of filter.authors) {
+          authors.add(author);
+        }
+      }
+    }
+
+    if (!authors.size) {
+      return routes;
+    }
+
+    const pubkey = await conf.signer.getPublicKey();
+    const map = new Map<string, NostrEvent>();
+
+    for (const event of await relay.query([{ kinds: [10002], authors: [pubkey, ...authors] }])) {
+      map.set(event.pubkey, event);
+    }
+
+    for (const filter of filters) {
+      if (filter.authors) {
+        const relayAuthors = new Map<`wss://${string}`, Set<string>>();
+
+        for (const author of filter.authors) {
+          const event = map.get(author) ?? map.get(pubkey);
+          if (event) {
+            for (const relayUrl of [...this.getEventRelayUrls(event, 'write')].slice(0, maxReqRelays)) {
+              const value = relayAuthors.get(relayUrl);
+              relayAuthors.set(relayUrl, value ? new Set([...value, author]) : new Set([author]));
+            }
+          }
+        }
+
+        for (const [relayUrl, authors] of relayAuthors) {
+          const value = routes.get(relayUrl);
+          const _filter = { ...filter, authors: [...authors] };
+          routes.set(relayUrl, value ? [...value, _filter] : [_filter]);
+        }
+      } else {
+        const event = map.get(pubkey);
+        if (event) {
+          for (const relayUrl of [...this.getEventRelayUrls(event, 'read')].slice(0, maxReqRelays)) {
+            const value = routes.get(relayUrl);
+            routes.set(relayUrl, value ? [...value, filter] : [filter]);
+          }
+        }
+      }
     }
 
     return routes;
   }
 
-  private async eventRouter(event: NostrEvent): Promise<string[]> {
+  async eventRouter(event: NostrEvent): Promise<string[]> {
-    const { conf, maxEventRelays = 4 } = this._opts;
+    const { conf, maxEventRelays = 10 } = this._opts;
     const { pubkey } = event;
 
     const relaySet = await this.getRelayUrls({ pubkey, marker: 'write' });
@@ -72,16 +121,26 @@ export class DittoPool extends NPool<NRelay1> {
     }
 
     for (const event of events) {
-      for (const [name, relayUrl, marker] of event.tags) {
+      for (const relayUrl of this.getEventRelayUrls(event, opts.marker)) {
-        if (name === 'r' && (!marker || !opts.marker || marker === opts.marker)) {
+        relays.add(relayUrl);
-          try {
+      }
-            const url = new URL(relayUrl);
+    }
-            if (url.protocol === 'wss:') {
+
-              relays.add(url.toString() as `wss://${string}`);
+    return relays;
-            }
+  }
-          } catch {
+
-            // fallthrough
+  private getEventRelayUrls(event: NostrEvent, marker?: 'read' | 'write'): Set<`wss://${string}`> {
+    const relays = new Set<`wss://${string}`>();
+
+    for (const [name, relayUrl, _marker] of event.tags) {
+      if (name === 'r' && (!marker || !_marker || marker === _marker)) {
+        try {
+          const url = new URL(relayUrl);
+          if (url.protocol === 'wss:') {
+            relays.add(url.toString() as `wss://${string}`);
           }
+        } catch {
+          // fallthrough
         }
       }
     }