/**
 * Going on and off air for one channel: a source produces one MPEG-TS
 * stream; it is copied to one small ffmpeg per destination (Twitch,
 * YouTube, …) so a failing destination reconnects without touching the
 * others, and cut into 2-second HLS segments that are uploaded to storage:
 * the live stream on the Vash site and the recording afterwards.
 *
 * Vash TV uses it with its own screen encoder; the Windows worker with an
 * RTMP ingest that OBS streams into.
 */
import { mkdir, readdir, readFile, rm } from 'node:fs/promises';
import { join } from 'node:path';
import type { Subprocess } from 'bun';
import type { StreamConfig, VashApi } from './vash-api';

export type Log = (...parts: unknown[]) => void;

export interface OutputStatus {
	id: string;
	label: string;
	state: Output['state'];
	error: string | null;
}

/** One consumer of the encoded stream: a destination, or the recorder. */
export class Output {
	state: 'starting' | 'live' | 'reconnecting' | 'error' = 'starting';
	error: string | null = null;
	private process: Subprocess<'pipe', 'ignore', 'pipe'> | null = null;
	private since = 0;
	private queue: Uint8Array[] = [];
	private queued = 0;
	private writing = false;
	private attempts = 0;
	private stopped = false;

	constructor(
		readonly id: string,
		readonly label: string,
		private args: string[],
		private ffmpeg: string,
		private log: Log,
		/** The destination URL, to notice when its settings change. */
		readonly url: string | null = null
	) {
		this.spawn();
	}

	private spawn() {
		if (this.stopped) return;
		const process = Bun.spawn(
			[
				this.ffmpeg,
				'-hide_banner',
				'-loglevel',
				'error',
				'-f',
				'mpegts',
				'-i',
				'pipe:0',
				...this.args
			],
			{ stdin: 'pipe', stdout: 'ignore', stderr: 'pipe', windowsHide: true }
		);
		this.process = process;
		this.since = Date.now();
		this.queue = [];
		this.queued = 0;
		void (async () => {
			let last = '';
			for await (const chunk of process.stderr)
				last = new TextDecoder().decode(chunk).trim().split('\n').pop() || last;
			const code = await process.exited;
			if (this.stopped || this.process !== process) return;
			this.error = last || `ffmpeg exited with ${code}`;
			this.state = this.attempts > 3 ? 'error' : 'reconnecting';
			const wait = Math.min(30_000, 2000 * 2 ** this.attempts++);
			this.log(`${this.label}: ${this.error}; retrying in ${Math.round(wait / 1000)}s`);
			setTimeout(() => this.spawn(), wait);
		})();
	}

	/**
	 * Queues bytes in order. Dropping any would corrupt the MPEG-TS timestamps,
	 * so a destination that falls far behind is restarted instead.
	 */
	write(chunk: Uint8Array) {
		const process = this.process;
		if (!process || this.stopped) return;
		if (this.state !== 'live' && Date.now() - this.since > 5000 && process.exitCode == null) {
			this.state = 'live';
			this.error = null;
			this.attempts = 0;
		}
		this.queue.push(chunk);
		this.queued += chunk.byteLength;
		if (this.queued > 32 * 1024 * 1024) {
			this.error = 'Destination too slow; restarting';
			this.queue = [];
			this.queued = 0;
			process.kill();
			return;
		}
		void this.drain(process);
	}

	private async drain(process: Subprocess<'pipe', 'ignore', 'pipe'>) {
		if (this.writing) return;
		this.writing = true;
		try {
			while (this.queue.length && this.process === process) {
				const chunk = this.queue.shift()!;
				this.queued -= chunk.byteLength;
				process.stdin.write(chunk);
				const flushed = process.stdin.flush();
				if (flushed instanceof Promise) await flushed;
			}
		} catch {
			// The process is exiting; its watcher restarts it.
		} finally {
			this.writing = false;
		}
	}

	status(): OutputStatus {
		return { id: this.id, label: this.label, state: this.state, error: this.error };
	}

	stop() {
		this.stopped = true;
		try {
			this.process?.stdin.end();
		} catch {
			/* already closed */
		}
		const process = this.process;
		setTimeout(() => process?.kill(), 3000);
	}
}

