fix tests, .seenOn() method for pools.

.github/workflows/test.yml

@@ -1,7 +1,9 @@
 name: test every commit
 on:
-  - push
-  - pull_request
+  push:
+    branches:
+      - master
+  pull_request:
 
 jobs:
   test:

README.md

@@ -111,6 +111,11 @@ pub.on('failed', reason => {
   console.log(`failed to publish to ${relay.url}: ${reason}`)
 })
 
+let events = await relay.list([{kinds: [0, 1]}])
+let event = await relay.get({
+  ids: ['44e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245']
+})
+
 await relay.close()
 ```
 
@@ -129,12 +134,7 @@ const pool = new SimplePool()
 
 let relays = ['wss://relay.example.com', 'wss://relay.example2.com']
 
-relays.forEach(async url => {
-  let relay = pool.ensureRelay(url)
-  await relay.connect()
-})
-
-let relay = pool.ensureRelay('wss://relay.example3.com')
+let relay = await pool.ensureRelay('wss://relay.example3.com')
 
 let subs = pool.sub([...relays, relay], {
   authors: ['32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245']
@@ -147,12 +147,22 @@ subs.forEach(sub =>
   })
 )
 
-let pubs = pool.publish(newEvent)
+let pubs = pool.publish(relays, newEvent)
 pubs.forEach(pub =>
   pub.on('ok', () => {
     // ...
   })
 )
+
+let events = await pool.list(relays, [{kinds: [0, 1]}])
+let event = await pool.get(relays, {
+  ids: ['44e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245']
+})
+
+let relaysForEvent = pool.seenOn(
+  '44e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245'
+)
+// relaysForEvent will be an array of URLs from relays a given event was seen on
 ```
 
 ### Querying profile data from a NIP-05 address

justfile

@@ -11,3 +11,6 @@ test: build
 
 testOnly file: build
     jest {{file}}
+
+publish: build
+    npm publish

magic.ts

@@ -0,0 +1,140 @@
+import {Relay, relayInit} from './relay'
+import {Event} from './event'
+import {normalizeURL} from './utils'
+
+export default function (
+  writeableRelays: string[],
+  fallbackRelays: string[],
+  safeRelays: string[]
+) {
+  return new MagicPool(fallbackRelays, writeableRelays, safeRelays)
+}
+
+class MagicPool {
+  private _conn: {[url: string]: Relay}
+  private _fallback: {[url: string]: Relay}
+  private _write: {[url: string]: Relay}
+  private _safe: {[url: string]: Relay}
+
+  private _profileRelays: {[pubkey: string]: RelayTableScore}
+  private _tempCache: {[id: string]: Event}
+
+  constructor(
+    fallbackRelays: string[],
+    writeableRelays: string[],
+    safeRelays: string[] = [
+      'wss://eden.nostr.land',
+      'wss://nostr.milou.lol',
+      'wss://relay.minds.com/nostr/v1/ws'
+    ]
+  ) {
+    this._conn = {}
+    this._write = {}
+    this._fallback = {}
+    this._profileRelays = {}
+    this._tempCache = {}
+
+    const hasEventId = (id: string): boolean => id in this._tempCache
+    const init = (url: string) => {
+      this._conn[normalizeURL(url)] = relayInit(normalizeURL(url), hasEventId)
+    }
+
+    fallbackRelays.forEach(init)
+    writeableRelays.forEach(init)
+    safeRelays.forEach(init)
+
+    this._write = Object.fromEntries(
+      writeableRelays.map(url => [
+        normalizeURL(url),
+        this._conn[normalizeURL(url)]
+      ])
+    )
+    this._fallback = Object.fromEntries(
+      fallbackRelays.map(url => [
+        normalizeURL(url),
+        this._conn[normalizeURL(url)]
+      ])
+    )
+    this._safe = Object.fromEntries(
+      safeRelays.map(url => [normalizeURL(url), this._conn[normalizeURL(url)]])
+    )
+  }
+
+  publish(event: Event) {
+    return Promise.all(
+      Object.entries(this._write).map(
+        ([url, relay]) =>
+          new Promise(async resolve => {
+            await relay.connect()
+            let pub = relay.publish(event)
+            let to = setTimeout(() => {
+              let end = setTimeout(() => {
+                resolve({url, success: false, reason: 'timeout'})
+              }, 2500)
+              pub.on('seen', () => {
+                clearTimeout(end)
+                resolve({url, success: true, reason: 'seen'})
+              })
+            }, 2500)
+            pub.on('ok', () => {
+              clearTimeout(to)
+              resolve({url, success: true, reason: 'ok'})
+            })
+            pub.on('failed', (reason: string) => {
+              clearTimeout(to)
+              resolve({url, success: false, reason})
+            })
+          })
+      )
+    )
+  }
+
+  profile(
+    pubkey: string,
+    onUpdate: (events: Event[]) => void
+  ): {
+    page(n: number): void
+  } {
+    var relays = new Set()
+    let rts = this._profileRelays[pubkey]
+    if (rts) {
+      relays = rts.get(3)
+    }
+
+    let fallback = Object.values(this._fallback)
+    for (let i = 0; i < fallback.length; i++) {
+      if (relays.size < 3) {
+        relays.add(fallback[Math.floor(Math.random() * fallback.length)])
+      } else break
+    }
+
+    // start subscription
+    for (let r in relays) {
+        r.
+    }
+
+    return {
+      page(n: number) {}
+    }
+  }
+}
+
+class RelayTableScore {
+  seen: string[] = []
+  hinted: string[] = []
+  explicit: string[] = []
+
+  get(n: number): Set<string> {
+    let relays = new Set<string>()
+    for (let i = 0; i < n; i++) {
+      for (let j = 0; j < 3; j++) {
+        let v = [this.seen, this.explicit, this.hinted][j][i]
+        if (v) {
+          relays.add(v)
+          if (relays.size >= n) return relays
+        }
+      }
+    }
+    return relays
+  }
+}

