import { test } from "@playwright/test";

import { serviceDo, getOk, refsOf } from "./api";
import { attach, fresh, note, within } from "./record";
import { checkGit } from "./cobgit";
import { config, type Account } from "./config";
import { lsRemote } from "./git";
import { knotFirehose, pdsFirehose, type Firehose } from "./firehose";
import type { PocketPreferences } from "./lexicons";
import { get, partsOf, recordUrl, until, xrpc, type Reply, type Written } from "./net";
import { checkStream, streamCarries, watcher } from "./stream";
import type { Held } from "./tabs";
import { checkVersions, isCob, versionsBefore, type Version } from "./versions";

// read while the write is still held or a hop cannot tell what it changed
export interface Before {
	versions?: Version[];
	mark?: number;
	knotMark?: number;
	pdsMark?: number;
	pdsExisted?: boolean;
	released?: number;
}

export interface Settling {
	how: Held["method"] | "settled";
	before?: Before;
	// whose notices get checked. with nobody here an "absent" check can't fail, so it proves nothing
	people?: Account[];
}

export interface Hop {
	name: string;
	present: (written: Written, who: Account, settling?: Settling) => Promise<unknown>;
	absent: (written: Pick<Written, "uri">, who?: Account, settling?: Settling) => Promise<unknown>;
	look?: (uri: string | undefined) => Promise<Before>;
}

// creates have no rkey yet, the knot mints it, so there is no uri to read
export const lookBefore = async (hops: Hop[], { body }: Held): Promise<Before> => {
	const uri = body.rkey ? `at://${body.repo}/${body.collection}/${body.rkey}` : undefined;
	return Object.assign({}, ...(await Promise.all(hops.map((hop) => hop.look?.(uri)))));
};

// one step per check so the report shows how long after the write each service had it
// the tidy replay runs outside a test, so test.info throws there
const step = <T>(title: string, check: () => Promise<T>): Promise<T> => {
	try {
		test.info();
	} catch {
		return check();
	}
	return test.step(title, async (info) => {
		const record = fresh();
		try {
			return await within(record, check);
		} finally {
			await attach(info, record);
		}
	});
};

// one hop is one row, so a failing sub-check is named on it
// the hop's own check keeps its plain message
const alongside = async <T>(
	own: Promise<T>,
	checks: Record<string, Promise<unknown> | undefined>
) => {
	const named = Object.entries(checks).filter(([, check]) => check);
	const [first, ...rest] = await Promise.allSettled([own, ...named.map(([, check]) => check)]);
	const failed = [first, ...rest].flatMap((check, at) =>
		check.status === "rejected"
			? [`${at ? `${named[at - 1][0]}: ` : ""}${(check.reason as Error).message}`]
			: []
	);
	if (failed.length) throw new Error(failed.join("\n      "));
	return (first as PromiseFulfilledResult<T>).value;
};

const recordHop = (name: string, base: string) => {
	const has = ({ uri, cid }: Written) =>
		until(`${name} should have ${uri} at ${cid}`, 10_000, async () => {
			const { status, body } = await get(recordUrl(base, uri));
			const saw =
				status === 200
					? `it has ${body.cid}`
					: `it answered ${status} ${JSON.stringify(body)}`;
			return [status === 200 && body.cid === cid, saw, body.value];
		});
	const lacks = ({ uri }: Pick<Written, "uri">) =>
		until(`${name} should have dropped ${uri}`, 10_000, async () => {
			const { status, body } = await get(recordUrl(base, uri));
			// down or erroring is not the same as gone, only a 404 counts
			const gone = status === 404 || body?.error === "RecordNotFound";
			return [
				gone,
				status === 200 ? "it still has it" : `it answered ${status} ${JSON.stringify(body)}`
			];
		});
	return { name, host: new URL(base).host, has, lacks, present: has, absent: lacks };
};

const firehosed = (
	watcher: Firehose,
	written: { uri: string; cid?: string },
	settling?: Settling,
	since?: number,
	existed = true
) =>
	step("firehose", async () => {
		const how = settling?.how;
		const action =
			how === "put" ? (existed ? "update" : "create") : how === "settled" ? undefined : how;
		const frame = await watcher.check(written, action, since);
		const released = settling?.before?.released;
		if (released) note("arrived", `${frame.at - released} ms after release`);
	});

