import assert from "node:assert/strict";
import { afterEach, test } from "node:test";
import { VoiceSessionController } from "../../src/voice_assistant/audio_gateway/web/voice-session.js";

const controllers = [];
class Node {
  constructor() { this.disconnected = false; }
  connect() {}
  disconnect() { this.disconnected = true; }
}
class Context {
  constructor() {
    this.destination = {};
    this.currentTime = 0;
    this.sources = [];
    this.audioWorklet = { addModule: async () => {} };
    this.nextResume = null;
  }
  createAnalyser() { return new Node(); }
  createGain() { return Object.assign(new Node(), { gain: { value: 1 } }); }
  createMediaStreamSource() { this.input = new Node(); return this.input; }
  async resume() { const pending = this.nextResume; this.nextResume = null; if (pending) await pending; }
  createBuffer(channels, length, sampleRate) {
    return { duration: length / sampleRate, copyToChannel(samples) { this.samples = samples; } };
  }
  createBufferSource() {
    const source = Object.assign(new Node(), {
      stopped: false, onended: null,
      start: () => this.sources.push(source),
      stop() { this.stopped = true; },
      end() { this.onended?.(); },
    });
    return source;
  }
}
class Worklet extends Node {
  constructor() {
    super();
    this.port = { messages: [], postMessage(data) { this.messages.push(data); }, onmessage: null };
  }
}
class Socket {
  static OPEN = 1;
  static CLOSED = 3;
  constructor(url) { this.url = url; this.readyState = 0; this.sent = []; this.bufferedAmount = 0; }
  send(data) { if (this.sendFailure) throw new Error("Transport failed"); this.sent.push(data); }
  open() { this.readyState = 1; this.onopen?.(); }
  close() { this.readyState = 3; this.onclose?.(); }
  receive(event) { this.onmessage?.({ data: JSON.stringify(event) }); }
}
globalThis.WebSocket = Socket;
globalThis.AudioWorkletNode = Worklet;
globalThis.location = { protocol: "http:", host: "localhost:8001" };
afterEach(() => { for (const controller of controllers.splice(0)) controller.close(); });
const tick = () => new Promise((resolve) => setImmediate(resolve));
const ready = (overrides = {}) => ({
  type: "session.ready", turn_detection: "server",
  audio: { encoding: "pcm_s16le", sample_rate: 16000, channels: 1 },
  output_audio: { encoding: "pcm_s16le", sample_rate: 48000, channels: 1 },
  max_audio_chunk_bytes: 65536, ...overrides,
});
function messages(socket) { return socket.sent.filter((data) => typeof data === "string").map(JSON.parse); }
async function setup(options = {}, confirm = true) {
  const errors = [], states = [], transcripts = [], replies = [];
  const controller = new VoiceSessionController({
    onError: (error) => errors.push(error), onStateChange: (state) => states.push(state),
    onTranscript: (...args) => transcripts.push(args), onAgentText: (...args) => replies.push(args),
    ...options,
  });
  controllers.push(controller);
  const context = new Context();
  await controller.start({ stream: {}, audioContext: context });
  const socket = controller.ws;
  socket.open();
  if (confirm) socket.receive(ready());
  return { controller, context, socket, errors, states, transcripts, replies };
}

test("requests server VAD and starts capture only after session.ready", async () => {
  const { controller, socket } = await setup({}, false);
  assert.equal(messages(socket)[0].turn_detection, "server");
  assert.deepEqual(controller.worklet.port.messages, []);
  controller.handlePCM(new ArrayBuffer(4));
  assert.equal(socket.sent.length, 1);
  socket.receive(ready());
  assert.deepEqual(controller.worklet.port.messages, [{ type: "start", chunkSamples: 1600 }]);
  assert.equal(controller.state, "listening");
});

