johnandersen777.bsky.social/atlibs-fedproxy

Switch repository

Loading…

import { Cubit } from 'atlibs/cubit/index.js'
import { createDidKey, didkeyCall, ingressHost } from 'atlibs/didkey/index.js'
import { runModule } from 'atlibs/run-module/index.js'
import { hostCall } from 'atlibs/run-module/host-call.js'

const KEEP = 50;
// A tab serves until stopped: an hour of wall time, keepalives included in the events.
export const BUDGET = { wallMs: 3_600_000, events: 100_000 };
const KEEPALIVE = '/xrpc/com.publicdomainrelay.temp.keepalive.ping';

// Everything the page knows. status: "loading" | "idle" | "running" | "done" | "error".
const INITIAL = { status: 'loading', did: null, url: null, served: 0, drops: 0, error: null, log: [] };

// One log line per event worth showing; relay keepalives are left out.
function describe(event) {
	if (event.type === 'message') {
		const { $type = '', path, nsid } = event.data ?? {};
		return path === KEEPALIVE ? null : `${event.id} ${$type.split('#')[1] ?? $type} ${path ?? nsid ?? ''}`;
	}
	if (event.type === 'result') return `${event.id} ${event.ok ? 'ok' : `failed: ${event.error}`}`;
	if (event.type === 'closed') return `${event.id} closed${event.error ? `: ${event.error}` : ''}`;
	if (event.type === 'timer') return event.id.startsWith('tick:') ? null : `timer ${event.id}`;
	return event.type;
}

// The ingress module's browser host: a did:key held in this tab (WebCrypto; its JWK sealed in the
// plugin-context secret setting `didkeys`, keyed by the module's identity, never given to wasm), the
// relay subscription over the browser WebSocket (service-auth rides in the query string), and
// public DID reads. No atproto session.
//   ctx        PluginContext: wasm, settings (didkeys: secret), http
//   module     Promise of { module, calls, didkey }
//   id         the module's identity: the wasm URL; a loaded record's own id (its at:// URI) wins, so a
//              ?module=<handle>/<name> is keyed and confirmed by the DID's URI, never the handle
//   WebSocket  for tests
export class IngressCubit extends Cubit {
	#ctx;
	#module;
	#id;
	#WebSocket;
	#key;
	#stop;

	constructor({ ctx, module, id, WebSocket = globalThis.WebSocket }) {
		super(INITIAL);
		this.#ctx = ctx;
		this.#module = module;
		this.#id = id;
		this.#WebSocket = WebSocket;
	}

	async init() {
		try {
			const mod = await this.#module;
			this.#id = mod.id ?? this.#id;
			if (!mod.didkey?.services?.length) throw new Error('the module declares no didkey services');
			// one tab at a time reads and writes the key map (Web Locks), so tabs first serving different modules keep every key
			const locked = (fn) => (globalThis.navigator?.locks ? navigator.locks.request('atlibs-didkeys', fn) : fn());
			this.#key = await locked(async () => {
				const keys = (await this.#ctx.settings.get('didkeys')) ?? {};
				const key = await createDidKey(keys[this.#id]);
				if (!keys[this.#id]) await this.#ctx.settings.set('didkeys', { ...keys, [this.#id]: key.jwk });
				return key;
			});
			this.emit({ status: 'idle', did: this.#key.did, url: `https://${ingressHost(this.#key.did, mod.didkey.services[0])}/` });
		} catch (error) {
			this.emit({ status: 'error', error: error.message });
		}
	}

	#line(text) {
		if (text) this.emit({ log: [`${new Date().toISOString().slice(11, 19)} ${text}`, ...this.state.log].slice(0, KEEP) });
	}

	// Runs `serve` until it is done, fails, or stop() is called.
	async start(input = {}) {
		if (this.state.status === 'running' || !this.#key) return;
		const mod = await this.#module;
		this.#stop = new AbortController();
		this.emit({ status: 'running', error: null, served: 0, drops: 0, log: [] });
		const call = didkeyCall(this.#key, this.#call, { services: mod.didkey.services, fetch: (url, init) => this.#ctx.http.fetch(url, init), canHeader: false });
		const ran = await runModule(this.#ctx, mod, 'serve', input, call, {
			signal: this.#stop.signal,
			budget: BUDGET,
			provides: [],
			onEvent: (event, out) => {
				this.#line(describe(event));
				if (out?.result) this.emit({ served: out.result.served ?? 0, drops: out.result.drops ?? 0 });
			},
		});
		this.#line(ran.error ? `run ended: ${ran.error}` : 'run done');
		this.emit({ status: ran.error && ran.error !== 'aborted' ? 'error' : 'done', error: ran.error ?? null });
		return ran;
	}

	stop() {
		this.#stop?.abort();
	}

	// &serve on load (main.js) serves at once only this page's own module. A link can name any module
	// record (?module=<handle>/<name> or at://...), so that one serves only after confirm(message) shows what and where.
	async mayAutoServe(confirm) {
		const mod = await this.#module;
		const id = mod.id ?? this.#id;
		if (!id.startsWith('at://')) return true;
		const services = mod.didkey?.services ?? [];
		return confirm(`Serve ${id} from this tab?\n\nIt registers at ${services.join(', ')} and answers requests from anyone there, in this tab, for up to an hour.`) === true;
	}

	// The run-module host call (atlibs/run-module/host-call.js), signed out: subscriptions over the browser
	// WebSocket (no headers), and public reads of a DID's PDS. didkeyCall in front answers the did:key
	// pseudo-calls, the relay nonce, and serviceAuth to its own services; any other needs a session.
	get #call() {
		return hostCall({ ctx: this.#ctx, WebSocket: this.#WebSocket, canHeader: false, signIn: 'this page has none' });
	}
}