/**
 * Discord voice I/O.
 *
 * - Joins the caller's voice channel.
 * - Receives each speaker's Opus stream, decodes to PCM, and on end-of-speech
 *   forwards the utterance (as a WAV) to the brain bridge.
 * - Plays the brain's spoken reply back into the channel.
 *
 * No AI logic here — capture in, audio out. The brain lives in bridge/.
 */
import { Readable } from "node:stream";
import {
  joinVoiceChannel,
  createAudioPlayer,
  createAudioResource,
  EndBehaviorType,
  StreamType,
  AudioPlayerStatus,
  VoiceConnection,
  VoiceConnectionStatus,
  entersState,
  type AudioPlayer,
} from "@discordjs/voice";
import prism from "prism-media";
import type { VoiceBasedChannel } from "discord.js";
import { converseStream } from "./bridge.ts";
import { config } from "./config.ts";

const DISCORD_RATE = 48000;
const DISCORD_CHANNELS = 2;

/** Build a minimal PCM16 mono WAV around raw little-endian samples. */
function pcm16MonoToWav(pcm: Buffer, sampleRate: number): Buffer {
  const header = Buffer.alloc(44);
  const dataLen = pcm.length;
  header.write("RIFF", 0);
  header.writeUInt32LE(36 + dataLen, 4);
  header.write("WAVE", 8);
  header.write("fmt ", 12);
  header.writeUInt32LE(16, 16);
  header.writeUInt16LE(1, 20); // PCM
  header.writeUInt16LE(1, 22); // mono
  header.writeUInt32LE(sampleRate, 24);
  header.writeUInt32LE(sampleRate * 2, 28); // byte rate (mono * 2 bytes)
  header.writeUInt16LE(2, 32); // block align
  header.writeUInt16LE(16, 34); // bits per sample
  header.write("data", 36);
  header.writeUInt32LE(dataLen, 40);
  return Buffer.concat([header, pcm]);
}

/** Downmix interleaved stereo PCM16 to mono PCM16. */
function stereoToMono(stereo: Buffer): Buffer {
  const samples = stereo.length / 4; // 2 ch * 2 bytes
  const mono = Buffer.alloc(samples * 2);
  for (let i = 0; i < samples; i++) {
    const l = stereo.readInt16LE(i * 4);
    const r = stereo.readInt16LE(i * 4 + 2);
    mono.writeInt16LE((l + r) >> 1, i * 2);
  }
  return mono;
}

export class VoiceSession {
  readonly guildId: string;
  private connection: VoiceConnection;
  private player: AudioPlayer;
  private listening = new Set<string>();
  /** Set once the session is torn down (user left / leave command). In-flight
   *  captures check this so we don't run STT/reply or post a transcript for
   *  audio that arrived after the user already left the channel. */
  private destroyed = false;
  /** Opus subscriptions currently capturing, so leave() can end them
   *  immediately instead of waiting out the silence timeout. */
  private activeStreams = new Set<{ destroy: () => void }>();
  /** Pending reply clips. Played one at a time so concurrent speakers don't
   *  cut each other's replies off. */
  private playQueue: Buffer[] = [];
  /** Optional callback to surface EVERY heard utterance (and its outcome) to a
   *  text channel — including ones dropped before/at STT — so misses are
   *  diagnosable. `note` says why (e.g. "음성 아님(VAD 차단)", "너무 짧음", "ok"). */
  onTurn?: (info: {
    user: string;
    /** Resolved display name (server nickname / global name) for the speaker,
     *  so logs show a human name instead of the raw Discord user ID. */
    userName?: string;
    transcript: string;
    reply: string;
    note?: string;
    /** Wall-clock epoch-ms markers for each pipeline stage, so the transcript
     *  channel can show what took long. Listening is measured here (capture
     *  start -> end of speech); LLM/TTS come from the brain bridge. STT shows
     *  up as the gap between listenEndMs and llmStartMs. */
    listenStartMs?: number;
    listenEndMs?: number;
    llmStartMs?: number;
    llmEndMs?: number;
    ttsStartMs?: number;
    ttsEndMs?: number;
  }) => void;
  /** Live screen-share state, sent with each turn so the brain routes search
   *  (Chrome while broadcasting, Gemini when off). */
  getBroadcasting?: () => boolean;
  /** Apply a broadcast directive the brain requested (start/stop the stream). */
  onBroadcastAction?: (action: "start" | "stop") => void | Promise<void>;