const pdsRecords = recordHop("pds", config.pdsUrl);
export const pds = {
	...pdsRecords,
	present: (written: Written, _who?: Account, settling?: Settling) =>
		alongside(pdsRecords.has(written), {
			firehose: firehosed(
				pdsFirehose,
				written,
				settling,
				settling?.before?.pdsMark,
				settling?.before?.pdsExisted
			)
		}),
	absent: ({ uri }: Pick<Written, "uri">, _who?: Account, settling?: Settling) =>
		alongside(pdsRecords.lacks({ uri }), {
			firehose: firehosed(pdsFirehose, { uri }, settling, settling?.before?.pdsMark)
		}),
	look: async (uri: string | undefined): Promise<Before> => {
		const pdsMark = pdsFirehose.mark();
		if (!uri) return { pdsMark };
		// a pds put can create too, so its method alone cannot tell the action
		const { status, body } = await get(recordUrl(config.pdsUrl, uri));
		if (status !== 200 && status !== 404 && body?.error !== "RecordNotFound")
			throw new Error(
				`pds should say whether ${uri} exists before the write but answered ${status} ${JSON.stringify(body)}`
			);
		return { pdsMark, pdsExisted: status === 200 };
	}
};

const hydrantDb = recordHop("hydrant db", config.hydrantUrl);
const streamed = ({ uri, cid }: { uri: string; cid?: string }, settling?: Settling) =>
	streamCarries(uri)
		? step("stream", async () => {
				const { mark, released } = settling?.before ?? {};
				const frame = await checkStream({ uri, cid }, settling?.how ?? "settled", mark);
				// checkStream only notices a frame on its next wake, so the arrival time comes from the watcher
				if (released && frame) note("arrived", `${frame.at - released} ms after release`);
			})
		: undefined;
export const hydrant = {
	...hydrantDb,
	name: "hydrant",
	present: (written: Written, _who?: Account, settling?: Settling) =>
		alongside(
			step("db", () => hydrantDb.has(written)),
			{ stream: streamed(written, settling) }
		),
	absent: ({ uri }: Pick<Written, "uri">, _who?: Account, settling?: Settling) =>
		alongside(
			step("db", () => hydrantDb.lacks({ uri })),
			{ stream: streamed({ uri }, settling) }
		),
	look: async () => ({ mark: watcher.mark() })
};

const knotRecords = recordHop("knot", config.knotUrl);
// a cob's uri authority is the repo, so the author lives in the record body
const editorIs = async (written: Written, did: string) => {
	const value = await knotRecords.has(written);
	if (value["x-tngl-editor"] !== did) {
		throw new Error(
			`knot should show ${did} as the editor of ${written.uri} but x-tngl-editor is ${value["x-tngl-editor"]} (use knotCopy for a record the knot wrote itself)`
		);
	}
};

// one listCobVersions read feeds both checks, no cid means the record was deleted
const cobChecks = (
	written: { uri: string; cid?: string },
	editor: string | undefined,
	settling?: Settling
) => {
	const firehose =
		isCob(written.uri) || partsOf(written.uri).collection === "org.tangled.git.refs"
			? firehosed(knotFirehose, written, settling, settling?.before?.knotMark)
			: undefined;
	if (!isCob(written.uri)) return { firehose };
	const chain = step("versions", () =>
		checkVersions(written, { editor, had: settling?.before?.versions })
	);
	return {
		firehose,
		versions: chain,
		// a chain that never settles fails as versions already, git would only repeat it
		// git's step starts once the chain is in, so its bar is only git's time
		git: chain.then(
			(versions) => step("git", () => checkGit(written, versions)),
			() => undefined
		)
	};
};
const versionsLook = async (uri: string | undefined) => ({
	knotMark: knotFirehose.mark(),
	versions: await versionsBefore(uri)
});

