import { type ChildProcessWithoutNullStreams, spawn } from 'node:child_process'
import { decodeSupervisorState, encodeSupervisorMessage, type SupervisorStateMessage } from './supervisor-protocol'

interface SupervisorProcessLike {
	readonly stdin: { write(data: string): boolean; once(event: string, listener: (...args: unknown[]) => void): unknown }
	readonly stdout: {
		setEncoding(encoding: string): unknown
		on(event: string, listener: (...args: unknown[]) => void): unknown
	}
	on(event: string, listener: (...args: unknown[]) => void): unknown
	once(event: string, listener: (...args: unknown[]) => void): unknown
}

type SupervisorProcessSpawner = (
	file: string,
	args: readonly string[],
	options: {
		stdio: ['pipe', 'pipe', 'inherit']
		env: NodeJS.ProcessEnv
	},
) => SupervisorProcessLike

interface SupervisorClientOptions {
	readonly executable?: string
	readonly parentPid?: number
	readonly environment?: NodeJS.ProcessEnv
	readonly spawn?: SupervisorProcessSpawner
	readonly onState?: (state: SupervisorStateMessage) => void
	readonly onFailure?: (error: Error) => void
}

const DEFAULT_SUPERVISOR_EXECUTABLE = '/usr/local/bin/firehose-supervisor'
const MAX_STATE_BUFFER = 64 * 1024

/** Worker-side client. It never owns or renews the Redis/PG authority itself. */
export class SupervisorClient {
	private readonly executable: string
	private readonly parentPid: number
	private readonly environment: NodeJS.ProcessEnv
	private readonly spawnProcess: SupervisorProcessSpawner
	private readonly onState?: (state: SupervisorStateMessage) => void
	private readonly onFailure?: (error: Error) => void
	private child: SupervisorProcessLike | null = null
	private inputBuffer = ''
	private stateMessage: SupervisorStateMessage | null = null
	private startPromise: Promise<void> | null = null
	private releasePromise: Promise<void> | null = null
	private acquiredResolve: ((state: SupervisorStateMessage) => void) | null = null
	private acquiredReject: ((error: Error) => void) | null = null
	private releaseResolve: (() => void) | null = null
	private releaseReject: ((error: Error) => void) | null = null
	private failureReported = false

	constructor(options: SupervisorClientOptions = {}) {
		this.executable = options.executable ?? process.env.FIREHOSE_SUPERVISOR_PATH ?? DEFAULT_SUPERVISOR_EXECUTABLE
		this.parentPid = options.parentPid ?? process.pid
		this.environment = options.environment ?? process.env
		this.spawnProcess =
			options.spawn ??
			((file, args, spawnOptions) => spawn(file, args, spawnOptions) as unknown as ChildProcessWithoutNullStreams)
		this.onState = options.onState
		this.onFailure = options.onFailure
	}

	get state(): SupervisorStateMessage | null {
		return this.stateMessage
	}

	get process(): SupervisorProcessLike | null {
		return this.child
	}

	start(): Promise<void> {
		if (this.startPromise) return this.startPromise
		this.startPromise = new Promise<void>((resolve, reject) => {
			let child: SupervisorProcessLike
			try {
				child = this.spawnProcess(this.executable, [`--parent-pid=${this.parentPid}`], {
					stdio: ['pipe', 'pipe', 'inherit'],
					env: {
						...this.environment,
						SUPERVISOR_PARENT_PID: String(this.parentPid),
					},
				})
			} catch (error) {
				reject(error instanceof Error ? error : new Error('Failed to start leadership supervisor'))
				return
			}
			this.child = child
			child.stdout.setEncoding('utf8')
			child.stdout.on('data', (chunk: unknown) => this.consumeOutput(String(chunk)))
			child.stdout.on('error', () => this.fail(new Error('Leadership supervisor output pipe failed')))
			child.on('error', (error: unknown) =>
				this.fail(error instanceof Error ? error : new Error('Leadership supervisor failed')),
			)
			child.on('close', (code: unknown) => {
				if (this.stateMessage?.state === 'released') return
				this.fail(new Error(`Leadership supervisor exited unexpectedly (${String(code)})`))
			})
			try {
				const accepted = child.stdin.write(
					encodeSupervisorMessage({ version: 1, type: 'hello', parentPid: this.parentPid }),
				)
				if (!accepted) throw new Error('Leadership supervisor command pipe backpressure')
			} catch (error) {
				this.fail(error instanceof Error ? error : new Error('Leadership supervisor command pipe failed'))
				reject(error instanceof Error ? error : new Error('Leadership supervisor command pipe failed'))
				return
			}
			resolve()
		})
		return this.startPromise
	}

