nostr-tools/abstract-relay.ts

/* global WebSocket */

import type { Event, EventTemplate, VerifiedEvent, NostrEvent } from './core.ts'
import { matchFilters, type Filter } from './filter.ts'
import { getHex64, getSubscriptionId } from './fakejson.ts'
import { normalizeURL } from './utils.ts'
import { makeAuthEvent } from './nip42.ts'

type RelayWebSocket = WebSocket & {
  ping?(): void
  on?(event: 'pong', listener: () => void): any
}

export type AbstractRelayConstructorOptions = {
  verifyEvent: (event: Event, url: string) => boolean
  websocketImplementation?: typeof WebSocket
  enablePing?: boolean
  enableReconnect?: boolean
  idleTimeout?: number
}

export class SendingOnClosedConnection extends Error {
  constructor(message: string, relay: string) {
    super(`Tried to send message '${message} on a closed connection to ${relay}.`)
    this.name = 'SendingOnClosedConnection'
  }
}

export class AbstractRelay {
  public readonly url: string
  private _connected: boolean = false

  public onclose: (() => void) | null = null
  public onnotice: (msg: string) => void = msg => console.debug(`NOTICE from ${this.url}: ${msg}`)
  public onauth: undefined | ((evt: EventTemplate) => Promise<VerifiedEvent>)

  public baseEoseTimeout: number = 4400
  public publishTimeout: number = 4400
  public pingFrequency: number = 29000
  public pingTimeout: number = 20000
  public resubscribeBackoff: number[] = [10000, 10000, 10000, 20000, 20000, 30000, 60000]
  public openSubs: Map<string, Subscription> = new Map()
  public enablePing: boolean | undefined
  public enableReconnect: boolean
  public idleTimeout: number = 0 // auto-close after ms of inactivity, 0 = disabled
  public idleSince: number | undefined = Date.now() // when undefined that means it isn't idle
  public ongoingOperations: number = 0 // used to compute idleness
  private reconnectTimeoutHandle: ReturnType<typeof setTimeout> | undefined
  private pingIntervalHandle: ReturnType<typeof setInterval> | undefined
  private reconnectAttempts: number = 0
  private skipReconnection: boolean = false
  private idleTimeoutHandle: ReturnType<typeof setTimeout> | undefined

  private connectionPromise: Promise<void> | undefined
  private openCountRequests = new Map<string, CountResolver>()
  private openEventPublishes = new Map<string, EventPublishResolver>()
  private ws: RelayWebSocket | undefined
  private challenge: string | undefined
  private authPromise: Promise<string> | undefined
  private serial: number = 0
  private verifyEvent: (event: Event, url: string) => boolean

  private _WebSocket: typeof WebSocket

  constructor(url: string, opts: AbstractRelayConstructorOptions) {
    this.url = normalizeURL(url)
    this.verifyEvent = opts.verifyEvent
    this._WebSocket = opts.websocketImplementation || WebSocket
    this.enablePing = opts.enablePing
    this.enableReconnect = opts.enableReconnect || false
    if (opts.idleTimeout) this.idleTimeout = opts.idleTimeout
  }

  static async connect(
    url: string,
    opts: AbstractRelayConstructorOptions & Parameters<AbstractRelay['connect']>[0],
  ): Promise<AbstractRelay> {
    const relay = new AbstractRelay(url, opts)
    await relay.connect(opts)
    return relay
  }

  private closeAllSubscriptions(reason: string) {
    for (let [_, sub] of this.openSubs) {
      sub.close(reason)
    }
    this.openSubs.clear()

    for (let [_, ep] of this.openEventPublishes) {
      ep.reject(new Error(reason))
    }
    this.openEventPublishes.clear()

    for (let [_, cr] of this.openCountRequests) {
      cr.reject(new Error(reason))
    }
    this.openCountRequests.clear()
  }

  public get connected(): boolean {
    return this._connected
  }

  private clearIdleTimeout() {
    if (this.idleTimeoutHandle) {
      clearTimeout(this.idleTimeoutHandle)
      this.idleTimeoutHandle = undefined
    }
  }

  public scheduleIdleClose() {
    this.clearIdleTimeout()
    if (this.idleTimeout > 0) {
      this.idleTimeoutHandle = setTimeout(() => {
        if (this.ongoingOperations === 0 && this.idleSince) {
          this.close()
        }
      }, this.idleTimeout)
    }
  }

