import * as fs from "node:fs";
import * as path from "node:path";
import {
	MAX_STEER_BYTES,
	STEER_ACK_TIMEOUT_MS,
	patchStatus,
	readFileIfExists,
	type RunPaths,
	type RunState,
	writeJsonAtomic,
} from "../worker/status.ts";
import { newControlRequestId } from "./plane.ts";
import { parseSteerCapability, parseSteerRequest, quarantinePath, steerFileName, type SteerRequestRecord } from "./request.ts";

export type SteerState = "delivered" | "queued" | "failed";

export type SteerRequest = SteerRequestRecord;

export interface SteerAck {
	reqId: string;
	state: "delivered" | "failed";
	ts: string;
	detail?: string;
}

export interface SteerResult {
	reqId: string;
	state: SteerState;
	detail?: string;
	/** The durable request could not be withdrawn and may still be delivered. */
	indeterminate?: boolean;
}

export interface SteerDeps {
	now?: () => number;
	requestId?: () => string;
	write?: (file: string, value: unknown) => void;
	remove?: (file: string) => void;
	move?: (from: string, to: string) => void;
	read?: (file: string) => string | undefined;
	delay?: (ms: number) => Promise<void>;
	afterWrite?: () => void;
}

const lastSequenceByInbox = new Map<string, number>();

export function formatSteering(message: string): string {
	return message;
}

export function steerRequestFile(paths: RunPaths, request: Pick<SteerRequest, "seq" | "reqId">): string {
	return path.join(paths.steer, steerFileName(request));
}

export function steerAckFile(paths: RunPaths, reqId: string): string {
	return path.join(paths.steerAck, `${Buffer.from(reqId).toString("base64url")}.json`);
}

export function readClosedState(paths: RunPaths, read: (file: string) => string | undefined = readFileIfExists): RunState | undefined {
	const raw = read(paths.closed);
	if (raw === undefined) return undefined;
	try {
		const parsed = JSON.parse(raw) as { state?: unknown };
		return typeof parsed.state === "string" ? (parsed.state as RunState) : "unknown";
	} catch {
		return "unknown";
	}
}

/**
 * Withdraw a request that must not be delivered.
 *
 * "Removed" is not the only acceptable outcome and neither is "the call threw".
 * ENOENT means a concurrent worker check already consumed it; a real I/O failure
 * means the file is still in the inbox where a poll or a later revive would execute
 * it, so it is moved out of the scanned directory instead. Only when both fail is
 * the withdrawal genuinely indeterminate, and then the caller must say so.
 */
function withdrawRequest(
	paths: RunPaths,
	file: string,
	deps: SteerDeps,
): { withdrawn: boolean; detail?: string } {
	const remove = deps.remove ?? ((target: string) => fs.rmSync(target, { force: true }));
	try {
		remove(file);
		return { withdrawn: true };
	} catch (error) {
		if ((error as NodeJS.ErrnoException).code === "ENOENT") return { withdrawn: true };
		const move = deps.move ?? ((from: string, to: string) => {
			fs.mkdirSync(path.dirname(to), { recursive: true, mode: 0o700 });
			fs.renameSync(from, to);
		});
		try {
			move(file, quarantinePath(paths, file, deps.now?.() ?? Date.now()));
			return { withdrawn: true, detail: "request could not be deleted and was quarantined instead" };
		} catch (moveError) {
			return {
				withdrawn: false,
				detail: `request at ${file} could not be removed (${(error as Error).message}) or quarantined (${(moveError as Error).message})`,
			};
		}
	}
}

function failAck(paths: RunPaths, reqId: string, detail: string): void {
	try {
		writeJsonAtomic(steerAckFile(paths, reqId), { reqId, state: "failed", ts: new Date().toISOString(), detail });
	} catch {
		// The call already reports the failure; the ack file is for a later reader.
	}
	try {
		patchStatus(paths, (current) => ({
			...current,
			steering: { ...current.steering, pending: Math.max(0, current.steering.pending - 1), failed: current.steering.failed + 1 },
		}));
	} catch {
		// Counters are observability only.
	}
}

