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(); } }