const knotHop = (checksEditor: boolean): Hop => ({
	name: "knot",
	present: (written, who, settling) => {
		const editor = checksEditor ? who.did : undefined;
		return alongside(
			step("record", () => (editor ? editorIs(written, editor) : knotRecords.has(written))),
			cobChecks(written, editor, settling)
		);
	},
	absent: ({ uri }, who, settling) =>
		alongside(
			step("record", () => knotRecords.lacks({ uri })),
			cobChecks({ uri }, checksEditor ? who?.did : undefined, settling)
		),
	look: versionsLook
});

// the knot wrote this one itself, so there is no x-tngl-editor to check
export const knotCopy = knotHop(false);

export const knot = { ...knotRecords, ...knotHop(true) };

// the knot rewrites the refs record a moment after a push, so wait for it
export const refsNaming = (
	uri: string,
	what: string,
	wanted: (named: [sha: string, ref: string][]) => boolean
) =>
	until(what, 10_000, async () => {
		const now = await refsOf(uri);
		if (!now) return [false, "it has no refs record"];
		const said = now.named.map(([sha, ref]) => `${ref} ${sha}`).join(", ");
		return [wanted(now.named), `it names ${said}`, now];
	});

export const git: Hop = {
	name: "git ls-remote",
	present: (written, who) =>
		until(`git should list exactly the refs ${written.uri} names`, 10_000, async () => {
			const stored = await refsOf(written.uri);
			if (stored?.cid !== written.cid)
				return [false, `the knot has ${stored?.cid ?? "no record"}`];
			const named = Object.fromEntries(stored.named.map(([sha, ref]) => [ref, sha]));
			if (Object.keys(named).length === 0) return [false, "the record names no refs"];
			// git lists HEAD, the refs record never does
			const listed = await lsRemote(who, partsOf(written.uri).repo, "refs/");
			return [
				JSON.stringify(Object.entries(named).sort()) ===
					JSON.stringify(Object.entries(listed).sort()),
				`git lists ${JSON.stringify(listed)}, the record names ${JSON.stringify(named)}`
			];
		}),
	absent: async () => {
		throw new Error("git has no refs record to lose");
	}
};

const awaitRecord = async ({ uri, cid }: { uri: string; cid?: string }) =>
	(
		await get(
			xrpc(config.bobbinUrl, "org.tangled.bobbin.awaitRecord", {
				uri,
				...(cid ? { cid } : {})
			})
		)
	).body;

const said = (body: { status?: string; reason?: string; detail?: string }) =>
	body.status
		? `it said ${body.status} ${body.reason ?? ""} ${body.detail ?? ""}`
		: JSON.stringify(body);

// parallel workers can push pds to hydrant to bobbin past 15 s, hence 30 s
const indexed = (written: Written) =>
	until(
		`bobbin should have indexed ${written.uri} at ${written.cid}`,
		30_000,
		async () => {
			const body = await awaitRecord(written);
			// a rejected record never gets indexed, so throw instead of polling
			if (body.status === "rejected")
				throw new Error(`bobbin should have indexed ${written.uri} but ${said(body)}`);
			return [body.status === "indexed" && body.cid === written.cid, said(body)];
		},
		{ longPoll: true }
	);

const deleted = ({ uri }: Pick<Written, "uri">) =>
	until(
		`bobbin should have deleted ${uri}`,
		15_000,
		async () => {
			const body = await awaitRecord({ uri });
			return [body.status === "deleted", said(body)];
		},
		{ longPoll: true }
	);

const mirrorList = async (kind: "Branches" | "Tags", repoDid: string) => {
	const { status, body } = await get(
		xrpc(config.bobbinUrl, `org.tangled.temp.git.list${kind}`, { repo: repoDid, limit: "100" })
	);
	const names: string[] = (body?.[kind.toLowerCase()] ?? []).map((one: any) => one.name);
	return { status, names };
};

