johnandersen777.bsky.social/atlibs

Switch repository

Loading…

// Browser implementation of the parts of emdash's PluginContext that
// atproto-integrations-emdash uses: storage, settings, kv, http, wasm, log,
// site, url and plugin. Storage lives in IndexedDB; every compare-and-set runs
// inside one readwrite transaction, so it is atomic across tabs too.

const WASM_LIMITS = { moduleBytes: 4 << 20, inputBytes: 1 << 20, outputBytes: 1 << 20, memoryPages: 1024, wallTimeMs: 5000 };
const ENTRY_PATTERN = /^[A-Za-z_][A-Za-z0-9_]{0,255}$/;

const done = (request) =>
	new Promise((resolve, reject) => {
		request.onsuccess = () => resolve(request.result);
		request.onerror = () => reject(request.error);
	});

function openDatabase(name) {
	const request = indexedDB.open(name, 1);
	request.onupgradeneeded = () => request.result.createObjectStore("rows");
	return done(request);
}

// Values round-trip through JSON, as they do in emdash's database.
const json = (value) => JSON.parse(JSON.stringify(value ?? null));

export function matches(data, where = {}) {
	return Object.entries(where).every(([field, cond]) => {
		const value = data?.[field] ?? null;
		if (cond === null || typeof cond !== "object") return value === cond;
		if ("in" in cond) return cond.in.includes(value);
		if ("startsWith" in cond) return typeof value === "string" && value.startsWith(cond.startsWith);
		return (
			value !== null &&
			(cond.gt === undefined || value > cond.gt) &&
			(cond.gte === undefined || value >= cond.gte) &&
			(cond.lt === undefined || value < cond.lt) &&
			(cond.lte === undefined || value <= cond.lte)
		);
	});
}

export function compareBy(orderBy = {}) {
	const fields = Object.entries(orderBy);
	return (a, b) => {
		for (const [field, direction] of fields) {
			const x = a.value?.[field];
			const y = b.value?.[field];
			if (x === y) continue;
			const order = x === undefined || x === null ? -1 : y === undefined || y === null ? 1 : x < y ? -1 : 1;
			return direction === "desc" ? -order : order;
		}
		return a.id < b.id ? -1 : a.id > b.id ? 1 : 0;
	};
}

// fn may only await IndexedDB requests; awaiting anything else lets the transaction commit early.
async function transaction(database, mode, fn) {
	const tx = (await database).transaction("rows", mode);
	const finished = new Promise((resolve, reject) => {
		tx.oncomplete = resolve;
		tx.onerror = tx.onabort = () => reject(tx.error);
	});
	const [result] = await Promise.all([fn(tx.objectStore("rows")), finished]);
	return result;
}

