// One server, two architectures, identical domain output.
//
//   --arch ws    : single WebSocket carries commands up and HTML patches down
//   --arch sse   : GET /events (SSE) down, POST /command up  (CQRS split)
//
// Flags: --tls --h1 --compress <none|gzip|br|deflate> --mode <narrow|fat>
//        --port --control-port --rooms
//
// A control port (plain HTTP, never behind the delay proxy) exposes /stats so
// the harness can read RSS/CPU without perturbing the measured path.

import fs from 'node:fs'
import http from 'node:http'
import https from 'node:https'
import http2 from 'node:http2'
import net from 'node:net'
import tls from 'node:tls'
import zlib from 'node:zlib'
import { fileURLToPath } from 'node:url'
import path from 'node:path'
import { WebSocketServer } from 'ws'
import { makeBoard, apply, renderPatch, renderBoard, sseEvent, wsMessage } from './domain.js'

const __dirname = path.dirname(fileURLToPath(import.meta.url))

function parseArgs(argv) {
  const out = {}
  for (let i = 0; i < argv.length; i++) {
    const a = argv[i]
    if (!a.startsWith('--')) continue
    const key = a.slice(2).replace(/-/g, '_')
    const next = argv[i + 1]
    if (next === undefined || next.startsWith('--')) out[key] = true
    else out[key] = argv[++i]
  }
  return out
}

const args = parseArgs(process.argv.slice(2))
const ARCH = args.arch ?? 'sse'
const USE_TLS = !!args.tls
const USE_H1 = !!args.h1
const COMPRESS = args.compress ?? 'none'
const MODE = args.mode ?? 'narrow'
// Brotli window exponent. Matters a lot: it sets both the ratio ceiling AND
// the per-connection memory, and those pull in opposite directions.
const LGWIN = Number(args.lgwin ?? 22)
const WS_NO_CTX = !!args.ws_no_context_takeover
const PORT = Number(args.port ?? 0)
const CONTROL_PORT = Number(args.control_port ?? 0)

const board = makeBoard()

// ---------------------------------------------------------------- subscribers

// room -> Set<subscriber>. A subscriber is whatever the transport needs to
// push bytes to one client. This registry is the thing both architectures
// need; the article implies WS needs it and SSE doesn't, which is false.
const rooms = new Map()
let subCount = 0

function join(room, sub) {
  let set = rooms.get(room)
  if (!set) rooms.set(room, (set = new Set()))
  set.add(sub)
  subCount++
  return () => {
    set.delete(sub)
    subCount--
    if (set.size === 0) rooms.delete(room)
  }
}

function broadcast(room, framer) {
  const set = rooms.get(room)
  if (!set) return 0
  let n = 0
  for (const sub of set) {
    try {
      sub.send(framer)
      n++
    } catch {}
  }
  return n
}

// ------------------------------------------------------------------ compress

// A per-connection compression context, flushed on every patch. This is where
// SSE quietly wins on HTML: the dictionary persists across the whole stream,
// so the Nth copy of the same markup costs almost nothing.
function makeCompressor(kind, sink) {
  if (kind === 'none') {
    return { write: (buf) => sink(buf), end: () => {} }
  }
  const z =
    kind === 'br'
      ? zlib.createBrotliCompress({
          params: {
            [zlib.constants.BROTLI_PARAM_QUALITY]: 4,
            [zlib.constants.BROTLI_PARAM_LGWIN]: LGWIN,
            [zlib.constants.BROTLI_PARAM_MODE]: zlib.constants.BROTLI_MODE_TEXT,
          },
        })
      : kind === 'deflate'
        ? zlib.createDeflate({ level: 6 })
        : zlib.createGzip({ level: 6 })
  z.on('data', sink)
  z.on('error', () => {})
  const FLUSH = kind === 'br' ? zlib.constants.BROTLI_OPERATION_FLUSH : zlib.constants.Z_SYNC_FLUSH
  return {
    write(buf) {
      z.write(buf)
      z.flush(FLUSH)
    },
    end() {
      z.end()
    },
  }
}

const contentEncoding = { none: null, gzip: 'gzip', br: 'br', deflate: 'deflate' }[COMPRESS]

// --------------------------------------------------------------- SSE + CQRS

let eventId = 0

function handleSse(req, res, roomOf) {
  const room = roomOf
  const headers = {
    'content-type': 'text/event-stream',
    'cache-control': 'no-cache',
  }
  if (contentEncoding) headers['content-encoding'] = contentEncoding
  res.writeHead(200, headers)
  if (res.flushHeaders) res.flushHeaders()

  const comp = makeCompressor(COMPRESS, (buf) => {
    if (!res.destroyed && res.writable) res.write(buf)
  })

  const leave = join(room, {
    send: (framer) => comp.write(framer()),
  })

  // Initial render, exactly as a hypermedia app would.
  comp.write(sseEvent(renderBoard(board), ++eventId))

  const done = () => {
    leave()
    comp.end()
  }
  res.on('close', done)
  res.on('error', done)
}

function handleCommand(room, id, nonce) {
  const row = apply(board, { id, nonce })
  const html = renderPatch(board, row, MODE)
  const evId = ++eventId
  // One render, many sends: framer is called per subscriber but the string is
  // built once, so we are not measuring template cost.
  const sseFrame = sseEvent(html, evId)
  return broadcast(room, () => sseFrame)
}

