/**
 * The file-backed store for one repository's session bus.
 *
 * Every session writes its own record, every message is its own file, and every
 * claim is an exclusive-create file. That keeps concurrent Pi processes off one
 * shared lock while still making the common operations atomic. The store knows
 * nothing about models or UI; it only reads and writes validated JSON.
 */

import { createHash, randomBytes } from "node:crypto";
import { mkdir, link, readdir, rename, rm, rmdir, stat, writeFile } from "node:fs/promises";
import { dirname, join } from "node:path";
import { asRecord, readJson } from "../../lib/config.ts";
import { claimsConflict } from "./overlap.ts";
import type {
	BusMessage,
	ClaimConflict,
	ClaimRecord,
	DeliveryFailure,
	InboxPeek,
	InboxRead,
	SessionRecord,
} from "./types.ts";
import { INBOX_DIR_NAME, SESSIONS_DIR_NAME } from "../../lib/session-roster.ts";

const DIR_MODE = 0o700;
const FILE_MODE = 0o600;

export interface BusStoreOptions {
	/** The repository bus directory, e.g. `<agent-dir>/session-bus/<key>`. */
	dir: string;
}

export interface BusStore {
	readonly dir: string;
	ensure(): Promise<void>;
	writeSession(record: SessionRecord): Promise<void>;
	sessions(): Promise<SessionRecord[]>;
	/** One session's record, or undefined when it is not on disk. */
	session(id: string): Promise<SessionRecord | undefined>;
	removeSession(id: string): Promise<void>;
	writeMessage(target: string, message: BusMessage): Promise<void>;
	/**
	 * Write one message for each recipient, and answer with those written and those a
	 * file could not be made for.
	 *
	 * A send is one file per inbox, not one transaction over several, so a failure part-way
	 * through leaves the recipients before it holding the message. Nothing here can make the
	 * set atomic — splitting the report from the write is the point — so the caller decides
	 * what a partial delivery means.
	 */
	writeTo(
		recipients: readonly { id: string; label: string }[],
		fields: Omit<BusMessage, "to" | "toLabel">,
	): Promise<{ delivered: string[]; failed: DeliveryFailure[] }>;
	/**
	 * Read up to `limit` messages from the front of a recipient's inbox without consuming
	 * them, skipping the ids the caller already has, with every id still unread.
	 */
	peek(id: string, options: { limit: number; known?: ReadonlySet<string> }): Promise<InboxPeek>;
	/**
	 * Read and consume up to `limit` messages from the front of a recipient's inbox, naming the
	 * files it could not validate instead of deleting them. A limit of zero consumes nothing.
	 */
	inbox(id: string, limit: number): Promise<InboxRead>;
	/**
	 * Every file in a recipient's inbox, by count — the number a peer's status row shows.
	 *
	 * It counts files, so a file this build cannot read counts until its TTL ends it: `inbox`
	 * is what says it cannot be read. That is the one promise this count makes.
	 */
	unread(id: string): Promise<number>;
	claims(): Promise<ClaimRecord[]>;
	claimAll(
		paths: readonly string[],
		owner: string,
		ownerLabel: string,
		at: number,
		/** Test seam: runs after each path is published, so a suite can interleave a rival. */
		afterClaim?: (path: string) => Promise<void>,
	): Promise<{ claimed: string[]; conflicts: ClaimConflict[] }>;
	release(path: string, owner: string): Promise<boolean>;
	releaseAll(owner: string): Promise<string[]>;
	prune(input: {
		now: number;
		/** The sessions known to be live when the caller read the bus. */
		activeIds: ReadonlySet<string>;
		/**
		 * Records whose process the caller found gone: they keep their row until their
		 * heartbeat ages out, and own no path and no mail in the meantime.
		 */
		goneIds?: ReadonlySet<string>;
		sessionTtlMs: number;
		messageTtlMs: number;
	}): Promise<void>;
}

/**
 * Release each path through `release`, reporting the ones this session did not
 * hold. One copy, because `release` and `leave` answer the same two lists, and a
 * second loop is how the two answers drift.
 */
