captoys/toybox/supervisor.js

/** @file toybox supervisor: vats with code installed at runtime, each app a facet; ingress by mount, egress by grant. */
import "@endo/init";
import { DurableObject, WorkerEntrypoint } from "cloudflare:workers";
import { RpcTarget, newWebSocketRpcSession, newWorkersRpcResponse } from "capnweb";
import { E, Far, passStyleOf } from "@endo/far";
import { makeClient, swissnumFromBytes } from "@endo/ocapn";
import { mergeLabels, urlLabels } from "./labels.js";
import { LOOKUP, MOUNT, MOUNT_LABELS, MOUNT_SCHEMA, OVERLAP, UNMOUNT, UNMOUNT_LABELS } from "./mounts-sql.js";
import { locationOf, makeRpcNetLayer } from "./netlayer.js";
import VAT_RUNTIME from "vat-runtime.txt";

/**
 * 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("");

/**
 * One vat, by DO id: holds its installed code and egress grant; the app
 * facet holds the app's state. Its mount, if any, is in the {@link Directory}.
 *
 * 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)",
		);
		// absent: an app vat (exports class App), from before OCapN vats
		ctx.storage.sql.exec("CREATE TABLE IF NOT EXISTS kind (k INTEGER PRIMARY KEY CHECK (k = 0), kind 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
	 * @param {"app" | "ocapn"} kind `app`: `src` exports `class App`; `ocapn`:
	 *   `src` exports `buildRootObject`, run by vat-runtime.js
	 * @returns {Promise<string>} the code's SHA-256, hex
	 */
	async start(src, egress, kind = "app") {
		const hash = hex(await crypto.subtle.digest("SHA-256", encodeUtf8(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);
		sql.exec("INSERT OR REPLACE INTO kind VALUES (0, ?)", kind);
		this.ctx.facets.abort("app", new Error("upgraded"));
		return hash;
	}

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

	#current() {
		const [row] = this.ctx.storage.sql
			.exec(
				`SELECT b.hash, b.src, coalesce(i.n, 0) AS n, i.egress, coalesce(k.kind, 'app') AS kind
				 FROM current c JOIN bundle b USING (hash)
				 LEFT JOIN incarnation i ON i.k = 0 LEFT JOIN kind k ON k.k = 0`,
			)
			.toArray();
		return row;
	}

	fetch(request) {
		const row = this.#current();
		if (!row) return new Response("no code installed\n", { status: 404 });
		return this.#app(row).fetch(request);
	}

	/** An OCapN vat's gateway, for the router. */
	gateway() {
		const row = this.#current();
		if (row?.kind !== "ocapn") throw new Error("not an OCapN vat");
		return this.#app(row).gateway();
	}
}

/**
 * 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 });
	}
}

/** The vatAdmin's designator, and the swissnum of its bootstrap object. */
const VAT_ADMIN = "vatAdmin";
/** The swissnum of an OCapN vat's bootstrap (see vat-runtime.js). */
const VAT_SWISSNUM = "vat";
/** UTF-8 bytes of a string: the method alone, so no encoder object is in module scope. */
const encodeUtf8 = harden(TextEncoder.prototype.encode.bind(new TextEncoder()));

/**
 * The vatAdmin: an OCapN peer, and the router through which every OCapN
 * peer here reaches every other. One instance, by name.
 *
 * Peers find each other by designator: the vatAdmin's is `vatAdmin`, a
 * vat's is its id, and a client registers its own when it connects to
 * the admin socket. Nothing here can dial out, so a handoff's receiver
 * reaches the exporter through this router.
 */
export class VatAdmin extends DurableObject {
	#clients = new Map();
	/** mount key → the handler serving it: live, so lost when this object is evicted */
	#handlers = new Map();
	/** mount key → resolvers waiting for a handler */
	#waiting = new Map();
	/** mount key ↔ its ingress object, made again as needed */
	#ingresses = new Map();
	#ingressKeys = new WeakMap();
	#reviving = new Map();
	#ready;

	constructor(ctx, env) {
		super(ctx, env);
		const { sql } = ctx.storage;
		// a vat's params, encoded (see #encode); a vat absent here can't be restarted
		sql.exec("CREATE TABLE IF NOT EXISTS vat_params (vat TEXT PRIMARY KEY, params TEXT)");
		// which vat was given which ingress, so a request can restart it
		sql.exec("CREATE TABLE IF NOT EXISTS ingress_vat (key TEXT PRIMARY KEY, vat TEXT)");
		this.#ready = this.#boot();
	}

