subscription.actor.ts raw

   1  import { Inbox, Lifecycle, spawn, loop, STOP } from 'anneal'
   2  import { EventTemplate, Filter, Event as NEvent, VerifiedEvent } from 'nostr-tools'
   3  import { SimplePool } from 'nostr-tools'
   4  import { AbstractRelay } from 'nostr-tools/abstract-relay'
   5  import { ISigner } from '@/types'
   6  
   7  type RelayEventMsg = { event: NEvent; url: string }
   8  type RelayEoseMsg = { url: string }
   9  type RelayCloseMsg = { reason: string; url: string }
  10  
  11  export type SubscriptionHandle = {
  12    lc: Lifecycle
  13    close: () => void
  14  }
  15  
  16  export type SubscriptionConfig = {
  17    urls: string[]
  18    filters: Filter[]
  19    pool: SimplePool
  20    signer?: ISigner
  21    startLogin?: () => void
  22    onevent?: (evt: NEvent) => void
  23    oneose?: (eosed: boolean) => void
  24    onclose?: (reasons: string[]) => void
  25    onReceived?: (id: string, relay: AbstractRelay) => void
  26  }
  27  
  28  export function createSubscription(config: SubscriptionConfig): SubscriptionHandle {
  29    const { urls, filters, pool, signer, startLogin, onReceived } = config
  30    let { onevent, oneose, onclose } = config
  31    const relays = Array.from(new Set(urls))
  32  
  33    const relayEvent = new Inbox<RelayEventMsg>(0)
  34    const relayEose = new Inbox<RelayEoseMsg>(0)
  35    const relayClose = new Inbox<RelayCloseMsg>(0)
  36  
  37    const lc = new Lifecycle()
  38    const subClosers: (() => void)[] = []
  39    const knownIds = new Set<string>()
  40  
  41    // Fix for race #1: count all relays BEFORE any async work.
  42    // EOSE check uses this fixed value - no interleaving possible.
  43    const startedCount = relays.length
  44    let eosedCount = 0
  45    let eosed = false
  46    let closedCount = 0
  47    const closeReasons: string[] = []
  48    const eosedRelays = new Set<string>()
  49    const authedRelays = new Set<string>()
  50  
  51    function connectRelay(url: string) {
  52      ;(async () => {
  53        let relay: AbstractRelay | undefined
  54        try {
  55          relay = await pool.ensureRelay(url, { connectionTimeout: 5000 })
  56        } catch {
  57          // Connection failed - count as EOSE so we don't block the threshold
  58          relayEose.send({ url })
  59          return
  60        }
  61        if (!relay) {
  62          relayEose.send({ url })
  63          return
  64        }
  65  
  66        const sub = relay.subscribe(filters, {
  67          receivedEvent: (r: AbstractRelay, id: string) => {
  68            onReceived?.(id, r)
  69          },
  70          alreadyHaveEvent: (id: string) => {
  71            if (knownIds.has(id)) return true
  72            knownIds.add(id)
  73            return false
  74          },
  75          onevent: (evt: NEvent) => {
  76            relayEvent.send({ event: evt, url })
  77          },
  78          oneose: () => {
  79            relayEose.send({ url })
  80          },
  81          onclose: (reason: string) => {
  82            relayClose.send({ reason, url })
  83          },
  84          eoseTimeout: 10_000
  85        })
  86        subClosers.push(() => sub.close())
  87      })()
  88    }
  89  
  90    // Fire-and-forget relay connections
  91    for (const url of relays) {
  92      connectRelay(url)
  93    }
  94  
  95    // Actor loop - processes relay messages one at a time
  96    spawn(lc, async () => {
  97      await loop(
  98        lc.on(() => {
  99          for (const closer of subClosers) closer()
 100          return STOP
 101        }),
 102        relayEvent.on(({ event }) => {
 103          onevent?.(event)
 104        }),
 105        relayEose.on(({ url }) => {
 106          if (eosed) return
 107          if (eosedRelays.has(url)) return
 108          eosedRelays.add(url)
 109          eosedCount++
 110          eosed = eosedCount >= startedCount
 111          oneose?.(eosed)
 112        }),
 113        relayClose.on(({ reason, url }) => {
 114          if (reason.startsWith('auth-required') && !authedRelays.has(url)) {
 115            if (signer) {
 116              authedRelays.add(url)
 117              ;(async () => {
 118                try {
 119                  const relay = await pool.ensureRelay(url)
 120                  await relay.auth(async (authEvt: EventTemplate) => {
 121                    const evt = await signer.signEvent(authEvt)
 122                    if (!evt) throw new Error('sign event failed')
 123                    return evt as VerifiedEvent
 124                  })
 125                  // Re-subscribe only if we haven't reached final EOSE yet.
 126                  // startedCount is NOT incremented - the re-sub fills the
 127                  // original relay's EOSE slot.
 128                  if (!eosed) connectRelay(url)
 129                } catch {
 130                  // Auth failed - count this relay's slot as EOSE
 131                  relayEose.send({ url })
 132                }
 133              })()
 134              return
 135            }
 136            if (startLogin) {
 137              startLogin()
 138              return
 139            }
 140          }
 141          closedCount++
 142          closeReasons.push(reason)
 143          if (closedCount >= startedCount) {
 144            onclose?.(closeReasons)
 145          }
 146        })
 147      )
 148    })
 149  
 150    return {
 151      lc,
 152      close: () => {
 153        onevent = undefined
 154        oneose = undefined
 155        onclose = undefined
 156        for (const closer of subClosers) closer()
 157        lc.stop()
 158      }
 159    }
 160  }
 161