import type { Database } from "bun:sqlite";
import {
	clearMirror,
	knownEventIds,
	knownObject,
	knownReleases,
	knownSave,
	land,
	mirrorState,
	openMirror,
	type StoredObject,
} from "../mirror/store.ts";
import { hiddenPaths, recordSync, registerRepo } from "../server/db.ts";
import { RESERVED_NAMES } from "../server/reserved.ts";
import { dropTarballs } from "../tarball/cache.ts";
import type { Bucket } from "./reader.ts";
import {
	type ArrivingSave,
	BucketError,
	checkFormat,
	type EventRecord,
	escapePath,
	isSafePath,
	listEventIds,
	listReleaseVersions,
	listRepos,
	OversizedObject,
	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;

export type SyncTarget = {
	bucket: Bucket;
	kepr: Database;
	mirrorDir: string;
};

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> {
	running ??= syncEach(target).finally(() => {
		running = null;
	});
	return running;
}

/** Rebuilds one repo's mirror from the bucket, once any sync under way has finished. */
export async function rebuildRepo(target: SyncTarget, repo: string): Promise<void> {
	await syncAll(target);
	clearMirror(openMirror(target.mirrorDir, repo));
	await syncAll(target);
}

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> {
	for (const repo of await listRepos(target.bucket)) {
		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);
	}
}

/**
 * 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. Returns the
 * events it landed.
 */
export async function syncRepo(
	bucket: Bucket,
	mirror: Database,
	repo: string,
	hidden: string[] = [],
	landingBytes = LANDING_BYTES,
): Promise<EventRecord[]> {
	await checkFormat(bucket, repo);
	const head = await readHead(bucket, repo);
	if (head === null) return [];
	const unknown: ArrivingSave[] = [];
	const seen = new Set<string>();
	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {
		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);
		seen.add(cursor);
		const save = await readSave(bucket, repo, cursor);
		unknown.push(save);
		cursor = save.parent;
	}
	// Oldest first, landing whenever the fetched objects pass `landingBytes`: a first sync of a big
	// repo never holds its whole history in memory, and one cut short resumes from the last batch.
	let saves: SaveRecord[] = [];
	let objects = new Map<string, StoredObject>();
	let quarantined: string[] = [];
	let fetched = 0;
	const oversized = new Set<string>();
	for (const save of unknown.reverse()) {
		quarantined.push(...save.quarantined);
		for (const file of save.files) {
			if (objects.has(file.hash) || oversized.has(file.hash) || knownObject(mirror, file.hash)) {
				continue;
			}
			try {
				const { compressed, bytes } = await readObject(bucket, repo, file.hash);
				objects.set(file.hash, { compressed, size: bytes.length });
				fetched += compressed.length;
			} catch (cause) {
				if (!(cause instanceof OversizedObject)) throw cause;
				oversized.add(file.hash);
			}
		}
		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 (fetched >= landingBytes && save.id !== head) {
			land(mirror, {
				head: save.id,
				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",
				saves,
				objects,
				events: [],
				releases: [],
				quarantined,
				hidden,
			});
			saves = [];
			objects = new Map();
			quarantined = [];
			fetched = 0;
		}
	}
	const known = knownEventIds(mirror);
	const events: EventRecord[] = [];
	for (const id of await listEventIds(bucket, repo)) {
		if (known.has(id)) continue;
		const event = await readEvent(bucket, repo, id);
		// Koh ignores it too: no save can hold an illegal path (§5.5).
		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);
	}
	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,
		hidden,
	});
	return events;
}

async function newReleases(
	bucket: Bucket,
	mirror: Database,
	repo: string,
): Promise<ReleaseRecord[]> {
	const known = knownReleases(mirror);
	const releases: ReleaseRecord[] = [];
	for (const version of await listReleaseVersions(bucket, repo)) {
		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));
	}
	return releases;
}