johnandersen777.bsky.social/atlibs-agents

Switch repository

Loading…

import { Cubit } from 'atlibs/cubit/index.js'
import { serviceAud, sessionXrpc } from 'atlibs/run-module/host-call.js'

export const LAUNCH = "com.fedfork.agents.launch";
export const NSID = { launch: "com.fedfork.agents.launchAgent", get: "com.fedfork.agents.getAgent", shutdown: "com.fedfork.agents.shutdownAgent" };
// The one OAuth session this page needs: the user's PDS minting service auth for the three methods.
// aud=* because the service's host name is the page's setting (a did:key ingress host), not a constant.
export const SCOPE = ["atproto", ...Object.values(NSID).map((nsid) => `rpc:${nsid}?aud=*`)].join(" ");
// The agents service this page talks to by default (service/ on xrpc.fedproxy.com).
export const DEFAULT_SERVICE = "https://did-key-zdnaewfbq92fbgek476eg2kmnykfbvzphxfwwexy7x3cp6dyc.xrpc.fedproxy.com";

// Page state.
//   session: "loading" | "signed-out" | "signed-in" | "error"
//   service: the agents service's https origin
//   agents: [{rkey, uri, record, guest?}] this account launched from this browser, newest first
//   busy / progress / results: per launch rkey ("launch" while a launch has no rkey yet)
const INITIAL = {
	session: "loading",
	did: null,
	handle: null,
	error: null,
	login: { busy: false, error: null },
	service: DEFAULT_SERVICE,
	agents: [],
	busy: {},
	progress: {},
	results: {},
};

// The agents service over XRPC as the signed-in account: each call carries a
// service-auth token the user's PDS mints (com.atproto.server.getServiceAuth, aud = the service
// host's did:web, lxm = the method). The service records the caller and answers only its launches,
// so the page keeps the rkeys it launched (localStorage, per account) and reads them back.
//   client     Promise of a BrowserOAuthClient
//   makeAgent  session => @atproto/api Agent
//   fetch      for the service (tests stub it)
//   storage    localStorage-like, for the service origin and the launched rkeys
export class AgentsCubit extends Cubit {
	#client;
	#makeAgent;
	#fetch;
	#storage;
	#agent;

	constructor({ client, makeAgent, fetch = globalThis.fetch.bind(globalThis), storage = globalThis.localStorage, service }) {
		super({ ...INITIAL, service: service ?? storage?.getItem("agents.service") ?? DEFAULT_SERVICE });
		this.#client = client;
		this.#makeAgent = makeAgent;
		this.#fetch = fetch;
		this.#storage = storage;
	}

	async init() {
		try {
			const result = 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.emit({ session: "signed-in", did: this.#agent.did, handle: res.data.handle });
		} catch (error) {
			return this.emit({ session: "error", error: `An error occurred: ${error.message}` });
		}
		await this.refresh();
	}

	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 = undefined;
		this.emit({ ...INITIAL, service: this.state.service, session: "signed-out" });
	}

	async setService(service) {
		const origin = new URL(service).origin;
		if (!origin.startsWith("https://")) throw new Error("the service must be an https origin");
		this.#storage?.setItem("agents.service", origin);
		this.emit({ service: origin, agents: [] });
		await this.refresh();
	}

	// ---- launched rkeys, per account and service -----------------------------------------------
	get #key() {
		return `agents.launches:${this.state.did}:${this.state.service}`;
	}
	#known() {
		try {
			return JSON.parse(this.#storage?.getItem(this.#key) ?? "[]");
		} catch {
			return [];
		}
	}
	#remember(rkeys) {
		this.#storage?.setItem(this.#key, JSON.stringify(rkeys));
	}

	// ---- XRPC -----------------------------------------------------------------------------------
	async #token(nsid) {
		const params = { aud: serviceAud(this.state.service), lxm: nsid };
		return (await sessionXrpc(this.#agent)("com.atproto.server.getServiceAuth", { params })).token;
	}

	// Errors keep the service's status and name ("403 Forbidden: ...") for the page and tests.
	async #xrpc(nsid, { params, body } = {}) {
		if (!this.#agent) throw new Error(`${nsid} needs a signed-in session`);
		const token = await this.#token(nsid);
		const res = await this.#fetch(`${this.state.service}/xrpc/${nsid}${params ? `?${new URLSearchParams(params)}` : ""}`, {
			method: body ? "POST" : "GET",
			headers: { authorization: `Bearer ${token}`, ...(body && { "content-type": "application/json" }) },
			body: body && JSON.stringify(body),
		});
		const out = await res.json().catch(() => ({}));
		if (!res.ok) throw Object.assign(new Error(`${res.status} ${out.error ?? "Error"}: ${out.message ?? ""}`), { status: res.status, name: out.error });
		return out;
	}

	#set(key, field, value) {
		this.emit({ [field]: { ...this.state[field], [key]: value } });
	}

	async #run(key, label, work) {
		this.#set(key, "busy", label);
		try {
			const result = await work();
			this.#set(result?.rkey ?? key, "results", result);
			return result;
		} catch (error) {
			this.#set(key, "results", { error: error.message });
			return { error: error.message };
		} finally {
			this.#set(key, "busy", null);
		}
	}

	// Every launch this browser remembers for the account, read back from the service; ones the
	// service no longer has (deleted, another service) are forgotten.
	async refresh() {
		if (!this.#agent) return this.emit({ agents: [] });
		const agents = [];
		const keep = [];
		for (const rkey of this.#known()) {
			try {
				agents.push(await this.#xrpc(NSID.get, { params: { rkey } }));
				keep.push(rkey);
			} catch (error) {
				if (!/RecordNotFound|Forbidden/.test(error.message)) {
					keep.push(rkey);
					agents.push({ rkey, record: { status: "unknown", error: error.message } });
				}
			}
		}
		this.#remember(keep);
		this.emit({ agents });
	}

	// bidders (DIDs, a list or space separated), name, exp_s: the service decides the rest (user_data,
	// the guest command `atirc agent run`).
	async launch({ bidders, name, exp_s }) {
		const body = { ...(bidders && { bidders }), ...(name && { name }), ...(exp_s && { exp_s: Number(exp_s) }) };
		const result = await this.#run("launch", "launch", () => this.#xrpc(NSID.launch, { body }));
		if (result.rkey) this.#remember([result.rkey, ...this.#known().filter((r) => r !== result.rkey)]);
		await this.refresh();
		return result;
	}

	async status(rkey) {
		const result = await this.#run(rkey, "status", () => this.#xrpc(NSID.get, { params: { rkey } }));
		if (result.record) this.emit({ agents: this.state.agents.map((a) => (a.rkey === rkey ? result : a)) });
		return result;
	}

	async shutdown(rkey, remove = false) {
		const result = await this.#run(rkey, "shutdown", () => this.#xrpc(NSID.shutdown, { body: { rkey, delete: remove } }));
		await this.refresh();
		return result;
	}
}