// The WebRTC transport for browser hosts: com.fedfork.exec.webrtc.connect (find the guest, seal and signal
// the offer, open the answer, check the DTLS fingerprint, open the datachannel, ssh over it) and the
// webrtc: scheme fetcher (HTTP/1.1 over a tag's datachannels). No deps (age, the peer connection and the
// ssh library are passed in). A page passes it in (hostCall's calls, runModule's fetchers).
import { httpChannel } from "./http-datachannel.js";
import { connectSsh, sshKeyPair } from "./ssh-datachannel.js";

export { connectSsh, httpChannel, sshKeyPair };
export * from "./record-signals.js";

// socialweb.computer guest transport: one STUN server, no TURN, non-trickle ICE.
const STUN = "stun:stun.cloudflare.com:3478";
const MAX_CIPHERTEXT = 16384;
const GATHER_MS = 20000;
const SRFLX_MS = 2000;
const POLL_MS = 5000;
const ANSWER_MS = 180000;
const OPEN_MS = 30000;
// runModule's fetchers for a page that only fetches (no store needed).
export const fetchers = (options) => new WebrtcHost(options).fetchers();

export const WEBRTC_NSID = "com.fedfork.exec.webrtc.connect";
// One fetch's cap: the guest agent's exec runs up to 600 s.
const FETCH_MS = 660000;
const TOKEN = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;
// httpChannel frames the request itself.
const FRAMING = new Set(["host", "content-length", "transfer-encoding", "connection"]);

// What the transport keeps in its store (a page's Cubit state, merged with the page's own):
//   phase: idle -> finding -> found -> offering -> offered -> answering -> connecting -> open -> closed,
//     error at any step (error holds why)
//   signal: the requester `signal` input of the sealed offer
//   ssh over the datachannel: authorizedKey to put in user_data, hostKey seen, hostKeys published
export const WEBRTC_STATE = Object.freeze({
	phase: "idle",
	receipt: null,
	tag: "",
	guest: null,
	signal: null,
	output: "",
	error: null,
	ssh: { authorizedKey: null, user: "root", hostKey: null, pinned: null, busy: false, error: null },
	hostKeys: [],
});

// A receipt arrives as its strong ref JSON {uri, cid}; the bidder is its repo unless given.
export function parseReceipt(text) {
	const ref = typeof text === "string" ? JSON.parse(text) : text;
	const bidder = ref?.bidder ?? ref?.uri?.match(/^at:\/\/(did:[^/]+)\//)?.[1];
	if (!bidder || !ref.cid) throw new Error("receipt must be a strong ref {uri, cid}");
	return { uri: ref.uri, cid: ref.cid, bidder };
}

const b64url = {
	encode: (text) => btoa(String.fromCharCode(...new TextEncoder().encode(text))).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, ""),
	decode: (text) => new TextDecoder().decode(Uint8Array.from(atob(text.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0))),
};

// webrtc:///<ref>/<tag>/<path...>, ref = base64url(JSON {receipt: {uri, cid}, bidder?}).
export const webrtcUrl = ({ receipt: { uri, cid }, bidder }, tag, path = "") =>
	`webrtc:///${b64url.encode(JSON.stringify({ receipt: { uri, cid }, ...(bidder && { bidder }) }))}/${tag}/${path.replace(/^\//, "")}`;

// A market endpoint (kcp offering endpoints {kind: webrtc, address}): webrtc://<address>/<path...>, address =
// base64url(JSON {did, listen}), the answerer's computer.socialweb.ingress.listen/<listen>. Its receipt is that
// listen record's uri; it is signalled over records (the host's `records`, recordSignals), datachannel label market.
export const LISTEN_COLLECTION = "computer.socialweb.ingress.listen";
export const endpointUrl = ({ did, listen }, path = "") => `webrtc://${b64url.encode(JSON.stringify({ did, listen }))}/${path.replace(/^\//, "")}`;

function parseEndpointUrl(u) {
	let ref;
	try {
		ref = JSON.parse(b64url.decode(u.host));
	} catch {
		throw new Error("a webrtc endpoint URL's host must be base64url JSON {did, listen}");
	}
	if (!(typeof ref?.did === "string" && /^did:[a-z]+:[\w.:%-]+$/.test(ref.did)) || !(typeof ref.listen === "string" && /^[A-Za-z0-9._:~-]{1,512}$/.test(ref.listen))) {
		throw new Error("a webrtc endpoint URL names {did, listen}");
	}
	return { receipt: { uri: `at://${ref.did}/${LISTEN_COLLECTION}/${ref.listen}`, cid: "listen", bidder: ref.did }, tag: "market", path: `${u.pathname || "/"}${u.search}`, records: true };
}

