captoys/toybox/supervisor.js

/** @file toybox supervisor: one Vat per name; its app runs as a facet on installed code. */
import { DurableObject, WorkerEntrypoint } from "cloudflare:workers";
import { newWebSocketRpcSession } from "capnweb";
import { mergeLabels, urlLabels } from "./labels.js";

/**
 * A stored grant: labels as JSON, or (before labels) a bare `host[:port]`.
 *
 * @param {string} stored
 */
const grantLabels = (stored) => {
	if (stored.startsWith("{")) return JSON.parse(stored);
	const { host, port } = urlLabels(`http://${stored}`);
	return stored.includes(":") ? { host, port } : { host };
};

const hex = (buf) => [...new Uint8Array(buf)].map((b) => b.toString(16).padStart(2, "0")).join("");

/**
 * Holds the vat's installed code and egress grant; the app facet holds the app's state.
 *
 * Code is stored by hash so the facet can be rebuilt after hibernation or a
 * restart. The app's global `fetch` throws; a granted host arrives as
 * `env.egress`, an {@link Egress} the app can pass along like any object.
 *
 * Each `start` begins a new incarnation, which names the loaded Worker: the
 * same code with a different grant must not share a Worker.
 */
export class Vat extends DurableObject {
	constructor(ctx, env) {
		super(ctx, env);
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS bundle (hash TEXT PRIMARY KEY, src TEXT)");
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS current (k INTEGER PRIMARY KEY CHECK (k = 0), hash TEXT)");
		ctx.storage.sql.exec(
			"CREATE TABLE IF NOT EXISTS incarnation (k INTEGER PRIMARY KEY CHECK (k = 0), n INTEGER, egress TEXT)",
		);
	}

	/**
	 * Install `src` and run the app on it; the app's storage carries over.
	 *
	 * @param {string} src an ES module exporting `class App extends DurableObject`
	 * @param {string | null} egress the app's grant: egress labels as JSON (see labels.js), or none
	 * @returns {Promise<string>} the code's SHA-256, hex
	 */
	async start(src, egress) {
		const hash = hex(await crypto.subtle.digest("SHA-256", new TextEncoder().encode(src)));
		const { sql } = this.ctx.storage;
		sql.exec("INSERT OR IGNORE INTO bundle VALUES (?, ?)", hash, src);
		sql.exec("INSERT OR REPLACE INTO current VALUES (0, ?)", hash);
		const [prev] = sql.exec("SELECT n FROM incarnation").toArray();
		sql.exec("INSERT OR REPLACE INTO incarnation VALUES (0, ?, ?)", (prev?.n ?? 0) + 1, egress);
		this.ctx.facets.abort("app", new Error("upgraded"));
		return hash;
	}

	#app({ hash, src, n, egress }) {
		return this.ctx.facets.get("app", () => {
			const worker = this.env.LOADER.get(`${this.ctx.id}:${n}`, () => ({
				compatibilityDate: "2026-10-01",
				mainModule: "app.js",
				modules: { "app.js": src },
				env: egress ? { egress: this.ctx.exports.Egress({ props: { labels: grantLabels(egress) } }) } : {},
				globalOutbound: null,
			}));
			return { class: worker.getDurableObjectClass("App") };
		});
	}

	fetch(request) {
		const [row] = this.ctx.storage.sql
			.exec(
				`SELECT b.hash, b.src, coalesce(i.n, 0) AS n, i.egress
				 FROM current c JOIN bundle b USING (hash) LEFT JOIN incarnation i ON i.k = 0`,
			)
			.toArray();
		if (!row) return new Response("no code installed\n", { status: 404 });
		return this.#app(row).fetch(request);
	}
}

/**
 * A vat's egress grant: a switchboard handle constrained to the labels in
 * `ctx.props`, which no holder can change. The switchboard enforces them;
 * a refusal comes back as a 403.
 */
export class Egress extends WorkerEntrypoint {
	/** A narrower grant: these labels too (conflicting ones throw). */
	constrain(more) {
		return this.ctx.exports.Egress({ props: { labels: mergeLabels(this.ctx.props.labels, more) } });
	}

	async #handle() {
		const resp = await this.env.SWITCHBOARD.fetch("http://switchboard/", { headers: { Upgrade: "websocket" } });
		const ws = resp.webSocket;
		if (!ws) throw new Error(`switchboard refused a session: ${resp.status}`);
		ws.accept();
		return newWebSocketRpcSession(ws).constrain(this.ctx.props.labels);
	}

	async fetch(request) {
		try {
			const handle = await this.#handle();
			if (request.headers.get("Upgrade")?.toLowerCase() !== "websocket") return await handle.fetch(request);
			const protocols = (request.headers.get("Sec-WebSocket-Protocol") ?? "").split(",").map((p) => p.trim()).filter(Boolean);
			const { readable, writable, protocol } = await handle.webSocket(request.url, protocols);
			return this.#bridge(readable, writable, protocol);
		} catch (e) {
			return new Response(`egress refused: ${e.message}\n`, { status: 403 });
		}
	}

	/** A WebSocket for the vat whose messages travel the switchboard's streams. */
	#bridge(readable, writable, protocol) {
		const [client, server] = Object.values(new WebSocketPair());
		server.accept();
		const writer = writable.getWriter();
		server.addEventListener("message", (e) => writer.write(e.data).catch(() => server.close(1011)));
		server.addEventListener("close", () => writer.close().catch(() => {}));
		this.ctx.waitUntil(
			(async () => {
				for await (const m of readable) server.send(m);
				server.close();
			})().catch(() => server.close(1011)),
		);
		const headers = protocol ? { "Sec-WebSocket-Protocol": protocol } : {};
		return new Response(null, { status: 101, webSocket: client, headers });
	}
}

const vatOf = (env, url) => env.VATS.getByName(new URL(url).pathname.split("/")[1]);

/** The http socket: `/<vat>/...` goes to that vat's app. */
export class Public extends WorkerEntrypoint {
	fetch(request) {
		return vatOf(this.env, request.url).fetch(request);
	}
}

/**
 * The admin socket: `PUT /<vat>` with an ES module body installs and starts
 * it; `?egress=<labels as JSON>` grants it egress within those labels.
 */
export class Admin extends WorkerEntrypoint {
	async fetch(request) {
		if (request.method !== "PUT") return new Response("PUT code to /<vat>\n", { status: 405 });
		const egress = new URL(request.url).searchParams.get("egress");
		return new Response((await vatOf(this.env, request.url).start(await request.text(), egress)) + "\n");
	}
}

export default Public;