  /** The selfbot client behind this voice connection + the negotiated voice
   *  session_id, so the broadcast (Go-Live) can ride the SAME session — one
   *  account hears/speaks AND broadcasts. */
  private readonly client: any;
  private readonly channelId: string;
  private sessionId?: string;

  constructor(channel: VoiceBasedChannel) {
    this.guildId = channel.guild.id;
    this.channelId = channel.id;
    this.client = (channel as any).client;
    // Wrap the gateway adapter to capture our own voice session_id (needed to
    // create the Go-Live stream on this same session).
    const realCreator = channel.guild.voiceAdapterCreator;
    const adapterCreator: typeof realCreator = (methods) =>
      realCreator({
        ...methods,
        onVoiceStateUpdate: (d: any) => {
          if (d?.session_id) this.sessionId = d.session_id;
          return methods.onVoiceStateUpdate(d);
        },
      });
    this.connection = joinVoiceChannel({
      channelId: channel.id,
      guildId: channel.guild.id,
      adapterCreator,
      selfDeaf: false, // we need to hear users
      selfMute: false,
    });
    this.player = createAudioPlayer();
    this.connection.subscribe(this.player);
    // Drain the queue when the current clip finishes.
    this.player.on(AudioPlayerStatus.Idle, () => this.pump());
    this.attachReceiver();
  }

  /** The shared session for single-account Go-Live, or null if session_id isn't
   *  captured yet / the client is unavailable. */
  getSharedSession(): {
    client: unknown;
    guildId: string;
    channelId: string;
    sessionId: string;
    botId: string;
  } | null {
    const botId = this.client?.user?.id;
    if (!this.sessionId || !this.client || !botId) return null;
    return { client: this.client, guildId: this.guildId, channelId: this.channelId, sessionId: this.sessionId, botId };
  }

  async ready(): Promise<void> {
    await entersState(this.connection, VoiceConnectionStatus.Ready, 20_000);
  }

  private attachReceiver() {
    const receiver = this.connection.receiver;
    receiver.speaking.on("start", (userId: string) => {
      if (this.listening.has(userId)) return;
      this.listening.add(userId);
      this.captureUtterance(userId).finally(() => this.listening.delete(userId));
    });
  }

  /** Resolve a speaker's Discord user ID to a human display name (server
   *  nickname, else global name / username), cached so we don't refetch every
   *  utterance. Falls back to the ID if lookup fails. */
  private nameCache = new Map<string, string>();
  private async displayName(userId: string): Promise<string> {
    const cached = this.nameCache.get(userId);
    if (cached) return cached;
    let name = userId;
    try {
      const guild: any = this.client?.guilds?.cache?.get(this.guildId);
      let member: any = guild?.members?.cache?.get(userId);
      if (!member && guild?.members?.fetch) member = await guild.members.fetch(userId).catch(() => null);
      if (member) {
        name = member.displayName || member.nickname || member.user?.globalName || member.user?.username || userId;
      } else {
        const u: any = this.client?.users?.cache?.get(userId) || (await this.client?.users?.fetch?.(userId).catch(() => null));
        name = u?.globalName || u?.username || userId;
      }
    } catch {
      /* fall back to id */
    }
    this.nameCache.set(userId, name);
    return name;
  }