// A webrtc: URL as { receipt (parseReceipt, bidder from the ref when given), tag, path (with the query) }; an
// endpoint URL (webrtc://<address>/...) also has records: true.
export function parseWebrtcUrl(url) {
	const u = new URL(url);
	if (u.protocol === "webrtc:" && u.host) return parseEndpointUrl(u);
	const [, ref, tag, ...rest] = u.pathname.split("/");
	if (u.protocol !== "webrtc:" || u.host || !ref || !tag) throw new Error("a webrtc URL is webrtc:///<ref>/<tag>/<path>");
	let parsed;
	try {
		parsed = JSON.parse(b64url.decode(ref));
	} catch {
		throw new Error("a webrtc URL's ref must be base64url JSON {receipt: {uri, cid}, bidder?}");
	}
	const { uri, cid } = parsed?.receipt ?? {};
	if (parsed.bidder !== undefined && !(typeof parsed.bidder === "string" && /^did:[a-z]+:[\w.:%-]+$/.test(parsed.bidder))) throw new Error("a webrtc URL's bidder must be a DID");
	const receipt = parseReceipt({ uri, cid, bidder: parsed.bidder });
	const at = decodeURIComponent(tag);
	if (!/^[\x21-\x7e]{1,32}$/.test(at) || at === "ssh") throw new Error("a webrtc URL's tag must be a tunnels.json tag (1 to 32 printable bytes, not ssh)");
	return { receipt, tag: at, path: `/${rest.join("/")}${u.search}` };
}

const normal = (fingerprint) => fingerprint.trim().split(/\s+/).join(" ").toLowerCase();

// Every a=fingerprint in the answer must be the one the guest reported in vm.onNetwork.
export function checkFingerprint(sdp, expected) {
	const found = sdp.split(/\r?\n/).filter((l) => l.startsWith("a=fingerprint:")).map((l) => normal(l.slice("a=fingerprint:".length)));
	if (!found.length) throw new Error("the answer has no a=fingerprint line");
	const bad = found.find((f) => f !== normal(expected));
	if (bad) throw new Error(`the answer's DTLS fingerprint ${bad} is not the guest's ${expected}`);
}

// The guest's age key and DTLS fingerprint for a tag, and the answers, from guest events.
export function guestOf(events, tag) {
	const values = (kind, any = false) => events.filter((e) => e.kind === kind && (any || e.value?.tag === tag)).map((e) => e.value)
		.sort((a, b) => String(a.createdAt).localeCompare(String(b.createdAt)));
	// The agent has one DTLS certificate for all tags, and hat-9 published vm.onNetwork for
	// only the first tag (2026-11-08), so fall back to any webrtc onNetwork of this receipt.
	const network = (any) => values("vm.onNetwork", any).filter((v) => v.kind === "webrtc").at(-1)?.address;
	return {
		key: values("guest.publicKey").at(-1)?.key ?? null,
		fingerprint: network(false) ?? network(true) ?? null,
		hostKeys: events.filter((e) => e.kind === "vm.hostKeys").at(-1)?.value?.keys ?? [],
		answers: values("guest.signalWebRTC").filter((v) => v.kind === "answer").map((v) => v.ciphertext).reverse(),
	};
}

// The `signals` option for a local guest (`webrtc-peer`) in place of
// the bidder's records: GET events answers what `watch` returns ({kind, value} guest events), POST signal
// takes the `signal` input ({receipt, tag, offer}). Same sealed offers and answers, no account, no PDS.
// The body goes as text/plain, so the cross-port fetch needs no CORS preflight (webrtc-peer --origin allows
// only this page's origin).
const json = async (what, response) => {
	if (!response.ok) throw new Error(`${what} failed with ${response.status}: ${await response.text()}`);
	return response.json();
};
export const localSignals = (base, fetchImpl = fetch) => ({
	events: async () => (await json("events", await fetchImpl(new URL("events", base)))).events,
	signal: async (input) => json("signal", await fetchImpl(new URL("signal", base), { method: "POST", body: JSON.stringify(input) })),
});

// The state of a peer the page does not show (a fetch peer): a Cubit's state, emit and change events.
class Store extends EventTarget {
	state = WEBRTC_STATE;
	emit(patch) {
		this.state = { ...this.state, ...patch };
		this.dispatchEvent(new Event("change"));
	}
}

