// The relay room (docs/CLOUD_INFRA.md §2): one Durable Object per `roomId`, holding the host's
// and the phones' WebSockets and brokering frames between them. It is **E2EE-opaque** — binary
// frame *payloads* are forwarded byte-for-byte and never inspected. A separate *text* control
// plane carries relay metadata (presence, push registration, wake requests) that is explicitly
// non-secret.
//
// **Per-peer routing (mesh P2).** The old code broadcast every binary frame to every other
// socket, so with more than two peers each client received (and had to AEAD-fail) frames meant
// for others, and the host had no way to tell which client a client→host frame came from. Now the
// DO owns a symmetric 9-byte routing envelope on the *host* leg only:
//   • client → host: DO prepends [ver][tag = sha256(senderDeviceId)[:8]] so the host can demux
//     which peer sent it, then sends to the single host socket.
//   • host → client: the host prepends [ver][tag = sha256(targetDeviceId)[:8]]; the DO reads the
//     tag, strips the envelope, and forwards the raw payload to exactly that client. Unknown tag
//     → dropped + counted.
// The client↔DO leg is always raw Noise frames — the envelope exists only between the host and
// the DO, so Noise and everything above see identical bytes on every transport.
//
// Uses the WebSocket Hibernation API so idle rooms evict from memory: handlers are methods the
// runtime calls on `webSocketMessage` / `webSocketClose`, and per-socket state rides along in the
// socket's serialized attachment rather than instance fields.

import type { Env } from "./env";
// `.ts` extension so `node --experimental-strip-types` can resolve this value import when the
// Room class is pulled into a test (test/room.test.ts); esbuild/wrangler accept it too.
import { apnsConfig, approvalPayload, clearPayload, sendApns, sendApnsClear } from "./apns.ts";

interface SocketMeta {
  role: "host" | "client";
  deviceId: string;
  /** Hex of the 8-byte routing tag, precomputed at accept. Routing used to await a WebCrypto
   * digest per frame (per *client* per frame on the host→client leg) — pure overhead on the
   * DO's single thread, and awaiting a non-storage promise inside `webSocketMessage` releases
   * the input gate, letting a later frame's handler interleave with (and potentially reorder
   * around) an earlier one — fatal to the Noise nonce sequence. Optional: sockets accepted
   * before this field shipped hibernate with attachments that lack it; routing lazily
   * backfills those. */
  tagHex?: string;
}

function bytesToHex(bytes: Uint8Array): string {
  let s = "";
  for (const b of bytes) s += b.toString(16).padStart(2, "0");
  return s;
}

/** Compact deviceId for room-log lines (they're already broadcast full-length in presence;
 * this just keeps the trail readable). */
function shortId(id: string): string {
  return id.length > 12 ? id.slice(0, 12) : id;
}

/** Routing-envelope version byte + a hard ceiling well above the app's 16 MiB frame cap but
 * under Cloudflare's 32 MiB WebSocket message limit — an oversized frame closes the socket.
 * (Connection-rate abuse is bounded elsewhere: 2-minute connection tokens gate admission, and
 * Cloudflare's platform limits cap raw socket churn — a message-rate counter here is unreliable
 * under the DO clock and risks false-closing a legitimately busy session.) */
const ENVELOPE_VERSION = 0x01;
const ENVELOPE_LEN = 9; // 1 version + 8 tag bytes
const MAX_FRAME_BYTES = 20 * 1024 * 1024;
const WS_CLOSE_POLICY = 1008; // policy violation (oversized frame)

/** One structural room-lifecycle event, kept in DO storage for `/v1/relay/roomlog` (the
 * host's mesh-log export pulls these). Structural metadata only — accepts, closes with
 * codes, supersedes, wake nudges — never frame payloads (which are E2EE ciphertext anyway).
 * Capped small: this is a debugging trail, not telemetry. */
export interface RoomLogEntry {
  at: number; // epoch ms
  ev: string;
}
const ROOM_LOG_KEY = "roomlog";
const ROOM_LOG_CAP = 200;

/** First 8 bytes of sha256(deviceId) — the routing tag the host addresses a peer by. */
async function deviceTag(deviceId: string): Promise<Uint8Array> {
  const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(deviceId));
  return new Uint8Array(digest).slice(0, 8);
}

export class Room implements DurableObject {
  // Plain field assignment (not constructor parameter-property shorthand) so the class can be
  // imported under `node --experimental-strip-types`, which the test runner uses.
  private readonly state: DurableObjectState;
  private readonly env: Env;

  constructor(state: DurableObjectState, env: Env) {
    this.state = state;
    this.env = env;
  }

