// A Jetstream for a reference PDS, which has only its firehose: each
// commit to a knotless collection is sent on as Jetstream would frame it,
// with the record read back from the PDS, to whoever is connected. Enough
// for knotless-run, which only needs to be told to look.
//
//   const js = await jetstream(pdsUrl); ... js.url; js.close()

import { cborDecodeMulti } from '@atproto/common'
import { WebSocket, WebSocketServer } from 'ws'

const KIND = 'network.bsky.jetstream.subscribeEvents#'

export async function jetstream(pds) {
  const server = new WebSocketServer({
    host: '127.0.0.1',
    port: 0,
    // Jetstream answers only a client that offers its framing.
    handleProtocols: (offered) => (offered.has('xrpc.v1.json') ? 'xrpc.v1.json' : false),
  })
  await new Promise((ok) => server.once('listening', ok))
  const send = (payload) => {
    const frame = JSON.stringify({ $type: 'message', payload })
    for (const client of server.clients) client.send(frame)
  }
  const record = async (did, collection, rkey) => {
    const url = `${pds}/xrpc/com.atproto.repo.getRecord?repo=${did}&collection=${collection}&rkey=${encodeURIComponent(rkey)}`
    const res = await fetch(url).catch(() => null)
    return res?.ok ? (await res.json()).value : undefined
  }
  let firehose
  let closed = false
  const follow = () => {
    firehose = new WebSocket(`${pds.replace(/^http/, 'ws')}/xrpc/com.atproto.sync.subscribeRepos`)
    firehose.on('message', async (data) => {
      let header, body
      try {
        ;[header, body] = cborDecodeMulti(data)
      } catch {
        return
      }
      if (header?.t !== '#commit') return
      for (const op of body.ops ?? []) {
        const [collection, rkey] = op.path.split('/')
        if (!collection.startsWith('dev.knotless.')) continue
        const value = op.action === 'delete' ? undefined : await record(body.repo, collection, rkey)
        send({
          $type: `${KIND}commit`,
          seq: body.seq,
          did: body.repo,
          time: body.time,
          rev: body.rev,
          operation: op.action,
          collection,
          rkey,
          ...(value ? { record: value } : {}),
        })
      }
    })
    firehose.on('close', () => closed || setTimeout(follow, 500))
    firehose.on('error', () => {})
  }
  follow()
  return {
    url: `ws://127.0.0.1:${server.address().port}`,
    close: () => {
      closed = true
      firehose.close()
      server.close()
    },
  }
}