// The transport: com.fedfork.exec.webrtc.connect for a browser host (lexicons/com/fedfork/exec/webrtc/connect.json),
// `call` for hostCall's calls: a request finds, offers, answers and opens (then a command over ssh, tag ssh),
// a subscribe is a channel session. The page brings the rest:
//   store     a Cubit (atlibs javascript/src/cubit) whose state holds WEBRTC_STATE (state, emit, change events)
//   events    ({receipt, tag}) => guest events ({kind, value}) of that receipt and tag: the requester `watch`, or localSignals
//   signal    (input) => posts the sealed offer { receipt, tag, offer }: the requester `signal` as the session, or
//             localSignals; resolving without posting leaves it in state.signal for a host that can
//   makePeer  config => RTCPeerConnection
//   age       the age-encryption (typage) module
//   ssh       @microsoft/dev-tunnels-ssh namespace plus its Buffer (optional: ssh over the channel)
export class WebrtcHost {
	#store;
	#events;
	#signal;
	#makePeer;
	#age;
	#ssh;
	#sleep;
	#identity;
	#pc;
	#dc;
	#sshKey;
	#sshSession;
	#received = [];
	// Protocol 2 subscriptions to the datachannel (#channel): each gets every message and the close.
	#listeners = new Set();
	// webrtc: fetches: "<receipt uri> <tag>" -> Promise of { host (its own peer), idle (httpChannels) }.
	#peers = new Map();
	// endpoint URLs' signalling: { events, signal } over records (recordSignals)
	#records;

	constructor({ store = new Store(), events, signal, records, makePeer, age, ssh, sleep = (ms) => new Promise((r) => setTimeout(r, ms)) }) {
		this.#store = store;
		this.#records = records;
		this.#events = events;
		this.#signal = signal;
		this.#makePeer = makePeer;
		this.#age = age;
		this.#ssh = ssh;
		this.#sleep = sleep;
	}

