import { Buffer } from 'node:buffer';
import { type ChildProcess, spawn } from 'node:child_process';
import { EventEmitter, once } from 'node:events';
import {
	createServer,
	type IncomingHttpHeaders,
	type IncomingMessage,
	type OutgoingHttpHeaders,
	request as httpRequest,
	type Server,
	type ServerResponse,
} from 'node:http';
import { type AddressInfo, connect as connectSocket } from 'node:net';
import type { Duplex } from 'node:stream';
import { json } from 'node:stream/consumers';
import { pipeline } from 'node:stream/promises';

import {
	type Connect,
	DevEnvironment,
	type FetchResult,
	type HotChannel,
	type HotChannelClient,
	type HotChannelListener,
	type HotPayload,
	loadEnv,
	type ResolvedConfig,
} from 'vite';

import { DENO_SPECIFIER_RE } from '../deno-specifiers.ts';
import {
	ERROR_HEADER,
	type HostMessage,
	type RunnerMessage,
	type SerializedError,
} from '../runtime/protocol.ts';

interface DenoDevEnvironmentOptions {
	/** absolute path to the server entrypoint */
	entry: string;
	/** flags passed to `deno run` */
	args: string[];
}

// connection-specific response headers must not reach HTTP/2 clients
const HOP_BY_HOP_HEADERS = new Set(['connection', 'keep-alive', 'proxy-connection', 'transfer-encoding']);

const getUpstreamHeaders = (headers: IncomingHttpHeaders): OutgoingHttpHeaders => {
	// preserve the public host; HTTP/2 carries it in :authority
	const result: OutgoingHttpHeaders = {};
	for (const [name, value] of Object.entries(headers)) {
		if (name === ':authority') {
			if (typeof value === 'string') {
				result.host ??= value;
			}
		} else if (!name.startsWith(':')) {
			// other HTTP/2 pseudo-headers aren't valid over HTTP/1
			result[name] = value;
		}
	}

	return result;
};

const getDownstreamHeaders = (headers: IncomingHttpHeaders): OutgoingHttpHeaders => {
	const result: OutgoingHttpHeaders = {};
	for (const [name, value] of Object.entries(headers)) {
		if (!HOP_BY_HOP_HEADERS.has(name)) {
			result[name] = value;
		}
	}

	return result;
};

/** shares a start promise; retries after failure */
const createLazyStart = <T>(start: () => Promise<T>) => {
	let current: Promise<T> | undefined;

	return {
		get(): Promise<T> {
			if (!current) {
				const promise = start();
				promise.catch(() => {
					if (current === promise) {
						current = undefined;
					}
				});

				current = promise;
			}

			return current;
		},
		clear(): void {
			current = undefined;
		},
	};
};

// #region hot channel

const createIpcChannel = () => {
	const emitter = new EventEmitter();
	let child: ChildProcess | undefined;
	let fullReloadListener: (() => void) | undefined;

	const post = (message: HostMessage) => {
		if (child?.connected) {
			child.send(message);
		}
	};

	const send = (payload: HotPayload) => {
		post({ type: 'hot', payload });
	};

	const client: HotChannelClient = { send };

	const channel: HotChannel = {
		// the channel is only reachable by the child process, never over the network
		skipFsCheck: true,
		send(payload: HotPayload) {
			// in-place reloads leave old workers, timers, and listeners running
			if (payload.type === 'full-reload') {
				fullReloadListener?.();
				return;
			}

			send(payload);
		},
		on(event: string, listener: HotChannelListener) {
			emitter.on(event, listener);
		},
		off(event: string, listener: HotChannelListener) {
			emitter.off(event, listener);
		},
	};

	return {
		channel,
		post,
		get child() {
			return child;
		},
		set child(proc: ChildProcess | undefined) {
			child = proc;
		},
		receive(payload: HotPayload) {
			if (payload.type === 'custom') {
				emitter.emit(payload.event, payload.data, client);
			}
		},
		onFullReload(listener: () => void) {
			fullReloadListener = listener;
		},
	};
};

// #endregion

// #region environment

/** runs server modules in Deno and proxies requests to their `fetch` handler */
export class DenoDevEnvironment extends DevEnvironment {
	#options: DenoDevEnvironmentOptions;
	#ipc: ReturnType<typeof createIpcChannel>;
	#port = createLazyStart(() => this.#spawn());
	#closed = false;
	#exited: Promise<void> | undefined;

	#assetServer: Server | undefined;
	#assetsOrigin = createLazyStart(() => this.#listenAssets());
	#assetHandler: Connect.Server | undefined;
	#assetRequests = new WeakSet<IncomingMessage>();

