// The two ways a page reaches the relay, behind one interface:
//
// - WebTransport over HTTP/3 (draft-ietf-webtrans-http3), offering one protocol,
//   `moqt-22` by default or `moqt-23` (`WT-Available-Protocols`), and, for a relay with a
//   development certificate, `serverCertificateHashes` (ECDSA P-256, valid at most
//   fourteen days);
// - the relay's WebSocket binding: a WebSocket with the subprotocol `moqt-22` or `moqt-23`, each
//   message one capsule of draft-ietf-webtrans-http2-15 §6, streams numbered as QUIC
//   numbers them, stream-count credit by WT_MAX_STREAMS, no data flow control (TCP's
//   backpressure bounds the sender).
//
// Each gives:
//   kind                      "webtransport" | "websocket"
//   protocol                  the negotiated protocol, or "" where none was signalled
//   draft                     the MOQT draft the session speaks (22 or 23): the one offered
//   openUni({sendOrder})      a Writer
//   openBidi()                { writer, reader }
//   acceptUni()               the next Reader the relay opened, or null once closed
//   acceptBidi()              the next { writer, reader }, or null once closed
//   close(code, reason)
//   closed                    a promise of { code, reason }
// A Writer has write(bytes), close() (a FIN) and reset(code); a Reader has read(), which
// gives { value, done } or throws a StreamReset with the code, and cancel(code).

import * as cap from "./capsule.js";

export class StreamReset extends Error {
  constructor(code, what) {
    super(`${what || "a stream"} was reset with code ${code}`);
    this.code = code;
  }
}

export const PROTOCOL = "moqt-22";
// The drafts a page speaks, and the one offered when a page names none: -22, which every
// relay this page has been deployed beside speaks; -23 when the page is asked for it.
export const DRAFTS = [23, 22];
export const DEFAULT_DRAFT = 22;
export function protocolOf(draft) {
  if (!DRAFTS.includes(draft)) throw new Error(`a page speaks drafts ${DRAFTS.join(" and ")}, not ${draft}`);
  return `moqt-${draft}`;
}

// A queue of things handed to whoever waits for them.
class Queue {
  constructor() { this.items = []; this.waiters = []; this.closed = false; }
  push(x) {
    if (this.closed) return;
    const w = this.waiters.shift();
    if (w) w(x); else this.items.push(x);
  }
  close() {
    this.closed = true;
    for (const w of this.waiters.splice(0)) w(null);
  }
  next() {
    if (this.items.length) return Promise.resolve(this.items.shift());
    if (this.closed) return Promise.resolve(null);
    return new Promise((r) => this.waiters.push(r));
  }
}

// ---------------------------------------------------------------------------------------
// WebTransport
// ---------------------------------------------------------------------------------------

function wtReader(readable) {
  const r = readable.getReader();
  return {
    async read() {
      try {
        const { value, done } = await r.read();
        return { value, done };
      } catch (e) {
        throw new StreamReset(e && e.streamErrorCode != null ? e.streamErrorCode : -1, "a WebTransport stream");
      }
    },
    cancel(code) {
      try { r.cancel(new WebTransportError({ streamErrorCode: code })); } catch { /* already closed */ }
    },
  };
}

function wtWriter(writable) {
  const w = writable.getWriter();
  return {
    async write(b) { await w.write(b); },
    async close() { try { await w.close(); } catch { /* reset or session closed */ } },
    reset(code) { try { w.abort(new WebTransportError({ streamErrorCode: code })); } catch { /* closed */ } },
  };
}