/** Produces the one MPEG-TS stream every output copies. */
export interface Source {
	/** Starts producing; each chunk goes to write. */
	start(config: StreamConfig, write: (chunk: Uint8Array) => void): Promise<void> | void;
	stop(): void;
	readonly running: boolean;
}

/** Listens for one RTMP publisher (OBS) and passes its stream on unchanged. */
export class RtmpIngest implements Source {
	private process: Subprocess<'ignore', 'pipe', 'pipe'> | null = null;
	/** When the current listener started; OBS should connect after it is up. */
	listeningSince = 0;
	/** When a publisher last sent data. */
	receivedAt = 0;

	constructor(
		private options: { ffmpeg: string; host: string; port: number; key: string; log: Log }
	) {}

	get running() {
		return !!this.process && this.process.exitCode == null && !this.process.killed;
	}

	/** Receiving data from OBS in the last few seconds. */
	get receiving() {
		return this.running && Date.now() - this.receivedAt < 5000;
	}

	get url() {
		return `rtmp://${this.options.host}:${this.options.port}/live`;
	}

	start(_config: StreamConfig, write: (chunk: Uint8Array) => void) {
		if (this.running) return;
		const { ffmpeg, host, port, key, log } = this.options;
		const process = Bun.spawn(
			[
				ffmpeg,
				'-hide_banner',
				'-loglevel',
				'error',
				'-listen',
				'1',
				'-i',
				`rtmp://${host}:${port}/live/${key}`,
				'-c',
				'copy',
				'-f',
				'mpegts',
				'pipe:1'
			],
			{ stdout: 'pipe', stderr: 'pipe', windowsHide: true }
		);
		this.process = process;
		this.listeningSince = Date.now();
		log(`Waiting for OBS on ${this.url}`);
		void (async () => {
			for await (const chunk of process.stdout) {
				this.receivedAt = Date.now();
				write(chunk);
			}
		})();
		void (async () => {
			let last = '';
			for await (const chunk of process.stderr)
				last = new TextDecoder().decode(chunk).trim() || last;
			const code = await process.exited;
			// OBS stopped or dropped; the next apply() listens again.
			if (this.process === process) {
				this.process = null;
				if (code !== 0 && last) log(`Ingest stopped (${code}): ${last}`);
			}
		})();
	}

	stop() {
		const process = this.process;
		this.process = null;
		process?.kill();
	}
}

interface Session {
	id: string;
	storagePrefix: string | null;
	startedAt: number;
}

export interface BroadcasterOptions {
	/** Null streams without Vash managing sessions (no recording). */
	api: VashApi | null;
	ffmpeg: string;
	hlsDir: string;
	source: Source;
	log: Log;
}

/** Follows a channel's desired state: on or off air, destinations, recording. */
export class Broadcaster {
	onAir = false;
	session: Session | null = null;
	private outputs = new Map<string, Output>();
	private vod: { name: string; duration: number }[] = [];
	private uploaded = new Set<string>();
	private lastVodUpload = 0;
	private posterDone = false;
	private lastReport = 0;
	private config: StreamConfig | null = null;
	private busy = false;

	constructor(private options: BroadcasterOptions) {}

	private get log() {
		return this.options.log;
	}

	/** Reconciles with the latest channel settings. Call every few seconds. */
	async apply(config: StreamConfig) {
		if (this.busy) return;
		this.busy = true;
		try {
			this.config = config;
			if (config.live && !this.onAir) await this.goLive(config);
			else if (!config.live && this.onAir) await this.goOffAir();
			else if (this.onAir) {
				this.syncDestinations(config);
				if (!this.options.source.running)
					await this.startSource(config).catch((err) =>
						this.log('The stream source did not start:', (err as Error).message)
					);
			}
		} finally {
			this.busy = false;
		}
	}

	/** Uploads recording segments and reports destination states. Call every second. */
	async tick() {
		if (!this.onAir) return;
		await this.pumpRecording().catch((err) =>
			this.log('Recording upload failed:', (err as Error).message)
		);
		if (Date.now() - this.lastReport > 10_000) {
			this.lastReport = Date.now();
			await this.report();
		}
	}

