improvements and fixes on pool.

now pools are much smarter.

.github/workflows/npm-publish.yml

@@ -12,9 +12,10 @@ jobs:
       - uses: actions/setup-node@v3
         with:
           node-version: 18
-      - run: yarn --ignore-engines
-      - run: node build.js
-      - run: yarn test
+      - uses: extractions/setup-just@v1
+      - run: just install-dependencies
+      - run: just build
+      - run: just test
       - uses: JS-DevTools/npm-publish@v1
         with:
           token: ${{ secrets.NPM_TOKEN }}

.github/workflows/test.yml

@@ -11,6 +11,7 @@ jobs:
       - uses: actions/setup-node@v3
         with:
           node-version: 18
-      - run: yarn --ignore-engines
-      - run: node build.js
-      - run: yarn test
+      - uses: extractions/setup-just@v1
+      - run: just install-dependencies
+      - run: just build
+      - run: just 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()
 ```
 
@@ -120,6 +125,41 @@ To use this on Node.js you first must install `websocket-polyfill` and import it
 import 'websocket-polyfill'
 ```
 
+### Interacting with multiple relays
+
+```js
+import {pool} from 'nostr-tools'
+
+const pool = new SimplePool()
+
+let relays = ['wss://relay.example.com', 'wss://relay.example2.com']
+
+let relay = await pool.ensureRelay('wss://relay.example3.com')
+
+let subs = pool.sub([...relays, relay], {
+  authors: ['32e1827635450ebb3c5a7d12c1f8e7b2b514439ac10a67eef3d9fd9c5c68e245']
+})
+
+subs.forEach(sub =>
+  sub.on('event', event => {
+    // this will only be called once the first time the event is received
+    // ...
+  })
+)
+
+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']
+})
+```
+
 ### Querying profile data from a NIP-05 address
 
 ```js
@@ -195,7 +235,7 @@ let event = {
 sendEvent(event)
 
 // on the receiver side
-sub.on('event', (event) => {
+sub.on('event', event => {
   let sender = event.tags.find(([k, v]) => k === 'p' && v && v !== '')[1]
   pk1 === sender
   let plaintext = await nip04.decrypt(sk2, pk1, event.content)

fakejson.test.js

@@ -33,3 +33,17 @@ test('match kind', () => {
     )
   ).toBeTruthy()
 })
+
+test('match subscription id', () => {
+  expect(fj.getSubscriptionId('["EVENT","",{}]')).toEqual('')
+  expect(fj.getSubscriptionId('["EVENT","_",{}]')).toEqual('_')
+  expect(fj.getSubscriptionId('["EVENT","subname",{}]')).toEqual('subname')
+  expect(fj.getSubscriptionId('["EVENT", "kasjbdjkav", {}]')).toEqual(
+    'kasjbdjkav'
+  )
+  expect(
+    fj.getSubscriptionId(
+      '  [ \n\n "EVENT"  , \n\n "y4d5ow45gfwoiudfÇA VSADLKAN KLDASB[12312535]SFMZSNJKLH" , {}]'
+    )
+  ).toEqual('y4d5ow45gfwoiudfÇA VSADLKAN KLDASB[12312535]SFMZSNJKLH')
+})

fakejson.ts

@@ -13,6 +13,21 @@ export function getInt(json: string, field: string): number {
   return parseInt(sliced.slice(0, end), 10)
 }
 
+export function getSubscriptionId(json: string): string | null {
+  let idx = json.slice(0, 22).indexOf(`"EVENT"`)
+  if (idx === -1) return null
+
+  let pstart = json.slice(idx + 7 + 1).indexOf(`"`)
+  if (pstart === -1) return null
+  let start = idx + 7 + 1 + pstart
+
+  let pend = json.slice(start + 1, 80).indexOf(`"`)
+  if (pend === -1) return null
+  let end = start + 1 + pend
+
+  return json.slice(start + 1, end)
+}
+
 export function matchEventId(json: string, id: string): boolean {
   return id === getHex64(json, 'id')
 }

index.ts

@@ -2,6 +2,7 @@ export * from './keys'
 export * from './relay'
 export * from './event'
 export * from './filter'
+export * from './pool'
 
 export * as nip04 from './nip04'
 export * as nip05 from './nip05'

