// Persistent carrier client (tunnel.mode:'relay') — replaces cloudflared.
//
// Opens ONE long-lived outbound WSS carrier to this bot's Durable Object at
// wss://<handle>/__morphy/carrier, authenticated by a short-lived Ed25519 ticket. The DO
// muxes browser HTTP + WebSocket traffic down the carrier; this client demuxes each stream
// and replays it to the local 127.0.0.1:<port> server (exactly where cloudflared delivered it).
//
// A reconnect is just a redial of the same DO — no random URL, no relay re-registration, no
// DNS propagation. See fluxy-5318/edge/src/{protocol,bot-do}.js for the DO side.
//
// Backpressure: response bodies stream to the DO in ≤64KB DATA frames; when the socket's
// bufferedAmount grows past a high-water mark we pause the local response and resume once it
// drains. No explicit credit windows (their accounting bug truncated any response >~700KB).
//
// SECURITY (report §5.4, fatal #2): every replayed HTTP request carries the real client IP in
// cf-connecting-ip AND an unconditional x-morphy-tunnel marker, and any client-supplied copies
// of those are stripped first. The supervisor's loopback guards reject both, so the seizure
// endpoints (/__bloby/control/*, channel mutations, agent api) stay unreachable from the public
// path even though carrier traffic arrives on 127.0.0.1.

import http from 'node:http';
import { WebSocket } from 'ws';
import { fetchTicket } from '../shared/relay.js';
import { log } from '../shared/logger.js';
import type { BotConfig } from '../shared/config.js';

// ── wire protocol (mirror of edge/src/protocol.js) ──────────────────────────
const T = { HELLO: 0x00, HELLO_ACK: 0x01, PING: 0x02, PONG: 0x03, GOAWAY: 0x04, OPEN: 0x10, RESP: 0x11, DATA: 0x12, CLOSE: 0x14, RESET: 0x15 } as const;
const F = { END: 0x01, WS_BINARY: 0x02 } as const;
const HEADER = 6;
const CHUNK = 64 * 1024;
const HIGH_WATER = 8 * 1024 * 1024;  // pause the local response when the socket buffer exceeds this
const LOW_WATER = 1 * 1024 * 1024;   // resume once it drains below this
const PING_MS = 15_000;
const PONG_TIMEOUT_MS = 30_000;

const DOMAIN = 'morphyagent.com';

function enc(type: number, flags: number, sid: number, payload?: Buffer | null): Buffer {
  const body = payload && payload.length ? payload : Buffer.alloc(0);
  const buf = Buffer.allocUnsafe(HEADER + body.length);
  buf[0] = type; buf[1] = flags; buf.writeUInt32BE(sid >>> 0, 2);
  if (body.length) body.copy(buf, HEADER);
  return buf;
}
function encJson(type: number, sid: number, obj: unknown, flags = 0): Buffer {
  return enc(type, flags, sid, Buffer.from(JSON.stringify(obj)));
}
function dec(buf: Buffer) {
  return { type: buf[0], flags: buf[1], sid: buf.readUInt32BE(2), payload: buf.subarray(HEADER) };
}

interface HttpStream {
  kind: 'http';
  req: http.ClientRequest;
  res?: http.IncomingMessage;
  drainTimer?: NodeJS.Timeout;
}
interface WsStream {
  kind: 'ws';
  local: WebSocket;
  opened: boolean;
  backlog: { data: Buffer; binary: boolean }[];
}
type Stream = HttpStream | WsStream;

export class RelayTunnel {
  private config: BotConfig;
  private host: string;
  private ws: WebSocket | null = null;
  private streams = new Map<number, Stream>();
  private ticket: string | null = null;
  private ticketAt = 0;
  private closed = false;
  private reconnectTimer: NodeJS.Timeout | null = null;
  private pingTimer: NodeJS.Timeout | null = null;
  private lastPong = 0;
  private attempt = 0;
  private generation = 0; // bumps on every teardown so stale socket callbacks are ignored
  private onFirstConnect: (() => void) | null = null;

  constructor(config: BotConfig) {
    this.config = config;
    const tier = config.relay?.tier || 'at';
    this.host = tier === 'at'
      ? `${config.username}.open.${DOMAIN}`
      : `${config.username}.${DOMAIN}`;
  }

  get publicUrl(): string { return `https://${this.host}`; }
  isConnected(): boolean { return !!this.ws && this.ws.readyState === WebSocket.OPEN; }

  /**
   * Dial and resolve true once the carrier opens, or false if it hasn't within timeoutMs
   * (the caller then falls back to a quick tunnel). Background reconnects continue regardless;
   * only the FIRST successful open resolves this.
   */
  connect(timeoutMs = 15_000): Promise<boolean> {
    this.closed = false;
    return new Promise<boolean>((resolve) => {
      let settled = false;
      const finish = (ok: boolean) => { if (settled) return; settled = true; this.onFirstConnect = null; resolve(ok); };
      this.onFirstConnect = () => finish(true);
      setTimeout(() => finish(false), timeoutMs);
      this.dial();
    });
  }

