johnandersen777.bsky.social/atlibs-socialweb-computer-requester

Switch repository

Loading…

import { Cubit } from 'atlibs/cubit/index.js'
import { attestHost } from 'atlibs/attest/index.js'
import { coerce } from 'atlibs/mcp/index.js'
import { runModule } from 'atlibs/run-module/index.js'
import { hostCall, sessionXrpc } from 'atlibs/run-module/host-call.js'
import { WEBRTC_NSID, WEBRTC_STATE, WebrtcHost } from 'atlibs/webrtc/index.js'
import { recordSignals } from 'atlibs/webrtc/record-signals.js'

// A contract polls the bid, receipt, guest reports and termination (poll_ms 3000 x max_polls 40 a wait):
// minutes of wall time and hundreds of reads, past runModule's defaults (60 s, 100 calls). A hook default
// (the bid policy) runs on public reads (callFor): without it every read of the policy is refused (a deny).
const CONTRACT_BUDGET = { wallMs: 20 * 60_000, calls: 4000, events: 40_000 };

// A form gives text: each field as its module.json type (hold_ms "60000" -> 60000; as text the module read
// no hold and released at once). Text that is not one goes as typed (the module parses lists and refs).
const typed = (module, entry, input) => {
	const fields = module.actions?.find((a) => a.entry === entry)?.fields ?? [];
	const type = (key) => fields.find((f) => f.key === key)?.type;
	return Object.fromEntries(Object.entries(input).map(([k, v]) => [k, (type(k) && coerce(type(k), v)) ?? v]));
};

// Everything the page knows.
//   session: "loading" | "signed-out" | "signed-in" | "error" (the OAuth session; reads work without it)
//   results: {entry: result | {error}} of requester module runs; running: the entry in flight
//   and the webrtc transport's (atlibs/webrtc WEBRTC_STATE): phase, receipt, tag, guest, output, error, ssh,
//     hostKeys, and signal: the requester `signal` input; signed in, the page runs it, else a host with the
//     attestation key does
const INITIAL = {
	session: "loading",
	did: null,
	handle: null,
	login: { busy: false, error: null },
	results: {},
	running: null,
	...WEBRTC_STATE,
};

// Page logic: the OAuth session, the requester wasm module run as that session (the run-module
// host contract, attest included), with the webrtc transport (atlibs/webrtc) as its webrtc.connect.
//   client     Promise of a BrowserOAuthClient (optional: without it only reads run)
//   makeAgent  session => @atproto/api Agent
//   secp256k1  @noble/curves secp256k1, for the attestation key
//   age        the age-encryption (typage) module
//   makePeer   config => RTCPeerConnection
//   module     Promise of { module: Uint8Array, calls: string[] } (requester)
//   ctx        PluginContext: wasm, http, settings with {attestKeys: {type: "secret"}}
//   ssh        @microsoft/dev-tunnels-ssh namespace plus its Buffer (optional: ssh over the channel)
//   signals    localSignals(url) (atlibs/webrtc, optional): guest events and offers over a local endpoint, not records
export class WebrtcCubit extends Cubit {
	#client;
	#makeAgent;
	#secp256k1;
	#module;
	#ctx;
	#agent;
	#attest;
	#signals;
	#webrtc;