	constructor(name: string, config: ResolvedConfig, options: DenoDevEnvironmentOptions) {
		const ipc = createIpcChannel();
		super(name, config, { hot: true, transport: ipc.channel });

		this.#options = options;
		this.#ipc = ipc;

		ipc.onFullReload(() => this.#restart());
	}

	override fetchModule(...args: Parameters<DevEnvironment['fetchModule']>): Promise<FetchResult> {
		// Vite prefixes unresolved ids with /@id/; Deno can import its own specifiers directly
		const id = args[0].replace(/^\/@id\//, '');
		if (DENO_SPECIFIER_RE.test(id)) {
			return Promise.resolve({ externalize: id, type: 'module' });
		}

		return super.fetchModule(...args);
	}

	override async close(): Promise<void> {
		this.#closed = true;
		this.#stop();
		this.#assetServer?.close();
		this.#assetServer = undefined;
		this.#assetsOrigin.clear();
		await super.close();
	}

	/**
	 * starts the Deno process in the background.
	 *
	 * @param middlewares Vite's middleware stack
	 */
	connect(middlewares: Connect.Server): void {
		this.#assetHandler = middlewares;
		this.#start().catch(() => {});
	}

	/**
	 * identifies asset requests to prevent dispatch back into the server entrypoint.
	 *
	 * @param req incoming request
	 * @returns whether `req` came from `fetchAsset()`
	 */
	isAssetRequest(req: IncomingMessage): boolean {
		return this.#assetRequests.has(req);
	}

	/**
	 * forwards a request to the server entrypoint.
	 *
	 * @param req incoming request
	 * @param res outgoing response
	 * @param next next middleware
	 */
	async dispatch(
		req: Connect.IncomingMessage,
		res: ServerResponse,
		next: Connect.NextFunction,
	): Promise<void> {
		if (this.#closed || this.isAssetRequest(req)) {
			return next();
		}

		try {
			await this.#proxy(req, res);
		} catch (error) {
			next(error);
		}
	}

	async #proxy(req: Connect.IncomingMessage, res: ServerResponse): Promise<void> {
		const port = await this.#start();
		const { promise, resolve, reject } = Promise.withResolvers<void>();

		const fail = (error: unknown) => {
			// client disconnects are not proxy failures
			if (res.destroyed) {
				resolve();
			} else {
				reject(error);
			}
		};

		const upstream = httpRequest(
			{
				hostname: '127.0.0.1',
				port,
				method: req.method,
				path: req.originalUrl ?? req.url,
				headers: getUpstreamHeaders(req.headers),
			},
			(response) => {
				if (response.headers[ERROR_HEADER] !== undefined) {
					// delegate logging and the error overlay to Vite's error middleware
					json(response).then((body) => {
						const { message, ...fields } = body as SerializedError;
						reject(Object.assign(new Error(message), fields));
					}, fail);
					return;
				}

				res.writeHead(response.statusCode ?? 500, getDownstreamHeaders(response.headers));
				pipeline(response, res).then(resolve, fail);
			},
		);

		res.once('close', () => {
			// cancel upstream streaming when the client disconnects
			if (!res.writableFinished) {
				upstream.destroy();
			}
		});

		upstream.on('error', fail);
		req.pipe(upstream);

		return promise;
	}

	/**
	 * forwards a WebSocket upgrade request to the server entrypoint.
	 *
	 * @param req upgrade request
	 * @param socket client socket
	 * @param head bytes received after the request headers
	 */
	async upgrade(req: IncomingMessage, socket: Duplex, head: Buffer): Promise<void> {
		if (this.#closed) {
			socket.destroy();
			return;
		}

		// Node removes its upgrade socket error handler; catch resets while Deno starts
		const onEarlyError = () => socket.destroy();
		socket.on('error', onEarlyError);

		let port: number;
		try {
			port = await this.#start();
		} catch (error) {
			this.logger.error(`[deno-server] WebSocket upgrade failed: ${error}`);
			socket.destroy();
			return;
		}

		// let Deno handle the handshake
		const lines = [`${req.method} ${req.url} HTTP/${req.httpVersion}`];
		for (let i = 0; i < req.rawHeaders.length; i += 2) {
			lines.push(`${req.rawHeaders[i]}: ${req.rawHeaders[i + 1]}`);
		}

		// avoid Nagle delays for small frames
		const upstream = connectSocket({ port, host: '127.0.0.1', noDelay: true });
		upstream.write(Buffer.concat([Buffer.from(`${lines.join('\r\n')}\r\n\r\n`), head]));

		socket.off('error', onEarlyError);
		pipeline(socket, upstream, socket).catch(() => {});
	}

	/** returns the port of the server in the Deno process, starting it if it isn't running */
	#start(): Promise<number> {
		if (this.#closed) {
			return Promise.reject(new Error(`the dev environment is closed`));
		}

		return this.#port.get();
	}

