// Uniform client facade over both architectures, so bench code is identical.
//
// The SSE client deliberately mirrors what a browser does per origin: ONE
// connection (h2 session / h1 keep-alive agent) carrying both the event stream
// and every command POST.

import http from 'node:http'
import https from 'node:https'
import http2 from 'node:http2'
import zlib from 'node:zlib'
import WebSocket from 'ws'

function makeDecompressor(kind, onText) {
  if (kind === 'none') {
    return { write: (b) => onText(b.toString('utf8')), end: () => {} }
  }
  const z =
    kind === 'br'
      ? zlib.createBrotliDecompress()
      : kind === 'deflate'
        ? zlib.createInflate()
        : zlib.createGunzip()
  z.on('data', (b) => onText(b.toString('utf8')))
  z.on('error', () => {})
  return { write: (b) => z.write(b), end: () => z.end() }
}

// Incremental SSE framing. Returns complete events as raw blocks.
function makeSseParser(onEvent) {
  let buf = ''
  return (text) => {
    buf += text
    let i
    while ((i = buf.indexOf('\n\n')) !== -1) {
      const block = buf.slice(0, i)
      buf = buf.slice(i + 2)
      onEvent(block)
    }
  }
}

// Matches both raw SSE payloads and the JSON-escaped form the WS envelope uses.
const NONCE_RE = /data-nonce=\\?"(\d+)\\?"/g

function noncesIn(s) {
  const out = []
  let m
  NONCE_RE.lastIndex = 0
  while ((m = NONCE_RE.exec(s))) out.push(Number(m[1]))
  return out
}

// ------------------------------------------------------------------ WS client

function wsClient({ host, port, tls, room, compress }) {
  const url = `${tls ? 'wss' : 'ws'}://${host}:${port}/?room=${room}`
  let sock
  const handlers = []
  let firstMsg
  return {
    connect() {
      return new Promise((resolve, reject) => {
        firstMsg = resolve
        sock = new WebSocket(url, {
          rejectUnauthorized: false,
          perMessageDeflate: compress !== 'none',
        })
        sock.on('error', reject)
        sock.on('message', (data) => {
          const s = data.toString('utf8')
          if (firstMsg) {
            const f = firstMsg
            firstMsg = null
            f()
            return
          }
          const ns = noncesIn(s)
          for (const h of handlers) h(ns, s)
        })
      })
    },
    onPatch(cb) {
      handlers.push(cb)
    },
    dispatch({ id, nonce }) {
      sock.send(JSON.stringify({ t: 'cmd', id, nonce, room }))
      return Promise.resolve()
    },
    close() {
      try {
        sock.close()
      } catch {}
    },
  }
}

// ------------------------------------------------------------- SSE/h2 client

function h2Client({ host, port, tls, room, compress }) {
  const origin = `${tls ? 'https' : 'http'}://${host}:${port}`
  let session
  let stream
  const handlers = []
  return {
    connect() {
      return new Promise((resolve, reject) => {
        session = http2.connect(origin, { rejectUnauthorized: false })
        session.on('error', reject)
        session.setMaxListeners(0)
        stream = session.request({
          ':path': `/events?room=${room}`,
          ':method': 'GET',
          'accept-encoding': compress === 'none' ? 'identity' : compress,
        })
        let first = true
        const parse = makeSseParser((block) => {
          if (first) {
            first = false
            resolve()
            return
          }
          const ns = noncesIn(block)
          for (const h of handlers) h(ns, block)
        })
        const dec = makeDecompressor(compress, parse)
        stream.on('data', (c) => dec.write(c))
        stream.on('error', reject)
        stream.on('response', () => {})
      })
    },
    onPatch(cb) {
      handlers.push(cb)
    },
    dispatch({ id, nonce }) {
      return new Promise((resolve, reject) => {
        const body = JSON.stringify({ id, nonce, room })
        // Same h2 session as the SSE stream -> new stream, not new TCP conn.
        const req = session.request({
          ':path': '/command',
          ':method': 'POST',
          'content-type': 'application/json',
          'content-length': Buffer.byteLength(body),
        })
        req.on('response', () => {})
        req.on('end', resolve)
        req.on('error', reject)
        req.resume()
        req.end(body)
      })
    },
    close() {
      try {
        stream?.close()
        session?.close()
      } catch {}
    },
  }
}

// ------------------------------------------------------------- SSE/h1 client

function h1Client({ host, port, tls, room, compress }) {
  const mod = tls ? https : http
  // A browser gives an origin 6 sockets on HTTP/1.1 and the SSE stream eats one.
  const agent = new mod.Agent({ keepAlive: true, maxSockets: 5, rejectUnauthorized: false })
  let streamReq
  const handlers = []
  return {
    connect() {
      return new Promise((resolve, reject) => {
        streamReq = mod.request(
          {
            host,
            port,
            path: `/events?room=${room}`,
            method: 'GET',
            rejectUnauthorized: false,
            headers: { 'accept-encoding': compress === 'none' ? 'identity' : compress },
            // The stream gets its own socket, exactly like the browser.
            agent: new mod.Agent({ keepAlive: true, maxSockets: 1, rejectUnauthorized: false }),
          },
          (res) => {
            let first = true
            const parse = makeSseParser((block) => {
              if (first) {
                first = false
                resolve()
                return
              }
              const ns = noncesIn(block)
              for (const h of handlers) h(ns, block)
            })
            const dec = makeDecompressor(compress, parse)
            res.on('data', (c) => dec.write(c))
            res.on('error', reject)
          },
        )
        streamReq.on('error', reject)
        streamReq.end()
      })
    },
    onPatch(cb) {
      handlers.push(cb)
    },
    dispatch({ id, nonce }) {
      return new Promise((resolve, reject) => {
        const body = JSON.stringify({ id, nonce, room })
        const req = mod.request(
          {
            host,
            port,
            path: '/command',
            method: 'POST',
            agent,
            rejectUnauthorized: false,
            headers: {
              'content-type': 'application/json',
              'content-length': Buffer.byteLength(body),
            },
          },
          (res) => {
            res.resume()
            res.on('end', resolve)
          },
        )
        req.on('error', reject)
        req.end(body)
      })
    },
    close() {
      try {
        streamReq?.destroy()
        agent.destroy()
      } catch {}
    },
  }
}

export function createClient(opts) {
  if (opts.arch === 'ws') return wsClient(opts)
  return opts.h1 ? h1Client(opts) : h2Client(opts)
}