import type { Database } from "bun:sqlite";
import {
	applyHidden,
	knownEventIds,
	knownObject,
	knownReleases,
	knownSave,
	land,
	mirrorState,
	openMirror,
	type StoredObject,
	storeObjects,
} from "../mirror/store.ts";
import {
	hiddenPaths,
	hiddenToHand,
	keepState,
	recordHiddenHanded,
	recordSync,
	registerRepo,
} from "../server/db.ts";
import { RESERVED_NAMES } from "../server/reserved.ts";
import { dropTarballs } from "../tarball/cache.ts";
import { type Bucket, BucketWriteError, inTurn, KEPR_PREFIX } from "./reader.ts";
import {
	BucketError,
	checkFormat,
	type EventRecord,
	escapePath,
	isSafePath,
	listEventIds,
	listReleaseVersions,
	listRepos,
	OversizedObject,
	objectSize,
	type ReleaseRecord,
	readEvent,
	readHead,
	readObject,
	readRelease,
	readReleasesLatest,
	readSave,
	type SaveRecord,
} from "./repo.ts";

const INTERVAL_MS = 2 * 60 * 1000;
// Compressed bytes a sync holds before landing them; sized for a 2 GB host.
const LANDING_BYTES = 32 * 1024 * 1024;
const LANDING_SAVES = 100;
// Bucket reads in flight at once. Each waits about 170 ms on Tylee's round trip to the bucket
// (measured 2026-10-02), so reading one at a time left the line idle.
const READS_AT_ONCE = 16;
// Compressed object bytes being read at once; a larger object waits and is read alone.
const READ_BYTES = 32 * 1024 * 1024;

export type SyncTarget = {
	bucket: Bucket;
	kepr: Database;
	mirrorDir: string;
	/** Called after each repo that synced without error. */
	synced?: (repo: string) => void;
};

let running: Promise<void> | null = null;

/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */
export function syncAll(target: SyncTarget): Promise<void> {
	return running ?? holdSync(() => syncEach(target));
}

/**
 * Runs `work` once any sync under way has finished, and keeps syncs waiting until it is done: a
 * rename moves a repo's keys, which a sync must never see half moved.
 */
export function holdSync(work: () => Promise<void>): Promise<void> {
	// Cleared before anyone awaiting `work` resumes, so a sync asked for next starts afresh.
	const mine = (running ?? Promise.resolve()).then(work).finally(() => {
		if (running === slot) running = null;
	});
	const slot = mine.catch(() => {});
	running = slot;
	return mine;
}

export function startSyncLoop(target: SyncTarget): void {
	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));
	tick();
	setInterval(tick, INTERVAL_MS);
}

async function syncEach(target: SyncTarget): Promise<void> {
	let repos: string[];
	try {
		repos = await listRepos(target.bucket);
	} catch (error) {
		keepState(target.kepr, "bucket_error", String(error));
		throw error;
	}
	// What the console's status shows, so it never asks the bucket itself.
	keepState(target.kepr, "bucket_error", null);
	keepState(target.kepr, "bucket_read_at", String(Math.floor(Date.now() / 1000)));
	keepState(target.kepr, "bucket_repos", JSON.stringify(repos));
	for (const repo of repos) {
		registerRepo(target.kepr, repo);
		let error: string | null = null;
		if (RESERVED_NAMES.has(repo)) {
			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;
		} else {
			try {
				const mirror = openMirror(target.mirrorDir, repo);
				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));
				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);
			} catch (cause) {
				error = cause instanceof Error ? cause.message : String(cause);
			}
		}
		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);
		if (error === null) target.synced?.(repo);
		await handHidden(target, repo);
	}
}

/**
 * Hands a repo's hidden paths to Koh at `.kepr/hidden/<repo>`, one per line, for it to add to
 * `.kohignore`, unless they are known to be there already, then drops the tarballs kept under the
 * old list. The sync does it rather than the console, so a failed write or a restart is retried at
 * the next pass, and it never overlaps a rename. Without a write key the list stays in Kepr: Koh
 * reads it only from the bucket anyway.
 */
async function handHidden(target: SyncTarget, repo: string): Promise<void> {
	const list = hiddenToHand(target.kepr, repo);
	if (list === null) return;
	const paths = list.split("\n").filter(Boolean);
	const bytes = new TextEncoder().encode(paths.map((path) => `${path}\n`).join(""));
	// Unknown until the write is confirmed, so a restart or failure mid-write is written again
	// even if the list changes back to the one recorded before.
	recordHiddenHanded(target.kepr, repo, null);
	try {
		await target.bucket.put(`${KEPR_PREFIX}hidden/${repo}`, bytes, "text/plain");
	} catch (error) {
		if (!(error instanceof BucketWriteError)) {
			console.error(`kepr: handing ${repo}'s hidden paths to Koh failed: ${error}`);
		}
		return;
	}
	recordHiddenHanded(target.kepr, repo, list);
	// Only storage: a kept tarball is never served once its generation's hidden list has changed.
	await dropTarballs(target.bucket, repo).catch((error) =>
		console.error(`kepr: dropping ${repo}'s kept tarballs failed: ${error}`),
	);
}

/**
 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,
 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a
 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`
 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates
 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. `hidden` is
 * applied first, so a list changed since the last sync takes effect even with nothing new. Returns
 * the events it landed.
 */
