import type { ChildProcessWithoutNullStreams } from "node:child_process";
import type * as NodeReadline from "node:readline";

type DynamicImport = (specifier: string) => Promise<unknown>;

const dynamicImport: DynamicImport = (specifier) => import(specifier);
const NODE_CHILD_PROCESS_SPECIFIER = "node:" + "child_process";
const NODE_READLINE_SPECIFIER = "node:" + "readline";

type RequestId = number;
type JsonObject = Record<string, unknown>;

interface JsonRpcError {
	code: number;
	message: string;
	data?: unknown;
}

interface JsonRpcResponse {
	id: RequestId;
	result?: unknown;
	error?: JsonRpcError;
}

interface JsonRpcNotification {
	method: string;
	params?: unknown;
}

interface JsonRpcServerRequest extends JsonRpcNotification {
	id: RequestId;
}

interface PendingRequest {
	resolve: (value: unknown) => void;
	reject: (reason: Error) => void;
}

export interface CodexAppServerClientOptions {
	cwd?: string;
	env?: NodeJS.ProcessEnv;
	extraArgs?: string[];
	clientInfo?: {
		name: string;
		title: string;
		version: string;
	};
}

export type CodexAppServerNotification = JsonRpcNotification;

export type CodexAppServerNotificationHandler = (
	notification: CodexAppServerNotification,
) => void;

const DEFAULT_CLIENT_INFO = {
	name: "singularity_forge_pi_ai",
	title: "Singularity Forge pi-ai",
	version: "0.0.0",
};

let sharedClientPromise: Promise<CodexAppServerClient> | undefined;

/**
 * Return the session-wide Codex app-server client. Spawns `codex app-server` lazily and reuses it.
 *
 * Purpose: delegate ChatGPT auth and protocol drift to the installed Codex CLI while keeping pi-ai provider calls cheap.
 * Consumer: openai-codex-responses.ts for every OpenAI Codex provider stream.
 */
export function getCodexAppServerClient(
	options?: CodexAppServerClientOptions,
): Promise<CodexAppServerClient> {
	if (!sharedClientPromise) {
		sharedClientPromise = CodexAppServerClient.connect(options);
	}
	return sharedClientPromise;
}

/**
 * Reset the session-wide Codex app-server client after the child process exits or is disposed.
 *
 * Purpose: allow the next provider call to recover from a crashed or deliberately closed Codex process.
 * Consumer: CodexAppServerClient lifecycle handlers in this module.
 */
export function clearCodexAppServerClient(client: CodexAppServerClient): void {
	if (sharedClientPromise) {
		sharedClientPromise.then(
			(current) => {
				if (current === client) sharedClientPromise = undefined;
			},
			() => {
				sharedClientPromise = undefined;
			},
		);
	}
}

/**
 * JSON-RPC client for a stdio `codex app-server` child process.
 *
 * Purpose: provide a small dependency-free transport that matches Codex's newline-delimited JSON protocol.
 * Consumer: getCodexAppServerClient and the OpenAI Codex provider adapter.
 */
export class CodexAppServerClient {
	private proc: ChildProcessWithoutNullStreams | undefined;
	private readline: NodeReadline.Interface | undefined;
	private nextId = 1;
	private stderr = "";
	private closed = false;
	private exitError: Error | undefined;
	private readonly pending = new Map<RequestId, PendingRequest>();
	private readonly notificationHandlers =
		new Set<CodexAppServerNotificationHandler>();

	private constructor(private readonly options: CodexAppServerClientOptions) {}

	/**
	 * Spawn and initialize a Codex app-server process.
	 *
	 * Purpose: complete Codex's required initialize/initialized handshake before any thread or turn RPC.
	 * Consumer: getCodexAppServerClient when creating the shared process.
	 */
	static async connect(
		options: CodexAppServerClientOptions = {},
	): Promise<CodexAppServerClient> {
		const client = new CodexAppServerClient(options);
		await client.initialize();
		return client;
	}

	/**
	 * Register a notification callback and return an unsubscribe function.
	 *
	 * Purpose: let provider streams observe only their own thread/turn notifications without owning the transport.
	 * Consumer: streamOpenAICodexResponses while a turn is active.
	 */
	onNotification(handler: CodexAppServerNotificationHandler): () => void {
		this.notificationHandlers.add(handler);
		return () => {
			this.notificationHandlers.delete(handler);
		};
	}