export async function releaseClaimed(
	paths: readonly string[],
	release: (path: string) => Promise<boolean>,
): Promise<{ released: string[]; notOwned: string[] }> {
	const released: string[] = [];
	const notOwned: string[] = [];
	for (const path of paths) {
		if (await release(path)) released.push(path);
		else notOwned.push(path);
	}
	return { released, notOwned };
}

const SHA = (value: string): string => createHash("sha256").update(value).digest("hex").slice(0, 32);

function isSession(value: unknown): value is SessionRecord {
	const entry = asRecord(value);
	if (!entry) return false;
	return (
		typeof entry.id === "string" &&
		typeof entry.label === "string" &&
		typeof entry.pid === "number" &&
		typeof entry.cwd === "string" &&
		typeof entry.repoRoot === "string" &&
		typeof entry.mode === "string" &&
		typeof entry.startedAt === "number" &&
		typeof entry.lastSeen === "number" &&
		(entry.startToken === undefined || typeof entry.startToken === "string") &&
		(entry.watching === undefined || isWatchingState(entry.watching))
	);
}

/** The optional listening field, validated only when it is there. */
function isWatchingState(value: unknown): boolean {
	const entry = asRecord(value);
	return entry !== undefined && typeof entry.armed === "boolean" && typeof entry.wake === "boolean";
}

function isMessage(value: unknown): value is BusMessage {
	const entry = asRecord(value);
	if (!entry) return false;
	return (
		typeof entry.id === "string" &&
		typeof entry.from === "string" &&
		typeof entry.fromLabel === "string" &&
		typeof entry.to === "string" &&
		typeof entry.toLabel === "string" &&
		typeof entry.kind === "string" &&
		typeof entry.text === "string" &&
		typeof entry.at === "number" &&
		(entry.paths === undefined || (Array.isArray(entry.paths) && entry.paths.every((path) => typeof path === "string")))
	);
}

function isClaim(value: unknown): value is ClaimRecord {
	const entry = asRecord(value);
	if (!entry) return false;
	return (
		typeof entry.path === "string" &&
		typeof entry.owner === "string" &&
		typeof entry.ownerLabel === "string" &&
		typeof entry.at === "number"
	);
}

function readTyped<T>(path: string, guard: (value: unknown) => value is T): T | undefined {
	const parsed = readJson(path);
	return parsed !== undefined && guard(parsed) ? parsed : undefined;
}

async function writeJson(path: string, value: unknown): Promise<void> {
	await mkdir(dirname(path), { recursive: true, mode: DIR_MODE });
	const temp = `${path}.${process.pid}.${randomBytes(4).toString("hex")}.tmp`;
	await writeFile(temp, `${JSON.stringify(value, null, 2)}\n`, { encoding: "utf-8", mode: FILE_MODE });
	try {
		await rename(temp, path);
	} catch {
		// Windows cannot replace an existing file with rename; removing first keeps
		// the write recoverable, and the window is only a heartbeat-sized race.
		await rm(path, { force: true });
		await rename(temp, path);
	}
}

const MESSAGE_SUFFIX = ".json";

async function listFiles(dir: string): Promise<string[]> {
	try {
		return (await readdir(dir)).filter((name) => name.endsWith(MESSAGE_SUFFIX)).sort();
	} catch {
		return [];
	}
}

/** The message id a file name holds, which is the id a reader compares against. */
const idOfFile = (name: string): string => name.slice(0, -MESSAGE_SUFFIX.length);

/**
 * A recipient's messages in the order they arrived, oldest first.
 *
 * The queue is the inbox's own clock, not the writer's: an id carries the sender's `now()`, so
 * its order is the sender's clock, and a peer whose clock runs ahead would write a name that
 * sorts after every other — at the back for as long as a backlog is deeper than one page, and
 * read last for as long as it keeps sending. A file's mtime is when the message landed here,
 * which every session shares, so the front is the same for every reader and every poll. Two
 * files published in one filesystem tick tie, and the name settles it so the page is stable.
 */
async function listMessages(dir: string): Promise<string[]> {
	const names = await listFiles(dir);
	const stamped = await Promise.all(
		names.map((name) =>
			stat(join(dir, name)).then(
				(info): { name: string; at: number } | undefined => ({ name, at: info.mtimeMs }),
				// Published between the readdir and the stat, and so already read by the next poll.
				() => undefined,
			),
		),
	);
	return stamped
		.filter((entry): entry is { name: string; at: number } => entry !== undefined)
		.sort((a, b) => a.at - b.at || a.name.localeCompare(b.name))
		.map((entry) => entry.name);
}

