#!/usr/bin/env node
// renders svfe's repo pages for a knot-static generation. reads only the knot's public xrpc
// surface and writes static html per repo, swapped in atomically per repo:
//   site/<repo did>         -> store/<token>/<generation>, the pages of one repo
//   www/<owner>/<name>      -> site/<repo did>, the urls svfe links to (owner = handle or did)
//   state/<token>.json      what each page read, so a diff of the knot's reads names the
//                           pages to render again
//   www/_app/immutable      the client chunks of every build a served page was rendered by
//   svfe-pages <hook.json> --server <svfe build-server> [--client <svfe build>] --out <dir>
//              [--read <xrpc base>]
import { fork } from "node:child_process";
import { createHash } from "node:crypto";
import { readFileSync } from "node:fs";
import { rm } from "node:fs/promises";
import { availableParallelism } from "node:os";
import path from "node:path";
import { parseArgs } from "node:util";
import { budgetFrom, BUDGET_OPTIONS } from "./budget.mjs";
import { collectClients, keepClient } from "./client.mjs";
import { atomicJson, ownerPrefixOf, prefixOf, readJson, RENDERER_FORMAT, tokenOf, unpublishOwner } from "./common.mjs";
import { countsChanged, dirtyPages, readDiff, stale } from "./diff.mjs";

const { values: opts, positionals } = parseArgs({
	allowPositionals: true,
	options: {
		server: { type: "string" },
		// the client build next to the server one, as svfe's package lays them out
		client: { type: "string" },
		out: { type: "string" },
		read: { type: "string" },
		// pages in flight per repo process
		jobs: { type: "string", default: "6" },
		// repo processes, and the syntax highlighting threads each one runs
		workers: { type: "string", default: String(Math.max(1, Math.floor(availableParallelism() / 2))) },
		highlighters: { type: "string", default: "0" },
		"worker-max-repos": { type: "string", default: "12" },
		// past it a worker is replaced between repos: 360 under bun, 650 under node
		"worker-max-rss-mb": { type: "string" },
		// v8's old space per worker process, under node only
		"worker-heap-mb": { type: "string", default: "128" },
		full: { type: "boolean", default: false },
		...BUDGET_OPTIONS
	}
});

await coordinate();