	destinations(): OutputStatus[] {
		return [...this.outputs.values()]
			.filter((output) => output.id !== 'recording')
			.map((output) => output.status());
	}

	get recording() {
		return this.outputs.has('recording');
	}

	private write = (chunk: Uint8Array) => {
		for (const output of this.outputs.values()) output.write(chunk);
	};

	private async startSource(config: StreamConfig) {
		await this.options.source.start(config, this.write);
	}

	private async goLive(config: StreamConfig) {
		const { api, hlsDir, ffmpeg } = this.options;
		if (api) {
			const started = await api.report<{ id: string; storagePrefix: string | null }>({
				event: 'start'
			});
			this.session = { ...started, startedAt: Date.now() };
			this.vod.length = 0;
			this.uploaded.clear();
			this.posterDone = false;
		}
		if (config.record && this.session?.storagePrefix) {
			await rm(hlsDir, { recursive: true, force: true });
			await mkdir(hlsDir, { recursive: true });
			this.outputs.set(
				'recording',
				new Output(
					'recording',
					'Recording',
					[
						'-c',
						'copy',
						'-f',
						'hls',
						'-hls_time',
						'2',
						'-hls_list_size',
						'8',
						'-hls_segment_filename',
						join(hlsDir, 'seg_%05d.ts'),
						join(hlsDir, 'index.m3u8')
					],
					ffmpeg,
					this.log
				)
			);
		}
		this.syncDestinations(config);
		// On air from here: the session exists, so a source that fails to start is
		// retried by apply() rather than opening another session.
		this.onAir = true;
		await this.startSource(config).catch((err) =>
			this.log('The stream source did not start:', (err as Error).message)
		);
		this.log(
			`On air: ${config.destinations.map((each) => each.label).join(', ') || 'recording only'}`
		);
	}

	private async goOffAir() {
		this.onAir = false;
		this.options.source.stop();
		for (const output of this.outputs.values()) output.stop();
		this.outputs.clear();
		await Bun.sleep(2500);
		await this.pumpRecording(true).catch((err) =>
			this.log('Final upload failed:', (err as Error).message)
		);
		const { api } = this.options;
		if (api && this.session) {
			await api
				.report({
					event: 'end',
					sessionId: this.session.id,
					durationSec: Math.round((Date.now() - this.session.startedAt) / 1000)
				})
				.catch(() => {});
		}
		this.session = null;
		this.log('Off air');
	}

	private syncDestinations(config: StreamConfig) {
		const wanted = new Map(config.destinations.map((destination) => [destination.id, destination]));
		for (const [id, output] of this.outputs) {
			if (id === 'recording') continue;
			const destination = wanted.get(id);
			if (!destination || output.url !== destination.url) {
				output.stop();
				this.outputs.delete(id);
				this.log(`${output.label}: ${destination ? 'settings changed, reconnecting' : 'removed'}`);
			}
		}
		for (const destination of config.destinations) {
			if (this.outputs.has(destination.id)) continue;
			this.outputs.set(
				destination.id,
				new Output(
					destination.id,
					destination.label,
					['-c', 'copy', '-f', 'flv', destination.url],
					this.options.ffmpeg,
					this.log,
					destination.url
				)
			);
			this.log(`${destination.label}: connecting`);
		}
	}

	private async upload(
		files: Record<string, { body: Uint8Array | string; type: string; cache?: string }>
	) {
		const { api } = this.options;
		if (!api || !this.session?.storagePrefix) return;
		const names = Object.keys(files);
		const { urls } = await api.uploadUrls(this.session.id, names);
		await Promise.all(
			names.map(async (name) => {
				const file = files[name];
				const response = await fetch(urls[name], {
					method: 'PUT',
					body: file.body,
					headers: {
						'content-type': file.type,
						...(file.cache ? { 'cache-control': file.cache } : {})
					}
				});
				if (!response.ok) throw new Error(`upload ${name}: ${response.status}`);
			})
		);
	}