justfile

@@ -0,0 +1,13 @@
+export PATH := "./node_modules/.bin:" + env_var('PATH')
+
+install-dependencies:
+    yarn --ignore-engines
+
+build:
+    node build.js
+
+test: build
+    jest
+
+testOnly file: build
+    jest {{file}}

nip05.ts

@@ -36,7 +36,7 @@ export async function queryProfile(
     name = '_'
   }
 
-  if (!name.match(/^[a-z0-9-_]+$/)) return null
+  if (!name.match(/^[A-Za-z0-9-_]+$/)) return null
 
   let res = await (
     await _fetch(`https://${domain}/.well-known/nostr.json?name=${name}`)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "nostr-tools",
-  "version": "1.2.1",
+  "version": "1.3.1",
   "description": "Tools for making a Nostr client.",
   "repository": {
     "type": "git",
@@ -38,10 +38,5 @@
     "tsd": "^0.22.0",
     "typescript": "^4.9.4",
     "websocket-polyfill": "^0.0.3"
-  },
-  "scripts": {
-    "build": "node build.js",
-    "pretest": "node build.js",
-    "test": "jest"
   }
 }

pool.test.js

@@ -0,0 +1,123 @@
+/* eslint-env jest */
+
+require('websocket-polyfill')
+const {
+  SimplePool,
+  generatePrivateKey,
+  getPublicKey,
+  getEventHash,
+  signEvent
+} = require('./lib/nostr.cjs')
+
+let pool = new SimplePool()
+
+let relays = [
+  'wss://nostr-dev.wellorder.net/',
+  'wss://relay.nostr.bg/',
+  'wss://nostr.fmt.wiz.biz/',
+  'wss://relay.nostr.band/',
+  'wss://nostr.zebedee.cloud/'
+]
+
+afterAll(async () => {
+  await pool.close([...relays, 'wss://nostr-relay.untethr.me'])
+})
+
+test('removing duplicates when querying', async () => {
+  let priv = generatePrivateKey()
+  let pub = getPublicKey(priv)
+
+  let sub = pool.sub(relays, [{authors: [pub]}])
+  let received = []
+
+  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,
+    created_at: Math.round(Date.now() / 1000),
+    content: 'test',
+    kind: 22345,
+    tags: []
+  }
+  event.id = getEventHash(event)
+  event.sig = signEvent(event, priv)
+
+  pool.publish(relays, event)
+
+  await new Promise(resolve => setTimeout(resolve, 1500))
+
+  expect(received).toHaveLength(1)
+})
+
+test('same with double querying', async () => {
+  let priv = generatePrivateKey()
+  let pub = getPublicKey(priv)
+
+  let sub1 = pool.sub(relays, [{authors: [pub]}])
+  let sub2 = pool.sub(relays, [{authors: [pub]}])
+
+  let received = []
+
+  sub1.on('event', event => {
+    received.push(event)
+  })
+
+  sub2.on('event', event => {
+    received.push(event)
+  })
+
+  let event = {
+    pubkey: pub,
+    created_at: Math.round(Date.now() / 1000),
+    content: 'test2',
+    kind: 22346,
+    tags: []
+  }
+  event.id = getEventHash(event)
+  event.sig = signEvent(event, priv)
+
+  pool.publish(relays, event)
+
+  await new Promise(resolve => setTimeout(resolve, 1500))
+
+  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
+  )
+})

pool.ts

@@ -0,0 +1,159 @@
+import {Relay, relayInit} from './relay'
+import {normalizeURL} from './utils'
+import {Filter} from './filter'
+import {Event} from './event'
+import {SubscriptionOptions, Sub, Pub} from './relay'
+
+export class SimplePool {
+  private _conn: {[url: string]: Relay}
+
+  constructor(defaultRelays: string[] = []) {
+    this._conn = {}
+    defaultRelays.forEach(this.ensureRelay)
+  }
+
+  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
+
+    const relay = relayInit(nm)
+    this._conn[nm] = relay
+
+    await relay.connect()
+
+    return relay
+  }
+
+  sub(relays: string[], filters: Filter[], opts?: SubscriptionOptions): Sub {
+    let _knownIds: Set<string> = new Set()
+    let modifiedOpts = opts || {}
+    modifiedOpts.alreadyHaveEvent = id => _knownIds.has(id)
+
+    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)
+        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) {
+        if (type === 'event') {
+          eventListeners.add(cb)
+        } else if (type === 'eose') eoseListeners.add(cb)
+      },
+      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[normalizeURL(relay)]
+      if (!r) return badPub(relay)
+      let s = r.publish(event)
+      return s
+    })
+  }
+}
+
+function badPub(relay: string): Pub {
+  return {
+    on(typ, cb) {
+      if (typ === 'failed') cb(`relay ${relay} not connected`)
+    },
+    off() {}
+  }
+}

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