	/** @param {string} key */
	#ingressFor(key) {
		let ingress = this.#ingresses.get(key);
		if (!ingress) {
			ingress = Far("ingress", {
				/** @param {{ handle: (request: object) => object }} handler */
				serve: (handler) => {
					this.#handlers.set(key, handler);
					for (const resolve of this.#waiting.get(key) ?? []) resolve();
					this.#waiting.delete(key);
				},
			});
			this.#ingresses.set(key, ingress);
			this.#ingressKeys.set(ingress, key);
		}
		return ingress;
	}

	/**
	 * `params` as JSON, an ingress by its key; `null` if anything in it can't
	 * be made again here (a remote object of another peer, say).
	 */
	#encode(params) {
		const keys = [];
		const walk = (v) => {
			const style = passStyleOf(v);
			if (style === "remotable") {
				const key = this.#ingressKeys.get(v);
				if (!key) throw new Error("not restartable");
				keys.push(key);
				return { "@ingress": key };
			}
			if (style === "copyArray") return v.map(walk);
			if (style === "copyRecord") return Object.fromEntries(Object.entries(v).map(([k, x]) => [k, walk(x)]));
			if (["string", "number", "boolean", "null"].includes(style)) return v;
			throw new Error("not restartable");
		};
		try {
			return { json: JSON.stringify(walk(params)), keys };
		} catch {
			return null;
		}
	}

	#decode(json) {
		const walk = (v) => {
			if (Array.isArray(v)) return v.map(walk);
			if (v && typeof v === "object") {
				if ("@ingress" in v) return this.#ingressFor(v["@ingress"]);
				return Object.fromEntries(Object.entries(v).map(([k, x]) => [k, walk(x)]));
			}
			return v;
		};
		return harden(walk(JSON.parse(json)));
	}

	/** (Re)start a vat's code with `params`: it rebuilds its root. */
	async #startVat(vat, params) {
		const vatBootstrap = await this.#enliven(vat, VAT_SWISSNUM);
		return E(vatBootstrap).start(params);
	}

	/**
	 * The handler for a mount whose handler went with an evicted instance:
	 * restart the vat given its ingress, and wait for it to serve again.
	 *
	 * @param {string} key
	 */
	#revive(key) {
		let reviving = this.#reviving.get(key);
		if (!reviving) {
			reviving = (async () => {
				const [row] = this.ctx.storage.sql
					.exec("SELECT i.vat, p.params FROM ingress_vat i JOIN vat_params p USING (vat) WHERE i.key = ?", key)
					.toArray();
				if (!row) return null;
				const served = new Promise((resolve) => this.#waiting.set(key, [...(this.#waiting.get(key) ?? []), resolve]));
				await this.#startVat(row.vat, this.#decode(row.params));
				await Promise.race([served, new Promise((resolve) => setTimeout(resolve, 5000))]);
				return this.#handlers.get(key) ?? null;
			})().finally(() => this.#reviving.delete(key));
			this.#reviving.set(key, reviving);
		}
		return reviving;
	}

	async #boot() {
		const vatAdmin = Far("vatAdmin", {
			/**
			 * Create an OCapN vat running `src`; `params` (any passable,
			 * remote objects included) reach its `buildRootObject`.
			 *
			 * If `params` holds only data and ingresses, they are kept, and the
			 * vat is restarted with them when a request needs it after an
			 * eviction or a restart.
			 *
			 * @param {string} src an ES module exporting `buildRootObject(vatPowers, params)`
			 */
			createVat: async (src, params = {}) => {
				const id = this.env.VATS.newUniqueId();
				const vat = id.toString();
				await this.env.VATS.get(id).start(src, null, "ocapn");
				const encoded = this.#encode(params);
				if (encoded) {
					const { sql } = this.ctx.storage;
					sql.exec("INSERT INTO vat_params VALUES (?, ?)", vat, encoded.json);
					for (const key of encoded.keys) sql.exec("INSERT OR REPLACE INTO ingress_vat VALUES (?, ?)", key, vat);
				}
				const root = await this.#startVat(vat, params);
				return harden({ id: vat, root });
			},
			/**
			 * Reserve a mount at `labels` and return the ingress for it: whoever
			 * holds the ingress (a vat it is passed to, say) serves requests
			 * that carry those labels.
			 *
			 * @param {Record<string, string>} labels
			 */
			makeIngress: async (labels) => {
				const key = `ingress:${crypto.randomUUID()}`;
				await this.env.DIRECTORY.getByName("directory").mount(key, harden({ ...labels }));
				return this.#ingressFor(key);
			},
		});
		// workerd runs FinalizationRegistry callbacks outside any request, where
		// sending the resulting GC message is forbidden I/O. So imports are
		// never collected here; they go when their session does.
		this.ocapn = makeClient({
			debugLabel: VAT_ADMIN,
			swissnumTable: new Map([[VAT_ADMIN, vatAdmin]]),
			enableImportCollection: false,
		});
		const netlayer = await this.ocapn.registerNetlayer(
			makeRpcNetLayer({ designator: VAT_ADMIN, openGateway: (designator) => this.gateway(designator) }),
		);
		this.own = netlayer.gateway;
	}

	#enliven(designator, swissnum) {
		return this.ocapn.enlivenSturdyRef(
			this.ocapn.makeSturdyRef(locationOf(designator), swissnumFromBytes(encodeUtf8(swissnum))),
		);
	}

	/**
	 * The gateway of the peer with this designator.
	 *
	 * @param {string} designator
	 */
	async gateway(designator) {
		await this.#ready;
		if (designator === VAT_ADMIN) return this.own;
		const client = this.#clients.get(designator);
		if (client) return client;
		let id;
		try {
			id = this.env.VATS.idFromString(designator);
		} catch {
			throw new Error("no such peer");
		}
		return this.env.VATS.get(id).gateway();
	}

	/**
	 * One HTTP request, as passable data, for the handler serving `key`;
	 * `null` when nothing serves it.
	 *
	 * @param {string} key
	 * @param {object} request
	 */
	async serveHttp(key, request) {
		await this.#ready;
		const handler = this.#handlers.get(key) ?? (await this.#revive(key));
		return handler ? E(handler).handle(harden(request)) : null;
	}

	/** A WebSocket here starts a Cap'n Web session whose main object is an {@link AdminSession}. */
	fetch(request) {
		return newWorkersRpcResponse(request, new AdminSession(this, this.#clients));
	}
}

/**
 * What an OCapN client on the admin socket gets: register its own
 * gateway under a designator, and reach any peer's.
 */
class AdminSession extends RpcTarget {
	#vatAdmin;
	#clients;

	constructor(vatAdmin, clients) {
		super();
		this.#vatAdmin = vatAdmin;
		this.#clients = clients;
	}

	/** @param {string} designator @param {any} gateway */
	register(designator, gateway) {
		if (designator === VAT_ADMIN || /^[0-9a-f]{64}$/.test(designator) || this.#clients.has(designator)) {
			throw new Error("designator taken");
		}
		this.#clients.set(designator, gateway.dup?.() ?? gateway);
	}

	/** @param {string} designator */
	gateway(designator) {
		return this.#vatAdmin.gateway(designator);
	}
}

/** A vat's way to the router: `env.router` in an OCapN vat. */
export class Router extends WorkerEntrypoint {
	gateway(designator) {
		return this.env.VAT_ADMIN.getByName(VAT_ADMIN).gateway(designator);
	}
}

/**
 * Mounts: which vat serves which label set. One instance, by name.
 *
 * Registration follows Flower: a mount must conflict with every other, so
 * no request can carry two mounts' labels. Lookup is stricter than
 * Flower's: the request must carry all of a mount's labels.
 */
export class Directory extends DurableObject {
	constructor(ctx, env) {
		super(ctx, env);
		for (const statement of MOUNT_SCHEMA) ctx.storage.sql.exec(statement);
	}

	/**
	 * Mount `vat` at `labels`, replacing its mount; `null` unmounts it.
	 *
	 * @param {string} vat a Vat's DO id, hex
	 * @param {Record<string, string> | null} labels
	 */
	mount(vat, labels) {
		const { sql } = this.ctx.storage;
		this.ctx.storage.transactionSync(() => {
			sql.exec(UNMOUNT_LABELS, vat);
			if (!labels) return void sql.exec(UNMOUNT, vat);
			const json = JSON.stringify(labels);
			if (sql.exec(OVERLAP, json, vat).toArray().length) {
				throw new Error("mount overlaps another: add a label whose value tells them apart");
			}
			sql.exec(MOUNT, vat, json);
			sql.exec(MOUNT_LABELS, vat, json);
		});
	}

	/**
	 * The mount whose labels `request` carries, if any. Registration keeps
	 * mounts pairwise conflicting, so at most one can match.
	 *
	 * @param {Record<string, string>} request
	 */
	lookup(request) {
		const [row] = this.ctx.storage.sql.exec(LOOKUP, JSON.stringify(request)).toArray();
		return row ? { vat: row.vat, labels: JSON.parse(row.labels) } : null;
	}
}

/**
 * A request's labels: its query parameters.
 *
 * Routing carries no secrets: a URL ends up in logs and `Referer`s. A
 * secret travels in-band instead, as on a webkey page: the page keeps it
 * in its fragment, reaches the vat by public labels, and sends it in the
 * body of a call, which the vat checks like a swissnum.
 *
 * @param {Request} request
 */
const requestLabels = (request) => {
	const labels = {};
	for (const [k, v] of new URL(request.url).searchParams) {
		if (k in labels) throw new Error(`label ${k} given twice`);
		labels[k] = v;
	}
	return Object.freeze(labels);
};

/**
 * The http socket: the ingress. A request reaches the vat mounted at labels
 * it carries, which sees the mount's and the request's labels together in
 * a `Toybox-Connection-Labels` header (JSON), as a Flower server gets its
 * connection labels.
 */
export class Public extends WorkerEntrypoint {
	async fetch(request) {
		let labels;
		try {
			labels = requestLabels(request);
		} catch (e) {
			return new Response(`bad labels: ${e.message}\n`, { status: 400 });
		}
		const mount = await this.env.DIRECTORY.getByName("directory").lookup(labels);
		if (!mount) return new Response("nothing mounted here\n", { status: 404 });
		if (mount.vat.startsWith("ingress:")) return this.#serveOcapn(mount, labels, request);
		const forwarded = new Request(request);
		forwarded.headers.set("Toybox-Connection-Labels", JSON.stringify(mergeLabels(mount.labels, labels)));
		const resp = await this.env.VATS.get(this.env.VATS.idFromString(mount.vat)).fetch(forwarded);
		if (resp.webSocket) return resp;
		const sealed = new Response(resp.body, resp);
		sealed.headers.set("Referrer-Policy", "no-referrer");
		return sealed;
	}

	/** A mount made by `makeIngress`: the request goes, as data, to whoever serves it. */
	async #serveOcapn(mount, labels, request) {
		const { pathname } = new URL(request.url);
		const body = ["GET", "HEAD"].includes(request.method) ? "" : await request.text();
		const req = { method: request.method, path: pathname, labels: mergeLabels(mount.labels, labels), body };
		const res = await this.env.VAT_ADMIN.getByName(VAT_ADMIN).serveHttp(mount.vat, req);
		if (!res) return new Response("mounted, but nothing serves it now\n", { status: 503 });
		const resp = new Response(res.body ?? "", { status: res.status ?? 200, headers: res.headers ?? {} });
		resp.headers.set("Referrer-Policy", "no-referrer");
		return resp;
	}
}

