stoat-mbot/src/core/player.ts

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<GuildPlayerEvents> {
  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<void> {
    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<void>((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<void> {
    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<void> {
    if (this.current || this.status === "buffering" || this.status === "connecting") return;
    await this.advance(false, false);
  }

  private async startPlayback(track: Track, seekSeconds = 0): Promise<void> {
    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<PlaybackInput["video"]>): Promise<void> {
    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<void> {
    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<void> {
    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<Track | null> {
    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<void> {
    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<void> {
    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<void> {
    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<void> {
    await this.leaveVoice();
    this.emit("destroyed", { serverId: this.serverId });
    this.removeAllListeners();
  }
}