package.json

@@ -1,6 +1,6 @@
 {
   "name": "nostr-tools",
-  "version": "1.2.4",
+  "version": "1.3.2",
   "description": "Tools for making a Nostr client.",
   "repository": {
     "type": "git",

pool.test.js

@@ -19,50 +19,28 @@ let relays = [
   'wss://nostr.zebedee.cloud/'
 ]
 
-beforeAll(async () => {
-  Promise.all(
-    relays.map(relay => {
-      try {
-        let r = pool.ensureRelay(relay)
-        return r.connect()
-      } catch (err) {
-        /***/
-      }
-    })
-  )
-})
-
 afterAll(async () => {
-  relays.forEach(relay => {
-    try {
-      let r = pool.ensureRelay(relay)
-      r.close()
-    } catch (err) {
-      /***/
-    }
-  })
+  await pool.close([
+    ...relays,
+    'wss://nostr-relay.untethr.me',
+    'wss://offchain.pub',
+    'wss://eden.nostr.land'
+  ])
 })
 
 test('removing duplicates when querying', async () => {
   let priv = generatePrivateKey()
   let pub = getPublicKey(priv)
 
-  let subs = pool.sub(relays, [
-    {
-      authors: [pub]
-    }
-  ])
-
+  let sub = pool.sub(relays, [{authors: [pub]}])
   let received = []
 
-  subs.forEach(sub =>
-    sub.on('event', event => {
-      // this should be called only once even though we're listening
-      // to multiple relays because the events will be catched and
-      // deduplicated efficiently (without even being parsed)
-      received.push(event)
-    })
-  )
+  sub.on('event', event => {
+    // this should be called only once even though we're listening
+    // to multiple relays because the events will be catched and
+    // deduplicated efficiently (without even being parsed)
+    received.push(event)
+  })
 
   let event = {
     pubkey: pub,
@@ -81,25 +59,22 @@ test('removing duplicates when querying', async () => {
   expect(received).toHaveLength(1)
 })
 
-test('removing duplicates correctly when double querying', async () => {
+test('same with double querying', async () => {
   let priv = generatePrivateKey()
   let pub = getPublicKey(priv)
 
-  let subs1 = pool.sub(relays, [{authors: [pub]}])
-  let subs2 = pool.sub(relays, [{authors: [pub]}])
+  let sub1 = pool.sub(relays, [{authors: [pub]}])
+  let sub2 = pool.sub(relays, [{authors: [pub]}])
 
   let received = []
 
-  subs1.forEach(sub =>
-    sub.on('event', event => {
-      received.push(event)
-    })
-  )
-  subs2.forEach(sub =>
-    sub.on('event', event => {
-      received.push(event)
-    })
-  )
+  sub1.on('event', event => {
+    received.push(event)
+  })
+
+  sub2.on('event', event => {
+    received.push(event)
+  })
 
   let event = {
     pubkey: pub,
@@ -117,3 +92,42 @@ test('removing duplicates correctly when double querying', async () => {
 
   expect(received).toHaveLength(2)
 })
+
+test('get()', async () => {
+  let event = await pool.get(relays, {
+    ids: ['d7dd5eb3ab747e16f8d0212d53032ea2a7cadef53837e5a6c66d42849fcb9027']
+  })
+
+  expect(event).toHaveProperty(
+    'id',
+    'd7dd5eb3ab747e16f8d0212d53032ea2a7cadef53837e5a6c66d42849fcb9027'
+  )
+})
+
+test('list()', async () => {
+  let events = await pool.list(
+    [...relays, 'wss://offchain.pub', 'wss://eden.nostr.land'],
+    [
+      {
+        authors: [
+          '3bf0c63fcb93463407af97a5e5ee64fa883d107ef9e558472c4eb9aaaefa459d'
+        ],
+        kinds: [1],
+        limit: 2
+      }
+    ]
+  )
+
+  // the actual received number will be greater than 2, but there will be no duplicates
+  expect(events.length).toEqual(
+    events
+      .map(evt => evt.id)
+      .reduce((acc, n) => (acc.indexOf(n) !== -1 ? acc : [...acc, n]), [])
+      .length
+  )
+
+  let relaysForAllEvents = events
+    .map(event => pool.seenOn(event.id))
+    .reduce((acc, n) => acc.concat(n), [])
+  expect(relaysForAllEvents.length).toBeGreaterThanOrEqual(events.length)
+})

pool.ts

@@ -6,13 +6,22 @@ import {SubscriptionOptions, Sub, Pub} from './relay'
 
 export class SimplePool {
   private _conn: {[url: string]: Relay}
+  private _seenOn: {[id: string]: Set<string>} = {} // a map of all events we've seen in each relay
 
-  constructor(defaultRelays: string[] = []) {
+  constructor() {
     this._conn = {}
-    defaultRelays.forEach(this.ensureRelay)
   }
 
-  ensureRelay(url: string): Relay {
+  async close(relays: string[]): Promise<void> {
+    await Promise.all(
+      relays.map(async url => {
+        let relay = this._conn[normalizeURL(url)]
+        if (relay) await relay.close()
+      })
+    )
+  }
+
+  async ensureRelay(url: string): Promise<Relay> {
     const nm = normalizeURL(url)
     const existing = this._conn[nm]
     if (existing) return existing
@@ -20,41 +29,131 @@ export class SimplePool {
     const relay = relayInit(nm)
     this._conn[nm] = relay
 
+    await relay.connect()
+
     return relay
   }
 
-  sub(relays: string[], filters: Filter[], opts?: SubscriptionOptions): Sub[] {
+  sub(relays: string[], filters: Filter[], opts?: SubscriptionOptions): Sub {
     let _knownIds: Set<string> = new Set()
     let modifiedOpts = opts || {}
-    modifiedOpts.alreadyHaveEvent = id => _knownIds.has(id)
+    modifiedOpts.alreadyHaveEvent = (id, url) => {
+      let set = this._seenOn[id] || new Set()
+      set.add(url)
+      this._seenOn[id] = set
+      return _knownIds.has(id)
+    }
 
-    return relays.map(relay => {
-      let r = this._conn[relay]
-      if (!r) return badSub()
+    let subs: Sub[] = []
+    let eventListeners: Set<(event: Event) => void> = new Set()
+    let eoseListeners: Set<() => void> = new Set()
+    let eosesMissing = relays.length
+
+    let eoseSent = false
+    let eoseTimeout = setTimeout(() => {
+      eoseSent = true
+      for (let cb of eoseListeners.values()) cb()
+    }, 2400)
+
+    relays.forEach(async relay => {
+      let r = await this.ensureRelay(relay)
+      if (!r) return
       let s = r.sub(filters, modifiedOpts)
-      s.on('event', (event: Event) => _knownIds.add(event.id as string))
-      return s
+      s.on('event', (event: Event) => {
+        _knownIds.add(event.id as string)
+        for (let cb of eventListeners.values()) cb(event)
+      })
+      s.on('eose', () => {
+        if (eoseSent) return
+
+        eosesMissing--
+        if (eosesMissing === 0) {
+          clearTimeout(eoseTimeout)
+          for (let cb of eoseListeners.values()) cb()
+        }
+      })
+      subs.push(s)
+    })
+
+    let greaterSub: Sub = {
+      sub(filters, opts) {
+        subs.forEach(sub => sub.sub(filters, opts))
+        return greaterSub
+      },
+      unsub() {
+        subs.forEach(sub => sub.unsub())
+      },
+      on(type, cb) {
+        switch (type) {
+          case 'event':
+            eventListeners.add(cb)
+            break
+          case 'eose':
+            eoseListeners.add(cb)
+            break
+        }
+      },
+      off(type, cb) {
+        if (type === 'event') {
+          eventListeners.delete(cb)
+        } else if (type === 'eose') eoseListeners.delete(cb)
+      }
+    }
+
+    return greaterSub
+  }
+
+  get(
+    relays: string[],
+    filter: Filter,
+    opts?: SubscriptionOptions
+  ): Promise<Event | null> {
+    return new Promise(resolve => {
+      let sub = this.sub(relays, [filter], opts)
+      let timeout = setTimeout(() => {
+        sub.unsub()
+        resolve(null)
+      }, 1500)
+      sub.on('event', (event: Event) => {
+        resolve(event)
+        clearTimeout(timeout)
+        sub.unsub()
+      })
+    })
+  }
+
+  list(
+    relays: string[],
+    filters: Filter[],
+    opts?: SubscriptionOptions
+  ): Promise<Event[]> {
+    return new Promise(resolve => {
+      let events: Event[] = []
+      let sub = this.sub(relays, filters, opts)
+
+      sub.on('event', (event: Event) => {
+        events.push(event)
+      })
+
+      // we can rely on an eose being emitted here because pool.sub() will fake one
+      sub.on('eose', () => {
+        sub.unsub()
+        resolve(events)
+      })
     })
   }
 
   publish(relays: string[], event: Event): Pub[] {
     return relays.map(relay => {
-      let r = this._conn[relay]
+      let r = this._conn[normalizeURL(relay)]
       if (!r) return badPub(relay)
       let s = r.publish(event)
       return s
     })
   }
-}
 
-function badSub(): Sub {
-  return {
-    on() {},
-    off() {},
-    sub(): Sub {
-      return badSub()
-    },
-    unsub() {}
+  seenOn(id: string): string[] {
+    return Array.from(this._seenOn[id]?.values?.() || [])
   }
 }
 

relay.test.js

@@ -32,7 +32,7 @@ test('connectivity', () => {
   ).resolves.toBe(true)
 })
 
-test('querying', () => {
+test('querying', async () => {
   var resolve1
   var resolve2
 
@@ -52,16 +52,42 @@ test('querying', () => {
     resolve2(true)
   })
 
-  return expect(
-    Promise.all([
-      new Promise(resolve => {
-        resolve1 = resolve
-      }),
-      new Promise(resolve => {
-        resolve2 = resolve
-      })
-    ])
-  ).resolves.toEqual([true, true])
+  let [t1, t2] = await Promise.all([
+    new Promise(resolve => {
+      resolve1 = resolve
+    }),
+    new Promise(resolve => {
+      resolve2 = resolve
+    })
+  ])
+
+  expect(t1).toEqual(true)
+  expect(t2).toEqual(true)
+})
+
+test('get()', async () => {
+  let event = await relay.get({
+    ids: ['d7dd5eb3ab747e16f8d0212d53032ea2a7cadef53837e5a6c66d42849fcb9027']
+  })
+
+  expect(event).toHaveProperty(
+    'id',
+    'd7dd5eb3ab747e16f8d0212d53032ea2a7cadef53837e5a6c66d42849fcb9027'
+  )
+})
+
+test('list()', async () => {
+  let events = await relay.list([
+    {
+      authors: [
+        '3bf0c63fcb93463407af97a5e5ee64fa883d107ef9e558472c4eb9aaaefa459d'
+      ],
+      kinds: [1],
+      limit: 2
+    }
+  ])
+
+  expect(events.length).toEqual(2)
 })
 
 test('listening (twice) and publishing', async () => {

relay.ts

@@ -12,6 +12,8 @@ export type Relay = {
   connect: () => Promise<void>
   close: () => Promise<void>
   sub: (filters: Filter[], opts?: SubscriptionOptions) => Sub
+  list: (filters: Filter[], opts?: SubscriptionOptions) => Promise<Event[]>
+  get: (filter: Filter, opts?: SubscriptionOptions) => Promise<Event | null>
   publish: (event: Event) => Pub
   on: (type: RelayEvent, cb: any) => void
   off: (type: RelayEvent, cb: any) => void
@@ -28,9 +30,9 @@ export type Sub = {
 }
 
 export type SubscriptionOptions = {
-  skipVerification?: boolean
-  alreadyHaveEvent?: null | ((id: string) => boolean)
   id?: string
+  skipVerification?: boolean
+  alreadyHaveEvent?: null | ((id: string, relay: string) => boolean)
 }
 
 export function relayInit(url: string): Relay {
@@ -106,8 +108,12 @@ export function relayInit(url: string): Relay {
 
         let subid = getSubscriptionId(json)
         if (subid) {
-          let {alreadyHaveEvent} = openSubs[subid]
-          if (alreadyHaveEvent && alreadyHaveEvent(getHex64(json, 'id'))) {
+          let so = openSubs[subid]
+          if (
+            so &&
+            so.alreadyHaveEvent &&
+            so.alreadyHaveEvent(getHex64(json, 'id'), url)
+          ) {
             return
           }
         }
@@ -231,6 +237,36 @@ export function relayInit(url: string): Relay {
       let index = listeners[type].indexOf(cb)
       if (index !== -1) listeners[type].splice(index, 1)
     },
+    list: (filters: Filter[], opts?: SubscriptionOptions): Promise<Event[]> =>
+      new Promise(resolve => {
+        let s = sub(filters, opts)
+        let events: Event[] = []
+        let timeout = setTimeout(() => {
+          s.unsub()
+          resolve(events)
+        }, 1500)
+        s.on('eose', () => {
+          s.unsub()
+          clearTimeout(timeout)
+          resolve(events)
+        })
+        s.on('event', (event: Event) => {
+          events.push(event)
+        })
+      }),
+    get: (filter: Filter, opts?: SubscriptionOptions): Promise<Event | null> =>
+      new Promise(resolve => {
+        let s = sub([filter], opts)
+        let timeout = setTimeout(() => {
+          s.unsub()
+          resolve(null)
+        }, 1500)
+        s.on('event', (event: Event) => {
+          s.unsub()
+          clearTimeout(timeout)
+          resolve(event)
+        })
+      }),
     publish(event: Event): Pub {
       if (!event.id) throw new Error(`event ${event} has no id`)
       let id = event.id
@@ -288,6 +324,7 @@ export function relayInit(url: string): Relay {
     },
     connect,
     close(): Promise<void> {
+      if (ws.readyState > 1) return Promise.resolve()
       ws.close()
       return new Promise<void>(resolve => {
         resolveClose = resolve