async function coordinate() {
	const hookPath = path.resolve(positionals[0]);
	const hook = JSON.parse(readFileSync(hookPath, "utf8"));
	const out = path.resolve(opts.out);
	const stateDir = path.join(out, "state");
	const statusPath = hook.paths?.status ?? path.join(out, "status.json");
	const knotState = readJson(path.join(stateDir, "knot.json"));
	// pages load the client chunks of the build that rendered them, so a new svfe build
	// renders everything again
	const build = createHash("sha256").update(readFileSync(path.join(opts.server, "manifest-full.js"))).digest("hex").slice(0, 16);
	const loadState = (did) => readJson(path.join(stateDir, tokenOf(did) + ".json"));
	const budget = budgetFrom(opts, hook.repos.length);
	const t0 = performance.now();

	const status = {
		generation: hook.generation,
		state: "rendering",
		started_at: new Date().toISOString(),
		repos: {},
		owners: {}
	};
	// one write at a time, so the file ends as the newest status
	let statusWrite = Promise.resolve();
	const writeStatus = (extra = {}) => {
		const snapshot = JSON.parse(JSON.stringify({ ...status, ...extra }));
		statusWrite = statusWrite.catch(() => {}).then(() => atomicJson(statusPath, snapshot));
		return statusWrite;
	};

	// what each repo's and owner's pages were rendered at, saved as each one finishes, so the
	// work of a run that is superseded or killed counts for the next. an entry carries the
	// formats and build it was rendered with: one rendered by another build renders again
	const stamp = `${hook.format}/${RENDERER_FORMAT}/${build}`;
	const current = (entry) => (!opts.full && entry?.stamp === stamp ? entry : undefined);
	// the build whose pages an entry's url serves: a failed render leaves the pages before it
	const servedBy = (entry) => entry?.served ?? (entry?.failed ? undefined : entry?.stamp?.split("/").at(-1));
	const nextRepos = {};
	const nextOwners = {};
	let knotWrite = Promise.resolve();
	const saveKnot = () => {
		const snapshot = JSON.parse(JSON.stringify({ generation: hook.generation, repos: nextRepos, owners: nextOwners }));
		knotWrite = knotWrite.catch(() => {}).then(() => atomicJson(path.join(stateDir, "knot.json"), snapshot));
		return knotWrite;
	};

	const children = new Set();
	// the knot stops a renderer it supersedes with SIGTERM to the process group: the repo in
	// flight is dropped (its next render clears the half-written copy), the finished ones stay
	process.on("SIGTERM", async () => {
		for (const child of children) child.kill("SIGTERM");
		await Promise.all([knotWrite, writeStatus({ state: "superseded" })]).catch(() => {});
		process.exit(3);
	});

	await writeStatus();
	const copied = await keepClient(opts.client ?? path.join(opts.server, "..", "build"), out, build);
	for (const did of hook.removed_repos ?? []) await removeRepo(out, did, loadState(did));

	// a page is current if every read it made is unchanged since the generation it was
	// rendered at, so each repo walks the knot's diffs back to its own generation
	const plans = [];
	let diffEntries = 0;
	for (const repo of hook.repos) {
		const prefix = prefixOf(repo);
		// until it renders again, a repo keeps the entry it had, whatever stamp it carries
		if (knotState?.repos?.[repo.did]) nextRepos[repo.did] = knotState.repos[repo.did];
		const summary = current(knotState?.repos?.[repo.did]);
		// a repo whose render failed last time keeps pages of an older run, maybe of another build
		if (summary?.failed) {
			plans.push({ repo, work: "all", weight: summary.pages ?? Infinity });
			continue;
		}
		if (summary && repo.change === "unchanged" && summary.prefix === prefix) {
			if (summary.generation === hook.generation) {
				nextRepos[repo.did] = summary;
				continue;
			}
			const diffs = await readDiff(hookPath, hook, summary.generation);
			if (diffs !== null) {
				diffEntries = Math.max(diffEntries, diffs.entries);
				if (!diffs.global && !diffs.repos.has(repo.did)) {
					nextRepos[repo.did] = { stamp, generation: hook.generation, prefix, pages: summary.pages };
					status.repos[repo.did] = { skipped: true };
					continue;
				}
			}
		}
		const prior = loadState(repo.did);
		const fresh = !summary || !prior || repo.change === "added" || prior.prefix !== prefix;
		if (fresh) {
			plans.push({ repo, work: "all", weight: prior ? Object.keys(prior.pages).length : (summary?.pages ?? Infinity) });
			continue;
		}
		const since = summary.generation;
		if (since === hook.generation) {
			nextRepos[repo.did] = { stamp, generation: hook.generation, prefix, pages: Object.keys(prior.pages).length };
			continue;
		}
		const diffs = await readDiff(hookPath, hook, since);
		if (diffs === null) {
			plans.push({ repo, work: "all", weight: Object.keys(prior.pages).length });
			continue;
		}
		diffEntries = Math.max(diffEntries, diffs.entries);
		const dirty = dirtyPages(prior, diffs);
		if (dirty === "all" || dirty.size > 0 || repo.change !== "unchanged" || countsChanged(prior, diffs)) {
			plans.push({ repo, work: dirty === "all" ? "all" : [...dirty], weight: dirty === "all" ? Object.keys(prior.pages).length : dirty.size });
		} else {
			// nothing it read changed: the pages stand for this generation too
			nextRepos[repo.did] = { stamp, generation: hook.generation, prefix, pages: Object.keys(prior.pages).length };
			status.repos[repo.did] = { skipped: true };
		}
	}

	// every owner of a repo here gets a user page, rendered again when a read it made changed
	const owners = new Map(hook.repos.map((repo) => [repo.owner_did, { did: repo.owner_did, prefix: ownerPrefixOf(repo) }]));
	// one with no repo left here, or under a new handle, loses the page at the old url
	for (const [did, known] of Object.entries(knotState?.owners ?? {})) {
		if (owners.get(did)?.prefix !== known.prefix) await unpublishOwner(out, known.prefix);
		else nextOwners[did] = known;
	}
	for (const owner of owners.values()) {
		const known = current(knotState?.owners?.[owner.did]);
		if (known && !known.failed && known.prefix === owner.prefix) {
			const diffs = await readDiff(hookPath, hook, known.generation);
			if (diffs !== null && !stale(known.deps, diffs)) {
				nextOwners[owner.did] = { ...known, generation: hook.generation };
				continue;
			}
		}
		plans.push({ owner, weight: 0 });
	}

	await saveKnot();

	// user pages first, they are few and cheap and every repo page links its owner's. then the
	// biggest repos, so the last worker to finish isn't the one holding ghostty
	plans.sort((a, b) => !!b.owner - !!a.owner || b.weight - a.weight);

	const workers = Math.min(Number(opts.workers), plans.length);
	const queue = [...plans];
	const familyOf = (plan) => (plan.repo?.name ?? plan.owner.prefix).toLowerCase().replace(/(-fork|\d+)$/, "");
	const activeFamilies = new Set();
	const takePlan = (lastFamily) => {
		if (queue.length === 0) return undefined;
		if (lastFamily) {
			const sameIdx = queue.findIndex((p) => familyOf(p) === lastFamily);
			if (sameIdx !== -1) return queue.splice(sameIdx, 1)[0];
		}
		const distinctIdx = queue.findIndex((p) => !activeFamilies.has(familyOf(p)));
		const plan = queue.splice(distinctIdx !== -1 ? distinctIdx : 0, 1)[0];
		activeFamilies.add(familyOf(plan));
		return plan;
	};
	// workers: the processes the render started, so each one past --workers is a recycle
	const totals = { rendered: 0, kept: 0, dropped: 0, over_budget: 0, pages: 0, bytes: 0, reads: 0, workers: 0, inlined_over_cap: 0 };
	const failures = [];
	await Promise.all(
		Array.from({ length: workers }, () =>
			runWorkerLoop(children, queue, takePlan, activeFamilies, familyOf, { hookPath, out, budget, totals }, async (plan, result, wallMs) => {
				if (plan.owner) {
					const { did, prefix } = plan.owner;
					if (result.error) {
						failures.push(did);
						nextOwners[did] = { stamp, failed: true, prefix, served: servedBy(nextOwners[did]) };
						status.owners[did] = { error: result.error };
					} else {
						const { deps, ...stats } = result;
						nextOwners[did] = { stamp, generation: hook.generation, prefix, deps };
						status.owners[did] = { ...stats, wall_ms: wallMs };
						for (const key of Object.keys(totals)) totals[key] += stats[key] ?? 0;
					}
				} else if (result.error) {
					failures.push(plan.repo.did);
					// its site link still names the pages of its last finished render
					nextRepos[plan.repo.did] = {
						stamp,
						failed: true,
						prefix: prefixOf(plan.repo),
						pages: nextRepos[plan.repo.did]?.pages,
						served: servedBy(nextRepos[plan.repo.did])
					};
					status.repos[plan.repo.did] = { error: result.error };
				} else {
					nextRepos[plan.repo.did] = { stamp, generation: hook.generation, prefix: prefixOf(plan.repo), pages: result.pages };
					status.repos[plan.repo.did] = { ...result, wall_ms: wallMs };
					for (const key of Object.keys(totals)) totals[key] += result[key] ?? 0;
				}
				await Promise.all([saveKnot(), writeStatus()]);
			})
		)
	);

	await knotWrite;
	const served = new Set([build, ...[...Object.values(nextRepos), ...Object.values(nextOwners)].map(servedBy)]);
	const collected = await collectClients(out, served);
	const summary = {
		state: failures.length === 0 ? "done" : "partial",
		failed: failures,
		finished_at: new Date().toISOString(),
		diff_entries: diffEntries,
		client: { copied, collected },
		totals,
		total_ms: Math.round(performance.now() - t0)
	};
	await writeStatus(summary);
	process.stdout.write(JSON.stringify({ ...summary, repos: status.repos, owners: status.owners }) + "\n", () => process.exit(failures.length === 0 ? 0 : 1));
}