  private async reconnect(): Promise<void> {
    const backoff = this.resubscribeBackoff[Math.min(this.reconnectAttempts, this.resubscribeBackoff.length - 1)]
    this.reconnectAttempts++

    this.reconnectTimeoutHandle = setTimeout(async () => {
      try {
        await this.connect()
      } catch (err) {
        // this will be called again through onclose/onerror
      }
    }, backoff)
  }

  private handleHardClose(reason: string) {
    // browsers and the node `ws` library fire BOTH onerror and onclose on a
    // failed/abnormal socket. detach the handlers on the first invocation so the
    // error->close pair collapses into a single terminal action -- otherwise
    // onclose would fire twice (terminal paths) and reconnect() would be
    // scheduled twice (reconnect paths, skipping a backoff slot).
    if (this.ws) {
      this.ws.onopen = null
      this.ws.onerror = null
      this.ws.onclose = null
    }

    if (this.pingIntervalHandle) {
      clearInterval(this.pingIntervalHandle)
      this.pingIntervalHandle = undefined
    }

    this._connected = false
    this.connectionPromise = undefined
    this.idleSince = undefined
    this.clearIdleTimeout()

    if (this.enableReconnect && !this.skipReconnection) {
      this.reconnect()
    } else {
      this.onclose?.()
      this.closeAllSubscriptions(reason)
    }
  }

  public async connect(opts?: { timeout?: number; abort?: AbortSignal }): Promise<void> {
    let connectionTimeoutHandle: ReturnType<typeof setTimeout> | undefined

    if (this.connectionPromise) return this.connectionPromise

    this.challenge = undefined
    this.authPromise = undefined
    this.skipReconnection = false
    this.connectionPromise = new Promise((resolve, reject) => {
      if (opts?.timeout) {
        connectionTimeoutHandle = setTimeout(() => {
          reject('connection timed out')
          this.connectionPromise = undefined
          // only give up on the initial connect; a slow reconnect
          // should fall through to the next backoff slot.
          if (this.reconnectAttempts === 0) {
            this.skipReconnection = true
          }
          this.ws?.close?.()
          this.handleHardClose('relay connection timed out')
        }, opts.timeout)
      }

      if (opts?.abort) {
        opts.abort.onabort = reject
      }

      try {
        this.ws = new this._WebSocket(this.url)
      } catch (err) {
        clearTimeout(connectionTimeoutHandle)
        reject(err)
        return
      }

      this.ws.onopen = () => {
        if (this.reconnectTimeoutHandle) {
          clearTimeout(this.reconnectTimeoutHandle)
          this.reconnectTimeoutHandle = undefined
        }
        clearTimeout(connectionTimeoutHandle)
        this._connected = true

        const isReconnection = this.reconnectAttempts > 0
        this.reconnectAttempts = 0

        // resubscribe to all open subscriptions
        for (const sub of this.openSubs.values()) {
          sub.eosed = false
          if (isReconnection) {
            for (let f = 0; f < sub.filters.length; f++) {
              if (sub.lastEmitted) {
                sub.filters[f].since = sub.lastEmitted + 1
              }
            }
          }
          sub.fire()
        }

        if (this.enablePing) {
          this.pingIntervalHandle = setInterval(() => this.pingpong(), this.pingFrequency)
        }
        resolve()
      }

      this.ws.onerror = () => {
        clearTimeout(connectionTimeoutHandle)
        reject('connection failed')
        this.connectionPromise = undefined
        // only give up on the initial connect. a failed reconnect
        // attempt must fall through so the next backoff slot fires;
        // otherwise one failed retry tears down every subscription.
        if (this.reconnectAttempts === 0) {
          this.skipReconnection = true
        }
        this.ws?.close?.()
        this.handleHardClose('relay connection failed')
      }

      this.ws.onclose = ev => {
        clearTimeout(connectionTimeoutHandle)
        reject((ev as any).message || 'websocket closed')
        this.handleHardClose('relay connection closed')
      }

      this.ws.onmessage = this._onmessage.bind(this)
    })

    return this.connectionPromise
  }

  private waitForPingPong() {
    return new Promise(resolve => {
      // listen for pong
      ;(this.ws as any).once('pong', () => resolve(true))
      // send a ping
      this.ws!.ping!()
    })
  }

  private waitForDummyReq() {
    return new Promise((resolve, reject) => {
      if (!this.connectionPromise) return reject(new Error(`no connection to ${this.url}, can't ping`))

      // make a dummy request with expected empty eose reply
      // ["REQ", "_", {"ids":["aaaa...aaaa"], "limit": 1}]
      try {
        const sub = this.subscribe(
          [{ ids: ['aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'], limit: 1 }],
          {
            label: '<forced-ping>',
            oneose: () => {
              resolve(true)
              sub.close()
            },
            onclose() {
              // if we get a CLOSED it's because the relay is alive
              resolve(true)
            },
            eoseTimeout: this.pingTimeout + 1000,
          },
        )
      } catch (err) {
        reject(err)
      }
    })
  }