  private async captureUtterance(userId: string): Promise<void> {
    // Don't start a new capture once we're tearing down (user left).
    if (this.destroyed) return;
    // "듣기 시작": the moment we begin capturing this speaker's utterance.
    const listenStartMs = Date.now();
    const opusStream = this.connection.receiver.subscribe(userId, {
      end: { behavior: EndBehaviorType.AfterSilence, duration: config.silenceMs },
    });
    this.activeStreams.add(opusStream);
    const decoder = new prism.opus.Decoder({
      frameSize: 960,
      channels: DISCORD_CHANNELS,
      rate: DISCORD_RATE,
    });
    const chunks: Buffer[] = [];
    const pcmStream = opusStream.pipe(decoder);
    pcmStream.on("data", (c: Buffer) => chunks.push(c));

    await new Promise<void>((resolve) => pcmStream.once("end", () => resolve()));
    this.activeStreams.delete(opusStream);
    // If the user left while we were capturing, drop this utterance entirely —
    // don't run STT/reply or report a (usually empty/VAD-blocked) turn for
    // audio that trailed in after they left.
    if (this.destroyed) return;
    // "듣기 종료": end of speech (silence detected). Anything after this is
    // STT + reply + TTS on the brain side.
    const listenEndMs = Date.now();

    if (!chunks.length) return;
    const mono = stereoToMono(Buffer.concat(chunks));
    // Ignore blips shorter than ~300ms (likely noise / key clicks) — but still
    // report them so the transcript channel shows every captured utterance.
    if (mono.length < DISCORD_RATE * 0.3 * 2) {
      this.onTurn?.({
        user: userId,
        userName: await this.displayName(userId),
        transcript: "",
        reply: "",
        note: "너무 짧음(<300ms)",
        listenStartMs,
        listenEndMs,
      });
      return;
    }
    const wav = pcm16MonoToWav(mono, DISCORD_RATE);

    try {
      // Streaming turn: the brain sends transcript/reply first, then one audio
      // clip per sentence as it is synthesised. We enqueue each clip on arrival
      // so the first sentence starts playing while the rest are still spoken.
      // The transcript-channel report is sent once the stream ends so it can
      // include TTS timing (synthesis runs after the meta line). Audio still
      // plays as it arrives — only the diagnostic text post waits.
      let metaSeen: {
        transcript: string;
        reply: string;
        note?: string;
        llm_start_ms?: number;
        llm_end_ms?: number;
      } | undefined;
      let endSeen: { tts_start_ms?: number; tts_end_ms?: number } | undefined;
      await converseStream(wav, this.getBroadcasting?.(), {
        onMeta: async (m) => {
          metaSeen = m;
          // Apply any broadcast directive the brain requested (e.g. user said
          // "방송 켜줘 / 꺼줘") before the reply audio plays. The meta line
          // always precedes the audio clips, so awaiting here preserves order.
          if (m.broadcast_action && this.onBroadcastAction) {
            try {
              await this.onBroadcastAction(m.broadcast_action);
            } catch (e) {
              console.error("[voice] broadcast action failed:", e);
            }
          }
        },
        onAudio: (clip) => this.play(clip),
        onEnd: (end) => {
          endSeen = end;
        },
      });
      // Report EVERY turn (even empty/VAD-dropped) so the transcript channel
      // explains why a turn did or didn't answer, with full stage timing.
      this.onTurn?.({
        user: userId,
        userName: await this.displayName(userId),
        transcript: metaSeen?.transcript ?? "",
        reply: metaSeen?.reply ?? "",
        note: metaSeen?.note,
        listenStartMs,
        listenEndMs,
        llmStartMs: metaSeen?.llm_start_ms,
        llmEndMs: metaSeen?.llm_end_ms,
        ttsStartMs: endSeen?.tts_start_ms,
        ttsEndMs: endSeen?.tts_end_ms,
      });
    } catch (err) {
      console.error("[voice] converse failed:", err);
    }
  }

  /** Queue a WAV buffer for playback (FIFO, one clip at a time). */
  play(wav: Buffer) {
    this.playQueue.push(wav);
    this.pump();
  }

  private pump() {
    if (this.player.state.status !== AudioPlayerStatus.Idle) return;
    const next = this.playQueue.shift();
    if (!next) return;
    const resource = createAudioResource(Readable.from(next), {
      inputType: StreamType.Arbitrary,
    });
    this.player.play(resource);
  }

  destroy() {
    this.destroyed = true;
    // End any in-flight captures now so their pcmStream resolves immediately
    // and the post-capture `destroyed` check drops them (no trailing
    // post-leave VAD turns).
    for (const s of this.activeStreams) {
      try {
        s.destroy();
      } catch {
        /* already ended */
      }
    }
    this.activeStreams.clear();
    try {
      this.connection.destroy();
    } catch {
      /* already gone */
    }
  }
}

/** One session per guild. */
const sessions = new Map<string, VoiceSession>();

export async function joinChannel(channel: VoiceBasedChannel): Promise<VoiceSession> {
  sessions.get(channel.guild.id)?.destroy();
  const session = new VoiceSession(channel);
  sessions.set(channel.guild.id, session);
  await session.ready();
  return session;
}

export function leaveGuild(guildId: string): boolean {
  const s = sessions.get(guildId);
  if (!s) return false;
  s.destroy();
  sessions.delete(guildId);
  return true;
}

export function getSession(guildId: string): VoiceSession | undefined {
  return sessions.get(guildId);
}