	/**
	 * Send a JSON-RPC request and resolve with the response result.
	 *
	 * Purpose: provide typed call sites for app-server methods while centralizing response/error handling.
	 * Consumer: provider setup, turn start, context injection, and cancellation paths.
	 */
	request(
		method: string,
		params?: unknown,
		signal?: AbortSignal,
	): Promise<unknown> {
		if (this.closed) {
			return Promise.reject(
				this.exitError ?? new Error("codex app-server is closed."),
			);
		}
		if (signal?.aborted) {
			return Promise.reject(new Error("Request was aborted"));
		}

		const id = this.nextId++;
		const message =
			params === undefined ? { id, method } : { id, method, params };

		return new Promise<unknown>((resolve, reject) => {
			const abort = () => {
				this.pending.delete(id);
				reject(new Error("Request was aborted"));
			};

			this.pending.set(id, {
				resolve: (value) => {
					signal?.removeEventListener("abort", abort);
					resolve(value);
				},
				reject: (error) => {
					signal?.removeEventListener("abort", abort);
					reject(error);
				},
			});

			signal?.addEventListener("abort", abort, { once: true });

			this.send(message).catch((error: unknown) => {
				this.pending.delete(id);
				signal?.removeEventListener("abort", abort);
				reject(error instanceof Error ? error : new Error(String(error)));
			});
		});
	}

	/**
	 * Send a JSON-RPC notification.
	 *
	 * Purpose: acknowledge initialization and support fire-and-forget app-server protocol calls.
	 * Consumer: initialize() for the required `initialized` notification.
	 */
	async notify(method: string, params?: unknown): Promise<void> {
		const message = params === undefined ? { method } : { method, params };
		await this.send(message);
	}

	/**
	 * Interrupt an active Codex turn.
	 *
	 * Purpose: translate an AbortSignal into Codex's cooperative turn cancellation RPC.
	 * Consumer: openai-codex-responses.ts abort handling.
	 */
	async interruptTurn(threadId: string, turnId: string): Promise<void> {
		await this.request("turn/interrupt", { threadId, turnId });
	}

	/**
	 * Dispose the child process and reject pending requests.
	 *
	 * Purpose: release the long-running Codex process when the owning session is shutting down.
	 * Consumer: tests, future host lifecycle hooks, and crash recovery.
	 */
	async dispose(): Promise<void> {
		if (this.closed) return;
		this.closed = true;
		clearCodexAppServerClient(this);
		this.readline?.close();
		if (this.proc && !this.proc.killed) {
			this.proc.kill("SIGTERM");
		}
		this.rejectPending(new Error("codex app-server was disposed."));
	}

	/**
	 * Dispose the client if it has no pending requests and no active notification
	 * handlers. The check is deferred by one event-loop turn so a consumer that is
	 * about to register a new handler wins the race.
	 *
	 * Purpose: allow short-lived processes (smoke tests, one-shot scripts) to exit
	 * cleanly without leaking the codex app-server child process, while keeping the
	 * client alive across back-to-back turns in a long-running session.
	 */
	releaseIfIdle(): Promise<void> {
		return new Promise((resolve) => {
			setImmediate(() => {
				if (
					this.closed ||
					this.pending.size > 0 ||
					this.notificationHandlers.size > 0
				) {
					resolve();
					return;
				}
				this.dispose().then(
					() => resolve(),
					() => resolve(),
				);
			});
		});
	}