  // nodejs requires this magic here to ensure connections are closed when internet goes off and stuff
  // in browsers it's done automatically. see https://github.com/nbd-wtf/nostr-tools/issues/491
  private async pingpong() {
    // if the websocket is connected
    if (this.ws?.readyState === 1) {
      // wait for either a ping-pong reply or a timeout
      const result = await Promise.any([
        // browsers don't have ping so use a dummy req
        this.ws && this.ws.ping && (this.ws as any).once ? this.waitForPingPong() : this.waitForDummyReq(),
        new Promise(res => setTimeout(() => res(false), this.pingTimeout)),
      ])

      if (!result) {
        // pingpong closing socket
        if (this.ws?.readyState === this._WebSocket.OPEN) {
          this.ws?.close()
        }
      }
    }
  }

  public async send(message: string) {
    if (!this.connectionPromise) throw new SendingOnClosedConnection(message, this.url)

    this.connectionPromise.then(() => {
      this.ws?.send(message)
    })
  }

  public async auth(signAuthEvent: (evt: EventTemplate) => Promise<VerifiedEvent>): Promise<string> {
    const challenge = this.challenge
    if (!challenge) throw new Error("can't perform auth, no challenge was received")
    if (this.authPromise) return this.authPromise

    const attempt = new Promise<string>(async (resolve, reject) => {
      try {
        let evt = await signAuthEvent(makeAuthEvent(this.url, challenge))
        let timeout = setTimeout(() => {
          let ep = this.openEventPublishes.get(evt.id) as EventPublishResolver
          if (ep) {
            ep.reject(new Error('auth timed out'))
            this.openEventPublishes.delete(evt.id)
          }
        }, this.publishTimeout)
        this.openEventPublishes.set(evt.id, { resolve, reject, timeout })
        this.send('["AUTH",' + JSON.stringify(evt) + ']')
      } catch (err) {
        reject(err)
      }
    })
    // a failed attempt is dropped so the next call retries
    attempt.catch(() => {
      if (this.authPromise === attempt) this.authPromise = undefined
    })
    this.authPromise = attempt
    return attempt
  }

  public async publish(event: Event): Promise<string> {
    this.idleSince = undefined
    this.clearIdleTimeout()
    this.ongoingOperations++

    const ret = new Promise<string>((resolve, reject) => {
      const timeout = setTimeout(() => {
        const ep = this.openEventPublishes.get(event.id) as EventPublishResolver
        if (ep) {
          ep.reject(new Error('publish timed out'))
          this.openEventPublishes.delete(event.id)
        }
      }, this.publishTimeout)
      this.openEventPublishes.set(event.id, { resolve, reject, timeout })
    })
    try {
      await this.send('["EVENT",' + JSON.stringify(event) + ']')
    } catch (err) {
      const ep = this.openEventPublishes.get(event.id)
      if (ep) {
        ep.reject(err as Error)
        this.openEventPublishes.delete(event.id)
      }
    }

    // compute idleness state
    this.ongoingOperations--
    if (this.ongoingOperations === 0) {
      this.idleSince = Date.now()
      this.scheduleIdleClose()
    }

    return ret
  }

  public async count(filters: Filter[], params: { id?: string | null }): Promise<number> {
    return (await this.countWithHLL(filters, params)).count
  }

  public async countWithHLL(
    filters: Filter[],
    params: { id?: string | null },
  ): Promise<{ count: number; hll?: string }> {
    this.serial++
    const id = params?.id || 'count:' + this.serial
    const ret = new Promise<{ count: number; hll?: string }>((resolve, reject) => {
      this.openCountRequests.set(id, { resolve, reject })
    })
    try {
      await this.send('["COUNT","' + id + '",' + JSON.stringify(filters).substring(1))
    } catch (err) {
      const cr = this.openCountRequests.get(id)
      if (cr) {
        cr.reject(err as Error)
        this.openCountRequests.delete(id)
      }
    }
    return ret
  }

  public subscribe(
    filters: Filter[],
    params: Partial<SubscriptionParams> & { label?: string; id?: string },
  ): Subscription {
    if (params.label !== '<forced-ping>') {
      this.idleSince = undefined
      this.clearIdleTimeout()
      this.ongoingOperations++
    }

    const sub = this.prepareSubscription(filters, params)
    sub.fire()

    if (params.abort) {
      params.abort.onabort = () => sub.close(String(params.abort!.reason || '<aborted>'))
    }

    return sub
  }