for (const state of ["listening", "recording", "processing", "speaking", "waiting-playback"]) {
  test(`sends continuous PCM while ${state}, without turn metadata`, async () => {
    const { controller, socket } = await setup();
    controller.setState(state);
    const audio = new ArrayBuffer(4);
    controller.handlePCM(audio);
    assert.deepEqual(messages(socket).at(-1), { type: "audio.input", sequence_number: 0 });
    assert.equal(socket.sent.at(-1), audio);
  });
}

test("global input sequence never resets when backend opens another turn", async () => {
  const { controller, socket } = await setup();
  controller.handlePCM(new ArrayBuffer(4));
  socket.receive({ type: "turn.started", turn_id: 1 });
  controller.handlePCM(new ArrayBuffer(4));
  socket.receive({ type: "interrupted", turn_id: 1 });
  socket.receive({ type: "turn.started", turn_id: 2 });
  controller.handlePCM(new ArrayBuffer(4));
  assert.deepEqual(messages(socket).filter((m) => m.type === "audio.input").map((m) => m.sequence_number), [0, 1, 2]);
  assert.equal(messages(socket).some((m) => ["audio.start", "audio.stop", "interrupt"].includes(m.type)), false);
  assert.equal(controller.worklet.port.messages.length, 1);
});

test("uses the negotiated chunk size for microphone capture", async () => {
  const { controller, socket } = await setup({}, false);
  socket.receive(ready({ max_audio_chunk_bytes: 1000 }));
  assert.deepEqual(controller.worklet.port.messages, [{ type: "start", chunkSamples: 500 }]);
});

test("turn.ended waits for speakers without pausing microphone input", async () => {
  const { controller, context, socket } = await setup();
  socket.receive({ type: "turn.started", turn_id: 1 });
  controller.handleAudioChunk(new ArrayBuffer(8));
  await controller.playbackPending;
  assert.equal(controller.state, "speaking");
  socket.receive({ type: "turn.ended", turn_id: 1 });
  await tick();
  assert.equal(controller.state, "waiting-playback");
  controller.handlePCM(new ArrayBuffer(4));
  assert.equal(messages(socket).at(-1).type, "audio.input");
  context.sources[0].end();
  await tick();
  assert.equal(controller.state, "listening");
  assert.equal(controller.worklet.port.messages.length, 1);
});

test("a new backend turn stops old playback and ignores its pending completion", async () => {
  const { controller, context, socket } = await setup();
  socket.receive({ type: "turn.started", turn_id: 1 });
  controller.handleAudioChunk(new ArrayBuffer(8));
  await controller.playbackPending;
  socket.receive({ type: "turn.ended", turn_id: 1 });
  await tick();
  socket.receive({ type: "turn.started", turn_id: 2 });
  await tick();
  assert.equal(context.sources[0].stopped, true);
  assert.equal(controller.turnId, 2);
  assert.equal(controller.state, "recording");
});

test("old queued PCM is discarded after playback generation changes", async () => {
  const { controller, context, socket } = await setup();
  let release;
  context.nextResume = new Promise((resolve) => { release = resolve; });
  socket.receive({ type: "turn.started", turn_id: 1 });
  controller.handleAudioChunk(new ArrayBuffer(8));
  const obsolete = controller.playbackPending;
  await tick();
  socket.receive({ type: "turn.started", turn_id: 2 });
  release();
  await obsolete;
  assert.equal(context.sources.length, 0);
  controller.handleAudioChunk(new ArrayBuffer(4));
  await controller.playbackPending;
  assert.equal(context.sources.length, 1);
});

test("a failure in obsolete playback does not close the new turn", async () => {
  const { controller, context, socket, errors } = await setup();
  let reject;
  context.nextResume = new Promise((resolve, fail) => { reject = fail; });
  socket.receive({ type: "turn.started", turn_id: 1 });
  controller.handleAudioChunk(new ArrayBuffer(8));
  const obsolete = controller.playbackPending;
  await tick();
  socket.receive({ type: "turn.started", turn_id: 2 });
  reject(new Error("Old playback failed"));
  await obsolete;
  assert.equal(errors.length, 0);
  assert.equal(controller.turnId, 2);
  assert.equal(socket.readyState, Socket.OPEN);
});