	private async initialize(): Promise<void> {
		const childProcessModule = (await dynamicImport(
			NODE_CHILD_PROCESS_SPECIFIER,
		)) as typeof import("node:child_process");
		const readlineModule = (await dynamicImport(
			NODE_READLINE_SPECIFIER,
		)) as typeof import("node:readline");
		const args = [
			"app-server",
			"--listen",
			"stdio://",
			...(this.options.extraArgs ?? []),
		];

		try {
			this.proc = childProcessModule.spawn("codex", args, {
				cwd: this.options.cwd ?? process.cwd(),
				env: this.options.env ?? process.env,
				stdio: ["pipe", "pipe", "pipe"],
				shell: process.platform === "win32",
				windowsHide: true,
			});
		} catch (error) {
			throw this.toSpawnError(error);
		}

		this.proc.stdout.setEncoding("utf8");
		this.proc.stderr.setEncoding("utf8");
		this.proc.stderr.on("data", (chunk: string) => {
			this.stderr = (this.stderr + chunk).slice(-12000);
		});
		this.proc.on("error", (error) => {
			this.handleExit(this.toSpawnError(error));
		});
		this.proc.on("exit", (code, signal) => {
			if (this.closed) return;
			const detail = signal ? `signal ${signal}` : `exit ${code ?? "unknown"}`;
			this.handleExit(
				new Error(
					`codex app-server exited unexpectedly (${detail}).${this.stderrSuffix()}`,
				),
			);
		});

		this.readline = readlineModule.createInterface({ input: this.proc.stdout });
		this.readline.on("line", (line) => this.handleLine(line));

		await this.request("initialize", {
			clientInfo: this.options.clientInfo ?? DEFAULT_CLIENT_INFO,
			capabilities: { experimentalApi: true },
		});
		await this.notify("initialized");
	}

	private async send(message: JsonObject): Promise<void> {
		if (!this.proc?.stdin || this.closed) {
			throw (
				this.exitError ?? new Error("codex app-server stdin is not available.")
			);
		}
		const line = `${JSON.stringify(message)}\n`;
		if (this.proc.stdin.write(line)) return;
		await new Promise<void>((resolve, reject) => {
			const stdin = this.proc?.stdin;
			if (!stdin) {
				reject(new Error("codex app-server stdin is not available."));
				return;
			}
			const onDrain = () => {
				cleanup();
				resolve();
			};
			const onError = (error: Error) => {
				cleanup();
				reject(error);
			};
			const cleanup = () => {
				stdin.off("drain", onDrain);
				stdin.off("error", onError);
			};
			stdin.once("drain", onDrain);
			stdin.once("error", onError);
		});
	}

	private handleLine(line: string): void {
		let message: unknown;
		try {
			message = JSON.parse(line);
		} catch {
			return;
		}
		if (!message || typeof message !== "object") return;
		const object = message as JsonObject;

		if (typeof object.id === "number" && !("method" in object)) {
			this.handleResponse(object as unknown as JsonRpcResponse);
			return;
		}
		if (typeof object.method === "string" && typeof object.id === "number") {
			this.handleServerRequest(object as unknown as JsonRpcServerRequest);
			return;
		}
		if (typeof object.method === "string") {
			this.handleNotification({ method: object.method, params: object.params });
		}
	}

	private handleResponse(response: JsonRpcResponse): void {
		const pending = this.pending.get(response.id);
		if (!pending) return;
		this.pending.delete(response.id);
		if (response.error) {
			pending.reject(
				new Error(`${response.error.message} (code ${response.error.code})`),
			);
			return;
		}
		pending.resolve(response.result);
	}

	private handleNotification(notification: CodexAppServerNotification): void {
		for (const handler of this.notificationHandlers) {
			handler(notification);
		}
	}

	private handleServerRequest(request: JsonRpcServerRequest): void {
		this.send({
			id: request.id,
			error: {
				code: -32601,
				message: `Unsupported codex app-server request: ${request.method}`,
			},
		}).catch(() => {});
	}

	private handleExit(error?: Error): void {
		if (this.closed && !error) return;
		this.closed = true;
		this.exitError = error ?? new Error("codex app-server connection closed.");
		clearCodexAppServerClient(this);
		this.readline?.close();
		this.rejectPending(this.exitError);
	}

	private rejectPending(error: Error): void {
		for (const pending of this.pending.values()) {
			pending.reject(error);
		}
		this.pending.clear();
	}

	private toSpawnError(error: unknown): Error {
		const err = error instanceof Error ? error : new Error(String(error));
		const nodeError = err as Error & { code?: string };
		if (nodeError.code === "ENOENT") {
			return new Error(
				"Codex CLI was not found. Install the OpenAI Codex CLI and make sure `codex` is on PATH, then run `codex login` if needed.",
			);
		}
		return err;
	}

	private stderrSuffix(): string {
		const trimmed = this.stderr.trim();
		return trimmed ? ` stderr: ${trimmed}` : "";
	}
}