  // The Worker has already verified the connection token; it forwards the resolved identity in
  // headers. We just accept the socket (hibernatable) and announce presence.
  async fetch(req: Request): Promise<Response> {
    // Room-log read (the Worker gates it behind the host's room credential — only this
    // room's owner can pull its trail).
    if (new URL(req.url).pathname === "/roomlog") {
      const log = ((await this.state.storage.get(ROOM_LOG_KEY)) as RoomLogEntry[] | undefined) ?? [];
      return new Response(JSON.stringify({ events: log }), {
        headers: { "content-type": "application/json" },
      });
    }
    if (req.headers.get("upgrade") !== "websocket") {
      return new Response("expected websocket", { status: 426 });
    }
    const role = req.headers.get("x-nucleic-role") === "host" ? "host" : "client";
    const deviceId = req.headers.get("x-nucleic-device") ?? "unknown";
    // The hibernation tags below are [role, deviceId] in one namespace, so a *client* claiming
    // the literal deviceId "host" would be matched by `getWebSockets("host")` — routing every
    // client→host frame into that client (a blackhole; E2EE keeps content safe, but the room
    // is dead) and getting closed as "superseded" by real host reconnects. The deviceId is
    // client-influenced (it originates in the device's own hello), so reject the collision.
    if (role !== "host" && deviceId === "host") {
      return new Response("reserved deviceId", { status: 403 });
    }

    // Relay presence → runner wake (docs/COVALENCE_RUNNER.md §3): a client arriving in a room
    // with no host may mean the host is an Covalence runner whose container is asleep — nudge
    // the runner control plane, which wakes the pool that registered this room (or 404s an
    // ordinary Mac's room, which is fine). Fire-and-forget so the connect never waits on it.
    if (role === "client" && this.state.getWebSockets("host").length === 0) {
      void this.maybeWakeRunner(req.headers.get("x-nucleic-room"));
    }

    // One host per room: a new host connection replaces any prior one (a stale socket the old
    // host never cleanly closed would otherwise shadow the live one in `getWebSockets("host")`).
    // Likewise a client reconnecting under the same deviceId supersedes its own stale socket.
    for (const existing of this.state.getWebSockets(role === "host" ? "host" : deviceId)) {
      const m = existing.deserializeAttachment() as SocketMeta | null;
      if (role === "host" || m?.deviceId === deviceId) {
        try {
          existing.close(1000, "superseded");
        } catch {
          /* already closing */
        }
        await this.appendLog(`supersede ${role} device=${shortId(m?.deviceId ?? deviceId)}`);
      }
    }

    const { 0: client, 1: server } = new WebSocketPair();
    const tagHex = bytesToHex(await deviceTag(deviceId));
    this.state.acceptWebSocket(server, [role, deviceId]);
    server.serializeAttachment({ role, deviceId, tagHex } satisfies SocketMeta);

    await this.appendLog(
      `accept ${role} device=${shortId(deviceId)} peers=${this.state.getWebSockets().length}`);
    this.broadcastPresence();
    return new Response(null, { status: 101, webSocket: client });
  }

  /** Append one structural event to the capped room log (best-effort — a storage hiccup
   * must never break the data path). */
  private async appendLog(ev: string): Promise<void> {
    try {
      const log =
        ((await this.state.storage.get(ROOM_LOG_KEY)) as RoomLogEntry[] | undefined) ?? [];
      log.push({ at: Date.now(), ev });
      if (log.length > ROOM_LOG_CAP) log.splice(0, log.length - ROOM_LOG_CAP);
      await this.state.storage.put(ROOM_LOG_KEY, log);
    } catch {
      /* log-only path */
    }
  }

  async webSocketMessage(ws: WebSocket, message: ArrayBuffer | string): Promise<void> {
    // Text = relay control plane (never E2EE payload).
    if (typeof message === "string") {
      await this.handleControl(ws, message);
      return;
    }
    if (message.byteLength > MAX_FRAME_BYTES) {
      try {
        ws.close(WS_CLOSE_POLICY, "frame too large");
      } catch {
        /* already closing */
      }
      return;
    }

    const me = ws.deserializeAttachment() as SocketMeta | null;
    if (!me) return;
    if (me.role === "host") {
      await this.routeFromHost(message);
    } else {
      await this.routeFromClient(ws, me, message);
    }
  }

