// atirc channels from a browser (or Node with an RTCPeerConnection), over WebRTC datachannels speaking atirc's
// `atirc/mesh/1` channel lines to `atirc mesh run` nodes. No atirc server, no token. Two ways in:
//
// - guests: your own node on a VM you rent (`atirc mesh run --local-listen 127.0.0.1:P`), through the contract's
//   WebRTC tunnel {tag: "atirc", port: P}: signalled by the guest agent (contract events), so only you reach it.
// - mesh: your account's nodes by their computer.socialweb.peer records (exact-key record signalling). The
//   account's own node (identity = its handle) is a bridge; without it, each WebRTC peer of the channel.
//
// A bridge (either way) carries the whole channel: every line attributed, member lists, channels joined on
// demand, your lines posted as that node.
import { MESH_LABEL, PEER, WebrtcHost, peerReceipt, recordSignals } from "@publicdomainrelay/atlibs-webrtc";

// atirc's channel for a handle or identity: `#` + lowercase, `.` `,` NUL and whitespace as `_`, 63 chars.
export function channelOf(name) {
	const channel = `#${[...String(name)].slice(0, 63).map((c) => (/[.,\0\s]/.test(c) ? "_" : c)).join("")}`.toLowerCase();
	return channel === "#" ? "#atirc" : channel;
}

export function normalizeChannel(channel) {
	const c = String(channel).trim().toLowerCase();
	return c.startsWith("#") ? c : `#${c}`;
}

export const GUEST_TAG = "atirc";
const keyOf = ({ did, subhandle }) => `${did} ${subhandle}`;
const nickOf = (identity) => String(identity ?? "").split(".")[0];
const RETRY_MS = 5000;
const MAX_RETRY_MS = 60000;
const BRIDGE_WAIT_MS = 30000;

// One channel stream over a datachannel subscription: hello, the node's reply, then JSON lines.
export async function channelOver(ch, { channel, since = 0, seenEpoch = 0, epoch = Date.now() }) {
	const iterator = lines(ch);
	await ch.send(`${JSON.stringify({ hello: 1, kind: "channel", channel, since, seen_epoch: seenEpoch, epoch })}\n`);
	const first = await iterator.next();
	if (first.done) throw new Error("the node closed the channel during hello");
	const reply = JSON.parse(first.value);
	if (reply.event !== "hello") throw new Error(`the node refused: ${reply.message ?? first.value}`);
	let seq = 0;
	return {
		reply,
		send: (text) => ch.send(`${JSON.stringify({ seq: ++seq, text, ts: new Date().toISOString() })}\n`),
		messages: (async function* () {
			for await (const line of iterator) yield JSON.parse(line);
		})(),
		close: () => ch.close(),
	};
}

async function* lines(ch) {
	let buffer = "";
	for await (const chunk of ch) {
		buffer += chunk;
		let at;
		while ((at = buffer.indexOf("\n")) >= 0) {
			const line = buffer.slice(0, at);
			buffer = buffer.slice(at + 1);
			if (line.trim()) yield line;
		}
	}
}