/**
 * Drop a recipient's directory once it holds no files, so a bus does not keep
 * one empty directory for every session that ever joined it. An inbox with mail
 * stays, because a session's id survives a resume and its mail is still its own;
 * non-recursive removal also loses the race to a message that lands meanwhile.
 */
async function dropIfEmpty(dir: string): Promise<void> {
	const entries = await readdir(dir).catch(() => null);
	if (!entries || entries.length > 0) return;
	await rmdir(dir).catch(() => {});
}

/**
 * How long a writer's temp file may sit before it is debris.
 *
 * A record and a claim are published by renaming or linking a temp file, so the file
 * exists for milliseconds; one left behind is a process that stopped between the
 * write and the publish. The window is measured on the filesystem's own clock, not
 * the bus's, because that is the clock the timestamps came from.
 */
const TEMP_TTL_MS = 60_000;

/**
 * Remove the temp files a crashed writer left in one directory.
 *
 * Readers only ever look at `*.json`, so this debris is invisible and would live
 * forever without a pass of its own.
 */
async function dropStaleTemps(dir: string): Promise<void> {
	const cut = Date.now() - TEMP_TTL_MS;
	for (const name of await readdir(dir).catch(() => [] as string[])) {
		if (name.endsWith(".json")) continue;
		const path = join(dir, name);
		const info = await stat(path).catch(() => undefined);
		if (info !== undefined && info.mtimeMs < cut) await rm(path, { force: true });
	}
}