	// hostCall's calls[WEBRTC_NSID]
	call = (request = {}) => (request.subscribe ? this.#channel(request.params ?? {}) : this.#connect(request.input ?? {}));

	// The NSIDs this transport serves, for hostCall's calls (their keys are what the host provides).
	get calls() {
		return { [WEBRTC_NSID]: this.call };
	}

	// runModule's fetchers: the webrtc: scheme.
	fetchers() {
		return { "webrtc:": this.fetch };
	}

	// The webrtc: scheme fetcher: {method, url, headers, body, signal} -> {status, headers, body (text)}, HTTP/1.1
	// to the guest's tunnel tag over a datachannel. Each (receipt, tag) has its own peer connection, kept for later
	// fetches and redialled once it fails, and a pool of datachannels, each one HTTP/1.1 connection serving one
	// request at a time: concurrent fetches (a long-poll read beside an exec) each take an idle one or open another.
	fetch = async ({ method = "GET", url, headers = {}, body, signal } = {}) => {
		const { receipt, tag, path, records } = parseWebrtcUrl(url);
		if (records && !this.#records) throw new Error("webrtc fetch: an endpoint URL needs the host's records signalling (recordSignals)");
		if (!TOKEN.test(method)) throw new Error(`webrtc fetch: a bad method ${method}`);
		const sent = {};
		for (const [k, v] of Object.entries(headers ?? {})) {
			if (!TOKEN.test(k) || typeof v !== "string" || /[\r\n\0]/.test(v)) throw new Error(`webrtc fetch: a bad header ${k}`);
			if (!FRAMING.has(k.toLowerCase())) sent[k] = v;
		}
		if (signal?.aborted) throw new Error("webrtc fetch: aborted");
		const peer = await this.#peer(receipt, tag, records ? this.#records : this);
		let conn;
		while ((conn = peer.idle.pop()) && !conn.reusable);
		if (!conn) {
			try {
				conn = httpChannel(await peer.host.#newChannel());
			} catch (error) {
				// The peer cannot open channels any more: the next fetch dials again.
				if (this.#peers.get(`${receipt.uri} ${tag}`) === peer.held) this.#forget(`${receipt.uri} ${tag}`);
				throw error;
			}
		}
		const stop = () => conn.close();
		const timer = setTimeout(stop, FETCH_MS);
		signal?.addEventListener("abort", stop, { once: true });
		let response;
		try {
			response = await conn.request({ method, path, headers: sent, body: body ?? "" });
		} catch (error) {
			throw new Error(`webrtc fetch: ${signal?.aborted ? "aborted" : error.message}`);
		} finally {
			clearTimeout(timer);
			signal?.removeEventListener("abort", stop);
		}
		if (conn.reusable) peer.idle.push(conn);
		return { status: response.status, headers: response.headers, body: new TextDecoder().decode(response.body) };
	};

	#forget(key) {
		this.#peers.get(key)?.then((p) => p.host.close(), () => {});
		this.#peers.delete(key);
	}

	// The peer of (receipt, tag): connected once, then reused while its connection holds.
	async #peer(ref, tag, via = this) {
		const key = `${ref.uri} ${tag}`;
		const held = this.#peers.get(key);
		const peer = held && (await held.catch(() => null));
		if (peer && !["failed", "closed"].includes(peer.host.#pc?.connectionState)) return peer;
		if (held && this.#peers.get(key) === held) this.#forget(key);
		if (!this.#peers.has(key)) {
			const signals = via === this ? { events: this.#events, signal: this.#signal } : via;
			const host = new WebrtcHost({ events: signals.events, signal: signals.signal, makePeer: this.#makePeer, age: this.#age, sleep: this.#sleep });
			const connecting = host.#connect({ receipt: ref, tag }).then(() => ({ host, idle: [httpChannel(host.#dc)], held: connecting }));
			connecting.catch(() => this.#peers.get(key) === connecting && this.#peers.delete(key));
			this.#peers.set(key, connecting);
		}
		return this.#peers.get(key);
	}

	// Another datachannel on this peer, labelled its tag; resolves open.
	#newChannel() {
		return new Promise((resolve, reject) => {
			const dc = this.#pc.createDataChannel(this.#store.state.tag, { ordered: true });
			const timer = setTimeout(() => (dc.close(), reject(new Error(`the datachannel did not open in ${OPEN_MS / 1000} s`))), OPEN_MS);
			dc.onopen = () => (clearTimeout(timer), resolve(dc));
			dc.onclose = () => (clearTimeout(timer), reject(new Error("the datachannel closed before it opened")));
		});
	}

	// com.fedfork.exec.webrtc.connect (lexicons/com/fedfork/exec/webrtc/connect.json): find, offer
	// (signal runs as this session), answer, open; then the command over ssh (tag ssh). The
	// channel stays open for the page.
	async #connect({ receipt, tag, command, stdin }) {
		if (command !== undefined && (tag !== "ssh" || !this.#ssh)) throw new Error(`${WEBRTC_NSID} runs a command only over tag ssh, with an ssh client in the page`);
		await this.find(receipt, tag);
		if (this.#store.state.phase === "found") await this.offer();
		if (this.#store.state.phase === "offered") await this.answer();
		await this.#opened();
		const { fingerprint } = this.#store.state.guest;
		if (command === undefined) return { tag, fingerprint };
		if (stdin !== undefined && !(typeof stdin === "string" && stdin.length <= 65536)) throw new Error("stdin must be a string of at most 65536 bytes");
		const output = await this.sshExec(command, undefined, stdin);
		return { tag, fingerprint, output, hostKey: { ...this.#store.state.ssh.hostKey, pinned: this.#store.state.ssh.pinned } };
	}

	// com.fedfork.exec.webrtc.connect with subscribe (params { receipt, tag }): connect as above, then
	// every datachannel message is a subscription message (text) and send(data) writes it (a string
	// as is, anything else as JSON). Closing the subscription leaves the channel open for the page.
	// Raw tags only: tag ssh carries the page's ssh session, not a byte stream.
	async #channel({ receipt, tag }) {
		if (tag === "ssh") throw new Error(`${WEBRTC_NSID} subscribe carries raw bytes; use a tag other than ssh`);
		await this.#connect({ receipt, tag });
		const dc = this.#dc;
		const queue = [];
		const decoder = new TextDecoder();
		let closed = false;
		let wake;
		const listener = (data) => {
			if (data === null) closed = true;
			else queue.push(typeof data === "string" ? data : decoder.decode(data, { stream: true }));
			wake?.();
		};
		this.#listeners.add(listener);
		const close = () => (this.#listeners.delete(listener), listener(null));
		return {
			close,
			send: async (data) => {
				if (closed || dc.readyState !== "open") throw new Error("the datachannel is closed");
				dc.send(new TextEncoder().encode(typeof data === "string" ? data : JSON.stringify(data)));
			},
			async *[Symbol.asyncIterator]() {
				for (;;) {
					while (!queue.length && !closed) await new Promise((resolve) => (wake = resolve));
					if (queue.length) yield queue.shift();
					else return;
				}
			},
		};
	}

	// Resolves when the datachannel opens; rejects on error, close, or OPEN_MS.
	#opened() {
		return new Promise((resolve, reject) => {
			const done = (settle) => {
				clearTimeout(timer);
				this.#store.removeEventListener("change", check);
				settle();
			};
			const check = () => {
				const { phase, error } = this.#store.state;
				if (phase === "open") done(resolve);
				else if (phase === "error" || phase === "closed") done(() => reject(new Error(error ?? "the datachannel closed")));
			};
			const timer = setTimeout(() => done(() => reject(new Error(`the datachannel did not open in ${OPEN_MS / 1000} s`))), OPEN_MS);
			this.#store.addEventListener("change", check);
			check();
		});
	}

	async find(receiptText, tag) {
		try {
			this.#close();
			this.#store.emit({ phase: "finding", receipt: parseReceipt(receiptText), tag: tag.trim(), guest: null, signal: null, output: "", error: null });
			const { receipt, tag: at } = this.#store.state;
			const { key, fingerprint, hostKeys } = guestOf(await this.#events({ receipt, tag: at }), at);
			this.#store.emit({ hostKeys });
			if (!key || !fingerprint) throw new Error(`the guest has not reported ${key ? "" : "a guest.publicKey "}${fingerprint ? "" : "a vm.onNetwork kind webrtc "}for tag ${this.#store.state.tag} yet`);
			this.#store.emit({ phase: "found", guest: { key, fingerprint } });
		} catch (error) {
			this.#store.emit({ phase: "error", error: error.message });
		}
	}

	// Non-trickle: gather every candidate, then seal {sdp, replyTo} to the guest's key.
	async offer() {
		try {
			this.#store.emit({ phase: "offering", error: null });
			const age = this.#age;
			this.#identity = await age.generateX25519Identity();
			const replyTo = await age.identityToRecipient(this.#identity);
			const pc = (this.#pc = this.#makePeer({ iceServers: [{ urls: STUN }] }));
			this.#wire(pc.createDataChannel(this.#store.state.tag, { ordered: true }));
			pc.addEventListener("connectionstatechange", () => {
				if (pc.connectionState !== "failed") return;
				// After the channel opened, a failure is the guest going away (e.g. released), not NAT.
				if (["open", "closed"].includes(this.#store.state.phase)) return this.#store.emit({ phase: "closed" });
				this.#store.emit({ phase: "error", error: `ICE ${pc.iceConnectionState}: a symmetric NAT on either side defeats STUN-only WebRTC (${STUN}, no TURN)` });
			});
			const gathered = this.#gathered(pc);
			await pc.setLocalDescription(await pc.createOffer());
			await gathered;
			const encrypter = new age.Encrypter();
			encrypter.addRecipient(this.#store.state.guest.key);
			const offer = age.armor.encode(await encrypter.encrypt(JSON.stringify({ sdp: pc.localDescription.sdp, replyTo })));
			if (offer.length > MAX_CIPHERTEXT) throw new Error(`the sealed offer is ${offer.length} bytes, over ${MAX_CIPHERTEXT} (too many local candidates)`);
			const { uri, cid } = this.#store.state.receipt;
			const signal = { receipt: { uri, cid }, tag: this.#store.state.tag, offer };
			await this.#signal(signal);
			this.#store.emit({ phase: "offered", signal });
		} catch (error) {
			this.#store.emit({ phase: "error", error: error.message });
		}
	}

	// Non-trickle, without waiting out the cap: done when gathering completes, on the null
	// candidate, or SRFLX_MS after the first server-reflexive (STUN) candidate; GATHER_MS caps it.
	// Attach before setLocalDescription, which starts gathering.
	#gathered(pc) {
		return new Promise((resolve) => {
			let srflx;
			const done = () => {
				clearTimeout(cap);
				clearTimeout(srflx);
				pc.removeEventListener("icegatheringstatechange", onState);
				pc.removeEventListener("icecandidate", onCandidate);
				resolve();
			};
			const cap = setTimeout(done, GATHER_MS);
			const onState = () => pc.iceGatheringState === "complete" && done();
			const onCandidate = ({ candidate }) => {
				if (!candidate) return done();
				if (srflx === undefined && / typ srflx /.test(candidate.candidate)) srflx = setTimeout(done, SRFLX_MS);
			};
			pc.addEventListener("icegatheringstatechange", onState);
			pc.addEventListener("icecandidate", onCandidate);
		});
	}

	// Signals name no key: open every answer for the tag and keep the one sealed to us.
	async answer() {
		this.#store.emit({ phase: "answering", error: null });
		try {
			for (const deadline = Date.now() + ANSWER_MS; Date.now() < deadline; await this.#sleep(POLL_MS)) {
				const { receipt, tag } = this.#store.state;
				for (const ciphertext of guestOf(await this.#events({ receipt, tag }), tag).answers) {
					const sdp = await this.#open(ciphertext);
					if (!sdp) continue;
					checkFingerprint(sdp, this.#store.state.guest.fingerprint);
					await this.#pc.setRemoteDescription({ type: "answer", sdp });
					if (this.#store.state.phase === "answering") this.#store.emit({ phase: "connecting" });
					return;
				}
				if (this.#store.state.phase !== "answering") return;
			}
			throw new Error("no answer sealed to this session arrived");
		} catch (error) {
			this.#store.emit({ phase: "error", error: error.message });
		}
	}

	async #open(ciphertext) {
		try {
			const decrypter = new this.#age.Decrypter();
			decrypter.addIdentity(this.#identity);
			return JSON.parse(await decrypter.decrypt(this.#age.armor.decode(ciphertext), "text")).sdp || null;
		} catch {
			return null;
		}
	}

	// The datachannel is raw TCP to the guest's tunnel port.
	#wire(dc) {
		this.#dc = dc;
		dc.binaryType = "arraybuffer";
		const decoder = new TextDecoder();
		dc.onopen = () => this.#store.emit({ phase: "open" });
		dc.onclose = () => {
			for (const listener of this.#listeners) listener(null);
			this.#listeners.clear();
			if (this.#store.state.phase !== "error") this.#store.emit({ phase: "closed" });
		};
		dc.onmessage = ({ data }) => {
			if (this.#listeners.size) return this.#listeners.forEach((listener) => listener(data));
			// Tag ssh: keep the bytes (sshd's banner) for connectSsh instead of printing them.
			if (this.#store.state.tag === "ssh" && this.#ssh) return this.#received.push(data);
			this.#store.emit({ output: this.#store.state.output + (typeof data === "string" ? data : decoder.decode(data, { stream: true })) });
		};
	}

	// Tag ssh: the line runs over ssh; any other tag: raw bytes to the guest port.
	send(line) {
		if (this.#dc?.readyState !== "open") return;
		if (this.#store.state.tag === "ssh" && this.#ssh) return this.sshExec(line).catch(() => undefined);
		this.#dc.send(new TextEncoder().encode(`${line}\n`));
	}

	// The authorized_keys line of this page's ssh key (memory only: a reload makes a new one).
	async sshKey() {
		this.#sshKey ??= sshKeyPair(this.#ssh);
		const { authorizedKey } = await this.#sshKey;
		this.#store.emit({ ssh: { ...this.#store.state.ssh, authorizedKey } });
		return authorizedKey;
	}

	// Runs a command over ssh on the open datachannel (connects on first use); output joins state.output.
	// stdin (optional) goes to the command, never into state.
	async sshExec(command, user = this.#store.state.ssh.user, stdin) {
		this.#store.emit({ ssh: { ...this.#store.state.ssh, busy: true, error: null } });
		try {
			if (!this.#sshSession) {
				const { keyPair } = await (this.#sshKey ??= sshKeyPair(this.#ssh));
				const session = await connectSsh(this.#ssh, this.#dc, keyPair, user, this.#received.splice(0));
				this.#sshSession = session;
				const pinned = this.#store.state.hostKeys.some((k) => k.fingerprint === session.hostKey.fingerprint);
				this.#store.emit({ ssh: { ...this.#store.state.ssh, user, hostKey: session.hostKey, pinned } });
			}
			const output = await this.#sshSession.exec(command, stdin);
			this.#store.emit({ output: `${this.#store.state.output}$ ${command}\n${output}`, ssh: { ...this.#store.state.ssh, busy: false } });
			return output;
		} catch (error) {
			this.#store.emit({ ssh: { ...this.#store.state.ssh, busy: false, error: error.message } });
			throw error;
		}
	}

	#close() {
		this.#sshSession?.close();
		this.#pc?.close();
		this.#pc = this.#dc = this.#identity = this.#sshSession = undefined;
	}

	// Closes the page's channel and every fetch peer.
	close() {
		for (const key of [...this.#peers.keys()]) this.#forget(key);
		this.#close();
		this.#store.emit({ phase: "closed" });
	}
}