export async function connectWebTransport(url, { hashes = [], timeoutMs = 10000, draft = DEFAULT_DRAFT } = {}) {
  if (typeof WebTransport !== "function") throw new Error("this browser has no WebTransport");
  const opts = { protocols: [protocolOf(draft)] };
  if (hashes.length) opts.serverCertificateHashes = hashes.map((value) => ({ algorithm: "sha-256", value }));
  const wt = new WebTransport(url, opts);
  let timer;
  try {
    await Promise.race([wt.ready, new Promise((_, no) => { timer = setTimeout(() => no(new Error(`WebTransport not ready in ${timeoutMs} ms`)), timeoutMs); })]);
  } catch (e) {
    try { wt.close(); } catch { /* never opened */ }
    throw e;
  } finally {
    clearTimeout(timer);
  }
  const uni = wt.incomingUnidirectionalStreams.getReader();
  const bidi = wt.incomingBidirectionalStreams.getReader();
  const closed = wt.closed.then(
    (i) => ({ code: i ? i.closeCode : 0, reason: i ? i.reason : "" }),
    (e) => ({ code: e && e.closeCode != null ? e.closeCode : -1, reason: String(e && e.message || e) }),
  );
  return {
    kind: "webtransport",
    // The URL the session opened: its authority and path are what a grant's proof names
    // (WAI live-rooms §10.5.1).
    url,
    // `protocol` is the server's answer to WT-Available-Protocols, where the browser has it.
    protocol: typeof wt.protocol === "string" ? wt.protocol : "",
    // One draft is offered, so it is the one the session speaks.
    draft,
    async openUni({ sendOrder } = {}) {
      const s = await wt.createUnidirectionalStream(sendOrder != null ? { sendOrder } : {});
      return wtWriter(s);
    },
    async openBidi() {
      const s = await wt.createBidirectionalStream();
      return { writer: wtWriter(s.writable), reader: wtReader(s.readable) };
    },
    async acceptUni() {
      try { const { value, done } = await uni.read(); return done ? null : wtReader(value); } catch { return null; }
    },
    async acceptBidi() {
      try {
        const { value, done } = await bidi.read();
        return done ? null : { writer: wtWriter(value.writable), reader: wtReader(value.readable) };
      } catch { return null; }
    },
    close(code = 0, reason = "") {
      try { wt.close({ closeCode: code, reason }); } catch { /* closed */ }
    },
    closed,
    stats: () => ({}),
  };
}

// ---------------------------------------------------------------------------------------
// The WebSocket binding
// ---------------------------------------------------------------------------------------

// What this end grants the relay: bidirectional streams (requests the relay opens to it)
// and unidirectional streams (the relay's data streams), cumulative as in QUIC, given
// back in batches of an eighth as streams finish.
const LOCAL_MAX_BIDI = 64;
const LOCAL_MAX_UNI = 1024;
// Bytes queued on the WebSocket before a writer waits.
const SEND_HIGH_WATER = 1 << 20;

class WsSession {
  constructor(ws) {
    this.ws = ws;
    this.kind = "websocket";
    this.protocol = ws.protocol;
    this.nextUni = 0;
    this.nextBidi = 0;
    this.peerMax = { bidi: null, uni: null };
    this.local = { bidi: { max: LOCAL_MAX_BIDI, pending: 0, initial: LOCAL_MAX_BIDI }, uni: { max: LOCAL_MAX_UNI, pending: 0, initial: LOCAL_MAX_UNI } };
    this.recv = new Map(); // stream id -> inbox
    this.halves = new Map(); // peer-initiated bidi id -> halves still open
    this.stopped = new Map(); // stream id -> STOP_SENDING code
    this.seen = { bidi: new Set(), uni: new Set() };
    this.creditWaiters = [];
    this.incomingUni = new Queue();
    this.incomingBidi = new Queue();
    this.counters = { capsules_in: 0, capsules_out: 0, bytes_in: 0, bytes_out: 0, credit_granted: 0, blocked_sent: 0 };
    this.isClosed = null;
    this.closed = new Promise((r) => { this._closed = r; });
    ws.onmessage = (ev) => this.onMessage(ev.data);
    ws.onclose = (ev) => this.finish({ code: this.closeCode != null ? this.closeCode : 0, reason: this.closeReason || ev.reason || "the WebSocket closed" });
    ws.onerror = () => {};
    // Each end first sends its WT_MAX_STREAMS for both kinds.
    this.push(cap.maxStreamsCapsule(true, LOCAL_MAX_BIDI));
    this.push(cap.maxStreamsCapsule(false, LOCAL_MAX_UNI));
  }

  push(b) {
    if (this.isClosed) return;
    this.counters.capsules_out++;
    this.counters.bytes_out += b.length;
    this.ws.send(b);
  }

  async room() {
    while (this.ws.bufferedAmount > SEND_HIGH_WATER && !this.isClosed) await new Promise((r) => setTimeout(r, 5));
  }

  finish(info) {
    if (this.isClosed) return;
    this.isClosed = info;
    for (const ib of this.recv.values()) ib.fail(-1);
    this.recv.clear();
    this.incomingUni.close();
    this.incomingBidi.close();
    for (const w of this.creditWaiters.splice(0)) w();
    this._closed(info);
  }