	private vodPlaylist(ended: boolean) {
		const target = Math.ceil(Math.max(2, ...this.vod.map((each) => each.duration)));
		return [
			'#EXTM3U',
			'#EXT-X-VERSION:3',
			`#EXT-X-TARGETDURATION:${target}`,
			'#EXT-X-MEDIA-SEQUENCE:0',
			`#EXT-X-PLAYLIST-TYPE:${ended ? 'VOD' : 'EVENT'}`,
			...this.vod.flatMap((each) => [`#EXTINF:${each.duration.toFixed(3)},`, `live/${each.name}`]),
			...(ended ? ['#EXT-X-ENDLIST'] : [])
		].join('\n');
	}

	/** Uploads new segments, then the live playlist, and now and then the growing VOD playlist. */
	private async pumpRecording(final = false) {
		if (!this.session?.storagePrefix) return;
		const { hlsDir, ffmpeg } = this.options;
		let playlist: string;
		try {
			playlist = await readFile(join(hlsDir, 'index.m3u8'), 'utf8');
		} catch {
			return;
		}
		const entries = parsePlaylist(playlist);
		const fresh = entries.filter((entry) => !this.uploaded.has(entry.name));
		if (fresh.length) {
			const files: Record<string, { body: Uint8Array; type: string }> = {};
			for (const entry of fresh)
				files[`live/${entry.name}`] = {
					body: new Uint8Array(await readFile(join(hlsDir, entry.name))),
					type: 'video/mp2t'
				};
			await this.upload(files);
			for (const entry of fresh) {
				this.uploaded.add(entry.name);
				this.vod.push(entry);
			}
			await this.upload({
				'live/index.m3u8': {
					body: playlist,
					type: 'application/vnd.apple.mpegurl',
					cache: 'no-cache, max-age=1'
				}
			});
		}
		if (final || Date.now() - this.lastVodUpload > 30_000) {
			this.lastVodUpload = Date.now();
			await this.upload({
				'vod.m3u8': {
					body: this.vodPlaylist(final),
					type: 'application/vnd.apple.mpegurl',
					cache: final ? 'max-age=3600' : 'no-cache'
				}
			});
		}
		// Local segments that left the live playlist are safely stored; free the disk.
		const live = new Set(entries.map((entry) => entry.name));
		for (const name of await readdir(hlsDir))
			if (name.endsWith('.ts') && this.uploaded.has(name) && !live.has(name))
				await rm(join(hlsDir, name), { force: true });
		const session = this.session;
		if (!this.posterDone && session && Date.now() - session.startedAt > 15_000 && entries.length) {
			this.posterDone = true;
			const latest = entries[entries.length - 1].name;
			const poster = join(hlsDir, 'poster.jpg');
			const grab = Bun.spawn(
				[
					ffmpeg,
					'-hide_banner',
					'-loglevel',
					'error',
					'-y',
					'-i',
					join(hlsDir, latest),
					'-frames:v',
					'1',
					'-vf',
					'scale=640:-2',
					poster
				],
				{ windowsHide: true }
			);
			if ((await grab.exited) === 0)
				await this.upload({
					'poster.jpg': { body: new Uint8Array(await readFile(poster)), type: 'image/jpeg' }
				}).catch(() => {});
		}
	}

	private async report() {
		const { api } = this.options;
		if (!api || !this.session) return;
		await api
			.report({
				event: 'status',
				sessionId: this.session.id,
				durationSec: Math.round((Date.now() - this.session.startedAt) / 1000),
				destinations: this.destinations()
			})
			.catch((err) => this.log('Status report failed:', (err as Error).message));
	}

	/** Goes off air (ending the session and finishing the recording) if on air. */
	async stop() {
		if (this.onAir) await this.goOffAir();
		this.options.source.stop();
	}
}

/** Segment names and durations from an HLS media playlist. */
export function parsePlaylist(playlist: string) {
	const entries: { name: string; duration: number }[] = [];
	const lines = playlist.split(/\r?\n/);
	for (let index = 0; index < lines.length; index++) {
		const match = /^#EXTINF:([\d.]+)/.exec(lines[index]);
		if (match && lines[index + 1]?.trim())
			entries.push({ name: lines[index + 1].trim(), duration: Number(match[1]) });
	}
	return entries;
}