  public prepareSubscription(
    filters: Filter[],
    params: Partial<SubscriptionParams> & { label?: string; id?: string },
  ): Subscription {
    this.serial++
    const id = params.id || (params.label ? params.label + ':' : 'sub:') + this.serial
    const sub = new Subscription(this, id, filters, params)
    this.openSubs.set(id, sub)
    return sub
  }

  public close() {
    this.skipReconnection = true
    if (this.reconnectTimeoutHandle) {
      clearTimeout(this.reconnectTimeoutHandle)
      this.reconnectTimeoutHandle = undefined
    }
    if (this.pingIntervalHandle) {
      clearInterval(this.pingIntervalHandle)
      this.pingIntervalHandle = undefined
    }
    this.closeAllSubscriptions('relay connection closed by us')
    this._connected = false
    this.connectionPromise = undefined
    this.idleSince = undefined
    this.clearIdleTimeout()
    this.onclose?.()
    if (this.ws) {
      // detach the handlers before closing so the resulting ws.onclose doesn't
      // route into handleHardClose and fire onclose / closeAllSubscriptions a
      // second time -- this user-initiated close() is the single owner of that.
      this.ws.onopen = null
      this.ws.onerror = null
      this.ws.onclose = null
      // close the socket unless it's already closing/closed; this also aborts a
      // still-CONNECTING socket so it can't open and linger untracked.
      if (this.ws.readyState !== this._WebSocket.CLOSING && this.ws.readyState !== this._WebSocket.CLOSED) {
        this.ws.close()
      }
    }
  }

  // this is the function assigned to this.ws.onmessage
  // it's exposed for testing and debugging purposes
  public _onmessage(ev: MessageEvent<any>): void {
    const json = ev.data
    if (!json) {
      return
    }

    // shortcut EVENT sub
    const subid = getSubscriptionId(json)
    if (subid) {
      const so = this.openSubs.get(subid as string)
      if (!so) {
        // this is an EVENT message, but for a subscription we don't have, so just stop here
        return
      }

      // this will be called only when this message is a EVENT message for a subscription we have
      // we do this before parsing the JSON to not have to do that for duplicate events
      //   since JSON parsing is slow
      const id = getHex64(json, 'id')
      const alreadyHave = so.alreadyHaveEvent?.(id)

      // notify any interested client that the relay has this event
      // (do this after alreadyHaveEvent() because the client may rely on this to answer that)
      so.receivedEvent?.(this, id)

      if (alreadyHave) {
        // if we had already seen this event we can just stop here
        return
      }
    }

    try {
      let data = JSON.parse(json)
      // we won't do any checks against the data since all failures (i.e. invalid messages from relays)
      // will naturally be caught by the encompassing try..catch block

      switch (data[0]) {
        case 'EVENT': {
          const so = this.openSubs.get(data[1] as string) as Subscription
          const event = data[2] as NostrEvent
          if (matchFilters(so.filters, event) && this.verifyEvent(event, this.url)) {
            so.onevent(event)
          } else {
            so.oninvalidevent?.(event)
          }
          if (!so.lastEmitted || so.lastEmitted < event.created_at) so.lastEmitted = event.created_at
          return
        }
        case 'COUNT': {
          const id: string = data[1]
          const payload = data[2] as { count: number; hll?: string }
          const cr = this.openCountRequests.get(id) as CountResolver
          if (cr) {
            cr.resolve(payload)
            this.openCountRequests.delete(id)
          }
          return
        }
        case 'EOSE': {
          const so = this.openSubs.get(data[1] as string)
          if (!so) return
          so.receivedEose()
          return
        }
        case 'OK': {
          const id: string = data[1]
          const ok: boolean = data[2]
          const reason: string = data[3]
          const ep = this.openEventPublishes.get(id) as EventPublishResolver
          if (ep) {
            clearTimeout(ep.timeout)
            if (ok) ep.resolve(reason)
            else ep.reject(new Error(reason))
            this.openEventPublishes.delete(id)
          }
          return
        }
        case 'CLOSED': {
          const id: string = data[1]
          const so = this.openSubs.get(id)
          if (!so) {
            const cr = this.openCountRequests.get(id) as CountResolver
            if (cr) {
              cr.reject(new Error(data[2] as string))
              this.openCountRequests.delete(id)
            }
            return
          }
          so.closed = true
          so.close(data[2] as string)
          return
        }
        case 'NOTICE': {
          this.onnotice(data[1] as string)
          return
        }
        case 'AUTH': {
          this.challenge = data[1] as string
          if (this.onauth) {
            this.auth(this.onauth).catch(err => {
              // If the connection closed before auth could be sent, just ignore it.
              // This is a race condition when relays close connections quickly
              // (e.g., WoT-enforced relays that reject unknown pubkeys).
              if (!(err instanceof SendingOnClosedConnection)) console.warn('automatic auth failed:', err)
            })
          }
          return
        }
        default: {
          const so = this.openSubs.get(data[1])
          so?.oncustom?.(data)
          return
        }
      }
    } catch (err) {
      try {
        const [_, __, event] = JSON.parse(json)
        console.warn(`[nostr] relay ${this.url} error processing message:`, err, event)
      } catch (_) {
        console.warn(`[nostr] relay ${this.url} error processing message:`, err)
      }
      return
    }
  }
}