export const bobbin = {
	name: "bobbin",
	indexed,
	list: mirrorList,
	present: indexed,
	absent: deleted,
	// a stall here means knot2 is rate limiting hydrant's backfill
	repoByName: (owner: string, name: string) =>
		get(xrpc(config.bobbinUrl, "org.tangled.temp.repo.getRepoByName", { owner, name })),
	hasRepo: (ownerDid: string, name: string) =>
		until(
			`bobbin should know the repo ${name} (hydrant may be backing off knot2)`,
			30_000,
			async () => {
				const { status } = await bobbin.repoByName(ownerDid, name);
				return [status === 200, `getRepoByName answered ${status}`];
			}
		),
	// the mirror 404s an unsynced repo, even an empty one, and the web reads that as a knot needing an upgrade
	knowsRepo: (repoDid: string) =>
		until(`bobbin should know ${repoDid}`, 60_000, async () => {
			const { status } = await mirrorList("Branches", repoDid);
			return [status === 200, `listBranches answered ${status}`];
		}),
	// the mirror picks a push up a moment after the knot, so pages that read git wait for this first
	hasBranch: (repoDid: string, branch: string, sha: string) =>
		until(`bobbin should have ${branch} at ${sha}`, 60_000, async () => {
			const { status, body } = await get(
				xrpc(config.bobbinUrl, "org.tangled.temp.git.getRef", {
					repo: repoDid,
					ref: `refs/heads/${branch}`
				})
			);
			return [
				status === 200 && body.oid === sha,
				`getRef answered ${status} ${JSON.stringify(body)}`
			];
		}),
	countStars: async (subject: string) => {
		const body = await getOk(
			xrpc(config.bobbinUrl, "org.tangled.feed.countStars", { subject })
		);
		return body.count as number;
	},
	listForks: (subject: string) =>
		getOk(xrpc(config.bobbinUrl, "org.tangled.repo.listForks", { subject, limit: "50" })),
	countForks: async (subject: string) => {
		const body = await getOk(
			xrpc(config.bobbinUrl, "org.tangled.repo.countForks", { subject })
		);
		return body.count as number;
	}
};

interface Workflow {
	name?: string;
	id: string;
	status: string;
	error?: string;
}
const FINISHED = ["success", "failed", "cancelled", "timeout"];

const finishedRun = (repoDid: string, sha: string) =>
	until<{ pipeline: string; workflow: Workflow }>(
		"spindle should finish the workflow",
		120_000,
		async () => {
			const { status, body } = await get(
				xrpc(config.spindleUrl, "org.tangled.ci.queryPipelines", {
					repo: repoDid,
					commits: sha
				})
			);
			const runs: { pipeline: string; workflow: Workflow }[] = (
				body?.pipelines ?? []
			).flatMap((pipeline: { id: string; workflows?: Workflow[] }) =>
				(pipeline.workflows ?? []).map((workflow) => ({ pipeline: pipeline.id, workflow }))
			);
			const done = runs.find((one) => FINISHED.includes(one.workflow.status));
			return [
				Boolean(done),
				`it answered ${status} with ${JSON.stringify(runs.map((one) => one.workflow))}`,
				done
			];
		}
	);

// triggers look like org.tangled.event.push, so match on the suffix
export const spindle = {
	hasPipeline: (repoDid: string, sha: string, event: "push" | "pullRequest", budget = 30_000) =>
		until(
			`spindle should have a ${event} pipeline for ${repoDid} at ${sha}`,
			budget,
			async () => {
				const { status, body } = await get(
					xrpc(config.spindleUrl, "org.tangled.ci.queryPipelines", {
						repo: repoDid,
						commits: sha
					})
				);
				const pipelines: { commit: string; trigger?: { $type?: string } }[] =
					body.pipelines ?? [];
				const found = pipelines.find(
					(p) => p.commit === sha && p.trigger?.$type?.endsWith(`.event.${event}`)
				);
				const saw =
					status === 200
						? `it listed ${pipelines.length} pipelines for the commit`
						: `it answered ${status}`;
				return [Boolean(found), saw, found];
			}
		),
	finishedRun,
	finishedWorkflow: async (repoDid: string, sha: string) =>
		(await finishedRun(repoDid, sha)).workflow
};