// guests: [{ receipt: {uri, cid}, events, signal, tag? }]: your VMs' nodes, newest first (events/signal: the
//   requester's watch and signal for that receipt, as atlibs-webrtc's WebrtcHost takes them)
// repo: { did, getRecord({repo, collection, rkey}) -> {value, cid} | null, putRecord, deleteRecord } for mesh
//   peers (sessionRepo makes one from an OAuth session); handle: the signed-in account's handle
// subhandle: this page's identity on mesh peers (default web.<handle>); owner: { did, handle } of the mesh owner
// bridge: { did, subhandle } or false; peers: [{ did, subhandle }] to dial directly
// age: the age-encryption module; makePeer: config -> RTCPeerConnection
export function atirc({
	guests = [],
	repo,
	handle,
	subhandle = handle && `web.${handle}`,
	owner = repo && { did: repo.did, handle },
	bridge = owner && { did: owner.did, subhandle: owner.handle },
	peers = [],
	age,
	makePeer = (config) => new RTCPeerConnection(config),
	host = (options) => new WebrtcHost(options),
	retryMs = RETRY_MS,
	bridgeWaitMs = BRIDGE_WAIT_MS,
} = {}) {
	if (!guests.length && (!repo?.did || !handle)) throw new Error("atirc needs guests, or repo (with did) and handle");
	const channels = new Map();
	const messageListeners = new Set();
	const presenceListeners = new Set();
	const emit = (listeners, event) => listeners.forEach((cb) => cb(event));
	const known = new Map(peers.map((p) => [keyOf(p), p]));
	const presence = (name, event, peer) => emit(presenceListeners, { channel: name, event, did: peer.did, identity: peer.subhandle });

	// Where a link goes: a guest's tunnel, or a mesh peer's record-signalled datachannel.
	const guestTarget = (g) => ({
		key: `guest ${g.receipt.uri}`,
		async connect(name, seen) {
			const webrtc = host({ events: g.events, signal: g.signal, makePeer, age });
			const ch = await webrtc.call({ subscribe: true, params: { receipt: g.receipt, tag: g.tag ?? GUEST_TAG } }).catch((e) => (webrtc.close?.(), Promise.reject(e)));
			return { webrtc, mesh: await channelOver(ch, { channel: name, ...seen }).catch((e) => (webrtc.close?.(), Promise.reject(e))) };
		},
	});
	const meshTarget = (peer) => ({
		key: keyOf(peer),
		peer,
		async connect(name, seen) {
			const signals = recordSignals({ repo, subhandle });
			const webrtc = host({ events: signals.events, signal: signals.signal, makePeer, age });
			try {
				const receipt = await peerReceipt(repo, peer.did, peer.subhandle);
				const ch = await webrtc.call({ subscribe: true, params: { receipt, tag: MESH_LABEL } });
				return { webrtc, mesh: await channelOver(ch, { channel: name, ...seen }) };
			} catch (error) {
				webrtc.close?.();
				throw error;
			}
		},
	});

	// A mesh peer we may dial for a channel: its record names a WebRTC answerer and carries the channel.
	async function dialable(peer, name) {
		const found = await repo.getRecord({ repo: peer.did, collection: PEER, rkey: peer.subhandle }).catch(() => null);
		const value = found?.value;
		return typeof value?.webrtc === "string" && Array.isArray(value.channels) && value.channels.includes(name);
	}

	async function open(name, ch, target) {
		const { key } = target;
		if (ch.links.has(key) || ch.dialing.has(key)) return ch.links.get(key);
		ch.dialing.add(key);
		try {
			const seen = ch.seen.get(key) ?? { since: 0, epoch: 0 };
			const { webrtc, mesh } = await target.connect(name, { since: seen.since, seenEpoch: seen.epoch });
			const peer = { did: mesh.reply.did ?? target.peer?.did, subhandle: mesh.reply.subhandle ?? target.peer?.subhandle };
			const link = { target, peer, bridge: mesh.reply.bridge === true, epoch: mesh.reply.epoch ?? 0, mesh, webrtc, members: new Map() };
			if (!ch.joined || (ch.bridge && !link.bridge)) {
				mesh.close();
				webrtc.close?.();
				return null;
			}
			ch.links.set(key, link);
			ch.retry = retryMs;
			presence(name, "join", peer);
			if (link.bridge) {
				ch.bridge = link;
				for (const [other, direct] of ch.links) if (other !== key) drop(name, ch, other, direct);
			} else {
				for (const p of mesh.reply.peers ?? []) if (p?.did && p?.subhandle) known.set(keyOf(p), { did: p.did, subhandle: p.subhandle });
				if (!ch.bridge && !guests.length) dialKnown(name, ch);
			}
			pump(name, ch, key, link);
			return link;
		} finally {
			ch.dialing.delete(key);
		}
	}

	function drop(name, ch, key, link) {
		ch.links.delete(key);
		link.mesh.close();
		link.webrtc.close?.();
		for (const member of link.members.values()) presence(name, "leave", member);
		presence(name, "leave", link.peer);
	}

	async function pump(name, ch, key, link) {
		try {
			for await (const line of link.mesh.messages) receive(name, ch, key, link, line);
		} catch {
			// a closed or broken channel ends like a clean close
		}
		if (ch.links.get(key) !== link) return;
		drop(name, ch, key, link);
		if (ch.bridge === link) ch.bridge = null;
		retry(name, ch, link.target);
	}

	// Redial with backoff while joined: the same target while others stay linked, else the whole plan.
	function retry(name, ch, target) {
		if (!ch.joined) return;
		const wait = ch.retry;
		ch.retry = Math.min(ch.retry * 2, MAX_RETRY_MS);
		setTimeout(async () => {
			if (!ch.joined) return;
			const again = ch.links.size ? open(name, ch, target) : plan(name, ch);
			if (!(await again.then(() => true, () => false)) && !ch.links.size) retry(name, ch, target);
		}, wait);
	}

	function receive(name, ch, key, link, line) {
		if (link.bridge && Array.isArray(line.members)) {
			const now = new Map(line.members.filter((m) => m?.did && m?.subhandle).map((m) => [keyOf(m), { did: m.did, subhandle: m.subhandle }]));
			for (const [k, m] of link.members) if (!now.has(k)) presence(name, "leave", m);
			for (const [k, m] of now) if (!link.members.has(k)) presence(name, "join", m);
			link.members = now;
			return;
		}
		if (typeof line.seq !== "number" || typeof line.text !== "string") return;
		const seen = ch.seen.get(key);
		if (seen?.epoch === link.epoch && line.seq <= seen.since) return;
		ch.seen.set(key, { since: line.seq, epoch: link.epoch });
		// Only a bridge speaks for others; a direct peer's lines are its own.
		const author = link.bridge && line.did && line.subhandle ? { did: line.did, subhandle: line.subhandle } : link.peer;
		emit(messageListeners, { channel: name, identity: author.subhandle, nick: nickOf(author.subhandle), did: author.did, text: line.text, ts: line.ts, seq: line.seq, via: link.peer.subhandle });
	}

	function dialKnown(name, ch) {
		for (const peer of known.values()) {
			if (peer.subhandle === subhandle && peer.did === repo.did) continue;
			if (ch.links.has(keyOf(peer)) || ch.dialing.has(keyOf(peer))) continue;
			dialable(peer, name).then((ok) => ok && !ch.bridge && open(name, ch, meshTarget(peer))).catch(() => {});
		}
	}

	const opened = (target, name, ch) => open(name, ch, target).then((l) => l ?? Promise.reject(new Error("closed")));

	// Guests in order (the first that opens); else the mesh bridge first, then every WebRTC peer that carries
	// the channel once the bridge fails or takes over bridgeWaitMs (a bridge that opens later takes over).
	async function plan(name, ch) {
		if (guests.length) {
			const errors = [];
			for (const g of guests) {
				try {
					return void (await opened(guestTarget(g), name, ch));
				} catch (error) {
					errors.push(error.message);
				}
			}
			throw new Error(`no guest node opened ${name}: ${errors.join("; ")}`);
		}
		const attempts = [];
		if (bridge) {
			const viaBridge = opened(meshTarget(bridge), name, ch);
			attempts.push(viaBridge);
			const link = await Promise.race([viaBridge.catch(() => null), new Promise((r) => setTimeout(r, bridgeWaitMs))]);
			if (link?.bridge) return;
		}
		for (const peer of known.values()) {
			if (keyOf(peer) === keyOf(bridge || {}) || !(await dialable(peer, name))) continue;
			attempts.push(opened(meshTarget(peer), name, ch));
		}
		if (!attempts.length) throw new Error(`no peer of ${name} takes WebRTC`);
		await Promise.any(attempts).catch((error) => {
			if (!ch.links.size) throw new Error(`no peer of ${name} answered: ${error.errors.map((e) => e.message).join("; ")}`);
		});
	}

	// Who this page posts as on a channel: its bridge node, else itself.
	const meOn = (ch) => (ch?.bridge ? ch.bridge.peer : { did: repo?.did, subhandle });

	const api = {
		// Join a channel; resolves once one link is open (history replays first).
		join(channel) {
			const name = normalizeChannel(channel);
			let ch = channels.get(name);
			if (ch) return ch.planning;
			channels.set(name, (ch = { links: new Map(), dialing: new Set(), seen: new Map(), bridge: null, joined: true, retry: retryMs }));
			ch.planning = plan(name, ch).catch((error) => {
				if (!ch.links.size) {
					ch.joined = false;
					if (channels.get(name) === ch) channels.delete(name);
				}
				throw error;
			});
			return ch.planning;
		},
		leave(channel) {
			const name = normalizeChannel(channel);
			const ch = channels.get(name);
			if (!ch) return;
			ch.joined = false;
			channels.delete(name);
			for (const [key, link] of ch.links) drop(name, ch, key, link);
		},
		// Post on every link of the channel; the page's own line is emitted too (self: true).
		async post(channel, text) {
			const name = normalizeChannel(channel);
			const ch = channels.get(name);
			const links = [...(ch?.links.values() ?? [])];
			if (!links.length) throw new Error(`not in ${name}`);
			await Promise.all(links.map((link) => link.mesh.send(String(text))));
			const me = meOn(ch);
			emit(messageListeners, { channel: name, identity: me.subhandle, nick: nickOf(me.subhandle), did: me.did, text: String(text), ts: new Date().toISOString(), self: true });
		},
		onMessage(cb) {
			messageListeners.add(cb);
			return () => messageListeners.delete(cb);
		},
		onPresence(cb) {
			presenceListeners.add(cb);
			return () => presenceListeners.delete(cb);
		},
		// { me, channels, peers: [{did, subhandle}] }: the linked nodes and their members, on one channel or all.
		async peers(channel) {
			const names = channel ? [normalizeChannel(channel)] : [...channels.keys()];
			const out = new Map();
			let me;
			for (const name of names) {
				const ch = channels.get(name);
				me ??= ch?.bridge?.peer.subhandle;
				for (const link of ch?.links.values() ?? []) {
					out.set(keyOf(link.peer), link.peer);
					for (const [k, m] of link.members) out.set(k, m);
				}
			}
			me ??= subhandle;
			for (const [k, p] of out) if (p.subhandle === me) out.delete(k);
			return { me, channels: [...channels.keys()], peers: [...out.values()] };
		},
		close() {
			for (const name of [...channels.keys()]) api.leave(name);
		},
	};
	return api;
}

