uniweb/web/src/bus/client.ts

// The browser-native bus client, ported from Go pkg/client. It does what the Go
// gateway used to do server-side — only now it runs in the browser, so the user's
// private key never leaves the device (issue 0001).
//
// The module is split so the security-critical part is pure and unit-testable
// without a live server:
//   - sealRoomMessage / openRoomMessage: the room ENVELOPE (build a frame, AEAD-seal
//     the payload with the room key using the subject as AAD, sign it; and the
//     inverse: verify the signature and open the payload). These are pure and pinned
//     by tests.
//   - NatsTransport: the data-plane transport interface. The concrete WebSocket
//     implementation (nats.ws) is thin glue wired and E2E-tested in a later phase.
//   - ControlPlane: the signed HTTP client for membershipd (rooms, keys, members).
//   - BusClient: orchestrates transport + control plane + envelope.

import { Policy, Room } from "./room.js";
import {
  Frame,
  FrameType,
  marshal,
  unmarshal,
  signingBytes,
} from "./frame.js";
import {
  sealAEAD,
  openAEAD,
  randomNonce,
  signEd25519,
  verifyEd25519,
  sealKeyBox,
  openKeyBox,
  endpointID,
  bytesToBase64,
} from "./crypto.js";
import { signedHeaders, freshNonce } from "./busauth.js";

// Identity is the user's full cryptographic identity. The private halves stay in
// memory in the browser and are NEVER serialized to the network.
export interface Identity {
  signPub: Uint8Array;
  signPriv: Uint8Array; // 64-byte Ed25519 (seed||pub)
  kexPub: Uint8Array;
  kexPriv: Uint8Array;
}

// --- ULID (message ids), Crockford base32, time-ordered ----------------------

const CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";

export function newULID(nowMs: number = Date.now()): string {
  let ts = "";
  let t = nowMs;
  for (let i = 0; i < 10; i++) {
    ts = CROCKFORD[t % 32] + ts;
    t = Math.floor(t / 32);
  }
  const rnd = crypto.getRandomValues(new Uint8Array(16));
  let r = "";
  for (let i = 0; i < 16; i++) r += CROCKFORD[rnd[i] & 31];
  return ts + r;
}

// --- room envelope (pure, the security-critical core) ------------------------

export interface SealOptions {
  type: FrameType;
  subject: string;
  sender: string; // this peer's endpoint id
  signPriv: Uint8Array;
  policy: Policy;
  epoch: number;
  plaintext: Uint8Array;
  roomKey?: Uint8Array; // required when policy.encrypt
  threadID?: string;
  replyTo?: string;
  msgID?: string; // defaults to a fresh ULID
}

// sealRoomMessage builds a wire frame from plaintext exactly as Go's publishFrame:
// for encrypted rooms the payload is ChaCha20-Poly1305-sealed with the room key and
// the SUBJECT as additional authenticated data; for signed rooms an Ed25519
// signature over the canonical bytes is attached.
export function sealRoomMessage(o: SealOptions): Frame {
  const f: Frame = {
    type: o.type,
    subject: o.subject,
    sender: o.sender,
    msgID: o.msgID ?? newULID(),
    epoch: o.epoch,
    threadID: o.threadID,
    replyTo: o.replyTo,
  };
  if (o.policy.encrypt) {
    if (!o.roomKey) throw new Error("sealRoomMessage: encrypted room requires roomKey");
    const nonce = randomNonce();
    f.nonce = nonce;
    f.payload = sealAEAD(o.roomKey, nonce, o.plaintext, new TextEncoder().encode(o.subject));
  } else {
    f.payload = o.plaintext;
  }
  if (o.policy.signMsgs) {
    f.sig = signEd25519(o.signPriv, signingBytes(f));
  }
  return f;
}

// openRoomMessage is the inverse: it verifies the signature (for signed rooms) and
// opens the AEAD payload (for encrypted rooms), returning the plaintext or null if
// verification/decryption fails (the caller drops the message).
export function openRoomMessage(
  f: Frame,
  policy: Policy,
  signerPub: Uint8Array | undefined,
  roomKey: Uint8Array | undefined,
): Uint8Array | null {
  if (policy.signMsgs) {
    if (!f.sig || !signerPub || !verifyEd25519(f.sig, signingBytes(f), signerPub)) return null;
  }
  if (policy.encrypt) {
    if (!f.nonce || !f.payload || !roomKey) return null;
    try {
      return openAEAD(roomKey, f.nonce, f.payload, new TextEncoder().encode(f.subject));
    } catch {
      return null;
    }
  }
  return f.payload ?? new Uint8Array(0);
}

// --- data-plane transport ----------------------------------------------------

export type MessageHandler = (subject: string, data: Uint8Array) => void;

