johnandersen777.bsky.social/atlibs-agents

Switch repository

Loading…

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

const KEEP = 100;

// Everything the page knows, in one place. Components render slices of it.
//   session: "loading" | "signed-out" | "signed-in" | "error"
//   messages.status: "idle" | "loading" | "loaded" | "error"; cursor is the last seq read
//   post.status: "idle" | "busy" | "done" | "error"
const INITIAL = {
	session: "loading",
	handle: null,
	error: null,
	login: { busy: false, error: null },
	channel: "",
	messages: { status: "idle", list: [], cursor: 0, error: null },
	post: { status: "idle", error: null },
};

// Page logic: the OAuth session, and the atirc wasm module's actions run as that session.
// Module calls carry `proxy`; it goes out as the atproto-proxy header, so the user's PDS
// forwards them to atirc with a service-auth token.
//   client     Promise of a BrowserOAuthClient
//   makeAgent  session => @atproto/api Agent
//   module     Promise of { module: Uint8Array, calls: string[] }
//   ctx        PluginContext providing ctx.wasm
export class AtircCubit extends Cubit {
	#client;
	#makeAgent;
	#module;
	#ctx;
	#agent;
	#loading;

	constructor({ client, makeAgent, module, ctx }) {
		super(INITIAL);
		this.#client = client;
		this.#makeAgent = makeAgent;
		this.#module = module;
		this.#ctx = ctx;
	}

	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));
			// atirc's handle channel: "#" + handle, dots to underscores, lowercased.
			const channel = `#${res.data.handle.replaceAll(".", "_").toLowerCase()}`;
			this.emit({ session: "signed-in", handle: res.data.handle, channel });
		} 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, session: "signed-out" });
	}

	async setChannel(channel) {
		if (channel === this.state.channel) return;
		await this.#loading;
		this.emit({ channel, messages: INITIAL.messages });
		await this.refresh();
	}

	// Reads everything after the cursor; one read at a time, so polls and posts never double up.
	refresh() {
		return (this.#loading ??= this.#read().finally(() => (this.#loading = undefined)));
	}

	async #read() {
		const { channel } = this.state;
		let { list, cursor } = this.state.messages;
		this.emit({ messages: { ...this.state.messages, status: list.length ? "loaded" : "loading", error: null } });
		try {
			for (let more = true; more; ) {
				const read = await this.#run("read", { channel, cursor, limit: KEEP });
				({ cursor, more } = read);
				list = [...list, ...read.messages].slice(-KEEP);
			}
			this.emit({ messages: { status: "loaded", list, cursor, error: null } });
		} catch (error) {
			this.emit({ messages: { ...this.state.messages, status: "error", error: error.message } });
		}
	}

	async post(text) {
		this.emit({ post: { status: "busy", error: null } });
		try {
			await this.#run("post", { channel: this.state.channel, text });
		} catch (error) {
			return this.emit({ post: { status: "error", error: error.message } });
		}
		this.emit({ post: { status: "done", error: null } });
		await this.refresh();
	}

	// The run-module host call as this session (atlibs/run-module/host-call.js): chat.atirc.* goes
	// through the PDS with atproto-proxy, over the session's fetchHandler (DPoP, token refresh).
	async #run(entry, input) {
		const agent = this.#agent;
		const call = hostCall({ ctx: this.#ctx, xrpc: sessionXrpc(agent), signIn: "sign in" });
		const ran = await runModule(this.#ctx, await this.#module, entry, { ...input, did: agent.did, handle: this.state.handle }, call, { provides: [] });
		if (ran.error !== undefined) throw new Error(ran.error);
		return ran.result;
	}
}