luma/src/modules/audioStreams.ts

import {
	AudioPlayer,
	createAudioPlayer,
	createAudioResource,
	StreamType,
	VoiceConnection,
	VoiceConnectionStatus
} from '@discordjs/voice';
import { AudioMixer } from 'node-audio-mixer';
import { PassThrough, Readable } from 'stream';

import prism from 'prism-media';

const DURATION_EXTRA_MS = 500;

export class StreamQueue {
	private queue: Readable[] = [];
	private isPlaying = false;
	private mixer: MixedStream;
	private currentStop: (() => void) | null = null;

	constructor(mixer: MixedStream) {
		this.mixer = mixer;
	}

	public enqueue(resource: Readable) {
		this.queue.push(resource);
		this.processQueue();
	}

	private async processQueue() {
		if (this.isPlaying || this.queue.length === 0) return;

		this.isPlaying = true;
		const nextStream = this.queue.shift();

		try {
			if (nextStream) {
				const { completion, stop } = this.mixer.playStream(nextStream);
				this.currentStop = stop;
				await completion;
				this.currentStop = null;
			}
		} catch (e) {
			console.error('Queue error:', e);
		} finally {
			this.isPlaying = false;
			this.processQueue();
		}
	}

	public clear() {
		this.queue = [];
		if (this.currentStop) {
			this.currentStop();
			this.currentStop = null;
		}
	}

	public flush() {
		this.mixer.flush();
	}
}

export class MixedStream {
	public readonly player: AudioPlayer;
	private mixer: AudioMixer;
	private output: PassThrough;
	private silenceInterval: NodeJS.Timeout;

	private queues: Map<string, StreamQueue> = new Map();

	public constructor() {
		this.player = createAudioPlayer();

		this.mixer = new AudioMixer({
			channels: 2,
			bitDepth: 16,
			sampleRate: 48000,
			autoClose: false,
			generateSilence: false // does not work :<
		});

		const silenceInput = this.mixer.createAudioInput({
			channels: 2,
			sampleRate: 48000,
			bitDepth: 16,
			volume: 100
		});

		const chunk = Buffer.alloc(3840);
		this.silenceInterval = setInterval(() => {
			if (silenceInput.writable && silenceInput.writableLength < 3840 * 10) {
				silenceInput.write(chunk);
			}
		}, 20);

		this.output = new PassThrough({ highWaterMark: 1024 * 16 });
		this.mixer.pipe(this.output);

		const resource = createAudioResource(this.output, {
			inputType: StreamType.Raw
		});

		this.player.play(resource);
		this.player.on('error', (error) => {
			console.error('Error: ', error.message);
		});
	}

	public getQueue(name: string): StreamQueue {
		let queue = this.queues.get(name);
		if (!queue) {
			queue = new StreamQueue(this);
			this.queues.set(name, queue);
		}
		return queue;
	}

	public playStream(source: Readable): {
		completion: Promise<void>;
		stop: () => void;
	} {
		let stopCallback: () => void = () => { };
		const completion = new Promise<void>((resolve) => {
			const mixerInput = this.mixer.createAudioInput({
				channels: 2,
				sampleRate: 48000,
				bitDepth: 16,
				volume: 100
			});

			const transcoder = new prism.FFmpeg({
				args: [
					'-analyzeduration',
					'0',
					'-loglevel',
					'0',
					'-f',
					's16le',
					'-ar',
					'48000',
					'-ac',
					'2'
				]
			});
			let totalBytes = 0;

			transcoder.on('data', (chunk: Buffer) => {
				totalBytes += chunk.length;
			});

			let resolved = false;
			const cleanup = () => {
				if (resolved) return;
				resolved = true;
				try {
					source.unpipe(transcoder);
					source.destroy();
				} catch (e) {
					console.error('Error destroying source:', e);
				}

				try {
					transcoder.unpipe(mixerInput);
					transcoder.destroy();
				} catch (e) {
					console.error('Error destroying transcoder:', e);
				}

				try {
					this.mixer.removeAudioinput(mixerInput);
				} catch (e) {
					console.error('Error removing audio input:', e);
				}

				try {
					// eslint-disable-next-line @typescript-eslint/no-explicit-any
					if (typeof (this.mixer as any).removeAudioInput === 'function') {
						// eslint-disable-next-line @typescript-eslint/no-explicit-any
						(this.mixer as any).removeAudioInput(mixerInput);
					}
				} catch (_) { /* ignore */ }

				try {
					mixerInput.destroy();
				} catch (e) {
					console.error('Error destroying mixer input:', e);
				}

				resolve();
			};

			stopCallback = cleanup;

			transcoder.on('end', () => {
				const durationMs =
					((totalBytes / 192000) * 1000) + DURATION_EXTRA_MS;

				setTimeout(() => {
					cleanup();
				}, durationMs);
			});

			transcoder.on('error', (err) => {
				console.error('Transcoder error:', err);
				cleanup();
			});

			source.pipe(transcoder).pipe(mixerInput);
		});

		return { completion, stop: stopCallback };
	}

	public destroy(): void {
		this.player.stop();
		this.output.destroy();
		this.mixer.destroy();
		clearInterval(this.silenceInterval);
	}

	public flush(): void {
		this.player.stop();

		this.mixer.unpipe(this.output);
		this.output.destroy();

		this.output = new PassThrough({ highWaterMark: 1024 * 16 });
		this.mixer.pipe(this.output);

		const resource = createAudioResource(this.output, {
			inputType: StreamType.Raw
		});

		this.player.play(resource);
	}
}

export class AudioStreamManager {
	private streams = new WeakMap<VoiceConnection, MixedStream>();

	public getOrCreateStream(conn: VoiceConnection): MixedStream {
		let stream = this.streams.get(conn);
		if (stream) return stream;

		stream = new MixedStream();
		this.streams.set(conn, stream);
		conn.subscribe(stream.player);

		conn.on('stateChange', (_, newState) => {
			if (
				newState.status === VoiceConnectionStatus.Disconnected ||
				newState.status === VoiceConnectionStatus.Destroyed
			) {
				this.destroyStream(conn);
			}
		});

		return stream;
	}

	public destroyStream(conn: VoiceConnection): void {
		const stream = this.streams.get(conn);
		if (stream) {
			stream.destroy();
			this.streams.delete(conn);
		}
	}

	/*
		singleton logic
	*/
	static #instance: AudioStreamManager | null = null;

	public static get get(): AudioStreamManager {
		if (!AudioStreamManager.#instance)
			AudioStreamManager.#instance = new AudioStreamManager();
		return AudioStreamManager.#instance;
	}
}