function collection(database, namespace) {
	const key = (id) => [namespace, id];
	const row = (id, data) => ({ id, value: json(data), revision: crypto.randomUUID() });
	const read = (fn) => transaction(database, "readonly", fn);
	const write = (fn) => transaction(database, "readwrite", fn);
	// Keys are [namespace, id]; an array sorts after every string, so this spans one namespace.
	const rows = () => read((store) => done(store.getAll(IDBKeyRange.bound([namespace], [namespace, []]))));

	const self = {
		get: async (id) => (await read((store) => done(store.get(key(id)))))?.value ?? null,
		getVersioned: async (id) => {
			const found = await read((store) => done(store.get(key(id))));
			return found ? { value: found.value, revision: found.revision } : null;
		},
		exists: async (id) => (await read((store) => done(store.count(key(id))))) > 0,
		put: (id, data) => write((store) => done(store.put(row(id, data), key(id)))).then(() => undefined),
		delete: (id) =>
			write(async (store) => {
				const found = await done(store.count(key(id)));
				await done(store.delete(key(id)));
				return found > 0;
			}),
		compareAndSet: (id, expectedRevision, data) =>
			write(async (store) => {
				const found = await done(store.get(key(id)));
				if ((found?.revision ?? null) !== expectedRevision) return { applied: false };
				const next = row(id, data);
				await done(store.put(next, key(id)));
				return { applied: true, revision: next.revision };
			}),
		compareAndDelete: (id, expectedRevision) =>
			write(async (store) => {
				const found = await done(store.get(key(id)));
				if (!found || found.revision !== expectedRevision) return { applied: false };
				await done(store.delete(key(id)));
				return { applied: true };
			}),
		getMany: (ids) =>
			read(async (store) => {
				const found = await Promise.all(ids.map((id) => done(store.get(key(id)))));
				return new Map(found.filter(Boolean).map((item) => [item.id, item.value]));
			}),
		putMany: (items) =>
			write((store) => Promise.all(items.map(({ id, data }) => done(store.put(row(id, data), key(id)))))).then(() => undefined),
		deleteMany: (ids) =>
			write(async (store) => {
				const counts = await Promise.all(ids.map((id) => done(store.count(key(id)))));
				await Promise.all(ids.map((id) => done(store.delete(key(id)))));
				return counts.reduce((sum, count) => sum + count, 0);
			}),
		// ponytail: scans the whole namespace and pages by offset; fine for plugin-sized tables, add IDB indexes if they grow.
		query: async ({ where, orderBy, limit = 50, cursor } = {}) => {
			const all = (await rows()).filter((item) => matches(item.value, where)).sort(compareBy(orderBy));
			const start = cursor ? Number(cursor) : 0;
			const end = start + Math.min(Math.max(limit, 1), 100);
			const hasMore = end < all.length;
			return { items: all.slice(start, end).map(({ id, value }) => ({ id, data: value })), hasMore, ...(hasMore && { cursor: String(end) }) };
		},
		count: async (where) => (await rows()).filter((item) => matches(item.value, where)).length,
		// Update-only: absent row, non-object document, failed guard or bad counter all return applied: false.
		updateIf: (id, { where, set = {}, delta = {} }) => {
			const changes = json(set);
			const steps = Object.entries(delta).map(([field, spec]) => {
				if (field in changes) throw new TypeError("Storage field appears in both set and delta");
				const amount = "inc" in spec ? spec.inc : -spec.dec;
				if (!Number.isSafeInteger(amount)) throw new TypeError("Storage delta must be a safe integer");
				return [field, amount];
			});
			return write(async (store) => {
				const data = (await done(store.get(key(id))))?.value;
				if (typeof data !== "object" || data === null || Array.isArray(data) || !matches(data, where)) return { applied: false };
				const next = { ...data, ...changes };
				for (const [field, amount] of steps) {
					next[field] = data[field] + amount;
					if (!Number.isSafeInteger(data[field]) || !Number.isSafeInteger(next[field])) return { applied: false };
				}
				const updated = row(id, next);
				await done(store.put(updated, key(id)));
				return { applied: true, data: updated.value };
			});
		},
		// SettingsAccess / KVAccess shape
		set: (id, value) => self.put(id, value),
		list: async (prefix = "") =>
			(await rows()).filter((item) => item.id.startsWith(prefix)).map(({ id, value }) => ({ key: id, value })),
	};
	return self;
}

// Non-extractable AES-GCM key kept in IndexedDB: page script can use it but never read its bytes.
function secretKey(database) {
	let key;
	return () =>
		(key ??= crypto.subtle.generateKey({ name: "AES-GCM", length: 256 }, false, ["encrypt", "decrypt"]).then((fresh) =>
			transaction(database, "readwrite", async (store) => {
				const found = await done(store.get(["meta", "secretKey"]));
				if (found) return found;
				await done(store.put(fresh, ["meta", "secretKey"]));
				return fresh;
			}),
		));
}

// Settings whose schema type is "secret" are stored encrypted, as emdash does, bound to plugin id and setting name.
export function settingsAccess(settings, schema, pluginId, key) {
	const additionalData = (name) => new TextEncoder().encode(`${pluginId}:settings:${name}`);
	const seal = async (name, value) => {
		if (schema[name]?.type !== "secret" || value === null || value === undefined) return value;
		const iv = crypto.getRandomValues(new Uint8Array(12));
		const data = await crypto.subtle.encrypt({ name: "AES-GCM", iv, additionalData: additionalData(name) }, await key(), new TextEncoder().encode(JSON.stringify(value)));
		return { $secret: { iv: iv.toBase64(), data: new Uint8Array(data).toBase64() } };
	};
	const open = async (name, value) => {
		if (!value?.$secret) return value;
		const { iv, data } = value.$secret;
		const plain = await crypto.subtle.decrypt({ name: "AES-GCM", iv: Uint8Array.fromBase64(iv), additionalData: additionalData(name) }, await key(), Uint8Array.fromBase64(data));
		return JSON.parse(new TextDecoder().decode(plain));
	};
	return {
		get: async (name) => open(name, await settings.get(name)),
		getVersioned: async (name) => {
			const found = await settings.getVersioned(name);
			return found && { ...found, value: await open(name, found.value) };
		},
		compareAndSet: async (name, revision, value) => settings.compareAndSet(name, revision, await seal(name, value)),
		compareAndDelete: settings.compareAndDelete,
		set: async (name, value) => settings.put(name, await seal(name, value)),
		delete: settings.delete,
		list: async (prefix) => Promise.all((await settings.list(prefix)).map(async ({ key, value }) => ({ key, value: await open(key, value) }))),
	};
}