  fatal(code, reason) {
    this.push(cap.closeSessionCapsule(code, reason));
    this.closeCode = code;
    this.closeReason = reason;
    try { this.ws.close(1000); } catch { /* closed */ }
    this.finish({ code, reason });
  }

  // A stream's credit comes back once both of its halves are done.
  returnCredit(bidi) {
    const l = bidi ? this.local.bidi : this.local.uni;
    l.pending++;
    if (l.pending >= Math.max(1, Math.floor(l.initial / 8))) {
      l.max += l.pending;
      l.pending = 0;
      this.counters.credit_granted++;
      this.push(cap.maxStreamsCapsule(bidi, l.max));
    }
  }

  halfDone(id) {
    if (cap.streamId.clientInitiated(id)) return; // credit for the streams this end opens is the relay's to give
    const bidi = cap.streamId.bidi(id);
    if (!bidi) return this.returnCredit(false);
    const left = (this.halves.get(id) || 2) - 1;
    if (left <= 0) {
      this.halves.delete(id);
      this.returnCredit(true);
    } else this.halves.set(id, left);
  }

  inbox(id) {
    const self = this;
    const ib = {
      chunks: [], done: false, code: null, waiter: null, finished: false,
      deliver() { if (this.waiter) { const w = this.waiter; this.waiter = null; w(); } },
      data(d) { this.chunks.push(d); this.deliver(); },
      end() { this.done = true; this.deliver(); },
      fail(code) { this.code = code; this.deliver(); },
      settle() { if (!this.finished) { this.finished = true; self.halfDone(id); } },
    };
    const reader = {
      async read() {
        for (;;) {
          if (ib.chunks.length) return { value: ib.chunks.shift(), done: false };
          if (ib.code != null) { ib.settle(); throw new StreamReset(ib.code, "a WebSocket-binding stream"); }
          if (ib.done) { ib.settle(); return { value: undefined, done: true }; }
          await new Promise((r) => { ib.waiter = r; });
        }
      },
      cancel(code) {
        if (ib.done || ib.code != null) return;
        self.push(cap.stopSendingCapsule(id, code));
        ib.fail(code);
        ib.settle();
        self.recv.delete(id);
      },
    };
    return { ib, reader };
  }

  writer(id) {
    const self = this;
    let ended = false;
    const end = () => { if (!ended) { ended = true; self.halfDone(id); } };
    return {
      async write(b) {
        if (self.isClosed) throw new Error("the session is closed");
        if (self.stopped.has(id)) throw new StreamReset(self.stopped.get(id), "a stopped stream");
        for (let at = 0; at < b.length; at += cap.CHUNK) {
          await self.room();
          self.push(cap.streamCapsule(id, b.subarray(at, Math.min(b.length, at + cap.CHUNK)), false));
        }
      },
      async close() {
        if (ended) return;
        self.push(cap.streamCapsule(id, new Uint8Array(0), true));
        end();
      },
      reset(code) {
        if (ended) return;
        self.push(cap.resetStreamCapsule(id, code));
        end();
      },
    };
  }

  onMessage(data) {
    if (typeof data === "string") return; // a text message carries nothing
    const msg = new Uint8Array(data);
    this.counters.capsules_in++;
    this.counters.bytes_in += msg.length;
    let c;
    try { c = cap.decodeCapsule(msg); } catch (e) { return this.fatal(0x1, String(e.message)); }
    switch (c.kind) {
      case "stream": return this.onStream(c.id, c.data, c.fin, null);
      case "reset": return this.onStream(c.id, null, false, c.code);
      case "stop": this.stopped.set(c.id, c.code); return;
      case "max_streams": {
        const k = c.bidi ? "bidi" : "uni";
        if (this.peerMax[k] != null && c.max < this.peerMax[k]) return this.fatal(cap.FLOW_CONTROL_ERROR, "a WT_MAX_STREAMS that lowers the limit");
        this.peerMax[k] = c.max;
        for (const w of this.creditWaiters.splice(0)) w();
        return;
      }
      case "close":
        this.closeCode = c.code;
        this.closeReason = c.message;
        this.finish({ code: c.code, reason: c.message });
        try { this.ws.close(1000); } catch { /* closed */ }
        return;
      default:
        return; // DATAGRAM, DRAIN, BLOCKED and unknown capsules: read and ignored
    }
  }