export const settle = async (
	written: Written,
	hops: Hop[],
	who: Account,
	nsid: string,
	how: Settling["how"],
	before?: Before,
	people?: Account[]
) => {
	const settling = { how, before, people };
	const checks = await Promise.allSettled(
		hops.map((hop) =>
			step(hop.name, () =>
				how === "delete"
					? hop.absent(written, who, settling)
					: hop.present(written, who, settling)
			)
		)
	);
	if (checks.every((check) => check.status === "fulfilled")) return;
	const table = checks.map((check, at) =>
		check.status === "fulfilled"
			? `  ✓ ${hops[at].name}`
			: `  ✗ ${hops[at].name}  ${(check.reason as Error).message}`
	);
	throw new Error([`${nsid} ${written.uri} ${written.cid} (${how})`, ...table].join("\n"));
};

// pocket keys the saved lists by the token's aud
const POCKET_AUD = "did:web:tangled.org";
const KNOT_PREF = "org.tangled.temp.actor.defs#savedKnotPref";
const SPINDLE_PREF = "org.tangled.temp.actor.defs#savedSpindlePref";
const RECENTS_PREF = "org.tangled.temp.actor.defs#recentsPref";

export interface Saved {
	knots: string[];
	spindles: string[];
}

type Stored = PocketPreferences["preferences"];

const storedBy = async (who: Account): Promise<Stored> => {
	const lxm = "com.bad-example.pocket.getPreferences";
	const reply = await serviceDo(who, config.bobbinUrl, POCKET_AUD, lxm);
	// a 400 NotFound means the account never saved anything
	if (reply.status === 400 && reply.body?.error === "NotFound") return [];
	if (reply.status !== 200) throw new Error(`pocket answered ${JSON.stringify(reply.body)}`);
	return reply.body.preferences;
};

export const savedBy = async (who: Account): Promise<Saved> => {
	const stored = await storedBy(who);
	const hosts = (items: { value: string }[]) => items.map((item) => item.value).sort();
	return {
		knots: hosts(stored.flatMap((pref) => (pref.$type === KNOT_PREF ? pref.items : []))),
		spindles: hosts(stored.flatMap((pref) => (pref.$type === SPINDLE_PREF ? pref.items : [])))
	};
};

export const recentsIn = async (who: Account) =>
	(await storedBy(who)).flatMap((pref) => (pref.$type === RECENTS_PREF ? pref.items : []));

// a single try, saveFor is the one that retries
export const putSaved = async (who: Account, saved: Saved): Promise<Reply> => {
	const lxm = "com.bad-example.pocket.putPreferences";
	const items = (hosts: string[]) => hosts.map((host) => ({ id: host, value: host }));
	return serviceDo(who, config.bobbinUrl, POCKET_AUD, lxm, {
		json: {
			preferences: [
				{ $type: KNOT_PREF, items: items(saved.knots) },
				{ $type: SPINDLE_PREF, items: items(saved.spindles) }
			]
		} satisfies PocketPreferences
	});
};

// pocket can't resolve a new account's did doc for a moment
export const saveFor = async (who: Account, saved: Saved) =>
	until(`pocket should take the save for ${who.handle}`, 15_000, async () => {
		const reply = await putSaved(who, saved);
		return [reply.status === 200, `it answered ${reply.status} ${JSON.stringify(reply.body)}`];
	});

// deliberi only answers the recipient, not other accounts
const deliberiGet = async (who: Account, lxm: string, params?: Record<string, string>) => {
	const aud = `did:web:${new URL(config.deliberiUrl).host}`;
	const reply = await serviceDo(who, config.deliberiUrl, aud, lxm, { params });
	if (reply.status !== 200) throw new Error(`deliberi answered ${JSON.stringify(reply.body)}`);
	return reply.body;
};

export type Toggles = Record<string, boolean>;

export const notificationsOf = (who: Account): Promise<Toggles> =>
	deliberiGet(who, "org.tangled.temp.notification.getPreferences");

export interface Notice {
	uri: string;
	type: string;
	read: boolean;
	actorDid: string;
	repoDid?: string;
	issueAt?: string;
	pullAt?: string;
}

export const noticesOf = async (who: Account): Promise<Notice[]> =>
	(await deliberiGet(who, "org.tangled.temp.notification.listNotifications", { limit: "50" }))
		.notifications;

export const aboutOf = (notice: Notice) => notice.issueAt ?? notice.pullAt;