// NatsTransport abstracts the NATS data plane so BusClient's logic is testable with
// a mock and the concrete WebSocket transport (nats.ws) stays swappable. The browser
// transport connects over ws(s):// using a NATS nkey authenticator built from the
// user's Ed25519 identity (see busauth.natsAuthenticator).
export interface NatsTransport {
  publish(subject: string, data: Uint8Array): void | Promise<void>;
  subscribe(subject: string, handler: MessageHandler): Promise<Subscription>;
  close(): Promise<void>;
}

export interface Subscription {
  unsubscribe(): void | Promise<void>;
}

// --- control plane (signed HTTP to membershipd) ------------------------------

interface RoomKeyResponse {
  sealed_key: string; // base64 sealed box of the room key for this peer
  epoch: number;
}

// PolicyWire is the control-plane JSON shape of a policy (snake_case sign_msgs).
interface PolicyWire {
  encrypt: boolean;
  persist: boolean;
  sign_msgs: boolean;
}

// RoomResp is GET /rooms/{id}: the room metadata WITHOUT the id (the caller knows it)
// and with the policy nested under snake_case keys.
interface RoomResp {
  subject: string;
  epoch: number;
  policy: PolicyWire;
}

interface MemberJSON {
  endpoint: string;
  sign_pub: string; // base64
}

// MemberRoomWire is one row of GET /members/{endpoint}/rooms.
interface MemberRoomWire {
  room_id: string;
  subject: string;
  epoch: number;
  policy: PolicyWire;
}

// ControlPlane is the signed HTTP client for the membershipd control plane. Every
// request carries the X-Unibus-* auth headers (busauth.signedHeaders). It pins no
// host so it can target any cluster node.
export class ControlPlane {
  constructor(
    private baseURL: string,
    private id: Identity,
  ) {}

  private async request<T>(method: string, path: string, body?: unknown): Promise<T> {
    const bodyBytes = body === undefined ? new Uint8Array(0) : new TextEncoder().encode(JSON.stringify(body));
    const headers = signedHeaders(
      this.id.signPub,
      this.id.signPriv,
      method,
      path,
      String(Math.floor(Date.now() / 1000)),
      freshNonce(),
      bodyBytes,
    );
    const init: RequestInit = { method, headers: { ...headers } };
    if (body !== undefined) {
      (init.headers as Record<string, string>)["Content-Type"] = "application/json";
      init.body = bodyBytes;
    }
    const resp = await fetch(this.baseURL + path, init);
    if (!resp.ok) {
      let msg = `${method} ${path} -> ${resp.status}`;
      try {
        const e = await resp.json();
        if (e?.error) msg = `${e.error} (HTTP ${resp.status})`;
      } catch {
        /* keep the generic message */
      }
      throw new Error(`control plane: ${msg}`);
    }
    return (await resp.json()) as T;
  }

  // fetchRoom resolves room metadata, mapping the control-plane wire shape
  // (snake_case policy, no id) to the SDK's Room type.
  async fetchRoom(roomID: string): Promise<Room> {
    const r = await this.request<RoomResp>("GET", `/rooms/${roomID}`);
    return {
      id: roomID,
      subject: r.subject,
      epoch: r.epoch,
      policy: { encrypt: r.policy.encrypt, persist: r.policy.persist, signMsgs: r.policy.sign_msgs },
    };
  }

  // createRoom creates a room owned by this peer. For an encrypted room it mints a
  // fresh 32-byte room key, seals it to the owner's own X25519 key (sealed box), and
  // ships it as sealed_key_self so the server can store the owner's copy without ever
  // seeing the key. Returns the new room id and (for encrypted rooms) the key.
  async createRoom(subject: string, policy: Policy): Promise<{ roomID: string; key?: Uint8Array }> {
    const body: Record<string, unknown> = {
      subject,
      policy: { encrypt: policy.encrypt, persist: policy.persist, sign_msgs: policy.signMsgs },
      owner: {
        endpoint: endpointID(this.id.signPub),
        sign_pub: bytesToBase64(this.id.signPub),
        kex_pub: bytesToBase64(this.id.kexPub),
      },
    };
    let key: Uint8Array | undefined;
    if (policy.encrypt) {
      key = crypto.getRandomValues(new Uint8Array(32));
      body.sealed_key_self = bytesToBase64(sealKeyBox(this.id.kexPub, key));
    }
    const resp = await this.request<{ room_id: string }>("POST", "/rooms", body);
    return { roomID: resp.room_id, key };
  }