  onStream(id, data, fin, reset) {
    let ib = this.recv.get(id);
    if (!ib) {
      const peerInitiated = !cap.streamId.clientInitiated(id);
      if (!peerInitiated) return; // a half this end stopped reading, or one already ended
      const bidi = cap.streamId.bidi(id);
      const k = bidi ? "bidi" : "uni";
      const idx = cap.streamId.index(id);
      if (idx >= this.local[k].max) return this.fatal(cap.FLOW_CONTROL_ERROR, "a stream past WT_MAX_STREAMS");
      if (this.seen[k].has(idx)) return;
      this.seen[k].add(idx);
      const { ib: inbox, reader } = this.inbox(id);
      ib = inbox;
      this.recv.set(id, ib);
      if (bidi) {
        this.halves.set(id, 2);
        this.incomingBidi.push({ writer: this.writer(id), reader });
      } else this.incomingUni.push(reader);
    }
    if (data && data.length) ib.data(data.slice());
    if (reset != null) { ib.fail(reset); this.recv.delete(id); } else if (fin) { ib.end(); this.recv.delete(id); }
  }

  async open(bidi) {
    const k = bidi ? "bidi" : "uni";
    for (;;) {
      if (this.isClosed) throw new Error("the session is closed");
      const next = bidi ? this.nextBidi : this.nextUni;
      const max = this.peerMax[k];
      if (max != null && next < max) {
        if (bidi) this.nextBidi++; else this.nextUni++;
        return cap.streamId.nth(next, true, bidi);
      }
      if (max != null && this.blockedAt !== `${k}:${max}`) {
        this.blockedAt = `${k}:${max}`;
        this.counters.blocked_sent++;
        this.push(cap.streamsBlockedCapsule(bidi, max));
      }
      await new Promise((r) => this.creditWaiters.push(r));
    }
  }

  async openUni() {
    return this.writer(await this.open(false));
  }

  async openBidi() {
    const id = await this.open(true);
    const { ib, reader } = this.inbox(id);
    this.recv.set(id, ib);
    return { writer: this.writer(id), reader };
  }

  acceptUni() { return this.incomingUni.next(); }
  acceptBidi() { return this.incomingBidi.next(); }

  close(code = 0, reason = "") {
    if (this.isClosed) return;
    this.push(cap.closeSessionCapsule(code, reason));
    this.closeCode = code;
    this.closeReason = reason;
    try { this.ws.close(1000); } catch { /* closed */ }
    this.finish({ code, reason });
  }

  stats() { return { ...this.counters, peer_max: { ...this.peerMax }, local_max: { bidi: this.local.bidi.max, uni: this.local.uni.max } }; }
}

export async function connectWebSocket(url, { timeoutMs = 10000, draft = DEFAULT_DRAFT } = {}) {
  const want = protocolOf(draft);
  const ws = new WebSocket(url, want);
  ws.binaryType = "arraybuffer";
  await new Promise((ok, no) => {
    const t = setTimeout(() => no(new Error(`WebSocket not open in ${timeoutMs} ms`)), timeoutMs);
    ws.onopen = () => { clearTimeout(t); ok(); };
    ws.onerror = () => { clearTimeout(t); no(new Error("the WebSocket did not open")); };
    ws.onclose = (e) => { clearTimeout(t); no(new Error(`the WebSocket closed: ${e.code} ${e.reason}`)); };
  });
  if (ws.protocol !== want) {
    ws.close();
    throw new Error(`the relay answered the subprotocol "${ws.protocol}"`);
  }
  const s = new WsSession(ws);
  s.url = url;
  s.draft = draft;
  // Until the relay's credit arrives this end opens nothing; it comes first.
  return s;
}

// Connect by WebTransport, falling back to the WebSocket binding when the browser has no
// WebTransport or the session does not open; `force` = "websocket" skips WebTransport.
export async function connect({ webtransport, websocket, hashes, force, draft = DEFAULT_DRAFT, log = () => {} }) {
  if (force !== "websocket" && webtransport) {
    try {
      const t = await connectWebTransport(webtransport, { hashes, draft });
      log(`transport: WebTransport ${webtransport} (protocol "${t.protocol}")`);
      return t;
    } catch (e) {
      if (!websocket || force === "webtransport") throw e;
      log(`transport: WebTransport failed (${e.message}); falling back to the WebSocket binding`);
    }
  }
  const t = await connectWebSocket(websocket, { draft });
  log(`transport: WebSocket binding ${websocket} (subprotocol "${t.protocol}")`);
  return t;
}