  /** A socket's routing-tag hex, from the attachment when present (the normal case — computed
   * once at accept) or lazily backfilled for a socket that hibernated with a pre-`tagHex`
   * attachment. Keeping the tag in the attachment makes frame routing synchronous: no awaited
   * digest per frame, and no input-gate release between reading a frame and forwarding it. */
  private async tagHexFor(ws: WebSocket, meta: SocketMeta): Promise<string> {
    if (meta.tagHex) return meta.tagHex;
    const tagHex = bytesToHex(await deviceTag(meta.deviceId));
    try {
      ws.serializeAttachment({ ...meta, tagHex } satisfies SocketMeta);
    } catch {
      /* socket closing — the computed value still serves this frame */
    }
    return tagHex;
  }

  /** client → host: prepend [ver][tag = sha256(senderDeviceId)[:8]] so the host can demux which
   * peer sent this frame, then deliver to the single host socket. */
  private async routeFromClient(ws: WebSocket, me: SocketMeta, payload: ArrayBuffer): Promise<void> {
    const host = this.state.getWebSockets("host")[0];
    if (!host) return; // no host present; the client will retry after reconnect
    const tagHex = await this.tagHexFor(ws, me);
    const framed = new Uint8Array(ENVELOPE_LEN + payload.byteLength);
    framed[0] = ENVELOPE_VERSION;
    for (let i = 0; i < 8; i++) framed[1 + i] = parseInt(tagHex.slice(i * 2, i * 2 + 2), 16);
    framed.set(new Uint8Array(payload), ENVELOPE_LEN);
    try {
      host.send(framed);
    } catch {
      /* host going away; its close handler reconciles presence */
    }
  }

  /** host → client: read the [ver][tag] envelope, strip it, and forward the raw Noise frame to
   * exactly the addressed client. An unknown/mismatched tag is dropped (and would show as a gap
   * to no one — the addressed peer simply isn't here). */
  private async routeFromHost(message: ArrayBuffer): Promise<void> {
    const bytes = new Uint8Array(message);
    if (bytes.length < ENVELOPE_LEN || bytes[0] !== ENVELOPE_VERSION) return; // malformed → drop
    const tagHex = bytesToHex(bytes.subarray(1, ENVELOPE_LEN));
    const raw = bytes.subarray(ENVELOPE_LEN);
    for (const client of this.state.getWebSockets("client")) {
      const m = client.deserializeAttachment() as SocketMeta | null;
      if (!m) continue;
      if ((await this.tagHexFor(client, m)) === tagHex) {
        try {
          client.send(raw);
        } catch {
          /* client going away */
        }
        return;
      }
    }
  }

  async webSocketClose(ws: WebSocket, code: number): Promise<void> {
    try {
      ws.close(code);
    } catch {
      /* already closed */
    }
    const m = ws.deserializeAttachment() as SocketMeta | null;
    await this.appendLog(
      `close ${m?.role ?? "?"} device=${shortId(m?.deviceId ?? "?")} code=${code}`);
    this.broadcastPresence();
  }

  async webSocketError(ws: WebSocket): Promise<void> {
    const m = ws?.deserializeAttachment?.() as SocketMeta | null;
    await this.appendLog(`error ${m?.role ?? "?"} device=${shortId(m?.deviceId ?? "?")}`);
    this.broadcastPresence();
  }

  // MARK: - Runner wake (relay presence → pool wake, docs/COVALENCE_RUNNER.md §3)

  /** Last runner-wake attempt (ms) — a reconnect storm from several devices must not hammer
   * the hook; the pool's `/wake` is idempotent anyway, this just saves the round-trips. Plain
   * instance state: a hibernated room simply starts a fresh window, which is fine. −∞ so the
   * FIRST nudge always fires regardless of the clock's origin. */
  private lastRunnerWakeAt = Number.NEGATIVE_INFINITY;
  static readonly RUNNER_WAKE_THROTTLE_MS = 30_000;

  /** Nudge the runner control plane that a client wants this (hostless) room's host. No-op
   * unless both `RUNNER_WAKE_URL` + `RUNNER_WAKE_SECRET` are configured and `roomId` came
   * through; throttled; errors are swallowed (the client keeps its normal retry loop either
   * way, and the next retry re-nudges). Returns whether a nudge was actually sent — the
   * seam the unit tests drive (with `fetcher` stubbed). */
  async maybeWakeRunner(
    roomId: string | null,
    now: number = Date.now(),
    fetcher: typeof fetch = fetch,
  ): Promise<boolean> {
    const { RUNNER_WAKE_URL: url, RUNNER_WAKE_SECRET: secret } = this.env;
    if (!url || !secret || !roomId) return false;
    if (now - this.lastRunnerWakeAt < Room.RUNNER_WAKE_THROTTLE_MS) return false;
    this.lastRunnerWakeAt = now;
    await this.appendLog("runner wake nudge (client arrived, no host)");
    try {
      await fetcher(url, {
        method: "POST",
        headers: { authorization: `Bearer ${secret}`, "content-type": "application/json" },
        body: JSON.stringify({ roomId }),
      });
    } catch {
      /* unreachable control plane — the client's reconnect loop re-nudges */
    }
    return true;
  }