  // fetchRoomKey fetches the sealed room key for this peer and opens it with the
  // user's X25519 private key. The server only ever stores the key sealed for each
  // member, so it cannot read it.
  async fetchRoomKey(roomID: string, epoch: number): Promise<{ key: Uint8Array; epoch: number }> {
    const q = epoch > 0 ? `&epoch=${epoch}` : "";
    const resp = await this.request<RoomKeyResponse>(
      "GET",
      `/rooms/${roomID}/key?endpoint=${endpointID(this.id.signPub)}${q}`,
    );
    const sealed = base64ToBytesLocal(resp.sealed_key);
    const key = openKeyBox(this.id.kexPub, this.id.kexPriv, sealed);
    if (!key) throw new Error("control plane: failed to open room key");
    return { key, epoch: resp.epoch };
  }

  // listMemberRooms returns the rooms a peer belongs to (GET /members/{endpoint}/rooms),
  // mapping the wire shape (room_id, snake_case policy) to the SDK Room type.
  async listMemberRooms(endpoint: string): Promise<Room[]> {
    const wire = await this.request<MemberRoomWire[]>("GET", `/members/${endpoint}/rooms`);
    return wire.map((r) => ({
      id: r.room_id,
      subject: r.subject,
      epoch: r.epoch,
      policy: { encrypt: r.policy.encrypt, persist: r.policy.persist, signMsgs: r.policy.sign_msgs },
    }));
  }

  // listMembers returns the room's members keyed by endpoint, so a receiver can find
  // a sender's signing public key to verify message signatures.
  async signerKeys(roomID: string): Promise<Map<string, Uint8Array>> {
    const members = await this.request<MemberJSON[]>("GET", `/rooms/${roomID}/members`);
    const m = new Map<string, Uint8Array>();
    for (const member of members) m.set(member.endpoint, base64ToBytesLocal(member.sign_pub));
    return m;
  }
}

// base64ToBytesLocal decodes standard base64 (kept local to avoid widening crypto's
// surface; identical behavior to crypto.base64ToBytes).
function base64ToBytesLocal(s: string): Uint8Array {
  const bin = atob(s);
  const out = new Uint8Array(bin.length);
  for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i);
  return out;
}

// --- BusClient ---------------------------------------------------------------

// BusClient ties the data plane (transport) and control plane together, applying the
// room envelope on publish and subscribe. It holds the user's identity in memory and
// never sends the private key anywhere.
export class BusClient {
  private endpoint: string;
  private keyCache = new Map<string, Map<number, Uint8Array>>(); // roomID -> epoch -> K
  private signCache = new Map<string, Map<string, Uint8Array>>(); // roomID -> endpoint -> signPub

  constructor(
    private id: Identity,
    private transport: NatsTransport,
    private control: ControlPlane,
  ) {
    this.endpoint = endpointID(id.signPub);
  }

  private async roomKey(roomID: string, epoch: number): Promise<Uint8Array> {
    const cached = this.keyCache.get(roomID)?.get(epoch);
    if (cached) return cached;
    const { key, epoch: ep } = await this.control.fetchRoomKey(roomID, epoch);
    let byEpoch = this.keyCache.get(roomID);
    if (!byEpoch) {
      byEpoch = new Map();
      this.keyCache.set(roomID, byEpoch);
    }
    byEpoch.set(ep, key);
    return key;
  }

  // publish seals plaintext per the room policy and publishes it on the data plane.
  async publish(roomID: string, plaintext: Uint8Array, opts: { threadID?: string; replyTo?: string; type?: FrameType } = {}): Promise<void> {
    const room = await this.control.fetchRoom(roomID);
    const roomKey = room.policy.encrypt ? await this.roomKey(roomID, room.epoch) : undefined;
    const f = sealRoomMessage({
      type: opts.type ?? FrameType.PUB,
      subject: room.subject,
      sender: this.endpoint,
      signPriv: this.id.signPriv,
      policy: room.policy,
      epoch: room.epoch,
      plaintext,
      roomKey,
      threadID: opts.threadID,
      replyTo: opts.replyTo,
    });
    await this.transport.publish(room.subject, marshal(f));
  }

  // subscribe delivers decoded, verified, decrypted messages for a room. Messages
  // that fail signature verification or decryption are dropped silently.
  async subscribe(roomID: string, handler: (f: Frame, plaintext: Uint8Array) => void): Promise<Subscription> {
    const room = await this.control.fetchRoom(roomID);
    if (room.policy.signMsgs) await this.loadSigners(roomID);
    return this.transport.subscribe(room.subject, async (_subject, data) => {
      const f = unmarshal(data);
      const signerPub = room.policy.signMsgs ? this.signCache.get(roomID)?.get(f.sender) : undefined;
      const roomKey = room.policy.encrypt ? await this.roomKey(roomID, f.epoch) : undefined;
      const plaintext = openRoomMessage(f, room.policy, signerPub, roomKey);
      if (plaintext) handler(f, plaintext);
    });
  }

  private async loadSigners(roomID: string): Promise<void> {
    this.signCache.set(roomID, await this.control.signerKeys(roomID));
  }
}