// Host loop for wasm modules speaking the abi step protocol.
// Give it bytes directly (a local build), or a com.fedfork.exec.wasm.module record fetched by loadModule().
//
// The entry is a handler: ctx.wasm.run(entry, { ...input, now, state, event }) once per event returns
// { calls?, send?, cancel?, timers?, state?, result?, done? } or { error }.

import { fromBase32 } from "../encoding/index.js";

export { decodeCbor, decodeFrame, encodeCbor, encodeFrame, subscribeXrpc, withServiceAuth, wsUrl } from "./subscribe.js";

const NSID = /^[a-zA-Z][a-zA-Z0-9-]*(\.[a-zA-Z0-9-]+){2,}$/;
// A service the PDS proxies the call to (the atproto-proxy header): did#service_id.
const PROXY = /^did:[a-z]+:[^#\s]+#[A-Za-z0-9_-]+$/;
// A service called directly: an https origin, with a service-auth token (aud did:web:<host>, lxm nsid) the
// host mints; or a DID, a public read (GET, no token) of the PDS its #atproto_pds names.
const MIME = /^[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]*\/[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]*$/;
const SERVICE = /^(https:\/\/[A-Za-z0-9.-]+(:\d+)?|did:(plc|web):[A-Za-z0-9._:%-]+(#[A-Za-z0-9_-]+)?)$/;

const PSEUDO = "com.fedfork.exec.";
const PARAM = /^[A-Za-z0-9_.-]{1,256}$/;
const MODULE_KEY = /^[A-Za-z0-9_-]{1,200}$/;

// With a did#id service: where that DID serves it, [{ kind, address }] in its record's order; the host picks one
// it can reach (a fetcher's scheme, else https) and sends service auth aud did#id (hostCall).
const ENDPOINTS = "Module call endpoints must be 1 to 8 { kind, address } (kind a URL scheme name, address up to 2048 printable ASCII)";
const endpointsOk = (list) =>
	Array.isArray(list) && list.length >= 1 && list.length <= 8 &&
	list.every((e) => e && typeof e === "object" && Object.keys(e).length === 2 && typeof e.kind === "string" && /^[a-z][a-z0-9]{0,15}$/.test(e.kind) && typeof e.address === "string" && /^[\x21-\x7e]{1,2048}$/.test(e.address));

function parseCall(call, allowed) {
	if (typeof call?.nsid !== "string" || !NSID.test(call.nsid)) throw new Error("Module call needs an NSID");
	if (allowed && !allowed.includes(call.nsid)) throw new Error(`Module does not allow ${call.nsid}`);
	const params = call.params && typeof call.params === "object" ? Object.fromEntries(Object.entries(call.params).map(([key, value]) => [key, String(value)])) : undefined;
	if (call.proxy !== undefined && (typeof call.proxy !== "string" || !PROXY.test(call.proxy))) throw new Error("Module call proxy must be did#service");
	if (call.service !== undefined && (typeof call.service !== "string" || !SERVICE.test(call.service))) throw new Error("Module call service must be an https origin, a DID or a did#service");
	if (call.proxy !== undefined && call.service !== undefined) throw new Error("Module call takes proxy or service, not both");
	if (call.endpoints !== undefined) {
		if (!(call.service?.startsWith("did:") && call.service.includes("#"))) throw new Error("Module call endpoints need a did#service service");
		if (!endpointsOk(call.endpoints)) throw new Error(ENDPOINTS);
	}
	if (call.service?.startsWith("did:") && call.input !== undefined && call.endpoints === undefined) throw new Error("Module call to a DID service is a read: no input");
	const planned = { nsid: call.nsid, params, input: call.input, proxy: call.proxy, service: call.service, subscribe: call.subscribe === true };
	if (call.endpoints !== undefined) planned.endpoints = call.endpoints;
	// The content-type of a { $bytes } input, which goes as those bytes (uploadBlob).
	if (call.encoding !== undefined) {
		if (typeof call.encoding !== "string" || !MIME.test(call.encoding)) throw new Error("Module call encoding must be a MIME type");
		if (!(typeof call.input?.$bytes === "string" && Object.keys(call.input).length === 1)) throw new Error("Module call encoding needs a $bytes input");
		planned.encoding = call.encoding;
	}
	if (call.timeoutMs !== undefined) {
		if (!(Number.isFinite(call.timeoutMs) && call.timeoutMs > 0)) throw new Error("Module call timeoutMs must be a positive number");
		planned.timeoutMs = call.timeoutMs;
	}
	if (planned.subscribe) {
		// A host pseudo-call session (iroh.connect) takes input; a WebSocket subscription is a GET.
		if (planned.input !== undefined && !planned.nsid.startsWith(PSEUDO)) throw new Error("Module subscribe takes params, not input");
		if (planned.proxy !== undefined) throw new Error("Module subscribe cannot go through atproto-proxy; call the service directly");
		if (call.framing !== undefined && call.framing !== "cbor" && call.framing !== "json") throw new Error("Module subscribe framing must be cbor or json");
		if (call.keepalive !== undefined && !(Number.isFinite(call.keepalive?.ms) && call.keepalive.ms >= 1000)) throw new Error("Module subscribe keepalive needs ms >= 1000");
		planned.framing = call.framing;
		planned.keepalive = call.keepalive && { ms: call.keepalive.ms, data: call.keepalive.data };
		if (call.overflow !== undefined && call.overflow !== "drop" && call.overflow !== "close") throw new Error("Module subscribe overflow must be drop or close");
		planned.overflow = call.overflow;
		// Authenticated subscriptions: {} sends a service-auth token as Authorization (not in
		// browsers), { param } puts it in that query parameter, for services that take it there.
		if (call.serviceAuth !== undefined) {
			const { param, ...extra } = call.serviceAuth ?? {};
			if (typeof call.serviceAuth !== "object" || call.serviceAuth === null || Array.isArray(call.serviceAuth) || Object.keys(extra).length || (param !== undefined && !(typeof param === "string" && PARAM.test(param)))) throw new Error("Module subscribe serviceAuth must be {} or { param }");
			if (!planned.service?.startsWith("https://")) throw new Error("Module subscribe serviceAuth needs an https service");
			planned.serviceAuth = param === undefined ? {} : { param };
		}
	}
	return planned;
}

// OAuth repo scopes stop at the collection, so the host narrows writes to rkeys: in a collection not listed
// in module.json anyRecords, a module may update or delete only records this host saw it create. The
// ledger is ctx.storage.managed_records, keyed by at:// URI, each value { createdAt, module: owner }
// (moduleOwner); without it nothing is managed and such updates and deletes are refused. An entry
// counts only for the module that made it: one with no module (written before owners were recorded)
// counts for none, so list its collection in anyRecords or delete it by hand.
const CREATE = "com.atproto.repo.createRecord";
const PUT = "com.atproto.repo.putRecord";
const DELETE = "com.atproto.repo.deleteRecord";
const APPLY = "com.atproto.repo.applyWrites";

const hex = (bytes) => Array.from(new Uint8Array(bytes), (b) => b.toString(16).padStart(2, "0")).join("");

/**
 * Who owns the ledger entries a run writes: mod.owner (a composition's, shared by its instances), else
 * mod.id (an at:// module URI, or a local module's absolute path), else the module's sha256.
 */
export async function moduleOwner(mod) {
	const bytes = mod.module;
	// bytes (Uint8Array, Buffer, ArrayBuffer); anything else (a compiled WebAssembly.Module) hashes as no bytes
	const view = ArrayBuffer.isView(bytes) ? new Uint8Array(bytes.buffer, bytes.byteOffset, bytes.byteLength) : bytes instanceof ArrayBuffer ? new Uint8Array(bytes) : new Uint8Array();
	return mod.owner ?? mod.id ?? mod.sha256 ?? hex(await crypto.subtle.digest("SHA-256", view));
}

// The ledger as one module sees it: an entry counts only when this module made it.
const ledgerFor = (store, owner) =>
	store && {
		get: async (uri) => (await store.get(uri))?.module === owner,
		put: (uri, value) => store.put(uri, { ...value, module: owner }),
		delete: (uri) => store.delete(uri),
	};

const CONFIG_TYPES = ["string", "integer", "number", "boolean", "object", "array"];
/** calls and anyRecords are absent or lists of strings: a string's includes() would match substrings. hooks and implements
 * are absent or lists of { nsid, reason, ... }; a hook default is a module key. Returns mod. */
export function checkLists(mod) {
	for (const key of ["calls", "anyRecords", "http"]) {
		const list = mod?.[key];
		if (list !== undefined && !(Array.isArray(list) && list.every((item) => typeof item === "string"))) throw new Error(`module ${key} must be a list of strings`);
	}
	const hookOk = (h) =>
		NSID.test(h?.nsid ?? "") &&
		typeof h.reason === "string" &&
		(h.required === undefined || typeof h.required === "boolean") &&
		(h.default === undefined || MODULE_KEY.test(h.default)) &&
		(h.combine === undefined || h.combine === "all-allow") &&
		(h.entries === undefined || (Array.isArray(h.entries) && h.entries.every((e) => typeof e === "string")));
	if (mod?.hooks !== undefined && !(Array.isArray(mod.hooks) && mod.hooks.every(hookOk))) throw new Error("module hooks must be a list of { nsid, reason, required?: boolean, entries?: [entry], default?: module key, combine?: all-allow }");
	const fieldOk = (f) => typeof f?.key === "string" && (f.type === undefined || CONFIG_TYPES.includes(f.type)) && (f.required === undefined || typeof f.required === "boolean");
	if (mod?.config !== undefined && !(Array.isArray(mod.config) && mod.config.every(fieldOk))) throw new Error(`module config must be a list of { key, label, type?: ${CONFIG_TYPES.join(" | ")}, required?: boolean }`);
	if (mod?.implements !== undefined && !(Array.isArray(mod.implements) && mod.implements.every((i) => NSID.test(i?.nsid ?? "") && typeof i.entry === "string"))) {
		throw new Error("module implements must be a list of { nsid, entry }");
	}
	return mod;
}

/** A hook route's answer as a kcp Verdict (com.publicdomainrelay.temp.market.evaluatePolicy): anything but allow: true, an
 * error included, is a deny; a deny without violations of its own names `policyId`. */
export function verdictOf(result, policyId, error) {
	if (error === undefined && result?.allow === true) return { allow: true, violations: [] };
	const own = error === undefined && Array.isArray(result?.violations) ? result.violations : [];
	return { allow: false, violations: own.length ? own : [{ msg: error ?? "no verdict", policyId }] };
}

/** The transport a host pseudo-call NSID names, com.fedfork.exec.<name>.connect -> <name>, else undefined (rust/host transport::name). */
export const transportName = (nsid) => /^com\.fedfork\.exec\.([^.]+)\.connect$/.exec(nsid)?.[1];

/** A hook the host serves itself: no default, a host pseudo-call NSID (com.fedfork.exec.*: a transport, code mode's
 * mcp.message) and in the module's calls (Rust hosts_hook). */
export const hostsHook = (hook, calls) => !hook.default && hook.nsid.startsWith(PSEUDO) && calls?.includes(hook.nsid) === true;

const NO_ROUTE = "no edge routes it";
/**
 * Why a hook nothing routes cannot be served on this host, or undefined. A required hook
 * the host serves (hostsHook: a transport) is served when `provides` (the NSIDs this host serves, from its
 * transports; undefined: none) has it. `entries`: the entries of the module that run; a hook with `entries`
 * binds only those.
 */
export function unservedHook(hook, calls, provides, entries) {
	if (!hook.required || hook.default) return undefined;
	// a hook that names its entries binds only runs of those; `entries` are the ones that run (undefined: not known yet)
	if (hook.entries && !entries?.some((e) => hook.entries.includes(e))) return undefined;
	if (!hostsHook(hook, calls)) return NO_ROUTE;
	if (new Set(provides ?? []).has(hook.nsid)) return undefined;
	return `this host was built without ${transportName(hook.nsid) ?? hook.nsid}`;
}

/**
 * Hooks in a run with no composition: one with a default is served by that module's implements
 * entry, as a nested run (`run`, default runModule) on the default's own call (`callFor(target)`: what the host
 * gives that module for its calls, never `call`; a default that lists no calls is refused); one without answers
 * null (skip); a required one without a default is `{ error }`. Hooks never reach `call`. Returns the module (hook
 * NSIDs allowed, hooks gone) and the call to run it with.
 */
export function withHooks(ctx, mod, call, { entry, load, callFor, budget, provides, http, fetchers, run = runModule } = {}) {
	const hooks = checkLists(mod).hooks ?? [];
	for (const hook of hooks) {
		const why = unservedHook(hook, mod.calls, provides, entry === undefined ? undefined : [entry]);
		if (why) return { error: `${mod.name ?? "module"} needs hook ${hook.nsid} (${hook.reason}): ${why === NO_ROUTE ? "run it in a composition that routes it" : why}` };
	}
	if (!hooks.length) return { mod, call };
	const loaded = new Map();
	const hooked = async (nsid, request = {}) => {
		const hook = mod.hooks.find((h) => h.nsid === nsid);
		// A transport the module lists in calls (iroh.connect) is the host's own: an ordinary call.
		if (!hook || hostsHook(hook, mod.calls)) return call(nsid, request);
		if (request.proxy || request.service || request.subscribe) throw new Error(`${nsid} is a hook: no proxy, service or subscribe`);
		if (!hook.default) return null;
		try {
			if (!loaded.has(hook.default)) {
				const loading = Promise.resolve().then(async () => {
					const target = await (load ?? siblingOf(ctx, mod))(hook.default);
					// a default runs on its own calls allowlist; one that lists none would get them all
					if (target.calls === undefined) throw new Error(`hook default ${hook.default}: its module lists no calls (a hook default must)`);
					return { target, call: callFor ? await callFor(target) : noCall(hook.default) };
				});
				loaded.set(hook.default, loading);
			}
			const { target, call: own } = await loaded.get(hook.default);
			const entry = target.implements?.find((i) => i.nsid === nsid)?.entry;
			if (!entry) throw new Error(`hook default ${hook.default} does not implement ${nsid}`);
			const body = request.input !== null && typeof request.input === "object" && !Array.isArray(request.input) ? request.input : { input: request.input };
			const caller = { instance: mod.name ?? "module", nsid, reason: hook.reason };
			const ran = await run(ctx, target, entry, { ...request.params, ...body, caller }, own, { signal: request.signal, budget, load, callFor, provides, http, fetchers });
			if (ran.error !== undefined) throw new Error(ran.error);
			return hook.combine === "all-allow" ? verdictOf(ran.result, hook.default) : ran.result;
		} catch (error) {
			if (hook.combine === "all-allow") return verdictOf(undefined, hook.default, messageOf(error));
			throw error;
		}
	};
	return { mod: { ...mod, calls: mod.calls && [...mod.calls, ...hooks.map((h) => h.nsid)], hooks: undefined }, call: hooked };
}

/**
 * Held values (Rust: run::Held): JSON a run keeps in the host by name instead of in its state, so a growing
 * value (a transcript) does not pass through the wasm every event. com.fedfork.exec.held.put { name, value },
 * .append { name, items } (to a list) -> { length, bytes }, .get { name } -> { value }, .delete { name }. Answered at
 * plan time, in order, so a later call in the same output sees them; { "$held": name } in an http.fetch body
 * or the run's result is replaced by the value ({ "$held": [names] }: those lists, concatenated). Per run, at
 * most budget.heldBytes, gone when the run ends; a host may start a run with some (runModule's held).
 */
export const HELD_NSID = "com.fedfork.exec.held.";
const utf8 = (value) => new TextEncoder().encode(JSON.stringify(value)).byteLength;
function heldValues(limit) {
	const values = new Map(); // name -> { value, bytes }
	let total = 0;
	const keep = (nsid, name, value, bytes) => {
		const was = values.get(name)?.bytes ?? 0;
		if (total - was + bytes > limit) throw new Error(`${nsid}: held values past ${limit} bytes`);
		total += bytes - was;
		values.set(name, { value, bytes });
	};
	const op = (nsid, input) => {
		const name = input?.name;
		if (typeof name !== "string" || !ID.test(name)) throw new Error(`${nsid}: input needs a name`);
		const now = values.get(name);
		switch (nsid.slice(HELD_NSID.length)) {
			case "put":
				if (input.value === undefined) throw new Error(`${nsid}: input needs a value`);
				// copies in and out: the module's output (onEvent's) and earlier get answers never alias what later appends grow
				keep(nsid, name, structuredClone(input.value), utf8(input.value));
				return {};
			case "append": {
				if (!Array.isArray(input.items)) throw new Error(`${nsid}: items must be a list`);
				if (now && !Array.isArray(now.value)) throw new Error(`${nsid}: ${name} is not a list`);
				const list = now?.value ?? [];
				// JSON bytes of the longer list: "[" + items joined by "," + "]"
				const bytes = (now?.bytes ?? 2) + input.items.reduce((sum, item) => sum + utf8(item), 0) + Math.max(0, list.length + input.items.length - 1) - Math.max(0, list.length - 1);
				keep(nsid, name, list, bytes);
				list.push(...structuredClone(input.items));
				return { length: list.length, bytes };
			}
			case "get":
				return { value: structuredClone(now?.value ?? null) };
			case "delete":
				if (now) (total -= now.bytes), values.delete(name);
				return {};
			default:
				throw new Error(`${nsid} is not a held-value call (put, append, get, delete)`);
		}
	};
	const get = (name) => {
		if (!values.has(name)) throw new Error(`no held value ${name}`);
		return values.get(name).value;
	};
	// { "$held": name } anywhere in value, replaced by that held value; { "$held": [names] }: those held lists, concatenated.
	const splice = (value) => {
		if (Array.isArray(value)) return value.map(splice);
		if (typeof value !== "object" || value === null) return value;
		const keys = Object.keys(value);
		const names = value.$held;
		if (keys.length === 1 && typeof names === "string") return structuredClone(get(names));
		if (keys.length === 1 && Array.isArray(names) && names.every((n) => typeof n === "string")) {
			return names.flatMap((name) => {
				const list = get(name);
				if (!Array.isArray(list)) throw new Error(`held value ${name} is not a list`);
				return structuredClone(list);
			});
		}
		return Object.fromEntries(keys.map((key) => [key, splice(value[key])]));
	};
	return { op, splice };
}

/** The host pseudo-call for plain HTTP (not XRPC): input { method?, url, headers?, body? } -> { status, headers, body }. */
export const HTTP_NSID = "com.fedfork.exec.http.fetch";
const METHODS = ["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE"];
const TOKEN_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/;
// An http(s) URL's origin, undefined for anything else or a URL carrying credentials (Rust: run::http_origin).
const httpOrigin = (url) => {
	try {
		const u = new URL(url);
		return (u.protocol === "https:" || u.protocol === "http:") && !u.username && !u.password ? u.origin : undefined;
	} catch {}
};
// An http list or allowlist entry (Rust: run::entry_key): an http(s) origin, or "<scheme>://*" for a host
// fetcher's scheme (webrtc:, iroh:).
const entryKey = (entry) => {
	const pattern = /^([a-z][a-z0-9+.-]*):\/\/\*$/.exec(entry ?? "");
	return pattern ? (pattern[1] === "http" || pattern[1] === "https" ? undefined : entry) : httpOrigin(entry);
};
// The entry a request URL needs (Rust: run::allow_key): its origin for http(s), else "<scheme>://*". Undefined with credentials.
const allowKey = (url) => {
	try {
		const u = new URL(url);
		return u.username || u.password ? undefined : u.protocol === "https:" || u.protocol === "http:" ? u.origin : `${u.protocol}//*`;
	} catch {}
};

const GET_SERVICE_AUTH = "com.atproto.server.getServiceAuth";

/**
 * HTTP_NSID served by the engine (Rust: run::HttpCall), everything else to `call`. A request goes out only to
 * an origin both the module declares (module.json `http`: origins, or "*" for any the host allows) and the
 * host allows (`allow`: { origin: { headers?, serviceAuth?: { aud, lxm } } }); the host's headers (API keys)
 * are added here and replace the module's of the same name, so the wasm never sees them. serviceAuth: before
 * each request the host mints a token with the run's own session (`mint`'s getServiceAuth, `call` unless the
 * host gives its session apart, as a composition does; an OAuth session's scope must grant rpc:<lxm>?aud=<aud>) and sends it as Authorization: Bearer; one per request
 * (a token may carry a jti nonce a service checks). Redirects are not followed; the response body arrives
 * whole as text (an event stream too, once it ends).
 * Any other scheme (webrtc:, iroh:) goes to `fetchers[scheme]` ({ method, url, headers, body, signal } -> { status,
 * headers, body }: a host transport's), allowed as "<scheme>://*" in both lists; the host's headers and serviceAuth
 * never go there, and a module's "*" does not cover them (it names each). No fetcher: "no fetcher for <scheme>".
 */
export function withHttp(ctx, declared, allow, call, fetchers, mint = call) {
	const hosts = new Map(Object.entries(allow ?? {}).map(([origin, grant]) => [entryKey(origin), grant ?? {}]));
	const fail = (why) => {
		throw new Error(`${HTTP_NSID}: ${why}`);
	};
	const serviceToken = async (origin, { aud, lxm } = {}, signal) => {
		if (typeof aud !== "string" || !aud.startsWith("did:") || typeof lxm !== "string" || !lxm) fail(`this host's serviceAuth for ${origin} must be { aud: a DID, lxm: an NSID }`);
		let answer;
		try {
			answer = await mint(GET_SERVICE_AUTH, { params: { aud, lxm }, signal });
		} catch (error) {
			fail(`${origin} needs a service-auth token (aud ${aud}, lxm ${lxm}) from the run's session: ${messageOf(error)} (an OAuth session's scope must include rpc:${lxm}?aud=${aud})`);
		}
		if (typeof answer?.token !== "string" || !answer.token) fail(`${GET_SERVICE_AUTH} answered no service-auth token`);
		return answer.token;
	};
	return async (nsid, request = {}) => {
		if (nsid !== HTTP_NSID) return call(nsid, request);
		if (request.subscribe) fail("cannot be subscribed to (a response arrives whole)");
		const input = request.input;
		if (typeof input !== "object" || input === null || Array.isArray(input)) fail("input must be { method?, url, headers?, body? }");
		const { method = "GET", url, headers = {}, body } = input;
		if (!METHODS.includes(method)) fail(`method must be one of ${METHODS.join(", ")}`);
		const origin = allowKey(url);
		if (!origin) fail("url must be a URL without credentials");
		if (typeof headers !== "object" || headers === null || Array.isArray(headers) || !Object.entries(headers).every(([k, v]) => TOKEN_NAME.test(k) && typeof v === "string" && !/[\r\n\0]/.test(v))) fail("headers must map header names to strings");
		if (body !== undefined && typeof body !== "string") fail("body must be a string");
		if (body !== undefined && (method === "GET" || method === "HEAD")) fail(`${method} takes no body`);
		if (!(declared ?? []).some((d) => (d === "*" && httpOrigin(url)) || entryKey(d) === origin)) fail(`${origin} is not in the module's http list`);
		if (!hosts.has(origin)) fail(`this host does not allow ${origin}`);
		if (!httpOrigin(url)) {
			const scheme = new URL(url).protocol;
			if (typeof fetchers?.[scheme] !== "function") fail(`no fetcher for ${scheme}`);
			const response = await fetchers[scheme]({ method, url, headers, ...(body !== undefined && { body }), signal: request.signal });
			return { status: response.status, headers: Object.fromEntries(new Headers(response.headers ?? {})), body: String(response.body ?? "") };
		}
		const grant = hosts.get(origin);
		const injected = { ...grant.headers };
		if (grant.serviceAuth !== undefined) injected.authorization = `Bearer ${await serviceToken(origin, grant.serviceAuth, request.signal)}`;
		const replaced = new Set(Object.keys(injected).map((k) => k.toLowerCase()));
		const sent = { ...Object.fromEntries(Object.entries(headers).filter(([k]) => !replaced.has(k.toLowerCase()))), ...injected };
		const response = await ctx.http.fetch(url, { method, headers: sent, ...(body !== undefined && { body }), signal: request.signal, redirect: "manual" });
		return { status: response.status, headers: Object.fromEntries(response.headers), body: await response.text() };
	};
}

// A hook default next to the module: the same publisher's record, or the same staged directory.
const siblingOf = (ctx, mod) => (key) => {
	const at = /^(at:\/\/[^/]+\/[^/]+\/)[^/]+$/.exec(mod.id ?? "");
	if (at) return loadModule(ctx, at[1] + key);
	if (/\/modules\/[^/]+\.wasm$/.test(mod.id ?? "")) return localModule(new URL(`${key}.wasm`, mod.id), new URL(`${key}.json`, mod.id));
	throw new Error(`hook default ${key}: no loader for a module loaded from ${mod.id ?? "bytes"} (pass the host's load option)`);
};

// A hook default's call when the host gave no callFor: it may call nothing.
const noCall = (key) => async (nsid) => {
	throw new Error(`hook default ${key}: no call for ${nsid} on this host (pass runModule's callFor option)`);
};

// The record writes one call makes: { action: create | update | delete, collection, uri when the rkey is known }.
function writesOf({ nsid, input }) {
	input = typeof input === "object" && input !== null ? input : {};
	const at = (collection, rkey) => (rkey === undefined ? undefined : `at://${input.repo}/${collection}/${rkey}`);
	if (nsid === CREATE) return [{ action: "create", collection: input.collection }];
	if (nsid === PUT) return [{ action: "update", collection: input.collection, uri: at(input.collection, input.rkey) }];
	if (nsid === DELETE) return [{ action: "delete", collection: input.collection, uri: at(input.collection, input.rkey) }];
	if (nsid === APPLY) return (Array.isArray(input.writes) ? input.writes : []).map((w) => ({ action: String(w?.$type).split("#")[1], collection: w?.collection, uri: at(w?.collection, w?.rkey) }));
	return [];
}

async function refusal(ledger, anyRecords, call) {
	for (const { action, collection, uri } of writesOf(call)) {
		if (action === "create" || anyRecords?.includes(collection)) continue;
		if (!uri || !ledger || !(await ledger.get(uri))) return `${call.nsid} refused: ${uri ?? collection} was not created by this module (list ${collection} in module.json anyRecords to allow)`;
	}
}

async function remember(ledger, call, output) {
	if (!ledger) return;
	const writes = writesOf(call);
	for (const [i, { action, uri }] of writes.entries()) {
		const created = call.nsid === APPLY ? output?.results?.[i]?.uri : output?.uri;
		if (action === "create" && typeof created === "string") await ledger.put(created, { createdAt: new Date().toISOString() });
		if (action === "delete" && uri) await ledger.delete(uri);
	}
}

// The host owns concurrency, subscriptions, timers and budgets.
// log: call records kept in the result, the last `log` (a long run, a server, stays bounded).
// callsPerMinute: calls per one-minute window (fixed, from the run's start), the guard for a run with calls unlimited.
// ioBytes: the wasm's input and output per event (JSON bytes), at most IO_BYTES_MAX.
// heldBytes: the run's held values together (HELD_NSID), JSON bytes.
export const BUDGET = Object.freeze({ inFlight: 6, subscriptions: 4, calls: 100, callsPerMinute: Infinity, events: 1000, wallMs: 60_000, queue: 256, log: 1000, ioBytes: 1 << 20, heldBytes: 16 << 20 });
/**
 * The ceiling on ioBytes, whatever a host sets: an event's input sits in the module's 64 MiB memory beside
 * its parsed copy, what it decodes and its output, so 16 MiB in and out still leaves room to work.
 */
export const IO_BYTES_MAX = 16 << 20;
/** The keys a host may set to Infinity (JSON surfaces: the string "unlimited"); a manifest never can. */
export const UNLIMITED = Object.freeze(["calls", "callsPerMinute", "events", "wallMs"]);
/**
 * A budget value: a positive finite number, or Infinity for an UNLIMITED `key` (anything else leaves the default).
 * Without a key (a manifest's values), finite only.
 */
export const budgetValue = (n, key) => typeof n === "number" && n > 0 && (Number.isFinite(n) || UNLIMITED.includes(key));
const ID = /^[A-Za-z0-9_.:-]{1,256}$/;
const DRAIN_MS = 1000;
// promise, or undefined after ms, whichever first; the timer never holds the process open.
const within = (promise, ms) => {
	let timer;
	return Promise.race([promise, new Promise((r) => (timer = setTimeout(r, ms)))]).finally(() => clearTimeout(timer));
};
const messageOf = (error) => (error instanceof Error ? error.message : String(error));
// setTimeout's limit (2^31 - 1 ms): past it Node fires at once. Every timer clamps to it (Rust: ms_duration).
const MAX_TIMER_MS = 2 ** 31 - 1;
const timerMs = (ms) => Math.min(Math.max(ms, 0), MAX_TIMER_MS) || 0;
const list = (value) => (Array.isArray(value) ? value : []);

const closeStream = (slot) => {
	try {
		if (typeof slot.stream?.close === "function") slot.stream.close();
		else Promise.resolve(slot.iterator?.return?.()).catch(() => {});
	} catch {}
};

/**
 * Run a module until it returns done, nothing is pending, or a budget runs out.
 * Never throws for module or call errors: they come back as `error` (the run) or as events (a call).
 * @param {import("emdash/plugin").PluginContext} ctx  ctx.wasm runs the module; ctx.storage.managed_records (when
 *   declared) records the at:// URIs the module created, so it may update and delete those
 * @param {{ module: Uint8Array, sha256?: string, calls?: string[], anyRecords?: string[], id?: string, owner?: string }} module  bytes, their expected
 *   sha256 (checked by ctx.wasm.run), the NSIDs module.json allows, and the collections where it may update and delete
 *   records it did not create (both absent or lists of strings, else this throws); id and owner: see moduleOwner;
 *   loadModule() returns this shape
 * @param {string} entry  the handler export
 * @param {Record<string, unknown>} input  passed with every event; `now`, `state` and `event` are added
 * @param {(nsid: string, request: { params?: Record<string, string>, input?: unknown, proxy?: string, service?: string, signal: AbortSignal,
 *   subscribe?: true, framing?: "cbor" | "json", keepalive?: { ms: number, data: unknown }, serviceAuth?: { param?: string },
 *   emit: (event: object) => void }) => Promise<unknown>} call
 *   performs one XRPC call with the user's session and resolves to its output; a proxy is sent as the atproto-proxy header;
 *   an https service is called directly at service/xrpc/nsid with getServiceAuth({ aud: did:web:<host>, lxm: nsid }) as Bearer;
 *   a DID service is a plain GET of params at its PDS (#atproto_pds), with no token. `signal`: timeout, cancel, run end.
 *   `subscribe`: then it resolves to an async iterable of
 *   messages, optionally with send(data) and close() (subscribeXrpc returns one); `emit` queues a `callback` event
 *   (`id` = this call's) into the run while the call is pending
 * @param {{ signal?: AbortSignal, inputs?: AsyncIterable<unknown>, budget?: Partial<typeof BUDGET>, onEvent?: (event: object, output: object) => void,
 *   load?: (key: string) => Promise<object>, callFor?: (mod: object) => Function | Promise<Function>, provides?: Iterable<string>,
 *   http?: Record<string, { headers?: Record<string, string>, serviceAuth?: { aud: string, lxm: string } }>, held?: Record<string, unknown>,
 *   fetchers?: Record<string, (request: { method: string, url: string, headers: Record<string, string>, body?: string, signal: AbortSignal }) => Promise<{ status: number, headers?: Record<string, string>, body?: string }>> }} [options]
 *   load: resolves a hook's default module key (default: the sibling record of an at:// id, or of a staged modules/<key>.wasm);
 *   callFor: the call a hook default runs on, built for that module as the host builds `call` for this one (its own
 *   calls, attest and keys); without it a default's calls fail;
 *   provides: the host pseudo-call NSIDs (transports) this host serves, checked against required hooks (absent: none);
 *   http: the origins this host lets HTTP_NSID reach, { "https://api.example": { headers?: { name: value }, serviceAuth?: { aud, lxm } } }
 *     (withHttp; absent: none), each with headers the host adds (an API key the wasm never sees) or a service-auth token it
 *     mints per request with the run's session; a module reaches those it declares too; "<scheme>://*" (webrtc://*) allows a fetcher's URLs
 *   session: the call that mints http's serviceAuth tokens (absent: `call`); a composition passes the instance's
 *     session (callFor), not its call, which holds the module to its own calls
 *   fetchers: HTTP_NSID for other URL schemes, { "webrtc:": fn } (a transport's fetchers(), spread in)
 *   held: the held values the run starts with, { name: value } (as held.put each; past heldBytes the run fails);
 *   inputs: each value becomes an `input` event, and the run stays open while inputs does;
 *   onEvent: called after each handler call, for hosts that show progress
 * @returns {Promise<{ entry: string, result: unknown, error?: string, events: number, calls: object[] }>}
 */
export async function runModule(ctx, mod, entry, input, call, { signal, inputs, budget: limits, onEvent, load, callFor, provides, http, fetchers, held: initial, session } = {}) {
	const hooked = withHooks(ctx, mod, call, { entry, load, callFor, budget: limits, provides, http, fetchers });
	if (hooked.error) return { entry, result: null, error: hooked.error, events: 0, calls: [] };
	({ mod, call } = hooked);
	call = withHttp(ctx, mod.http, http, call, fetchers, session ?? call);
	const { module, sha256, calls: allowed, anyRecords } = mod;
	const budget = { ...BUDGET };
	for (const key of Object.keys(BUDGET)) if (budgetValue(limits?.[key], key)) budget[key] = limits[key];
	budget.ioBytes = Math.min(budget.ioBytes, IO_BYTES_MAX);
	const held = heldValues(budget.heldBytes);
	try {
		for (const [name, value] of Object.entries(initial ?? {})) held.op(`${HELD_NSID}put`, { name, value });
	} catch (error) {
		return { entry, result: null, error: `held: ${messageOf(error)}`, events: 0, calls: [] };
	}
	const ledger = ctx.storage?.managed_records && ledgerFor(ctx.storage.managed_records, await moduleOwner(mod));
	const ended = new AbortController();
	const pending = new Map(); // id -> slot { kind: queued | call | sub | timer, ... }
	const closing = []; // cancelled subscriptions flushing their sends, then closing
	const waiting = []; // [id, slot] past budget.inFlight, FIFO
	const queue = [];
	const dropped = new Map();
	const log = [];
	let logDropped = 0;
	// Only the last budget.log call records are kept; the result counts the rest (callsDropped).
	const record = (entry) => {
		log.push(entry);
		if (log.length > budget.log) {
			log.shift();
			logDropped++;
		}
	};
	let state = null;
	let result = null;
	let error;
	let wake;
	let total = 0;
	let minute = { start: Date.now(), calls: 0 }; // the callsPerMinute window, from the run's start
	let events = 0;
	let inputsOpen = false;
	const notify = () => wake?.();
	const count = (kind) => [...pending.values()].filter((slot) => slot.kind === kind).length;

	// Past budget.queue the oldest message is dropped (results, closes and timers never are) and the
	// next message on its subscription carries the count. A subscription with overflow "close" never
	// loses a message: a message arriving for it on a full queue closes it (closed, error "overflow").
	const closes = (id) => pending.get(id)?.planned?.overflow === "close";
	const push = (event) => {
		if (ended.signal.aborted) return;
		if (queue.length >= budget.queue && event.type === "message") {
			if (closes(event.id)) return overflow(event.id);
			const oldest = queue.findIndex((e) => e.type === "message" && !closes(e.id));
			const victim = oldest >= 0 ? queue.splice(oldest, 1)[0] : event;
			dropped.set(victim.id, (dropped.get(victim.id) ?? 0) + 1);
			if (victim === event) return;
		}
		queue.push(event);
		notify();
	};
	const overflow = (id) => {
		const slot = pending.get(id);
		slot.abort.abort(new Error("overflow"));
		closeStream(slot);
		settle(id, slot, { type: "closed", id, error: "overflow" });
	};
	const refuse = (id, nsid, error) => {
		record({ id, nsid, ok: false, error });
		push({ type: "result", id, nsid, ok: false, error });
	};
	// A call or subscription ended on its own (not cancelled).
	const settle = (id, slot, event) => {
		if (pending.get(id) !== slot) return;
		pending.delete(id);
		const { nsid, subscribe } = slot.planned;
		record(subscribe ? { id, nsid, subscribe, messages: slot.messages, ok: !event.error, ...(event.error && { error: event.error }) } : { id, nsid, ok: event.ok, ...(!event.ok && { error: event.error }) });
		push(event);
		pump();
	};

	const start = (id, slot) => {
		const { planned } = slot;
		slot.kind = planned.subscribe ? "sub" : "call";
		const callSignal = AbortSignal.any([slot.abort.signal, ended.signal]);
		const aborted = new Promise((_, reject) => callSignal.addEventListener("abort", () => reject(callSignal.reason), { once: true }));
		aborted.catch(() => {});
		// a call cannot outlive the run: at most wallMs
		const timer = planned.timeoutMs && setTimeout(() => slot.abort.abort(new Error(`${planned.nsid} timed out after ${planned.timeoutMs} ms`)), timerMs(Math.min(planned.timeoutMs, budget.wallMs)));
		const request = {
			params: planned.params,
			input: planned.input,
			...(planned.encoding && { encoding: planned.encoding }),
			proxy: planned.proxy,
			...(planned.service && { service: planned.service }),
			...(planned.endpoints && { endpoints: planned.endpoints }),
			signal: callSignal,
			...(planned.subscribe && { subscribe: true, framing: planned.framing, keepalive: planned.keepalive, ...(planned.serviceAuth && { serviceAuth: planned.serviceAuth }) }),
			// A callback into this run while the call is pending (javascript/src/compose).
			emit: (event) => void (pending.get(id) === slot && push({ ...event, type: "callback", id })),
		};
		const made = Promise.resolve().then(() => call(planned.nsid, request));
		if (!planned.subscribe) {
			// The write is remembered even when the module stopped waiting for it.
			const remembered = made.then(async (output) => (await remember(ledger, planned, output), output));
			remembered.catch(() => {});
			Promise.race([remembered, aborted]).then(
				(output) => ({ type: "result", id, nsid: planned.nsid, ok: true, output }),
				(error) => ({ type: "result", id, nsid: planned.nsid, ok: false, error: messageOf(error) }),
			).then((event) => (clearTimeout(timer), settle(id, slot, event)));
			return;
		}
		slot.messages = 0;
		slot.opened = made.then((stream) => {
			slot.stream = stream;
			if (callSignal.aborted) closeStream(slot);
			return stream;
		});
		slot.opened.catch(() => {});
		(async () => {
			let failure;
			try {
				slot.iterator = (await Promise.race([slot.opened, aborted]))[Symbol.asyncIterator]();
				for (;;) {
					const next = await Promise.race([slot.iterator.next(), aborted]);
					if (next.done || pending.get(id) !== slot) break;
					slot.messages++;
					push({ type: "message", id, data: next.value });
				}
			} catch (error) {
				failure = messageOf(error);
			}
			clearTimeout(timer);
			if (pending.get(id) !== slot) return; // cancelled: cancel() flushes sends, then closes
			closeStream(slot);
			settle(id, slot, { type: "closed", id, ...(failure && { error: failure }) });
		})();
	};
	const pump = () => {
		while (waiting.length && count("call") < budget.inFlight) start(...waiting.shift());
	};

	const plan = async (raw) => {
		const id = raw?.id;
		const nsid = raw?.nsid;
		if (typeof id !== "string" || !ID.test(id)) return refuse(id, nsid, "Module call needs an id");
		if (pending.has(id)) return refuse(id, nsid, `Module call id ${id} is already pending`);
		let planned;
		try {
			planned = parseCall(raw, allowed);
		} catch (error) {
			return refuse(id, nsid, messageOf(error));
		}
		if (++total > budget.calls) return refuse(id, nsid, `Module exceeded ${budget.calls} calls`);
		if (Date.now() - minute.start >= 60_000) minute = { start: Date.now(), calls: 0 };
		if (++minute.calls > budget.callsPerMinute) return refuse(id, nsid, `Module exceeded ${budget.callsPerMinute} calls per minute`);
		if (planned.subscribe && count("sub") >= budget.subscriptions) return refuse(id, nsid, `Module exceeded ${budget.subscriptions} subscriptions`);
		if (planned.nsid.startsWith(HELD_NSID)) {
			let output;
			try {
				if (planned.subscribe) throw new Error(`${planned.nsid} cannot be subscribed to`);
				output = held.op(planned.nsid, planned.input);
			} catch (failure) {
				return refuse(id, nsid, messageOf(failure));
			}
			record({ id, nsid, ok: true });
			return push({ type: "result", id, nsid, ok: true, output });
		}
		// An http.fetch body that is JSON (not text) goes as its JSON, held values spliced in.
		if (planned.nsid === HTTP_NSID && planned.input?.body !== undefined && typeof planned.input.body !== "string") {
			try {
				planned.input = { ...planned.input, body: JSON.stringify(held.splice(planned.input.body)) };
			} catch (failure) {
				return refuse(id, nsid, `${HTTP_NSID}: ${messageOf(failure)}`);
			}
		}
		// A ledger read that fails refuses the call with its error (runModule never throws for a call).
		const refused = await refusal(ledger, anyRecords, planned).catch(messageOf);
		if (refused) return refuse(id, nsid, refused);
		const slot = { kind: "queued", planned, abort: new AbortController(), sending: Promise.resolve() };
		pending.set(id, slot);
		if (planned.subscribe) start(id, slot);
		else waiting.push([id, slot]);
	};

	// Drops id's events already queued: what a cancel or a re-armed timer replaces must not arrive later.
	const purge = (id, types) => {
		for (let i = queue.length; i--; ) if (queue[i].id === id && types.includes(queue[i].type)) queue.splice(i, 1);
	};
	// A cancelled call yields nothing, even when its result is already queued; a cancelled subscription
	// yields closed now and no message after that.
	const cancel = (id) => {
		purge(id, ["result", "message", "timer", "callback"]);
		dropped.delete(id);
		const slot = pending.get(id);
		if (!slot) return;
		pending.delete(id);
		if (slot.kind === "timer") return clearTimeout(slot.timer);
		if (slot.kind === "queued") return void waiting.splice(waiting.findIndex(([, s]) => s === slot), 1);
		record({ id, nsid: slot.planned.nsid, cancelled: true, ...(slot.kind === "sub" && { subscribe: true, messages: slot.messages }) });
		if (slot.kind !== "sub") return slot.abort.abort(new Error("cancelled"));
		push({ type: "closed", id, error: "cancelled" });
		// Frames already sent go out first (bounded by DRAIN_MS), then the stream closes; the run's end waits for it.
		closing.push(within(slot.sending, DRAIN_MS).then(() => {
			slot.abort.abort(new Error("cancelled"));
			closeStream(slot);
		}));
	};

	// Frames go out in order per subscription; a send that fails closes the subscription.
	const send = (id, data) => {
		const slot = pending.get(id);
		if (slot?.kind !== "sub") return refuse(id, undefined, `send: ${id} is not an open subscription`);
		slot.sending = slot.sending
			.then(() => slot.opened)
			.then((stream) => {
				if (typeof stream?.send !== "function") return refuse(id, slot.planned.nsid, `send: ${slot.planned.nsid} cannot send`);
				return stream.send(data);
			})
			.catch((error) => slot.abort.abort(new Error(`send failed: ${messageOf(error)}`)));
	};

	const setTimer = ({ id, ms } = {}) => {
		if (typeof id !== "string" || !ID.test(id) || !(Number.isFinite(ms) && ms >= 0)) return "Module timer needs an id and ms";
		const old = pending.get(id);
		if (old && old.kind !== "timer") return `Module timer id ${id} is already pending`;
		clearTimeout(old?.timer);
		purge(id, ["timer"]); // re-armed after it fired: the earlier firing is not delivered
		const slot = { kind: "timer" };
		slot.timer = setTimeout(() => {
			if (pending.get(id) !== slot) return;
			pending.delete(id);
			push({ type: "timer", id });
		}, timerMs(Math.min(ms, budget.wallMs)));
		pending.set(id, slot);
	};

	let iterator;
	if (inputs) {
		inputsOpen = true;
		iterator = inputs[Symbol.asyncIterator]();
		(async () => {
			try {
				for (let next = await iterator.next(); !next.done && !ended.signal.aborted; next = await iterator.next()) push({ type: "input", data: next.value });
			} catch (failure) {
				error ??= `inputs failed: ${messageOf(failure)}`;
			} finally {
				inputsOpen = false;
				notify();
			}
		})();
	}
	// wallMs unlimited: no deadline (a server runs until its signal aborts).
	const deadline = Number.isFinite(budget.wallMs) && setTimeout(() => {
		error ??= `Module exceeded ${budget.wallMs} ms`;
		notify();
	}, timerMs(budget.wallMs));
	const onAbort = () => {
		error ??= "aborted";
		notify();
	};
	signal?.addEventListener("abort", onAbort, { once: true });
	if (signal?.aborted) onAbort();
	push({ type: "start" });
	try {
		while (error === undefined) {
			if (!queue.length) {
				if (!pending.size && !inputsOpen) break;
				await new Promise((resolve) => (wake = resolve));
				wake = undefined;
				continue;
			}
			const event = queue.shift();
			if (event.type === "message" && dropped.has(event.id)) event.dropped = dropped.get(event.id), dropped.delete(event.id);
			if (++events > budget.events) {
				error = `Module exceeded ${budget.events} events`;
				break;
			}
			let out;
			try {
				out = await ctx.wasm.run({ module, sha256, entry, input: { ...input, now: new Date().toISOString(), state, event }, ioBytes: budget.ioBytes });
			} catch (failure) {
				error = messageOf(failure);
				break;
			}
			if (typeof out?.error === "string") {
				error = out.error;
				break;
			}
			if (typeof out !== "object" || out === null || Array.isArray(out)) {
				error = "Module output must be an object";
				break;
			}
			if ("state" in out) state = out.state ?? null;
			if ("result" in out) result = out.result ?? null;
			onEvent?.(event, out);
			// Sends to an open subscription go before this output's cancels (a last frame, then close);
			// sends to a subscription this output opens go after its calls.
			const [early, late] = list(out.send).reduce((parts, frame) => (parts[pending.get(frame?.id)?.kind === "sub" ? 0 : 1].push(frame), parts), [[], []]);
			for (const frame of early) send(frame?.id, frame?.data);
			list(out.cancel).forEach(cancel);
			for (const timer of list(out.timers)) error ??= setTimer(timer);
			if (error !== undefined) break;
			for (const raw of list(out.calls)) await plan(raw);
			for (const frame of late) send(frame?.id, frame?.data);
			pump();
			if (out.done === true) break;
		}
	} finally {
		clearTimeout(deadline);
		signal?.removeEventListener("abort", onAbort);
		// Frames the last handler sent (a final response) go out before the streams close, unless aborted.
		const sending = [...closing, ...[...pending.values()].filter((slot) => slot.kind === "sub").map((slot) => slot.sending)];
		if (sending.length && !signal?.aborted) await within(Promise.allSettled(sending), DRAIN_MS);
		ended.abort(new Error(error ?? "run ended"));
		for (const slot of pending.values()) clearTimeout(slot.timer), closeStream(slot);
		pending.clear();
		if (inputsOpen) Promise.resolve(iterator.return?.()).catch(() => {});
	}
	// The result may name held values ({ "$held": name }): they go in now, before the run's are dropped.
	try {
		result = held.splice(result);
	} catch (failure) {
		error ??= `result: ${messageOf(failure)}`;
	}
	return { entry, result, ...(error !== undefined && { error }), events, calls: log, ...(logDropped && { callsDropped: logDropped }) };
}

const MODULE_COLLECTION = "com.fedfork.exec.wasm.module";
// A blob's CID (raw codec, sha2-256, base32 CIDv1) carries the sha256 its bytes must hash to.
function blobSha256(cid) {
	let bytes;
	try {
		if (typeof cid !== "string" || !cid.startsWith("b")) throw new Error();
		bytes = fromBase32(cid.slice(1));
	} catch {
		throw new Error("Blob CID must be base32 CIDv1");
	}
	if (bytes.length !== 36 || bytes[0] !== 0x01 || bytes[1] !== 0x55 || bytes[2] !== 0x12 || bytes[3] !== 0x20) {
		throw new Error("Blob CID must be a raw sha2-256 CIDv1");
	}
	return Array.from(bytes.subarray(4), (byte) => byte.toString(16).padStart(2, "0")).join("");
}

// Loopback, private, link-local, shared, multicast and unspecified hosts: localhost, IPv4 and IPv6 literals.
// ponytail: literal addresses only; a public name that resolves to one is not caught (needs a resolver hook).
function privateHost(hostname) {
	const host = hostname.toLowerCase().replace(/\.$/, "");
	if (host === "localhost" || host.endsWith(".localhost") || host === "") return true;
	if (host.startsWith("[")) {
		const first = parseInt(host.slice(1, -1).split(":")[0] || "0", 16);
		// ::/16 (::, ::1, v4-mapped and -compatible), fc00::/7 unique local, fe80::/10 link-local, ff00::/8 multicast
		return first === 0 || (first & 0xfe00) === 0xfc00 || (first & 0xffc0) === 0xfe80 || (first & 0xff00) === 0xff00;
	}
	const v4 = /^(\d+)\.(\d+)\.(\d+)\.(\d+)$/.exec(host)?.slice(1).map(Number);
	if (!v4) return false;
	const [a, b] = v4;
	return a === 0 || a === 10 || a === 127 || a >= 224 || (a === 169 && b === 254) || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168) || (a === 100 && b >= 64 && b <= 127);
}

// A DID document names its own endpoints, and DID reads run for anonymous visitors (ingress), so a
// lookup must not reach the host's own network: https, and not a privateHost. allowPrivate: local dev, tests.
function publicUrl(url, allowPrivate) {
	let parsed;
	try {
		parsed = new URL(url);
	} catch {
		throw new Error(`${url} is not a URL`);
	}
	if (allowPrivate) return url;
	if (parsed.protocol !== "https:") throw new Error(`${url} is not https`);
	if (privateHost(parsed.hostname)) throw new Error(`${url} is a private address`);
	return url;
}

/**
 * The PDS a did:plc or did:web account lives on, or for did#id that service's endpoint (did:web:x#svc):
 * an https URL off the host's network unless allowPrivate (default ctx.allowPrivate; the CLIs set it
 * from ATLIBS_ALLOW_PRIVATE=1).
 */
export async function pdsOf(ctx, ref, { allowPrivate = ctx?.allowPrivate === true } = {}) {
	const [did, id = "atproto_pds"] = ref.split("#");
	const service = (await didDoc(ctx, did, allowPrivate)).service?.find((entry) => entry.id === `#${id}` || entry.id === `${did}#${id}`);
	if (typeof service?.serviceEndpoint !== "string") throw new Error(id === "atproto_pds" ? `${did} has no PDS` : `${did} has no #${id} service`);
	return publicUrl(service.serviceEndpoint, allowPrivate);
}

async function didDoc(ctx, did, allowPrivate = ctx?.allowPrivate === true) {
	const url = did.startsWith("did:plc:")
		? `https://plc.directory/${did}`
		: did.startsWith("did:web:")
			? `https://${decodeURIComponent(did.slice("did:web:".length))}/.well-known/did.json`
			: null;
	if (!url) throw new Error(`${did} is not a did:plc or did:web`);
	const response = await ctx.http.fetch(publicUrl(url, allowPrivate));
	if (!response.ok) throw new Error(`Could not resolve ${did}`);
	return response.json();
}

/**
 * handle -> DID (Rust: Net::resolve_handle), checked both ways: the DID document must claim the handle, so a
 * lying resolver cannot substitute another account. A DID is returned as is.
 */
export async function resolveHandle(ctx, handle) {
	if (handle.startsWith("did:")) return handle;
	const url = `https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle?${new URLSearchParams({ handle })}`;
	const response = await ctx.http.fetch(url);
	const did = response.ok ? (await response.json())?.did : undefined;
	if (typeof did !== "string" || !did.startsWith("did:")) throw new Error(`${handle} did not resolve`);
	if (!(await didDoc(ctx, did)).alsoKnownAs?.includes(`at://${handle}`)) throw new Error(`${handle} resolves to ${did}, whose DID document does not claim it`);
	return did;
}

/**
 * Fetch a published module: its com.fedfork.exec.wasm.module record and blob, from the publisher's PDS.
 * @param {import("emdash/plugin").PluginContext} ctx  only ctx.http is used
 * @param {string} ref  <handle>/<rkey>, <did>/<rkey> or at://<did>/com.fedfork.exec.wasm.module/<rkey>; a
 *   <handle|did>/<rkey> may end in @<record CID> or @<32+ hex of its pin> (modulePin), checked here.
 *   The module's id (its ledger owner) is always the at:// URI.
 * @param {string} [pin]  a composition instance's pin, checked too. A 32+ hex pin loads the record publish keeps at
 *   <rkey>.<its first 32 hex> (Rust registry::version_rkey), so a republish leaves it in place.
 */
export async function loadModule(ctx, ref, pin) {
	const short = /^([^/@\s]+)\/([^/@\s]+)(?:@([A-Za-z0-9]+))?$/.exec(ref);
	const pins = [...new Set([pin, short?.[3]].filter(Boolean))];
	const uri = short ? `at://${await resolveHandle(ctx, short[1])}/${MODULE_COLLECTION}/${short[2]}` : ref;
	const [, did, collection, rkey] = /^at:\/\/(did:[^/]+)\/([^/]+)\/([^/]+)$/.exec(uri) ?? [];
	if (collection !== MODULE_COLLECTION) throw new Error(`${ref} is not a ${MODULE_COLLECTION} record (<handle>/<name>, <did>/<name> or at://<did>/${MODULE_COLLECTION}/<name>)`);
	const pds = await pdsOf(ctx, did);
	const xrpc = (nsid, params) => `${new URL(`/xrpc/${nsid}`, pds).href}?${new URLSearchParams(params)}`;
	const hex = pins.find((p) => /^[0-9a-fA-F]{32,64}$/.test(p));
	const version = hex ? `${rkey}.${hex.slice(0, 32).toLowerCase()}` : rkey;
	const record = await ctx.http.fetch(xrpc("com.atproto.repo.getRecord", { repo: did, collection, rkey: version }));
	if (!record.ok) throw new Error(hex ? `${uri}: pin ${hex} not published (at://${did}/${collection}/${version} not found)` : `${ref} not found`);
	const found = await record.json();
	const value = found?.value; // any JSON: not an object is an unsupported ABI
	if (value?.abi !== "emdash-wasm-json-v1") throw new Error(`${ref}: unsupported module ABI ${value?.abi}`);
	checkLists(value);
	const cid = value.module?.ref?.$link;
	const sha256 = blobSha256(cid);
	const blob = await ctx.http.fetch(xrpc("com.atproto.sync.getBlob", { did, cid }));
	if (!blob.ok) throw new Error(`${ref} blob not found`);
	const mod = { module: new Uint8Array(await blob.arrayBuffer()), sha256, id: uri, ...specOf(value) };
	for (const pinned of pins.filter((p) => p !== found.cid)) {
		// Rust: registry::check_pin
		if (!/^[0-9a-fA-F]{32,64}$/.test(pinned)) throw new Error(`${ref}: give the record CID, or at least 32 hex characters of the pin`);
		const is = await modulePin(mod);
		if (!is.startsWith(pinned.toLowerCase())) throw new Error(`${ref}: pinned to ${pinned}, but it is pin ${is} (record ${found.cid ?? "-"})`);
	}
	return mod;
}

// The record fields that grant or route authority; a pin covers them with the wasm (Rust: run::GRANT_FIELDS).
const GRANT_FIELDS = ["calls", "anyRecords", "hooks", "implements", "anyHooks", "didkey", "http"];
const sorted = (v) => (Array.isArray(v) ? v.map(sorted) : v !== null && typeof v === "object" ? Object.fromEntries(Object.keys(v).sort().map((k) => [k, sorted(v[k])])) : v);

/**
 * A module's pin (Rust run::module_pin): hex sha256 of its wasm's hex sha256, a newline, and its grant fields
 * (calls, anyRecords, hooks, implements, anyHooks, didkey, http: those present) as JSON with sorted keys. A composition
 * instance's `pin` must match it, so a record whose calls or hooks change no longer does.
 */
export async function modulePin(mod) {
	const bytes = mod.module;
	const view = ArrayBuffer.isView(bytes) ? new Uint8Array(bytes.buffer, bytes.byteOffset, bytes.byteLength) : new Uint8Array(bytes);
	const grants = Object.fromEntries(GRANT_FIELDS.filter((k) => mod[k] !== undefined).map((k) => [k, mod[k]]));
	const text = `${hex(await crypto.subtle.digest("SHA-256", view))}\n${JSON.stringify(sorted(grants))}`;
	return hex(await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text)));
}

// The module record fields a host takes from a record or a module.json (Rust: registry.rs SPEC_FIELDS).
const SPEC_FIELDS = ["name", "description", "actions", "config", "calls", "anyRecords", "didkey", "session", "hooks", "implements", "anyHooks", "http"];
const specOf = (value) => Object.fromEntries(SPEC_FIELDS.filter((key) => value?.[key] !== undefined).map((key) => [key, value[key]]));

/**
 * A local build in loadModule's shape: the wasm and its module.json, fetched (Node: server fileFetch for
 * file: URLs). Only the record fields are taken from the json; the id (the ledger owner) is the wasm's URL.
 * @param {string | URL} wasmUrl
 * @param {string | URL} specUrl
 */
export async function localModule(wasmUrl, specUrl, fetchImpl = globalThis.fetch) {
	const get = async (url) => {
		const response = await fetchImpl(url);
		if (!response.ok) throw new Error(`${url}: ${response.status}`);
		return response;
	};
	const [wasm, spec] = await Promise.all([get(wasmUrl).then((r) => r.arrayBuffer()), get(specUrl).then((r) => r.json())]);
	return { module: new Uint8Array(wasm), id: String(wasmUrl), ...checkLists(specOf(spec)) };
}