export class Subscription {
  public readonly relay: AbstractRelay
  public readonly id: string

  public lastEmitted: number | undefined
  public closed: boolean = false
  public eosed: boolean = false
  public filters: Filter[]
  public alreadyHaveEvent: ((id: string) => boolean) | undefined
  public receivedEvent: ((relay: AbstractRelay, id: string) => void) | undefined

  public onevent: (evt: Event) => void
  public oninvalidevent: ((evt: unknown) => void) | undefined
  public oneose: (() => void) | undefined
  public onclose: ((reason: string) => void) | undefined

  // will get any messages that have this subscription id as their second item and are not default standard
  public oncustom: ((msg: string[]) => void) | undefined

  public eoseTimeout: number
  private eoseTimeoutHandle: ReturnType<typeof setTimeout> | undefined

  constructor(relay: AbstractRelay, id: string, filters: Filter[], params: SubscriptionParams) {
    if (filters.length === 0) throw new Error("subscription can't be created with zero filters")

    this.relay = relay
    this.filters = filters
    this.id = id
    this.alreadyHaveEvent = params.alreadyHaveEvent
    this.receivedEvent = params.receivedEvent
    this.eoseTimeout = params.eoseTimeout || relay.baseEoseTimeout

    this.oneose = params.oneose
    this.onclose = params.onclose
    this.oninvalidevent = params.oninvalidevent
    this.onevent =
      params.onevent ||
      (event => {
        console.warn(
          `onevent() callback not defined for subscription '${this.id}' in relay ${this.relay.url}. event received:`,
          event,
        )
      })
  }

  public fire() {
    this.relay.send('["REQ","' + this.id + '",' + JSON.stringify(this.filters).substring(1))

    // only now we start counting the eoseTimeout
    this.eoseTimeoutHandle = setTimeout(this.receivedEose.bind(this), this.eoseTimeout)
  }

  public receivedEose() {
    if (this.eosed) return
    clearTimeout(this.eoseTimeoutHandle)
    this.eosed = true
    this.oneose?.()
  }

  public close(reason: string = 'closed by caller') {
    if (!this.closed && this.relay.connected) {
      // if the connection was closed by the user calling .close() we will send a CLOSE message
      // otherwise this._open will be already set to false so we will skip this
      try {
        this.relay.send('["CLOSE",' + JSON.stringify(this.id) + ']')
      } catch (err) {
        if (err instanceof SendingOnClosedConnection) {
          /* doesn't matter, it's ok */
        } else {
          throw err
        }
      }
      this.closed = true
    }
    this.relay.openSubs.delete(this.id)

    // compute idleness state
    if (!this.id.startsWith('<forced-ping>')) {
      this.relay.ongoingOperations--
      if (this.relay.ongoingOperations === 0) {
        this.relay.idleSince = Date.now()
        this.relay.scheduleIdleClose()
      }
    }

    this.onclose?.(reason)
  }
}

export type SubscriptionParams = {
  onevent?: (evt: Event) => void
  oninvalidevent?: (evt: unknown) => void
  oneose?: () => void
  onclose?: (reason: string) => void
  alreadyHaveEvent?: (id: string) => boolean
  receivedEvent?: (relay: AbstractRelay, id: string) => void
  eoseTimeout?: number
  abort?: AbortSignal
}

export type CountResolver = {
  resolve: (payload: { count: number; hll?: string }) => void
  reject: (err: Error) => void
}

export type EventPublishResolver = {
  resolve: (reason: string) => void
  reject: (err: Error) => void
  timeout: ReturnType<typeof setTimeout>
}