// A repo for atirc() from an OAuth session (@atproto/oauth-client-browser's OAuthSession: did + fetchHandler):
// writes go through the session, reads are public getRecords at the record's PDS (did:plc or did:web).
export function sessionRepo(session, { fetch = globalThis.fetch, plc = "https://plc.directory" } = {}) {
	const did = session.did ?? session.sub;
	const pds = new Map();
	const pdsOf = (repo) => {
		if (!pds.has(repo)) {
			const url = repo.startsWith("did:web:") ? `https://${repo.slice(8).replaceAll(":", "/")}/.well-known/did.json` : `${plc}/${repo}`;
			const found = fetch(url)
				.then((r) => (r.ok ? r.json() : Promise.reject(new Error(`${repo}: DID document ${r.status}`))))
				.then((doc) => {
					const endpoint = doc.service?.find((s) => s.id === "#atproto_pds" || s.id === `${repo}#atproto_pds`)?.serviceEndpoint;
					if (!endpoint) throw new Error(`${repo} names no PDS`);
					return endpoint;
				});
			found.catch(() => pds.delete(repo));
			pds.set(repo, found);
		}
		return pds.get(repo);
	};
	const write = async (method, body) => {
		const response = await session.fetchHandler(`/xrpc/com.atproto.repo.${method}`, {
			method: "POST",
			headers: { "content-type": "application/json" },
			body: JSON.stringify({ repo: did, ...body }),
		});
		if (!response.ok) throw new Error(`${method} failed with ${response.status}: ${await response.text()}`);
		return response.json();
	};
	return {
		did,
		async getRecord({ repo, collection, rkey }) {
			const url = new URL("/xrpc/com.atproto.repo.getRecord", await pdsOf(repo));
			url.search = new URLSearchParams({ repo, collection, rkey });
			const response = await fetch(url);
			if (response.status === 400 || response.status === 404) return null;
			if (!response.ok) throw new Error(`getRecord ${collection}/${rkey} failed with ${response.status}`);
			return response.json();
		},
		putRecord: (input) => write("putRecord", input),
		deleteRecord: (input) => write("deleteRecord", input),
	};
}