/** Create a store rooted at one repository's bus directory. */
export function createBusStore(options: BusStoreOptions): BusStore {
	const { dir } = options;
	const sessionsDir = join(dir, SESSIONS_DIR_NAME);
	const inboxRoot = join(dir, INBOX_DIR_NAME);
	const claimsDir = join(dir, "claims");

	const sessionFile = (id: string): string => join(sessionsDir, `${SHA(id)}.json`);
	const inboxDir = (id: string): string => join(inboxRoot, SHA(id));
	const claimFile = (path: string): string => join(claimsDir, `${SHA(path)}.json`);

	/**
	 * One message file in one inbox, created where it is not there yet.
	 *
	 * A local function rather than a method, because two entries of the store write through
	 * it and `this` is not a receiver a destructured caller keeps.
	 */
	async function writeMessageFile(target: string, message: BusMessage): Promise<void> {
		const directory = inboxDir(target);
		await mkdir(directory, { recursive: true, mode: DIR_MODE });
		await writeJson(join(directory, `${message.id}.json`), message);
	}

	async function readSessions(): Promise<SessionRecord[]> {
		const sessions: SessionRecord[] = [];
		for (const name of await listFiles(sessionsDir)) {
			const record = readTyped(join(sessionsDir, name), isSession);
			if (record) sessions.push(record);
		}
		return sessions.sort((a, b) => b.lastSeen - a.lastSeen || a.label.localeCompare(b.label));
	}

	/**
	 * Up to `limit` messages from the front of a recipient's inbox, oldest first.
	 *
	 * The front is the queue, not a detail: consuming starts there so a limit never starves,
	 * and so does a read, because the caller names the ids it already has. The order is when
	 * the mail arrived, not the clock the sender stamped it with, so no peer's clock decides
	 * where its mail sits. A file name holds the id, so a skip costs no read — which is what
	 * lets a watch page a backlog one poll at a time without reading the same file twice.
	 *
	 * Only the consuming read deletes, and it deletes only what it read. A file this build
	 * cannot validate is named back to the caller instead: it may be mail a newer build can
	 * read — a renamed field, a stricter guard — and destroying it on the way past is how one
	 * build's parser silently eats another's message. The message TTL ends it through its
	 * mtime, so nothing it cannot read accumulates.
	 *
	 * The walk stops at `limit` messages, so the names it reports are the ones it reached: a
	 * refusal behind a full page is named once the mail in front of it is consumed. Reading
	 * every name to find one would cost the whole backlog on every poll, which is the read this
	 * store exists to page.
	 */
	async function readMessages(
		directory: string,
		names: readonly string[],
		options: { limit: number; known?: ReadonlySet<string>; remove: boolean },
	): Promise<{ messages: BusMessage[]; unreadable: string[] }> {
		const { limit, known, remove } = options;
		const messages: BusMessage[] = [];
		const unreadable: string[] = [];
		for (const name of names) {
			if (known?.has(idOfFile(name))) continue;
			const path = join(directory, name);
			const message = readTyped(path, isMessage);
			if (!message) {
				unreadable.push(name);
				continue;
			}
			messages.push(message);
			if (remove) await rm(path, { force: true });
			if (messages.length >= limit) break;
		}
		// In the order it walked: the page a read shows is the front it deleted, and a message's
		// own `at` cannot reorder it after the fact.
		return { messages, unreadable };
	}

	async function readClaims(): Promise<ClaimRecord[]> {
		const claims: ClaimRecord[] = [];
		for (const name of await listFiles(claimsDir)) {
			const claim = readTyped(join(claimsDir, name), isClaim);
			if (claim) claims.push(claim);
		}
		return claims.sort((a, b) => a.path.localeCompare(b.path));
	}

	async function claimOne(
		path: string,
		owner: string,
		ownerLabel: string,
		at: number,
	): Promise<{ claim: ClaimRecord; conflict?: ClaimRecord; created: boolean }> {
		await mkdir(claimsDir, { recursive: true, mode: DIR_MODE });
		const file = claimFile(path);
		const claim: ClaimRecord = { path, owner, ownerLabel, at };
		// Publish a written temp file instead of creating the claim and filling it
		// afterwards: the link is the exclusive create, so no reader can observe a
		// half-written claim and mistake it for a crashed owner's.
		const temp = `${file}.${randomBytes(4).toString("hex")}.tmp`;
		await writeFile(temp, `${JSON.stringify(claim, null, 2)}\n`, { encoding: "utf-8", mode: FILE_MODE });
		try {
			try {
				await link(temp, file);
				return { claim, created: true };
			} catch (error) {
				if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
			}
			const existing = readTyped(file, isClaim);
			if (existing) {
				return existing.owner === owner
					? { claim: existing, created: false }
					: { claim, conflict: existing, created: false };
			}
			// A claim file that does not parse is debris from before claims were
			// published whole; clear it once so a stale path cannot block forever.
			await rm(file, { force: true });
			await link(temp, file);
			return { claim, created: true };
		} finally {
			await rm(temp, { force: true });
		}
	}

	async function releaseOne(path: string, owner: string): Promise<boolean> {
		const file = claimFile(path);
		const existing = readTyped(file, isClaim);
		if (!existing || existing.owner !== owner) return false;
		await rm(file, { force: true });
		return true;
	}

	/**
	 * Claim every path for one owner, or nothing.
	 *
	 * The pre-check reads the claims once and reports every blocking owner, so a
	 * batch that cannot be satisfied touches no file. A create that loses a race
	 * after the pre-check gives back only what this batch published: a path the
	 * owner already held carries an older `at` and is not this batch's to release.
	 *
	 * `afterClaim` is a test seam: it runs after a path is published, which is the
	 * only way a suite can put a rival claim in the window between the pre-check and
	 * the next create.
	 */
	async function claimAll(
		paths: readonly string[],
		owner: string,
		ownerLabel: string,
		at: number,
		afterClaim?: (path: string) => Promise<void>,
	): Promise<{ claimed: string[]; conflicts: ClaimConflict[] }> {
		const existing = await readClaims();
		const conflicts: ClaimConflict[] = [];
		for (const path of paths) {
			for (const held of existing) {
				if (held.owner !== owner && claimsConflict(path, held.path)) {
					conflicts.push({ path, owner: held.owner, ownerLabel: held.ownerLabel });
				}
			}
		}
		if (conflicts.length > 0) return { claimed: [], conflicts };
		const claimed: string[] = [];
		const created: string[] = [];
		for (const path of paths) {
			const result = await claimOne(path, owner, ownerLabel, at);
			if (result.conflict) {
				for (const done of created) await releaseOne(done, owner);
				const { owner: holder, ownerLabel: holderLabel } = result.conflict;
				return { claimed: [], conflicts: [{ path, owner: holder, ownerLabel: holderLabel }] };
			}
			claimed.push(path);
			if (result.created) created.push(path);
			if (afterClaim) await afterClaim(path);
		}
		return { claimed, conflicts: [] };
	}

	return {
		dir,
		async ensure() {
			await Promise.all([
				mkdir(sessionsDir, { recursive: true, mode: DIR_MODE }),
				mkdir(inboxRoot, { recursive: true, mode: DIR_MODE }),
				mkdir(claimsDir, { recursive: true, mode: DIR_MODE }),
			]);
		},

		async writeSession(record) {
			await writeJson(sessionFile(record.id), record);
		},

		sessions: readSessions,

		session: async (id) => readTyped(sessionFile(id), isSession),

		async removeSession(id) {
			await rm(sessionFile(id), { force: true });
			await dropIfEmpty(inboxDir(id));
		},

		writeMessage: writeMessageFile,

		async writeTo(recipients, fields) {
			const delivered: string[] = [];
			const failed: DeliveryFailure[] = [];
			for (const recipient of recipients) {
				try {
					await writeMessageFile(recipient.id, { ...fields, to: recipient.id, toLabel: recipient.label });
					delivered.push(recipient.label);
				} catch (error) {
					// The recipient's own sentence, kept as it came: it is the reason a caller
					// prints, and a filesystem refusal means nothing once it is rewritten.
					failed.push({ label: recipient.label, error: error instanceof Error ? error.message : String(error) });
				}
			}
			return { delivered, failed };
		},

		async peek(id, options) {
			const directory = inboxDir(id);
			const names = await listMessages(directory);
			const { messages, unreadable } = await readMessages(directory, names, { ...options, remove: false });
			return { mail: messages, unread: new Set(names.map(idOfFile)), unreadable };
		},

		async inbox(id, limit) {
			const directory = inboxDir(id);
			return readMessages(directory, await listMessages(directory), { limit, remove: true });
		},

		async unread(id) {
			return (await listFiles(inboxDir(id))).length;
		},

		claims: readClaims,

		claimAll,

		release: releaseOne,

		async releaseAll(owner) {
			const released: string[] = [];
			for (const claim of await readClaims()) {
				if (claim.owner !== owner) continue;
				await rm(claimFile(claim.path), { force: true });
				released.push(claim.path);
			}
			return released;
		},

		async prune({ now: at, activeIds, goneIds, sessionTtlMs, messageTtlMs }) {
			// The caller's set can lag: a peer that joined and claimed since that read is
			// live on disk, and its claims are its own. This read is the answer, so the
			// set only adds to what it found rather than deciding who survives — except for
			// a record whose process is gone, which the caller knows and the file cannot.
			// That record keeps its row until the heartbeat ages out and owns nothing.
			const live = new Set(activeIds);
			const gone = new Set(goneIds ?? []);
			for (const session of await readSessions()) {
				if (at - session.lastSeen > sessionTtlMs) {
					await rm(sessionFile(session.id), { force: true });
					await dropIfEmpty(inboxDir(session.id));
					live.delete(session.id);
				} else if (gone.has(session.id)) {
					live.delete(session.id);
				} else {
					live.add(session.id);
				}
			}
			for (const claim of await readClaims()) {
				if (!live.has(claim.owner)) await rm(claimFile(claim.path), { force: true });
			}
			await dropStaleTemps(sessionsDir);
			await dropStaleTemps(claimsDir);
			for (const name of await readdir(inboxRoot).catch(() => [] as string[])) {
				const directory = join(inboxRoot, name);
				for (const file of await listFiles(directory)) {
					const path = join(directory, file);
					const message = readTyped(path, isMessage);
					if (message) {
						if (at - message.at > messageTtlMs) await rm(path, { force: true });
						continue;
					}
					// A file with no readable `at` has no clock of its own, so its mtime is the one it
					// gets: the TTL still ends mail this build cannot read instead of keeping it forever.
					const written = await stat(path).then(
						(info) => info.mtimeMs,
						() => at,
					);
					if (at - written > messageTtlMs) await rm(path, { force: true });
				}
				await dropStaleTemps(directory);
				await dropIfEmpty(directory);
			}
		},
	};
}