export async function syncRepo(
	bucket: Bucket,
	mirror: Database,
	repo: string,
	hidden: string[] = [],
	landingBytes = LANDING_BYTES,
	readBytes = READ_BYTES,
): Promise<EventRecord[]> {
	applyHidden(mirror, hidden);
	await checkFormat(bucket, repo);
	const head = await readHead(bucket, repo);
	if (head === null) return [];
	const unknown = new Set<string>();
	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {
		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);
		unknown.add(cursor);
		cursor = (await readSave(bucket, repo, cursor)).parent;
	}
	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES
	// records: a first sync of a big repo never holds its whole history in memory, and one cut short
	// resumes from the last batch. Records are read again here rather than kept from the walk, since
	// each lists every file of its tree.
	let saves: SaveRecord[] = [];
	let objects = new Map<string, StoredObject>();
	let quarantined: string[] = [];
	let fetched = 0;
	const oversized = new Set<string>();
	const landBatch = (batchHead: string) => {
		land(mirror, {
			head: batchHead,
			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",
			saves,
			objects,
			events: [],
			releases: [],
			quarantined,
		});
		saves = [];
		objects = new Map();
		quarantined = [];
		fetched = 0;
	};
	const gate = byteGate(readBytes);
	let failure: unknown = null;
	// Reads one object into `ahead`. A read the gate admits after the batch filled, or after another
	// read failed, puts its hash back for the next batch instead.
	const readInto = async (ahead: Map<string, StoredObject>, wanted: string[], hash: string) => {
		const size = await objectSize(bucket, repo, hash);
		await gate.enter(size);
		try {
			if (failure !== null || fetched >= landingBytes) {
				wanted.unshift(hash);
				return;
			}
			const object = await readObject(bucket, repo, hash);
			ahead.set(hash, object);
			fetched += object.compressed.length;
		} catch (cause) {
			if (!(cause instanceof OversizedObject)) throw cause;
			oversized.add(hash);
		} finally {
			gate.leave(size);
		}
	};
	const ids = [...unknown].reverse();
	for (let start = 0; start < ids.length; start += READS_AT_ONCE) {
		const records = await inTurn(ids.slice(start, start + READS_AT_ONCE), READS_AT_ONCE, (id) =>
			readSave(bucket, repo, id),
		);
		for (const save of records) {
			quarantined.push(...save.quarantined);
			// One save can hold more than a batch, as a first import does: its objects are stored
			// ahead of it, after the saves before it have landed, so no landing's cleanup sees them
			// unreferenced.
			const ahead = new Map<string, StoredObject>();
			const wanted = [...new Set(save.files.map((file) => file.hash))].filter(
				(hash) => !objects.has(hash) && !oversized.has(hash) && !knownObject(mirror, hash),
			);
			while (wanted.length > 0) {
				// No read starts once the batch is full, so a batch passes `landingBytes` by at most
				// the reads under way, which the gate holds to `readBytes` or one larger object.
				const reader = async () => {
					while (failure === null && fetched < landingBytes && wanted.length > 0) {
						await readInto(ahead, wanted, wanted.shift() as string).catch((cause) => {
							failure ??= cause;
						});
					}
				};
				// Every reader settles before a failure is reported, so none outlives this sync.
				await Promise.all(Array.from({ length: READS_AT_ONCE }, reader));
				if (failure !== null) throw failure;
				if (fetched >= landingBytes) {
					const last = saves.at(-1);
					if (last) landBatch(last.id);
					storeObjects(mirror, ahead);
					ahead.clear();
					fetched = 0;
				}
			}
			for (const [hash, object] of ahead) objects.set(hash, object);
			quarantined.push(
				...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),
			);
			save.files = save.files.filter((file) => !oversized.has(file.hash));
			saves.push(save);
			if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);
		}
	}
	const known = knownEventIds(mirror);
	const newEvents = (await listEventIds(bucket, repo)).filter((id) => !known.has(id));
	const events = (
		await inTurn(newEvents, READS_AT_ONCE, (id) => readEvent(bucket, repo, id))
	).filter(
		// Koh ignores it too: no save can hold an illegal path (§5.5).
		(event) => event.kind !== "strip" || isSafePath(event.path),
	);
	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";
	const moved =
		mirrorState(mirror, "head") !== head ||
		mirrorState(mirror, "releases_latest") !== releasesLatest;
	const releases = moved ? await newReleases(bucket, mirror, repo) : [];
	if (!moved && events.length === 0) return [];
	land(mirror, {
		head,
		releasesLatest,
		saves,
		objects,
		events,
		releases,
		quarantined,
	});
	return events;
}

/** Admits reads while their bytes in flight stay within `limit`; a larger one waits to go alone. */
function byteGate(limit: number) {
	let inFlight = 0;
	let waiting: (() => void)[] = [];
	return {
		async enter(size: number): Promise<void> {
			while (inFlight > 0 && inFlight + size > limit) {
				await new Promise<void>((resolve) => waiting.push(resolve));
			}
			inFlight += size;
		},
		leave(size: number): void {
			inFlight -= size;
			const woken = waiting;
			waiting = [];
			for (const wake of woken) wake();
		},
	};
}

async function newReleases(
	bucket: Bucket,
	mirror: Database,
	repo: string,
): Promise<ReleaseRecord[]> {
	const known = knownReleases(mirror);
	const versions = (await listReleaseVersions(bucket, repo)).filter((v) => !known.has(v));
	return inTurn(versions, READS_AT_ONCE, (version) => readRelease(bucket, repo, version));
}