async function sha256Hex(bytes) {
	const digest = await crypto.subtle.digest("SHA-256", bytes);
	return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("");
}

const wasmError = (code, message) => Object.assign(new Error(`${code}: ${message}`), { name: "WasmRunError", code });

const uleb = (n) => {
	const out = [];
	do {
		const low = n % 128;
		n = Math.floor(n / 128);
		out.push(n ? low | 128 : low);
	} while (n);
	return out;
};

/**
 * The memory cap, enforced inside the call. V8 has no hook on a module's own memory.grow, and a
 * defined memory's maximum is not visible through WebAssembly.Module, so the host rewrites the
 * memory section's limits before compiling: maximum = `pages` (or the module's own, if lower;
 * rustc declares none). memory.grow past it then returns -1 inside the call, as the Rust host's
 * wasmtime StoreLimits does; a module that cannot handle that traps (WASM_TRAP). A minimum over
 * `pages` is refused (WASM_TOO_LARGE), as are a second memory and memory64 (WASM_ABI_INVALID, as
 * the Rust host's engine config). Other bytes are copied as they are.
 * @param {Uint8Array} bytes  a wasm binary
 * @param {number} pages
 */
export function capMemory(bytes, pages) {
	let at = 8; // magic + version
	const leb = () => {
		let value = 0;
		let scale = 1;
		let byte;
		do {
			byte = bytes[at++] ?? 0;
			value += (byte & 127) * scale;
			scale *= 128;
		} while (byte & 128);
		return value;
	};
	while (at < bytes.length) {
		const start = at;
		const id = bytes[at++];
		const size = leb();
		const end = at + size;
		if (id === 5) {
			const count = leb();
			// One memory, 32-bit: the cap is per memory, so a second memory (or memory64) would escape it.
			if (count > 1) throw wasmError("WASM_ABI_INVALID", "a module may declare at most one memory");
			const body = [...uleb(count)];
			for (let i = 0; i < count; i++) {
				const flags = bytes[at++];
				if (flags & 4) throw wasmError("WASM_ABI_INVALID", "memory64 is not supported");
				const min = leb();
				const max = flags & 1 ? leb() : pages;
				if (min > pages) throw wasmError("WASM_TOO_LARGE", `memory exceeds ${pages} pages`);
				body.push(flags | 1, ...uleb(min), ...uleb(Math.min(max, pages)));
			}
			const section = [5, ...uleb(body.length), ...body];
			const out = new Uint8Array(start + section.length + bytes.length - end);
			out.set(bytes.subarray(0, start));
			out.set(section, start);
			out.set(bytes.subarray(end), start + section.length);
			return out;
		}
		at = end;
	}
	return bytes;
}