test("stale turn events cannot change the current response", async () => {
  const { controller, socket, transcripts, replies } = await setup();
  socket.receive({ type: "turn.started", turn_id: 2 });
  socket.receive({ type: "stt.final", turn_id: 1, text: "old" });
  socket.receive({ type: "agent.response", turn_id: 1, text: "old" });
  socket.receive({ type: "interrupted", turn_id: 1 });
  socket.receive({ type: "turn.ended", turn_id: 1 });
  socket.receive({ type: "turn.started", turn_id: 1 });
  assert.equal(controller.turnId, 2);
  assert.deepEqual(transcripts, []);
  assert.deepEqual(replies, []);
  socket.receive({ type: "stt.final", turn_id: 2, text: "new" });
  socket.receive({ type: "agent.response", turn_id: 2, text: "reply" });
  assert.deepEqual(transcripts, [["new", true]]);
  assert.deepEqual(replies, [["reply", true]]);
});

test("interrupted returns to listening without stopping capture or starting a client turn", async () => {
  const { controller, socket } = await setup();
  socket.receive({ type: "turn.started", turn_id: 1 });
  socket.receive({ type: "interrupted", turn_id: 1 });
  assert.equal(controller.state, "listening");
  assert.equal(controller.turnId, null);
  assert.deepEqual(messages(socket).map((m) => m.type), ["session.start"]);
  assert.equal(controller.worklet.port.messages.length, 1);
});

for (const metadata of [{ turn_detection: "client" }, { turn_detection: undefined }, { max_audio_chunk_bytes: 1 }]) {
  test(`rejects incompatible session metadata: ${JSON.stringify(metadata)}`, async () => {
    const { controller, socket, errors } = await setup({}, false);
    socket.receive(ready(metadata));
    assert.equal(controller.state, "closed");
    assert.equal(errors.length, 1);
  });
}

test("backpressure closes the session before queuing more PCM", async () => {
  const { controller, socket, errors } = await setup({ maxBufferedBytes: 8 });
  socket.bufferedAmount = 8;
  controller.handlePCM(new ArrayBuffer(4));
  assert.equal(controller.state, "closed");
  assert.equal(errors.length, 1);
  assert.equal(messages(socket).some((m) => m.type === "audio.input"), false);
});

test("invalid PCM closes the session without emitting metadata", async () => {
  const { controller, socket, errors } = await setup();
  controller.handlePCM(new ArrayBuffer(3));
  assert.equal(errors.length, 1);
  assert.equal(messages(socket).some((m) => m.type === "audio.input"), false);
});

test("close stops capture and playback even if session.end cannot be sent", async () => {
  const { controller, socket, context } = await setup();
  const worklet = controller.worklet;
  socket.receive({ type: "turn.started", turn_id: 1 });
  controller.handleAudioChunk(new ArrayBuffer(8));
  await controller.playbackPending;
  socket.sendFailure = true;
  controller.close();
  assert.equal(controller.state, "closed");
  assert.equal(worklet.disconnected, true);
  assert.deepEqual(worklet.port.messages.at(-1), { type: "stop" });
  assert.equal(context.sources[0].stopped, true);
  assert.equal(socket.readyState, Socket.CLOSED);
});

test("closing during worklet loading prevents orphan microphone graphs and sockets", async () => {
  let release;
  const context = new Context();
  context.audioWorklet.addModule = () => new Promise((resolve) => { release = resolve; });
  const controller = new VoiceSessionController();
  controllers.push(controller);
  const starting = controller.start({ stream: {}, audioContext: context });
  controller.close();
  release();
  await starting;
  assert.equal(controller.ws, null);
  assert.equal(context.input, undefined);
});

test("missing session.ready times out and closes the connection", async () => {
  const { controller, errors } = await setup({ handshakeTimeoutMs: 5 }, false);
  await new Promise((resolve) => setTimeout(resolve, 20));
  assert.equal(controller.state, "closed");
  assert.equal(errors.length, 1);
});