async function runWorkerLoop(children, queue, takePlan, activeFamilies, familyOf, { hookPath, out, budget, totals }, onRepoDone) {
	const maxRepos = Math.max(1, Number(opts["worker-max-repos"]) || 12);
	const maxRssMb = Math.max(128, Number(opts["worker-max-rss-mb"]) || (process.versions.bun ? 360 : 650));
	let lastFamily = null;
	while (queue.length > 0) {
		const hl = Number(opts.highlighters) > 0 && queue.every((p) => p.weight <= 12) ? "1" : opts.highlighters;
		const args = [hookPath, "--out", out, "--server", opts.server, "--jobs", opts.jobs, "--highlighters", hl];
		if (opts.read) args.push("--read", opts.read);
		const child = fork(new URL("./worker.mjs", import.meta.url).pathname, args, {
			execPath: process.execPath,
			stdio: ["ignore", "inherit", "inherit", "ipc"],
			env: process.versions.bun
				? {
						...process.env,
						BUN_JSC_forceMiniVMMode: process.env.BUN_JSC_forceMiniVMMode ?? "1",
						BUN_JSC_useWasmFastMemory: process.env.BUN_JSC_useWasmFastMemory ?? "0"
					}
				: process.env,
			// SVFE_PAGES_PROFILE=<dir> writes a cpu profile per worker there
			execArgv: process.versions.bun
				? ["--bun", "--smol"]
				: [
						`--max-old-space-size=${Number(opts["worker-heap-mb"]) || 128}`,
						"--max-semi-space-size=4",
						...(process.env.SVFE_PAGES_PROFILE ? ["--cpu-prof", "--cpu-prof-dir", process.env.SVFE_PAGES_PROFILE] : []),
						// SVFE_PAGES_HEAPSNAPSHOT=<dir> writes a heap snapshot there when a worker nears its heap cap
						...(process.env.SVFE_PAGES_HEAPSNAPSHOT
							? ["--heapsnapshot-near-heap-limit=1", `--diagnostic-dir=${process.env.SVFE_PAGES_HEAPSNAPSHOT}`]
							: [])
					]
		});
		children.add(child);
		totals.workers++;
		await new Promise((resolve) => {
			let current = null;
			let tStart = 0;
			let served = 0;
			const next = () => {
				const plan = takePlan(lastFamily);
				if (!plan) {
					current = null;
					if (child.connected) child.disconnect();
					return;
				}
				current = plan;
				lastFamily = familyOf(plan);
				tStart = performance.now();
				child.send({ repo: plan.repo, owner: plan.owner, work: plan.work, budget });
			};
			child.on("message", async (message) => {
				if (message.ready) {
					next();
					return;
				}
				const finished = current;
				const wallMs = Math.round(performance.now() - tStart);
				current = null;
				served++;
				const fam = familyOf(finished);
				const hasSameFamilyNext = queue.some((p) => familyOf(p) === fam);
				if (!hasSameFamilyNext) activeFamilies.delete(fam);
				const rssMb = message.rss_mb ?? 0;
				await onRepoDone(finished, message, wallMs);
				const recycle =
					message.error ||
					served >= maxRepos ||
					rssMb >= (hasSameFamilyNext ? maxRssMb * 1.25 : maxRssMb);
				if (recycle) {
					if (child.connected) child.disconnect();
				} else {
					next();
				}
			});
			child.on("exit", async (code, signal) => {
				children.delete(child);
				const failed = current ?? (served === 0 ? takePlan(lastFamily) : null);
				if (failed) {
					const wallMs = tStart ? Math.round(performance.now() - tStart) : 0;
					current = null;
					activeFamilies.delete(familyOf(failed));
					await onRepoDone(failed, { error: `worker for ${(failed.repo ?? failed.owner).did} exited ${signal ?? code}` }, wallMs);
				}
				resolve();
			});
		});
	}
}

async function removeRepo(out, did, prior) {
	await rm(path.join(out, "site", did), { force: true });
	for (const alias of prior?.aliases ?? []) await rm(path.join(out, "www", alias), { force: true });
	await rm(path.join(out, "store", tokenOf(did)), { recursive: true, force: true });
	await rm(path.join(out, "state", tokenOf(did) + ".json"), { force: true });
}