@@ -2,7 +2,7 @@
 
 import {Event, verifySignature, validateEvent} from './event'
 import {Filter, matchFilters} from './filter'
-import {getHex64} from './fakejson'
+import {getHex64, getSubscriptionId} from './fakejson'
 
 type RelayEvent = 'connect' | 'disconnect' | 'error' | 'notice'
 
@@ -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
@@ -27,15 +29,13 @@ export type Sub = {
   off: (type: 'event' | 'eose', cb: any) => void
 }
 
-type SubscriptionOptions = {
+export type SubscriptionOptions = {
   skipVerification?: boolean
+  alreadyHaveEvent?: null | ((id: string) => boolean)
   id?: string
 }
 
-export function relayInit(
-  url: string,
-  alreadyHaveEvent: (id: string) => boolean = () => false
-): Relay {
+export function relayInit(url: string): Relay {
   var ws: WebSocket
   var resolveClose: () => void
   var setOpen: (value: PromiseLike<void> | void) => void
@@ -104,8 +104,18 @@ export function relayInit(
         }
 
         var json = incomingMessageQueue.shift()
-        if (!json || alreadyHaveEvent(getHex64(json, 'id'))) {
-          return
+        if (!json) return
+
+        let subid = getSubscriptionId(json)
+        if (subid) {
+          let so = openSubs[subid]
+          if (
+            so &&
+            so.alreadyHaveEvent &&
+            so.alreadyHaveEvent(getHex64(json, 'id'))
+          ) {
+            return
+          }
         }
 
         try {
@@ -173,6 +183,7 @@ export function relayInit(
     filters: Filter[],
     {
       skipVerification = false,
+      alreadyHaveEvent = null,
       id = Math.random().toString().slice(2)
     }: SubscriptionOptions = {}
   ): Sub => {
@@ -181,7 +192,8 @@ export function relayInit(
     openSubs[subid] = {
       id: subid,
       filters,
-      skipVerification
+      skipVerification,
+      alreadyHaveEvent
     }
     trySend(['REQ', subid, ...filters])
 
@@ -189,6 +201,7 @@ export function relayInit(
       sub: (newFilters, newOpts = {}) =>
         sub(newFilters || filters, {
           skipVerification: newOpts.skipVerification || skipVerification,
+          alreadyHaveEvent: newOpts.alreadyHaveEvent || alreadyHaveEvent,
           id: subid
         }),
       unsub: () => {
@@ -224,6 +237,36 @@ export function relayInit(
       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
@@ -281,6 +324,7 @@ export function relayInit(
     },
     connect,
     close(): Promise<void> {
+      if (ws.readyState > 1) return Promise.resolve()
       ws.close()
       return new Promise<void>(resolve => {
         resolveClose = resolve

utils.ts

@@ -3,6 +3,20 @@ import {Event} from './event'
 export const utf8Decoder = new TextDecoder('utf-8')
 export const utf8Encoder = new TextEncoder()
 
+export function normalizeURL(url: string): string {
+  let p = new URL(url)
+  p.pathname = p.pathname.replace(/\/+/g, '/')
+  if (p.pathname.endsWith('/')) p.pathname = p.pathname.slice(0, -1)
+  if (
+    (p.port === '80' && p.protocol === 'ws:') ||
+    (p.port === '443' && p.protocol === 'wss:')
+  )
+    p.port = ''
+  p.searchParams.sort()
+  p.hash = ''
+  return p.toString()
+}
+
 //
 // fast insert-into-sorted-array functions adapted from https://github.com/terrymorse58/fast-sorted-array
 //