johnandersen777.bsky.social/atlibs-webrtc

Switch repository

Loading…

// An HTTP/1.1 client over one RTCDataChannel: the channel is the connection (the guest agent bridges a
// tag to its unix socket, e.g. shell to its MCP server; webrtc-peer bridges a tag to a TCP port). One request
// at a time; keep-alive unless the response says close or is delimited by the close. No deps.

const HEAD_MAX = 65536;
// A response body over this is refused (the guest's tools cap their output well under it).
export const BODY_MAX = 1 << 22;
// The datachannel's message size every browser and pion take.
const CHUNK = 16384;
const enc = new TextEncoder();
const dec = new TextDecoder();

// Bytes from the channel, read as lines and exact counts.
class Bytes {
	#parts = [];
	#size = 0;
	#wake;
	ended = null;

	push(chunk) {
		this.#parts.push(chunk);
		this.#size += chunk.length;
		this.#wake?.();
	}

	end(error) {
		this.ended ??= error;
		this.#wake?.();
	}

	#flat() {
		if (this.#parts.length !== 1) {
			const all = new Uint8Array(this.#size);
			let at = 0;
			for (const p of this.#parts) all.set(p, (at += p.length) - p.length);
			this.#parts = [all];
		}
		return this.#parts[0];
	}

	async #more() {
		if (this.ended) throw this.ended;
		await new Promise((resolve) => (this.#wake = resolve));
		this.#wake = undefined;
	}

	async take(n) {
		while (this.#size < n) await this.#more();
		const all = this.#flat();
		this.#parts = [all.subarray(n)];
		this.#size -= n;
		return all.subarray(0, n);
	}

	async line(max = HEAD_MAX) {
		for (;;) {
			const all = this.#flat();
			for (let i = 0; i + 1 < all.length; i++) if (all[i] === 13 && all[i + 1] === 10) return dec.decode((await this.take(i + 2)).subarray(0, i));
			if (this.#size > max) throw new Error("the HTTP response head is too long");
			await this.#more();
		}
	}

	// Everything until the channel closes (a response with neither length nor chunks).
	async rest(max) {
		while (!this.ended) {
			if (this.#size > max) throw new Error(`the HTTP response body is over ${max} bytes`);
			await new Promise((resolve) => (this.#wake = resolve));
		}
		return this.take(this.#size);
	}
}

const tooBig = (n) => {
	if (n > BODY_MAX) throw new Error(`the HTTP response body is over ${BODY_MAX} bytes`);
	return n;
};

async function readBody(bytes, status, headers) {
	if (status === 204 || status === 304) return [new Uint8Array(0), true];
	if (/chunked/i.test(headers["transfer-encoding"] ?? "")) {
		const parts = [];
		let size = 0;
		for (;;) {
			const n = parseInt((await bytes.line()).split(";")[0], 16);
			if (!(n >= 0)) throw new Error("a bad HTTP chunk size");
			if (n === 0) break;
			tooBig((size += n));
			parts.push(await bytes.take(n));
			await bytes.take(2);
		}
		while ((await bytes.line()) !== ""); // trailers
		const body = new Uint8Array(size);
		parts.reduce((at, p) => (body.set(p, at), at + p.length), 0);
		return [body, true];
	}
	if (headers["content-length"] !== undefined) {
		const n = Number(headers["content-length"]);
		if (!Number.isSafeInteger(n) || n < 0) throw new Error("a bad HTTP content-length");
		return [await bytes.take(tooBig(n)), true];
	}
	return [await bytes.rest(BODY_MAX), false];
}

/**
 * The client on an open datachannel (binaryType arraybuffer). It takes the channel's onmessage and onclose.
 * request({method, path, headers, body}) -> {status, headers (lowercase names), body: Uint8Array};
 * reusable: false after a response that ends the connection (or once the channel closed).
 */
export function httpChannel(dc) {
	const bytes = new Bytes();
	let busy = false;
	let keep = true;
	dc.binaryType = "arraybuffer";
	dc.onmessage = ({ data }) => bytes.push(typeof data === "string" ? enc.encode(data) : new Uint8Array(data));
	dc.onclose = () => bytes.end(new Error("the datachannel closed"));
	return {
		get reusable() {
			return keep && !busy && !bytes.ended && dc.readyState === "open";
		},
		close: () => ((keep = false), dc.close()),
		async request({ method = "POST", path = "/", headers = {}, body = "" }) {
			if (busy) throw new Error("one HTTP request at a time on a datachannel");
			busy = true;
			try {
				const payload = typeof body === "string" ? enc.encode(body) : body;
				const head = [`${method} ${path} HTTP/1.1`, "host: guest", ...Object.entries({ ...headers, "content-length": payload.length }).map(([k, v]) => `${k}: ${v}`), "", ""].join("\r\n");
				const out = new Uint8Array(enc.encode(head).length + payload.length);
				out.set(enc.encode(head));
				out.set(payload, out.length - payload.length);
				for (let at = 0; at < out.length; at += CHUNK) dc.send(out.subarray(at, at + CHUNK));
				const [version, code] = (await bytes.line()).split(" ");
				const status = Number(code);
				if (!/^HTTP\/1\.[01]$/.test(version) || !(status >= 100 && status < 600)) throw new Error(`not an HTTP response: ${version} ${code}`);
				const response = {};
				for (let line; (line = await bytes.line()) !== ""; ) {
					const i = line.indexOf(":");
					if (i > 0) response[line.slice(0, i).trim().toLowerCase()] = line.slice(i + 1).trim();
				}
				const [data, framed] = await readBody(bytes, status, response);
				keep &&= framed && version === "HTTP/1.1" && !/close/i.test(response.connection ?? "");
				return { status, headers: response, body: data };
			} catch (error) {
				keep = false;
				throw error;
			} finally {
				busy = false;
				if (!keep) dc.close();
			}
		},
	};
}