// a state change notice is the ticket uri with #rev on the end
const madeFrom = (notice: Notice, uri: string) =>
	notice.uri === uri || notice.uri.startsWith(`${uri}#`);

// deliberi only shows each person their own notices, so ask as each one
export const noticesMadeFrom = async (uri: string, people: Account[]) =>
	(
		await Promise.all(
			people.map(async (who) =>
				(await noticesOf(who))
					.filter((n) => madeFrom(n, uri))
					.map((notice) => ({ who, notice }))
			)
		)
	).flat();

// deliberi saves the rev with its notices, so once _ingested shows the rev the notices are written
const ingested = async (written: Written) => {
	// an edit keeps the uri, so only a rev at least as new as the write proves it landed
	const watcher = isCob(written.uri) ? knotFirehose : pdsFirehose;
	const { rev } = await watcher.check(written);
	return until(
		`deliberi should have ingested ${written.uri} at ${written.cid}`,
		20_000,
		async () => {
			const { status, body } = await get(
				`${config.deliberiUrl}/xrpc/_ingested?${new URLSearchParams({ uri: written.uri })}`
			);
			// a json 404 is just unseen, a plain text 404 means the route is missing
			if (status === 404 && typeof body === "string")
				throw new Error(
					`deliberi has no /xrpc/_ingested route, run it with DELIBERI_DEV=true, nothing else can say it processed ${written.uri}`
				);
			// only tickets carry a cid, so a missing one doesn't count against it
			const has =
				status === 200 &&
				body.rev >= rev &&
				(body.cid === undefined || body.cid === written.cid);
			return [
				has,
				`/_ingested answered ${status} ${JSON.stringify(body)}, the write is at rev ${rev}`
			];
		}
	);
};

const told = (found: { who: Account; notice: Notice }[]) =>
	found.map(({ who, notice }) => `${notice.type} to ${who.name}`);

export const deliberi: Hop = {
	name: "deliberi",
	present: async (written, _who, settling) => {
		await step("ingested", () => ingested(written));
		const found = await noticesMadeFrom(written.uri, settling?.people ?? []);
		note("deliberi made", told(found).join(", ") || "no notice");
	},
	absent: async ({ uri }, _who, settling) => {
		await until(`deliberi should have no notice made from ${uri}`, 15_000, async () => {
			const found = await noticesMadeFrom(uri, settling?.people ?? []);
			return [found.length === 0, `it still has ${told(found).join(", ")}`];
		});
	}
};

// the opener is the editor of the first version, read from the knot's git
export const knotmirror = {
	// the mirror lags the knot a moment, so this polls
	openerOf: (uri: string) =>
		until(`knotmirror should list who opened ${uri}`, 30_000, async () => {
			const { repo, collection, rkey } = partsOf(uri);
			const { status, body } = await get(
				xrpc(config.knotmirrorUrl, "org.tangled.temp.repo.listRecordOrigins", {
					repo,
					collection,
					limit: "1000"
				})
			);
			const found =
				status === 200
					? body.origins?.find((o: { rkey: string }) => o.rkey === rkey)
					: undefined;
			return [
				Boolean(found),
				`listRecordOrigins answered ${status} ${JSON.stringify(body)}`,
				found?.author as string
			];
		})
};

export const openers = {
	view: async (uri: string): Promise<string> =>
		(await getOk(xrpc(config.bobbinUrl, "org.tangled.track.getTicketView", { uri }))).author
			.did,
	listed: async (repoDid: string, list: "listIssues" | "listPulls", uri: string) => {
		const body = await getOk(
			xrpc(config.bobbinUrl, `org.tangled.repo.${list}`, { subject: repoDid, limit: "100" })
		);
		return body.items.find((item: { uri: string }) => item.uri === uri)?.author as
			string | undefined;
	},
	searched: async (repoDid: string, kind: "issue" | "pull", query: string) => {
		const body = await getOk(
			xrpc(config.bobbinUrl, "org.tangled.search.query", {
				repo: repoDid,
				nsid: "org.tangled.track.ticket",
				kind,
				...(query ? { q: query } : {}),
				limit: "100"
			})
		);
		return body.hits as { uri: string; author?: string }[];
	}
};