	#restart(): void {
		if (this.#closed) {
			return;
		}

		this.logger.info(`[deno-server] restarting Deno`, { timestamp: true });

		this.#stop();
		this.#start().catch(() => {});
	}

	#stop(): void {
		const child = this.#ipc.child;
		// preserve a pending start while the previous process exits
		if (!child) {
			return;
		}

		this.#reset();

		// a failed spawn emits `error` rather than `exit`; either completes the wait
		const settle = () => {};
		this.#exited = once(child, 'exit').then(settle, settle);

		// the runner exits on disconnect, even when the entrypoint traps SIGTERM
		if (child.connected) {
			child.disconnect();
		}
		child.kill();
	}

	#reset(): void {
		this.#ipc.child = undefined;
		this.#port.clear();
	}

	/**
	 * starts the asset server on demand and returns its origin.
	 *
	 * @returns HTTP origin on `127.0.0.1`
	 * @throws after {@link close}, or before {@link connect} configures the asset middleware
	 */
	getAssetsOrigin(): Promise<string> {
		if (this.#closed) {
			return Promise.reject(new Error(`the dev environment is closed`));
		}

		return this.#assetsOrigin.get();
	}

	async #listenAssets(): Promise<string> {
		const handler = this.#assetHandler;
		if (!handler) {
			throw new Error(`call connect() before starting the asset server`);
		}

		const server = createServer((req, res) => {
			this.#assetRequests.add(req);
			handler(req, res, (error?: unknown) => {
				if (error) {
					this.logger.error(`[deno-server] fetchAsset() failed: ${error}`);
					res.writeHead(500).end();
				} else {
					res.writeHead(404).end();
				}
			});
		});

		this.#assetServer = server;
		server.listen(0, '127.0.0.1');
		await once(server, 'listening');

		return `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
	}

	async #spawn(): Promise<number> {
		// wait for the previous process to release entrypoint ports and locks
		await this.#exited;

		if (this.#closed) {
			throw new Error(`the dev environment is closed`);
		}

		const { entry, args } = this.#options;
		const { promise, resolve, reject } = Promise.withResolvers<number>();

		const { envDir, mode, root } = this.config;
		const child = spawn('deno', ['run', ...args, import.meta.resolve('../runtime/runner.ts')], {
			cwd: root,
			// load all variables for Deno.env; process.env takes precedence over .env files
			env: loadEnv(mode, envDir, ''),
			stdio: ['ignore', 'inherit', 'inherit', 'ipc'],
			serialization: 'json',
		});

		this.#ipc.child = child;
		this.#ipc.post({
			type: 'init',
			entry,
			moduleRunner: import.meta.resolve('vite/module-runner'),
		});

		child.on('message', (message: RunnerMessage) => {
			// a stopped process can still deliver messages until it exits
			if (this.#ipc.child !== child) {
				return;
			}

			switch (message.type) {
				case 'hot': {
					this.#ipc.receive(message.payload);
					break;
				}
				case 'invoke': {
					// handleInvoke returns handler failures as { error }
					this.hot.handleInvoke(message.payload).then((result) => {
						this.#ipc.post({ type: 'result', id: message.id, result });
					});
					break;
				}
				case 'log': {
					this.logger[message.level](message.message, { timestamp: true });
					break;
				}
				case 'ready': {
					resolve(message.port);
					break;
				}
				case 'restart': {
					this.#restart();
					break;
				}
			}
		});

		// log process failures here so background starts can ignore rejections
		child.once('exit', (code, signal) => {
			const reason = signal ? `signal ${signal}` : `code ${code}`;
			reject(new Error(`Deno process exited with ${reason}`));

			if (this.#ipc.child === child) {
				this.#reset();
				this.logger.error(`[deno-server] Deno exited with ${reason}; restarting on next request`);
			}
		});

		child.once('error', (error) => {
			reject(error);

			if (this.#ipc.child === child) {
				this.#reset();
				this.logger.error(`[deno-server] failed to start Deno: ${error.message}; retrying on next request`);
			}
		});

		return promise;
	}
}

// #endregion