	constructor({ client, makeAgent, secp256k1, age, makePeer, module, ctx, ssh, signals, sleep = (ms) => new Promise((r) => setTimeout(r, ms)) }) {
		super(INITIAL);
		this.#client = client;
		this.#makeAgent = makeAgent;
		this.#secp256k1 = secp256k1;
		this.#module = module;
		this.#ctx = ctx;
		this.#signals = signals;
		// The offer goes as the session's `signal` run (signed out: left in state.signal), or to the local endpoint.
		const signal = (input) => (signals ? signals.signal(input) : this.state.session === "signed-in" ? this.run("signal", input) : undefined);
		// A bidder's webrtc endpoint (offering endpoints, webrtc://<{did, listen}>): offers and answers as
		// computer.socialweb.ingress records, the offer written as this session.
		const records = recordSignals({ repo: this.#repo(), subhandle: "requester" });
		this.#webrtc = new WebrtcHost({ store: this, events: (at) => this.#events(at), signal, records, makePeer, age, ssh, sleep });
	}

	async init() {
		try {
			const result = this.#client && (await (await this.#client).init());
			if (!result) return this.emit({ session: "signed-out" });
			this.#agent = this.#makeAgent(result.session);
			const res = await this.#agent.com.atproto.server.getSession();
			if (!res.success) throw new Error(JSON.stringify(res));
			this.#attest = attestHost(this.#ctx, this.#secp256k1, this.#agent.did);
			this.emit({ session: "signed-in", did: this.#agent.did, handle: res.data.handle });
		} catch (error) {
			this.emit({ session: "error", error: `An error occurred: ${error.message}` });
		}
	}

	async login(identifier) {
		this.emit({ login: { busy: true, error: null } });
		try {
			await (await this.#client).signIn(identifier, { signal: new AbortController().signal });
		} catch (error) {
			this.emit({ login: { busy: false, error: `Login error: ${error.message}` } });
		}
	}

	async logout() {
		await (await this.#client).revoke(this.#agent.did).catch(() => undefined);
		this.#agent = this.#attest = undefined;
		this.emit({ session: "signed-out", did: null, handle: null, results: {} });
	}

	// recordSignals' repo: reads from each repo's own PDS (hostCall resolves the DID), writes as the session.
	// The offer's rkey is a fresh TID, so its put is a create (scope: ingress.offer create, delete).
	#repo() {
		const cubit = this;
		return {
			get did() {
				return cubit.state.did;
			},
			getRecord: async ({ repo, collection, rkey }) => {
				try {
					return await cubit.#call("com.atproto.repo.getRecord", { params: { repo, collection, rkey }, service: repo });
				} catch (error) {
					if (/RecordNotFound|not found|404/i.test(error.message)) return null;
					throw error;
				}
			},
			putRecord: ({ collection, rkey, record }) => cubit.#xrpc("com.atproto.repo.createRecord", { input: { repo: cubit.state.did, collection, rkey, record } }),
			deleteRecord: ({ collection, rkey }) => cubit.#xrpc("com.atproto.repo.deleteRecord", { input: { repo: cubit.state.did, collection, rkey } }),
		};
	}

	// XRPC as the session (its fetchHandler: DPoP, refresh).
	#xrpc(nsid, init) {
		if (!this.#agent) throw new Error(`${nsid} needs a signed-in session`);
		return sessionXrpc(this.#agent)(nsid, init);
	}

	// The run-module host call (atlibs/run-module/host-call.js) as this
	// session. webrtc.connect is the atlibs/webrtc transport: a request finds, offers, answers and opens,
	// a subscribe is a channel session; it needs a session (or local signals). In a browser a WebSocket takes no headers,
	// so no atproto-proxy on a subscribe and serviceAuth only as { param }.
	#call = (nsid, request) =>
		hostCall({
			ctx: this.#ctx,
			xrpc: this.#agent && ((n, init) => this.#xrpc(n, init)),
			attest: this.#attest,
			did: this.state.did,
			canHeader: false,
			signIn: "sign in",
			calls: { [WEBRTC_NSID]: (r) => this.#webrtc.call(this.#need(r)) },
			// a bidder's did#pdr_temp_market endpoints: webrtc://<{did, listen}>, over record signals
			fetchers: this.fetchers,
		})(nsid, request);

	// runModule's fetchers: webrtc: URLs (a kept VM's shell, contract's mcp) over this session's signals.
	get fetchers() {
		return { "webrtc:": async (r) => this.#webrtc.fetch(this.#need(r)) };
	}

	// The host call, for pages that run other modules (or compositions) as this session.
	get call() {
		return this.#call;
	}

	#need(request) {
		if (!this.#agent && !this.#signals) throw new Error(`${WEBRTC_NSID} needs a signed-in session`);
		return request;
	}

	// A run returns errors as data; surface them as throws.
	#runModule(entry, input) {
		return this.#module
			.then((module) => runModule(this.#ctx, module, entry, { did: this.state.did, ...typed(module, entry, input) }, this.#call, { provides: [WEBRTC_NSID], fetchers: this.fetchers, budget: CONTRACT_BUDGET, callFor: () => hostCall({ ctx: this.#ctx, canHeader: false }) }))
			.then((ran) => {
				if (ran.error) throw new Error(ran.error);
				return ran;
			});
	}

	// One requester entry as the session; the result (or error) lands in state.results[entry].
	async run(entry, input = {}) {
		this.emit({ running: entry });
		try {
			const { result } = await this.#runModule(entry, input);
			this.emit({ running: null, results: { ...this.state.results, [entry]: result } });
			return result;
		} catch (error) {
			this.emit({ running: null, results: { ...this.state.results, [entry]: { error: error.message } } });
			throw error;
		}
	}

	// Cleanup: deleteRecord each at:// URI in the session's repo (not part of the module).
	async deleteRecords(uris) {
		const deleted = [];
		for (const uri of uris) {
			const [repo, collection, rkey] = uri.replace("at://", "").split("/");
			if (repo !== this.state.did) throw new Error(`${uri} is not in your repo`);
			await this.#xrpc("com.atproto.repo.deleteRecord", { input: { repo, collection, rkey } });
			deleted.push(uri);
		}
		return deleted;
	}

	// Every event the bidder reported for the receipt, in one `watch` run (the derived reports and the
	// guest's event streams, getRecord each). watch only reads, so it runs signed out too; did is used by
	// write entries only. Its default tags (ssh, shell) plus `tag`: a guest has one DTLS certificate for
	// all tags and the bidder can drop one tag's vm.onNetwork, so guestOf falls back to another tag's.
	// at: {receipt, tag} of the peer asking (the page's, or a webrtc: fetch peer's).
	async #events({ receipt, tag } = this.state) {
		if (this.#signals) return this.#signals.events(receipt, tag);
		// A fetch peer's receipt carries its URL's bidder, else the bidder is the receipt's repo.
		const { uri } = receipt;
		const bidder = receipt.bidder ?? uri.split("/")[2];
		const ran = await this.#runModule("watch", { did: this.state.did ?? bidder, bidder, receipt: uri, tag, max_polls: 1 });
		return ran.result.events;
	}

	// The transport's steps, for the page's controls (atlibs/webrtc WebrtcHost).
	find(receiptText, tag) {
		return this.#webrtc.find(receiptText, tag);
	}
	offer() {
		return this.#webrtc.offer();
	}
	answer() {
		return this.#webrtc.answer();
	}
	send(line) {
		return this.#webrtc.send(line);
	}
	sshKey() {
		return this.#webrtc.sshKey();
	}
	sshExec(command, user, stdin) {
		return this.#webrtc.sshExec(command, user, stdin);
	}

	close() {
		this.#webrtc.close();
	}
}