/**
 * Grants in a query parameter, as JSON labels; absent means none.
 *
 * @param {URL} url
 * @param {string} name
 */
const grantParam = (url, name) => {
	const v = url.searchParams.get(name);
	return v ? JSON.parse(v) : null;
};

/**
 * The admin socket: a WebSocket upgrade reaches the {@link VatAdmin} (OCapN
 * over Cap'n Web); otherwise, app vats over HTTP:
 *
 * - `POST /` with an ES module body creates a vat and starts it; the reply
 *   is `{"vat": <id>, "hash": <code hash>}`, and the id addresses it here.
 * - `PUT /<id>` with a body upgrades it; its storage carries over.
 * - `PUT /by-name/<name>` does the same for a vat from before ids, when
 *   vats were named.
 *
 * Every start states the vat's grants, absent meaning none:
 * `?egress=<labels>` (see {@link Egress}) and `?mount=<labels>` (ingress:
 * see {@link Public}).
 */
export class Admin extends WorkerEntrypoint {
	async fetch(request) {
		if (request.headers.get("Upgrade")?.toLowerCase() === "websocket") {
			return this.env.VAT_ADMIN.getByName(VAT_ADMIN).fetch(request);
		}
		const url = new URL(request.url);
		const [, first, second] = url.pathname.split("/");
		let id;
		if (request.method === "POST" && !first) id = this.env.VATS.newUniqueId();
		else if (request.method === "PUT" && first === "by-name" && second) id = this.env.VATS.idFromName(second);
		else if (request.method === "PUT" && first && !second) {
			try {
				id = this.env.VATS.idFromString(first);
			} catch {
				return new Response("no such vat\n", { status: 404 });
			}
		} else return new Response("POST / or PUT /<id> or PUT /by-name/<name>, with code\n", { status: 405 });
		const vat = id.toString();
		try {
			await this.env.DIRECTORY.getByName("directory").mount(vat, grantParam(url, "mount"));
		} catch (e) {
			return new Response(`${e.message}\n`, { status: 409 });
		}
		const egress = url.searchParams.get("egress");
		const hash = await this.env.VATS.get(id).start(await request.text(), egress);
		return Response.json({ vat, hash });
	}
}

export default Public;