import { EventEmitter } from "node:events"; import { config } from "../config.js"; import { logger } from "../logger.js"; import { openPlayback, type PlaybackInput } from "../sources/index.js"; import { UserFacingError, type LoopMode, type PlayerSnapshot, type PlayerStatus, type Track, } from "../types.js"; import { VideoPublisher } from "./video.js"; import { MediaPlayer, parseFfmpegDuration, VOICE_STATE_OFFLINE, type MediaPlayerLike, type RevoiceLike, type VoiceConnectionLike, } from "./revoice.js"; const HISTORY_LIMIT = 50; const JOIN_TIMEOUT_MS = 20_000; export interface PositionUpdate { serverId: string; position: number; duration: number; status: PlayerStatus; } export interface Notice { serverId: string; textChannelId: string | null; text: string; } export interface GuildPlayerEvents { update: [PlayerSnapshot]; position: [PositionUpdate]; notice: [Notice]; destroyed: [{ serverId: string }]; } export interface GuildPlayerOptions { serverId: string; serverName: string | null; revoice: RevoiceLike; } /** * Owns everything about music playback for one Stoat server: the voice * connection, the queue and the ffmpeg-backed media player. Chat commands and * the web panel both drive playback exclusively through this class, so the two * can never drift apart. */ export class GuildPlayer extends EventEmitter { readonly serverId: string; serverName: string | null; textChannelId: string | null = null; voiceChannelId: string | null = null; voiceChannelName: string | null = null; queue: Track[] = []; history: Track[] = []; current: Track | null = null; volume = config.DEFAULT_VOLUME; loop: LoopMode = "off"; shuffleUsed = false; /** Per-server switch for publishing the clip; the env flag gates it too. */ videoEnabled = false; private status: PlayerStatus = "idle"; private readonly revoice: RevoiceLike; private connection: VoiceConnectionLike | null = null; /** Tracked from connection events: revoice's own `connected` getter is broken. */ private voiceReady = false; private media: MediaPlayerLike | null = null; private videoPublisher: VideoPublisher | null = null; private currentInput: PlaybackInput | null = null; private seekOffset = 0; /** Set while we tear playback down ourselves, so the resulting `finish` is ignored. */ private expectingStop = false; private leaveTimer: NodeJS.Timeout | null = null; private presenceTimer: NodeJS.Timeout | null = null; private ticker: NodeJS.Timeout | null = null; private readonly log; constructor(options: GuildPlayerOptions) { super(); this.serverId = options.serverId; this.serverName = options.serverName; this.revoice = options.revoice; this.log = logger.child({ mod: "player", server: options.serverId }); } // ---------------------------------------------------------------- state --- get position(): number { if (!this.media) return 0; return this.seekOffset + this.media.seconds; } snapshot(): PlayerSnapshot { return { serverId: this.serverId, serverName: this.serverName, voiceChannelId: this.voiceChannelId, voiceChannelName: this.voiceChannelName, textChannelId: this.textChannelId, status: this.status, current: this.current, position: Math.round(this.position * 10) / 10, queue: this.queue, history: this.history.slice(0, 10), volume: this.volume, loop: this.loop, shuffleUsed: this.shuffleUsed, videoEnabled: this.videoEnabled, updatedAt: Date.now(), }; } private setStatus(status: PlayerStatus): void { if (this.status === status) return; this.status = status; this.publish(); } publish(): void { this.emit("update", this.snapshot()); } private notify(text: string): void { this.emit("notice", { serverId: this.serverId, textChannelId: this.textChannelId, text }); } // ------------------------------------------------------------ connection --- isConnected(): boolean { return this.voiceReady; } async connect(channelId: string, channelName: string | null): Promise { if (this.voiceReady && this.voiceChannelId === channelId) { this.voiceChannelName = channelName ?? this.voiceChannelName; return; } if (this.connection) await this.leaveVoice(); this.setStatus("connecting"); this.log.info({ channelId }, "joining voice channel"); const connection = await this.revoice.join(channelId); try { // The room connects asynchronously inside revoice's constructor, so we wait // for it to report readiness rather than polling a state getter. await new Promise((resolve, reject) => { const timer = setTimeout( () => reject(new UserFacingError("Не удалось подключиться к голосовому каналу")), JOIN_TIMEOUT_MS, ); const done = () => { clearTimeout(timer); resolve(); }; connection.once("join", done); connection.once("roomfetched", done); }); } catch (err) { // Leave no orphan: an abandoned connection keeps the bot registered in the // channel, and Stoat then refuses the next join with AlreadyConnected. await connection.destroy().catch(() => {}); connection.removeAllListeners(); this.setStatus("idle"); throw err; } this.connection = connection; this.voiceReady = true; this.voiceChannelId = channelId; this.voiceChannelName = channelName; connection.on("state", (state) => { if (state !== VOICE_STATE_OFFLINE) return; this.voiceReady = false; this.log.warn({ channelId }, "voice connection went offline"); }); connection.on("userleave", () => this.checkEmptyChannel()); connection.on("userLeave", () => this.checkEmptyChannel()); connection.on("userJoin", () => this.cancelLeaveTimer()); const media = new MediaPlayer(true); // revoice's #cleanUp() dereferences this.fProc unconditionally, so a second // stop() (which its own ffmpeg error handler triggers) throws and would take // the whole process down. Everything it calls goes through this instance // method, so guarding it here covers its internal paths too. const stop = media.stop.bind(media); media.stop = (init?: boolean) => { try { stop(init); } catch (err) { this.log.debug({ err }, "revoice cleanup threw, ignoring"); } }; media.on("startplay", () => { this.setStatus(media.paused ? "paused" : "playing"); this.startTicker(); }); media.on("buffer", () => this.setStatus("buffering")); media.on("pause", () => this.setStatus("paused")); media.on("unpause", () => this.setStatus("playing")); media.on("finish", () => { void this.handleFinish(); }); this.media = media; await connection.play(media); this.setStatus("idle"); this.checkEmptyChannel(); // Belt and braces: a participant event we never receive would otherwise // leave the bot sitting in an empty channel indefinitely. this.presenceTimer = setInterval(() => this.checkEmptyChannel(), 30_000); this.presenceTimer.unref?.(); this.log.info({ channelId }, "voice connection established"); } async leaveVoice(): Promise { this.cancelLeaveTimer(); if (this.presenceTimer) { clearInterval(this.presenceTimer); this.presenceTimer = null; } this.stopScreenShare(); this.stopTicker(); this.teardownPlayback(); this.current = null; const connection = this.connection; this.connection = null; this.voiceReady = false; this.media?.removeAllListeners(); this.media = null; this.voiceChannelId = null; this.voiceChannelName = null; if (connection) { try { await connection.destroy(); } catch (err) { this.log.warn({ err }, "failed to leave voice channel cleanly"); } connection.removeAllListeners(); } this.setStatus("idle"); this.publish(); } /** * Counted straight from the LiveKit room. revoice's own list cannot be * trusted: it stores users by participant identity but removes them by * participant name, so nobody is ever dropped and the channel never looks * empty — the bot would sit in it forever. */ /** Whether this user is in the channel the bot is sitting in, per LiveKit. */ hasParticipant(userId: string): boolean { const participants = this.connection?.room?.remoteParticipants; if (!participants) return false; for (const participant of participants.values()) { if (participant.identity === userId) return true; } return false; } private listenersInChannel(): number { return this.connection?.room?.remoteParticipants?.size ?? 0; } private checkEmptyChannel(): void { if (!this.connection) return; if (this.listenersInChannel() > 0) { this.cancelLeaveTimer(); return; } this.startLeaveTimer(); } // -------------------------------------------------------------- playback --- private assertReady(): MediaPlayerLike { if (!this.media || !this.voiceReady) { throw new UserFacingError("Бот не подключён к голосовому каналу"); } return this.media; } enqueue(tracks: Track[], position?: number): void { if (this.queue.length + tracks.length > config.MAX_QUEUE_SIZE) { throw new UserFacingError(`Очередь ограничена ${config.MAX_QUEUE_SIZE} треками`); } if (position === undefined) this.queue.push(...tracks); else this.queue.splice(Math.max(0, position), 0, ...tracks); this.publish(); } /** Starts playback if nothing is currently playing. */ async ensurePlaying(): Promise { if (this.current || this.status === "buffering" || this.status === "connecting") return; await this.advance(false, false); } private async startPlayback(track: Track, seekSeconds = 0): Promise { const media = this.assertReady(); this.cancelLeaveTimer(); this.teardownPlayback(); this.current = track; this.seekOffset = seekSeconds; this.setStatus("buffering"); this.publish(); try { const input = await openPlayback(track, seekSeconds, { video: this.videoEnabled }); this.currentInput = input; this.expectingStop = false; await media.playStream(input.input, input.inputOptions); // stop() rebuilds the volume transformer, so volume is applied per track. media.setVolume(this.volume / 100); this.startTicker(); if (input.video && this.connection) { await this.startScreenShare(input.video); } // A downloader that dies mid-stream just looks like a very short track, so // say why instead of silently moving on. void input.failure?.then((reason) => { if (!reason || this.current?.id !== track.id) return; this.log.warn({ reason, track: track.title }, "source failed while streaming"); this.notify(`⚠️ **${track.title}** — источник отдал ошибку: ${reason}`); }); } catch (err) { this.log.warn({ err, track: track.title }, "playback failed"); const message = err instanceof UserFacingError ? err.message : "неизвестная ошибка"; this.notify(`⚠️ Не удалось воспроизвести **${track.title}** (${message}), пропускаю.`); this.current = null; await this.advance(true); } } /** Tears down ffmpeg/yt-dlp for the current track without advancing the queue. */ private teardownPlayback(): void { if (this.media) { this.expectingStop = true; try { // Detach revoice's own handlers first: killing ffmpeg makes it emit // "error", and its handler would call stop() again on a half-reset player. const proc = this.media.fProc; if (proc) { proc.removeAllListeners("error"); proc.removeAllListeners("end"); proc.kill("SIGKILL"); } } catch { // ffmpeg may already be gone. } try { this.media.stop(); } catch (err) { this.log.debug({ err }, "media.stop() threw"); } } this.stopScreenShare(); this.currentInput?.cleanup(); this.currentInput = null; this.seekOffset = 0; } private async startScreenShare(video: NonNullable): Promise { const room = this.connection?.room; if (!room) return; const publisher = new VideoPublisher({ width: video.width, height: video.height, fps: video.fps, // Frames are released against the sound people have actually heard, not // the sound handed to LiveKit: its audio source buffers up to a second, // and gating on the queued position showed every frame that much early. audioClock: () => this.heardSeconds(), }); try { await publisher.start(room, video.stream, (reason) => { this.notify(`📺 Видео отключено: ${reason}.`); }); this.videoPublisher = publisher; } catch (err) { this.log.warn({ err }, "could not publish screen share"); this.notify( "📺 Не удалось показать клип: инстанс не разрешает видео (нужны право Video у бота и включённое видео в конфигурации Stoat). Звук играет как обычно.", ); await publisher.stop().catch(() => {}); } } /** Playback position as heard, i.e. minus whatever is still queued in LiveKit. */ private heardSeconds(): number { const media = this.media; if (!media) return 0; const queuedMs = media.source?.queuedDuration ?? 0; return Math.max(0, media.seconds - queuedMs / 1000); } private stopScreenShare(): void { const publisher = this.videoPublisher; this.videoPublisher = null; if (publisher) void publisher.stop().catch(() => {}); } private async handleFinish(): Promise { if (this.expectingStop) { this.expectingStop = false; return; } await this.advance(false); } /** Moves to the next track. `skipLoop` ignores per-track looping (used by skip). */ private async advance(skipLoop: boolean, announce = true): Promise { const finished = this.current; this.current = null; // The track has to go with the sound: a published video track keeps showing // its last frame, so the clip would hang on screen after the music stopped. this.stopScreenShare(); this.currentInput?.cleanup(); this.currentInput = null; this.seekOffset = 0; if (finished) { this.history.unshift(finished); this.history = this.history.slice(0, HISTORY_LIMIT); if (!skipLoop && this.loop === "track") this.queue.unshift(finished); else if (this.loop === "queue") this.queue.push(finished); } const next = this.queue.shift(); if (!next) { this.stopTicker(); this.setStatus("idle"); this.publish(); if (finished) this.notify("⏹️ Очередь закончилась."); return; } await this.startPlayback(next); // A command that started this track has already said so; announcing again // would double every line in the channel. if (announce) this.notify(`▶️ Сейчас играет: **${next.title}**`); } async skip(count = 1): Promise { if (!this.current && this.queue.length === 0) throw new UserFacingError("Нечего пропускать"); for (let i = 1; i < count; i += 1) this.queue.shift(); this.teardownPlayback(); await this.advance(true, false); return this.current; } async stop(): Promise { this.queue = []; this.loop = "off"; this.teardownPlayback(); this.current = null; this.stopTicker(); this.setStatus("idle"); this.publish(); } pause(): void { const media = this.assertReady(); if (!this.current) throw new UserFacingError("Сейчас ничего не играет"); media.pause(); this.setStatus("paused"); this.publish(); } resume(): void { const media = this.assertReady(); if (!this.current) throw new UserFacingError("Сейчас ничего не играет"); media.resume(); this.setStatus("playing"); this.publish(); } setVolume(volume: number): void { const clamped = Math.min(200, Math.max(0, Math.round(volume))); this.volume = clamped; this.media?.setVolume(clamped / 100); this.publish(); } /** Takes effect on the next track: switching mid-stream would cut playback. */ setVideo(enabled: boolean): void { this.videoEnabled = enabled; this.publish(); } setLoop(mode: LoopMode): void { this.loop = mode; this.publish(); } shuffle(): void { for (let i = this.queue.length - 1; i > 0; i -= 1) { const j = Math.floor(Math.random() * (i + 1)); const a = this.queue[i]; const b = this.queue[j]; if (a && b) { this.queue[i] = b; this.queue[j] = a; } } this.shuffleUsed = true; this.publish(); } remove(trackId: string): Track { const index = this.queue.findIndex((track) => track.id === trackId); if (index === -1) throw new UserFacingError("Трек не найден в очереди"); const [removed] = this.queue.splice(index, 1); this.publish(); return removed as Track; } move(trackId: string, toIndex: number): void { const from = this.queue.findIndex((track) => track.id === trackId); if (from === -1) throw new UserFacingError("Трек не найден в очереди"); const target = Math.min(this.queue.length - 1, Math.max(0, toIndex)); const [track] = this.queue.splice(from, 1); if (track) this.queue.splice(target, 0, track); this.publish(); } clearQueue(): void { this.queue = []; this.publish(); } async seek(seconds: number): Promise { const track = this.current; if (!track) throw new UserFacingError("Сейчас ничего не играет"); if (track.isLive) throw new UserFacingError("Нельзя перематывать прямой эфир"); if (track.duration > 0 && seconds >= track.duration) { throw new UserFacingError("Позиция за пределами трека"); } this.teardownPlayback(); await this.startPlayback(track, Math.max(0, seconds)); } async playNow(tracks: Track[]): Promise { if (tracks.length === 0) return; this.queue.unshift(...tracks); this.teardownPlayback(); await this.advance(true, false); } // ---------------------------------------------------------- housekeeping --- private startTicker(): void { if (this.ticker) return; this.ticker = setInterval(() => { if (!this.current || !this.media) return; // ffmpeg reports the real duration once it has probed the input, which is // the only way we learn how long a local file or a direct URL is. if (this.current.duration === 0 && !this.current.isLive) { const probed = parseFfmpegDuration(this.media.codecData?.duration); if (probed > 0) { this.current.duration = Math.round(probed); this.publish(); } } this.emit("position", { serverId: this.serverId, position: Math.round(this.position * 10) / 10, duration: this.current.duration, status: this.status, }); }, 1000); this.ticker.unref?.(); } private stopTicker(): void { if (!this.ticker) return; clearInterval(this.ticker); this.ticker = null; } private cancelLeaveTimer(): void { if (!this.leaveTimer) return; clearTimeout(this.leaveTimer); this.leaveTimer = null; } /** * Leaving is tied to the channel being empty, never to an idle queue: the bot * stays put with people around, waiting for the next request. */ private startLeaveTimer(): void { this.cancelLeaveTimer(); if (config.EMPTY_TIMEOUT_SECONDS <= 0 || !this.connection) return; this.leaveTimer = setTimeout(() => { if (this.listenersInChannel() > 0) return; this.notify("👋 В канале никого не осталось, выхожу."); void this.leaveVoice(); }, config.EMPTY_TIMEOUT_SECONDS * 1000); this.leaveTimer.unref?.(); } async destroy(): Promise { await this.leaveVoice(); this.emit("destroyed", { serverId: this.serverId }); this.removeAllListeners(); } }