	waitForAcquired(signal?: AbortSignal): Promise<SupervisorStateMessage> {
		if (this.stateMessage?.state === 'acquired') return Promise.resolve(this.stateMessage)
		if (this.stateMessage?.state === 'fatal')
			return Promise.reject(new Error(this.stateMessage.reason ?? 'Supervisor failed'))
		return new Promise<SupervisorStateMessage>((resolve, reject) => {
			this.acquiredResolve = resolve
			this.acquiredReject = reject
			if (signal) {
				if (signal.aborted) {
					reject(new Error('Leadership supervisor wait aborted'))
					return
				}
				signal.addEventListener('abort', () => reject(new Error('Leadership supervisor wait aborted')), { once: true })
			}
		})
	}

	/** Send release only after the worker has stopped all firehose/cursor activity. */
	requestRelease(): Promise<void> {
		if (this.stateMessage?.state === 'released') return Promise.resolve()
		if (this.releasePromise) return this.releasePromise
		this.releasePromise = new Promise<void>((resolve, reject) => {
			this.releaseResolve = resolve
			this.releaseReject = reject
			const child = this.child
			if (!child) {
				reject(new Error('Leadership supervisor is not running'))
				return
			}
			try {
				const accepted = child.stdin.write(encodeSupervisorMessage({ version: 1, type: 'release' }))
				if (!accepted) throw new Error('Leadership supervisor command pipe backpressure')
			} catch (error) {
				reject(error instanceof Error ? error : new Error('Leadership supervisor release failed'))
			}
		})
		return this.releasePromise
	}

	private consumeOutput(chunk: string): void {
		this.inputBuffer += chunk
		if (this.inputBuffer.length > MAX_STATE_BUFFER) {
			this.fail(new Error('Leadership supervisor state exceeded limit'))
			return
		}
		while (true) {
			const newline = this.inputBuffer.indexOf('\n')
			if (newline < 0) return
			const line = this.inputBuffer.slice(0, newline).trim()
			this.inputBuffer = this.inputBuffer.slice(newline + 1)
			if (!line) {
				this.fail(new Error('Leadership supervisor sent an empty state'))
				return
			}
			let message: SupervisorStateMessage
			try {
				message = decodeSupervisorState(line)
			} catch {
				this.fail(new Error('Leadership supervisor sent malformed state'))
				return
			}
			this.stateMessage = message
			this.onState?.(message)
			if (message.state === 'acquired') {
				this.acquiredResolve?.(message)
				this.acquiredResolve = null
				this.acquiredReject = null
			}
			if (message.state === 'fatal') {
				this.acquiredReject?.(new Error(message.reason ?? 'Leadership supervisor failed'))
				this.acquiredResolve = null
				this.acquiredReject = null
				this.releaseReject?.(new Error(message.reason ?? 'Leadership supervisor failed'))
				this.releaseResolve = null
				this.releaseReject = null
			}
			if (message.state === 'released') {
				this.releaseResolve?.()
				this.releaseResolve = null
				this.releaseReject = null
			}
		}
	}

	private fail(error: Error): void {
		this.acquiredReject?.(error)
		this.acquiredResolve = null
		this.acquiredReject = null
		this.releaseReject?.(error)
		this.releaseResolve = null
		this.releaseReject = null
		if (!this.failureReported) {
			this.failureReported = true
			this.onFailure?.(error)
		}
	}
}