export async function queueSteer(
	paths: RunPaths,
	message: string,
	options: {
		waitForAck: boolean;
		interrupt?: boolean;
		source?: "orchestrator" | "user";
		spawning?: boolean;
		/** The live worker pid, for the R-CTRL-10 capability cross-check. */
		livePid?: number | null;
	},
	deps: SteerDeps = {},
): Promise<SteerResult> {
	const bytes = Buffer.byteLength(message, "utf8");
	if (bytes === 0) throw new Error("steering message is empty");
	if (bytes > MAX_STEER_BYTES) throw new Error(`steering message is ${bytes} bytes; maximum is ${MAX_STEER_BYTES}`);
	const read = deps.read ?? readFileIfExists;
	const closedBefore = readClosedState(paths, read);
	if (closedBefore !== undefined) return { reqId: "", state: "failed", detail: `run is ${closedBefore}; no longer accepts steering` };

	const now = deps.now?.() ?? Date.now();
	const priorSequence = lastSequenceByInbox.get(paths.steer) ?? -1;
	const sequence = Math.max(now, priorSequence + 1);
	lastSequenceByInbox.set(paths.steer, sequence);
	const reqId = deps.requestId?.() ?? newControlRequestId(now);
	const request: SteerRequest = {
		reqId,
		seq: String(sequence).padStart(13, "0"),
		ts: new Date(now).toISOString(),
		message,
		source: options.source ?? "orchestrator",
		interrupt: options.interrupt ?? false,
	};
	const file = steerRequestFile(paths, request);
	const write = deps.write ?? writeJsonAtomic;
	write(file, request);
	deps.afterWrite?.();
	// R-CTRL-14: closure is checked again after the write, because a steer written
	// into the closing window would otherwise be consumed by a later revive.
	const closedAfter = readClosedState(paths, read);
	if (closedAfter !== undefined) {
		const withdrawn = withdrawRequest(paths, file, deps);
		const base = `run is ${closedAfter}; no longer accepts steering`;
		if (!withdrawn.withdrawn) {
			return { reqId, state: "failed", indeterminate: true, detail: `${base}; ${withdrawn.detail}` };
		}
		return { reqId, state: "failed", detail: withdrawn.detail === undefined ? base : `${base} (${withdrawn.detail})` };
	}
	try {
		patchStatus(paths, (current) => ({ ...current, steering: { ...current.steering, pending: current.steering.pending + 1 } }));
	} catch {
		// Status counters are observability; the durable request remains authoritative.
	}

	// R-CTRL-10. `supported: false` is the only reason to destroy a queued steer, so
	// it is believed only when the record demonstrably describes the live worker. A
	// stale record from an earlier run of the same task, a malformed one, or one
	// naming another pid reads as "not ready yet" — which keeps the request queued.
	const capabilityRaw = read(paths.steerCapability);
	if (capabilityRaw !== undefined) {
		const capability = parseSteerCapability(capabilityRaw);
		if (
			capability.ok &&
			capability.value.supported === false &&
			options.livePid !== undefined &&
			options.livePid !== null &&
			capability.value.pid === options.livePid
		) {
			const detail = "worker does not support steering";
			const withdrawn = withdrawRequest(paths, file, deps);
			failAck(paths, reqId, detail);
			if (!withdrawn.withdrawn) return { reqId, state: "failed", indeterminate: true, detail: `${detail}; ${withdrawn.detail}` };
			return { reqId, state: "failed", detail };
		}
	}

	if (options.spawning || !options.waitForAck) return { reqId, state: "queued" };

	const delay = deps.delay ?? ((ms: number) => new Promise<void>((resolve) => {
		const timer = setTimeout(resolve, ms);
		timer.unref?.();
	}));
	const deadline = (deps.now?.() ?? Date.now()) + STEER_ACK_TIMEOUT_MS;
	const ackPath = steerAckFile(paths, reqId);
	while ((deps.now?.() ?? Date.now()) < deadline) {
		const raw = read(ackPath);
		if (raw !== undefined) {
			const ack = parseSteerAck(raw, reqId);
			if (ack !== undefined) {
				return { reqId, state: ack.state, ...(ack.detail === undefined ? {} : { detail: ack.detail }) };
			}
		}
		const closed = readClosedState(paths, read);
		if (closed !== undefined) {
			// The ack may still land: closure and acknowledgment race, and the ack file
			// is the only proof. Re-read once before reporting.
			const late = read(ackPath);
			if (late !== undefined) {
				const ack = parseSteerAck(late, reqId);
				if (ack !== undefined) return { reqId, state: ack.state, ...(ack.detail === undefined ? {} : { detail: ack.detail }) };
			}
			return { reqId, state: "failed", detail: `run is ${closed}; no longer accepts steering` };
		}
		await delay(25);
	}
	const closed = readClosedState(paths, read);
	if (closed !== undefined) {
		const late = read(ackPath);
		if (late !== undefined) {
			const ack = parseSteerAck(late, reqId);
			if (ack !== undefined) return { reqId, state: ack.state, ...(ack.detail === undefined ? {} : { detail: ack.detail }) };
		}
		return { reqId, state: "failed", detail: `run is ${closed}; no longer accepts steering` };
	}
	return { reqId, state: "queued", detail: "worker has not durably acknowledged this request yet" };
}

export function parseSteerAck(raw: string, reqId: string): SteerAck | undefined {
	try {
		const parsed = JSON.parse(raw) as Record<string, unknown>;
		if (parsed.reqId !== reqId || (parsed.state !== "delivered" && parsed.state !== "failed") || typeof parsed.ts !== "string") return undefined;
		return {
			reqId,
			state: parsed.state,
			ts: parsed.ts,
			...(typeof parsed.detail === "string" ? { detail: parsed.detail } : {}),
		};
	} catch {
		return undefined;
	}
}

/** Re-exported so the worker inbox and the orchestrator share one parser. */
export { parseSteerRequest };