function sseRequestHandler(req, res) {
  const url = new URL(req.url, 'http://x')
  if (url.pathname === '/events') {
    handleSse(req, res, url.searchParams.get('room') ?? '0')
    return
  }
  if (url.pathname === '/command') {
    // Read the body — a real command has one, and it costs bytes + a turn.
    let body = ''
    req.on('data', (c) => (body += c))
    req.on('end', () => {
      let cmd
      try {
        cmd = JSON.parse(body || '{}')
      } catch {
        cmd = {}
      }
      const n = handleCommand(String(cmd.room ?? '0'), Number(cmd.id ?? 0), cmd.nonce ?? 0)
      res.writeHead(204, { 'x-fanout': String(n) })
      res.end()
    })
    return
  }
  res.writeHead(404)
  res.end()
}

// ------------------------------------------------------------------- WS side

function attachWs(server) {
  const wss = new WebSocketServer({
    server,
    perMessageDeflate:
      COMPRESS === 'none'
        ? false
        : {
            zlibDeflateOptions: { level: 6 },
            // Context takeover is what makes WS competitive with SSE on
            // repeated HTML — and what costs ~256 KB of zlib state per
            // connection. Proxies routinely force it off to bound that.
            serverNoContextTakeover: WS_NO_CTX,
            clientNoContextTakeover: WS_NO_CTX,
            threshold: 0,
          },
  })

  wss.on('connection', (socket, req) => {
    const url = new URL(req.url, 'http://x')
    const room = url.searchParams.get('room') ?? '0'
    const leave = join(room, { send: (framer) => socket.send(framer()) })

    socket.send(wsMessage(renderBoard(board), ++eventId))

    socket.on('message', (data) => {
      let cmd
      try {
        cmd = JSON.parse(data)
      } catch {
        return
      }
      if (cmd.t !== 'cmd') return
      const row = apply(board, { id: Number(cmd.id ?? 0), nonce: cmd.nonce ?? 0 })
      const html = renderPatch(board, row, MODE)
      const msg = wsMessage(html, ++eventId)
      broadcast(String(cmd.room ?? room), () => msg)
    })
    socket.on('close', leave)
    socket.on('error', leave)
  })
  return wss
}

// ------------------------------------------------------------------- listen

const tlsOpts = () => ({
  key: fs.readFileSync(path.join(__dirname, '../certs/key.pem')),
  cert: fs.readFileSync(path.join(__dirname, '../certs/cert.pem')),
})

let server

if (ARCH === 'ws') {
  server = USE_TLS
    ? https.createServer({ ...tlsOpts(), ALPNProtocols: ['http/1.1'] }, (_, res) => {
        res.writeHead(404)
        res.end()
      })
    : http.createServer((_, res) => {
        res.writeHead(404)
        res.end()
      })
  attachWs(server)
} else if (USE_H1) {
  server = USE_TLS
    ? https.createServer({ ...tlsOpts(), ALPNProtocols: ['http/1.1'] }, sseRequestHandler)
    : http.createServer(sseRequestHandler)
  server.keepAliveTimeout = 120000
} else {
  // HTTP/2. allowHTTP1 stays off so we can be certain what we measured.
  const h2opts = { settings: { maxConcurrentStreams: 1000 } }
  server = USE_TLS
    ? http2.createSecureServer({ ...tlsOpts(), ALPNProtocols: ['h2'], ...h2opts })
    : http2.createServer(h2opts)
  server.on('stream', (stream, headers) => {
    const req = Object.assign(stream, {
      url: headers[':path'],
      method: headers[':method'],
    })
    const res = {
      writeHead(status, h) {
        stream.respond({ ':status': status, ...h })
      },
      write: (b) => stream.write(b),
      end: (b) => stream.end(b),
      on: (e, f) => stream.on(e, f),
      get destroyed() {
        return stream.destroyed
      },
      get writable() {
        return stream.writable
      },
    }
    sseRequestHandler(req, res)
  })
  // A long-lived SSE stream must not be reaped, and we need lots of them.
  server.on('session', (s) => {
    s.setLocalWindowSize?.(1 << 20)
  })
}

server.on('error', (e) => {
  console.error('server error', e.message)
  process.exit(1)
})

// ------------------------------------------------------------------ control

let cpuBase = process.cpuUsage()
const control = http
  .createServer((req, res) => {
    if (req.url === '/reset') {
      cpuBase = process.cpuUsage()
      if (global.gc) global.gc()
      res.writeHead(204)
      return res.end()
    }
    if (req.url === '/shutdown') {
      res.writeHead(204)
      res.end()
      return setTimeout(() => process.exit(0), 50)
    }
    if (global.gc && req.url === '/stats') global.gc()
    const mem = process.memoryUsage()
    const cpu = process.cpuUsage(cpuBase)
    res.writeHead(200, { 'content-type': 'application/json' })
    res.end(
      JSON.stringify({
        arch: ARCH,
        subscribers: subCount,
        rooms: rooms.size,
        rss: mem.rss,
        heapUsed: mem.heapUsed,
        external: mem.external,
        arrayBuffers: mem.arrayBuffers,
        cpuUserMs: cpu.user / 1000,
        cpuSystemMs: cpu.system / 1000,
      }),
    )
  })

// Both listeners bind port 0 by default and report what the OS handed out, so
// concurrent/aborted runs can never collide on a fixed port.
control.on('error', (e) => {
  console.error('control error', e.message)
  process.exit(1)
})
control.listen(CONTROL_PORT, '127.0.0.1', () => {
  server.listen(PORT, '127.0.0.1', () => {
    console.log(
      `[server] arch=${ARCH} tls=${USE_TLS} h1=${USE_H1} compress=${COMPRESS} mode=${MODE} ` +
        `lgwin=${LGWIN} port=${server.address().port} control=${control.address().port}`,
    )
  })
})