  /** Force a full teardown + immediate reconnect (called by the wake/network-change hook). */
  reconnectNow(): void {
    if (this.closed) return;
    log.info('[carrier] forced reconnect');
    this.teardown();
    this.scheduleReconnect(true);
  }

  close(): void {
    this.closed = true;
    this.teardown();
    if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; }
  }

  // ── ticket ────────────────────────────────────────────────────────────────
  private async getTicket(): Promise<string> {
    if (this.ticket && Date.now() - this.ticketAt < 240_000) return this.ticket;
    const { ticket } = await fetchTicket(this.config.relay.token);
    this.ticket = ticket; this.ticketAt = Date.now();
    return ticket;
  }

  // ── dial the single carrier ─────────────────────────────────────────────────
  private async dial(): Promise<void> {
    const gen = ++this.generation;
    let ticket: string;
    try { ticket = await this.getTicket(); }
    catch (e) { log.warn(`[carrier] ticket: ${e instanceof Error ? e.message : e}`); return this.scheduleReconnect(); }
    if (gen !== this.generation || this.closed) return;

    const ws = new WebSocket(`wss://${this.host}/__morphy/carrier?role=control`, {
      headers: { Authorization: `Bearer ${ticket}` },
    });
    this.ws = ws;

    const onFail = (why: string) => {
      if (gen !== this.generation) return;
      log.warn(`[carrier] ${why}`);
      this.teardown();
      this.scheduleReconnect();
    };

    ws.on('open', () => {
      if (gen !== this.generation) return;
      ws.send(encJson(T.HELLO, 0, { v: 1 }));
      log.ok(`[carrier] connected → ${this.host}`);
      this.attempt = 0;
      this.lastPong = Date.now();
      this.startPing();
      const f = this.onFirstConnect; this.onFirstConnect = null; if (f) f();
    });
    ws.on('message', (data: Buffer, isBinary: boolean) => { if (gen === this.generation && isBinary) this.onFrame(data); });
    ws.on('close', () => onFail('closed'));
    ws.on('error', (e) => onFail(`error: ${e.message}`));
  }

  private startPing(): void {
    if (this.pingTimer) clearInterval(this.pingTimer);
    this.pingTimer = setInterval(() => {
      if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return;
      if (Date.now() - this.lastPong > PONG_TIMEOUT_MS) { this.reconnectNow(); return; }
      try { this.ws.send(enc(T.PING, 0, 0, Buffer.alloc(0))); } catch {}
    }, PING_MS);
  }

  private teardown(): void {
    this.generation++; // invalidate in-flight socket callbacks
    if (this.pingTimer) { clearInterval(this.pingTimer); this.pingTimer = null; }
    for (const [, s] of this.streams) {
      try {
        if (s.kind === 'http') { if (s.drainTimer) clearInterval(s.drainTimer); s.req.destroy(); }
        else s.local.close();
      } catch {}
    }
    this.streams.clear();
    if (this.ws) {
      const ws = this.ws;
      try { ws.removeAllListeners(); } catch {}
      // terminate() on a still-CONNECTING socket makes ws emit 'error' (abortHandshake). With no
      // listener that would be an uncaughtException — keep a no-op error sink through teardown.
      ws.on('error', () => {});
      try { ws.terminate(); } catch {}
    }
    this.ws = null;
  }

  private scheduleReconnect(immediate = false): void {
    if (this.closed || this.reconnectTimer) return;
    const base = immediate ? 0 : Math.min(30_000, 500 * 2 ** this.attempt);
    const delay = immediate ? 0 : Math.floor(Math.random() * (base || 500));
    this.attempt = Math.min(this.attempt + 1, 6);
    this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; this.dial(); }, delay);
  }

  private send(frame: Buffer): void {
    try { this.ws?.send(frame); } catch {}
  }

  // ── frame dispatch ──────────────────────────────────────────────────────────
  private onFrame(data: Buffer): void {
    const { type, flags, sid, payload } = dec(data);
    switch (type) {
      case T.HELLO_ACK: return;
      case T.PING: this.send(enc(T.PONG, 0, 0, Buffer.from(payload))); return;
      case T.PONG: this.lastPong = Date.now(); return;
      case T.GOAWAY: { const r = safeJson(payload); log.warn(`[carrier] GOAWAY: ${r?.reason || 'server'}`); return; }
      case T.OPEN: return this.onOpen(sid, safeJson(payload));
      case T.DATA: return this.onData(sid, flags, payload);
      case T.CLOSE: return this.onClose(sid);
      case T.RESET: return this.onClose(sid);
    }
  }

  private onOpen(sid: number, o: any): void {
    if (!o) return;
    if (o.kind === 'ws') return this.openWs(sid, o);
    return this.openHttp(sid, o);
  }

  // ── HTTP stream ─────────────────────────────────────────────────────────────
  private openHttp(sid: number, o: any): void {
    const headers = this.replayHeaders(o.headers, o.remoteIp);
    const req = http.request({ host: '127.0.0.1', port: this.config.port, method: o.method, path: o.url, headers });
    const st: HttpStream = { kind: 'http', req };
    this.streams.set(sid, st);

    req.on('response', (res) => {
      st.res = res;
      const hobj: Record<string, string> = {};
      for (const [k, v] of Object.entries(res.headers)) { if (v != null) hobj[k] = Array.isArray(v) ? v.join(', ') : String(v); }
      this.send(encJson(T.RESP, sid, { status: res.statusCode, statusText: res.statusMessage || '', headers: hobj }));

      res.on('data', (chunk: Buffer) => {
        for (let off = 0; off < chunk.length; off += CHUNK) {
          this.send(enc(T.DATA, 0, sid, chunk.subarray(off, Math.min(off + CHUNK, chunk.length))));
        }
        // Backpressure: if the socket is backing up, pause the local response until it drains.
        if (this.ws && this.ws.bufferedAmount > HIGH_WATER && !st.drainTimer) {
          res.pause();
          st.drainTimer = setInterval(() => {
            if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { if (st.drainTimer) clearInterval(st.drainTimer); st.drainTimer = undefined; return; }
            if (this.ws.bufferedAmount < LOW_WATER) { if (st.drainTimer) clearInterval(st.drainTimer); st.drainTimer = undefined; res.resume(); }
          }, 25);
        }
      });
      res.on('end', () => { this.send(enc(T.DATA, F.END, sid, null)); this.cleanupHttp(sid); });
      res.on('error', () => this.reset(sid));
    });
    req.on('error', () => this.reset(sid));
  }

  private onData(sid: number, flags: number, payload: Buffer): void {
    const st = this.streams.get(sid);
    if (!st) return;
    if (st.kind === 'http') {
      if (payload.length) { try { st.req.write(Buffer.from(payload)); } catch {} }
      if (flags & F.END) { try { st.req.end(); } catch {} }
    } else {
      const buf = Buffer.from(payload);
      const binary = !!(flags & F.WS_BINARY);
      if (!st.opened) { st.backlog.push({ data: buf, binary }); return; }
      try { st.local.send(binary ? buf : buf.toString('utf8'), { binary }); } catch {}
    }
  }

  // ── WS stream ───────────────────────────────────────────────────────────────
  private openWs(sid: number, o: any): void {
    const proto = o.proto ? [o.proto] : undefined;
    const headers = this.replayHeaders(o.headers, o.remoteIp, true);
    const local = new WebSocket(`ws://127.0.0.1:${this.config.port}${o.url}`, proto, { headers });
    const st: WsStream = { kind: 'ws', local, opened: false, backlog: [] };
    this.streams.set(sid, st);

    local.on('open', () => {
      st.opened = true;
      for (const b of st.backlog) { try { local.send(b.binary ? b.data : b.data.toString('utf8'), { binary: b.binary }); } catch {} }
      st.backlog = [];
    });
    local.on('message', (data: Buffer, isBinary: boolean) => {
      const buf = Buffer.isBuffer(data) ? data : Buffer.from(data as any);
      this.send(enc(T.DATA, isBinary ? F.WS_BINARY : 0, sid, buf));
    });
    local.on('close', () => { this.send(enc(T.CLOSE, 0, sid, null)); this.streams.delete(sid); });
    local.on('error', () => this.reset(sid));
  }

  private cleanupHttp(sid: number): void {
    const st = this.streams.get(sid);
    if (st && st.kind === 'http' && st.drainTimer) clearInterval(st.drainTimer);
    this.streams.delete(sid);
  }

  private onClose(sid: number): void {
    const st = this.streams.get(sid);
    if (!st) return;
    try {
      if (st.kind === 'http') { if (st.drainTimer) clearInterval(st.drainTimer); st.req.destroy(); }
      else st.local.close();
    } catch {}
    this.streams.delete(sid);
  }

  private reset(sid: number): void {
    if (this.streams.has(sid)) { this.send(enc(T.RESET, 0, sid, Buffer.from([1]))); this.onClose(sid); }
  }

  // ── header replay + the security markers ────────────────────────────────────
  private replayHeaders(incoming: Record<string, string>, remoteIp: string, isWs = false): Record<string, string> {
    const h: Record<string, string> = {};
    for (const k in incoming) {
      const lk = k.toLowerCase();
      // Strip anything the client could smuggle to defeat the loopback guards, plus the
      // ws-handshake headers (the ws client regenerates its own).
      if (lk === 'cf-connecting-ip' || lk === 'cf-ray' || lk === 'x-morphy-tunnel') continue;
      // We reframe the body over the mux, so ask the local server for identity and let Cloudflare
      // re-compress fresh at the edge — avoids a content-encoding mismatch across the DO boundary.
      if (lk === 'accept-encoding') continue;
      if (isWs && (lk === 'connection' || lk === 'upgrade' || lk.startsWith('sec-websocket') || lk === 'host')) continue;
      h[k] = incoming[k];
    }
    // Unconditional markers: the guards reject BOTH, so seizure endpoints stay private.
    h['x-morphy-tunnel'] = '1';
    if (remoteIp) h['cf-connecting-ip'] = remoteIp;
    if (!isWs) h['accept-encoding'] = 'identity';
    return h;
  }
}

function safeJson(p: Buffer): any { try { return JSON.parse(p.toString('utf8')); } catch { return null; } }