// Validates and compiles here (cached by digest), then runs the call in a fresh Worker
// (wasm-worker.js) that is terminated after timeoutMs, so a module that never returns
// costs one worker, not the page. Matches emdash's WASM_LIMITS.wallTimeMs by default.
// ponytail: one worker per call (a few ms each); pool them if calls get hot.
// The default spawn starts a same-origin blob that imports wasm-worker.js: a page may load atlibs from a
// CDN, and a Worker's script must be the page's own origin (a module worker imports from anywhere with CORS).
const shims = new Map();
const spawnModuleWorker = (url) => {
	if (!shims.has(url.href)) shims.set(url.href, URL.createObjectURL(new Blob([`import ${JSON.stringify(url.href)};`], { type: "text/javascript" })));
	return new Worker(shims.get(url.href), { type: "module" });
};
export function wasmAccess(timeoutMs, spawn = spawnModuleWorker) {
	const cache = new Map();
	return {
		// ioBytes: the input and output cap for this call (the run's budget.ioBytes), at most 16 MiB.
		async run({ module, entry, input, sha256, ioBytes }) {
			const io = Math.min(ioBytes > 0 ? ioBytes : WASM_LIMITS.inputBytes, 16 << 20);
			const limits = { ...WASM_LIMITS, inputBytes: io, outputBytes: io };
			if (!(module instanceof Uint8Array)) throw wasmError("WASM_ABI_INVALID", "module must be a Uint8Array");
			if (module.byteLength > WASM_LIMITS.moduleBytes) throw wasmError("WASM_TOO_LARGE", `module exceeds ${WASM_LIMITS.moduleBytes} bytes`);
			if (typeof entry !== "string" || !ENTRY_PATTERN.test(entry)) throw wasmError("WASM_ABI_INVALID", "entry must be an export name");
			const inputJson = JSON.stringify(input ?? null);
			if (new TextEncoder().encode(inputJson).byteLength > io) throw wasmError("WASM_TOO_LARGE", `input exceeds ${io} bytes`);
			const digest = await sha256Hex(module);
			if (sha256 !== undefined && String(sha256).toLowerCase() !== digest) throw wasmError("WASM_DIGEST_MISMATCH", "module digest does not match sha256");

			let compiled = cache.get(digest);
			if (!compiled) {
				const capped = capMemory(module, WASM_LIMITS.memoryPages);
				try {
					compiled = await WebAssembly.compile(capped);
				} catch (error) {
					throw wasmError("WASM_ABI_INVALID", error.message);
				}
				cache.set(digest, compiled);
			}
			if (WebAssembly.Module.imports(compiled).length > 0) throw wasmError("WASM_IMPORTS_NOT_ALLOWED", "modules must not declare imports");

			const worker = spawn(new URL("./wasm-worker.js", import.meta.url));
			try {
				const reply = await new Promise((resolve, reject) => {
					const timer = setTimeout(() => reject(wasmError("WASM_TIMEOUT", `${entry} ran longer than ${timeoutMs} ms`)), timeoutMs);
					worker.onmessage = ({ data }) => (clearTimeout(timer), resolve(data));
					worker.onerror = (event) => (clearTimeout(timer), reject(wasmError("WASM_UNAVAILABLE", event.message ?? "worker failed")));
					worker.postMessage({ compiled, entry, inputJson, limits });
				});
				if (reply.error === undefined) return reply.output;
				const [, code, message] = /^(WASM_[A-Z_]+): ([\s\S]*)$/.exec(reply.error) ?? [, "WASM_TRAP", reply.error];
				throw wasmError(code, message);
			} finally {
				worker.terminate();
			}
		},
	};
}

/**
 * @param {{ id: string, version?: string, storage?: Record<string, unknown>, settingsSchema?: Record<string, { type: string }>, siteUrl?: string, siteName?: string, fetch?: typeof fetch, dbName?: string, wasmTimeoutMs?: number }} options
 *   `storage` and `settingsSchema` take the `storage` and `admin.settingsSchema` blocks of emdash-plugin.jsonc;
 *   storage indexes are not needed here.
 * @returns {import("emdash/plugin").PluginContext}
 */
export function createPluginContext({ id, version = "0.0.0", storage = {}, settingsSchema = {}, siteUrl = globalThis.location?.origin ?? "", siteName = "", fetch: fetchImpl, dbName = `emdash-plugin:${id}`, wasmTimeoutMs = WASM_LIMITS.wallTimeMs }) {
	const database = openDatabase(dbName);
	const log = Object.fromEntries(
		["debug", "info", "warn", "error"].map((level) => [level, (message, data) => console[level](`[${id}] ${message}`, ...(data === undefined ? [] : [data]))]),
	);
	return {
		plugin: { id, version },
		storage: Object.fromEntries(Object.keys(storage).map((name) => [name, collection(database, `storage:${name}`)])),
		settings: settingsAccess(collection(database, "settings"), settingsSchema, id, secretKey(database)),
		kv: collection(database, "kv"),
		http: { fetch: (url, init) => (fetchImpl ?? globalThis.fetch)(url, init) },
		wasm: wasmAccess(wasmTimeoutMs),
		log,
		site: { name: siteName, url: siteUrl, locale: "en", trailingSlash: "ignore" },
		url: (path) => new URL(path, siteUrl).href,
	};
}