  // MARK: - Control plane (text JSON)

  private async handleControl(ws: WebSocket, raw: string): Promise<void> {
    let msg: { t?: string; pushToken?: unknown; env?: unknown; topic?: unknown; deviceId?: unknown; kind?: unknown };
    try {
      msg = JSON.parse(raw);
    } catch {
      return;
    }
    const me = ws.deserializeAttachment() as SocketMeta | null;

    switch (msg.t) {
      case "register": {
        // The phone hands the relay its APNS token (not secret) so the relay can wake it later.
        // Owner-tagged with this room's DO id: in-band wakes (below) are authorized by room
        // membership, and the tag keeps out-of-band hosts (push.ts) from waking it.
        if (me?.deviceId && typeof msg.pushToken === "string") {
          const env = msg.env === "production" ? "production" : "sandbox";
          const hostId = `room:${this.state.id.toString()}`;
          // Per-channel builds have suffixed bundle ids; store the phone's topic so `wake`
          // addresses APNs correctly (omitted → the wake falls back to the configured topic).
          const record: { token: string; env: string; topic?: string; hostId: string } = {
            token: msg.pushToken,
            env,
            hostId,
          };
          if (typeof msg.topic === "string" && msg.topic) record.topic = msg.topic;
          await this.env.NUCLEIC_PUSH_TOKENS.put(me.deviceId, JSON.stringify(record), {
            expirationTtl: 90 * 24 * 60 * 60,
          });
        }
        break;
      }
      case "wake": {
        // Only the host may ask the relay to wake a phone, and only when it isn't connected.
        if (me?.role === "host" && typeof msg.deviceId === "string") {
          await this.wake(msg.deviceId, msg.kind === "question");
        }
        break;
      }
      case "clear": {
        // The counterpart to "wake": recall a disconnected phone's approval tickle once the
        // approval was resolved elsewhere. Host-only, same as wake.
        if (me?.role === "host" && typeof msg.deviceId === "string") {
          await this.clear(msg.deviceId);
        }
        break;
      }
    }
  }

  private async wake(deviceId: string, question = false): Promise<void> {
    const present = this.state.getWebSockets().some((w) => {
      const m = w.deserializeAttachment() as SocketMeta | null;
      return m?.deviceId === deviceId;
    });
    if (present) return; // a live socket will deliver the frame; no push needed

    const cfg = apnsConfig(this.env);
    if (!cfg) return;
    const rec = await this.env.NUCLEIC_PUSH_TOKENS.get(deviceId);
    if (!rec) return;
    const { token, env, topic } = JSON.parse(rec) as {
      token: string;
      env?: "sandbox" | "production";
      topic?: string;
    };
    // The record's environment + topic win (set when the phone/host registered — they know the
    // build's env and per-channel bundle id); the global APNS_ENV / APNS_TOPIC only backstop
    // records that predate them. A wrong topic is `BadTopic`-rejected by APNs.
    await sendApns({ ...cfg, env: env ?? cfg.env, topic: topic ?? cfg.topic }, token, approvalPayload(question));
  }

  /** Recall a phone's approval tickle (silent background push). Unlike `wake` this is worth
   * sending even to a *present* device — a phone that saw the tickle while its socket was down and
   * only just reconnected still has the notification on screen; a live socket's `approvalResolved`
   * frame also clears it, so the redundant clear is harmless. */
  private async clear(deviceId: string): Promise<void> {
    const cfg = apnsConfig(this.env);
    if (!cfg) return;
    const rec = await this.env.NUCLEIC_PUSH_TOKENS.get(deviceId);
    if (!rec) return;
    const { token, env, topic } = JSON.parse(rec) as {
      token: string;
      env?: "sandbox" | "production";
      topic?: string;
    };
    await sendApnsClear(
      { ...cfg, env: env ?? cfg.env, topic: topic ?? cfg.topic }, token, clearPayload());
  }

  private broadcastPresence(): void {
    const sockets = this.state.getWebSockets();
    const peers = sockets.map((w) => {
      const m = w.deserializeAttachment() as SocketMeta | null;
      return { deviceId: m?.deviceId ?? "?", role: m?.role ?? "client" };
    });
    const frame = JSON.stringify({ t: "presence", peers });
    for (const w of sockets) {
      try {
        w.send(frame);
      } catch {
        /* closing */
      }
    }
  }
}
