/**
 * dsh-voice-server — local streaming ASR for the DSH Web GUI.
 *
 * Registers the exact-path upgrade route `/api/voice/asr` on the harness
 * webserver. The browser client opens a WebSocket and streams 16 kHz Int16
 * PCM; this plugin feeds sherpa-onnx's streaming Zipformer recognizer and
 * replies with partial/final transcripts. Fully offline — no cloud service.
 *
 * Wire protocol (JSON frames over the socket):
 *   client → server: binary frames = Int16LE PCM @ 16 kHz mono
 *                    { "type": "stop" } → flush the final result and close
 *   server → client: { "type": "partial", "text" }  (interim)
 *                    { "type": "final",   "text" }  (committed segment)
 *                    { "type": "error",   "message" }
 */
import { existsSync, appendFileSync, readFileSync, mkdtempSync, rmSync } from "node:fs";
import { join } from "node:path";
import { tmpdir, homedir } from "node:os";
import { spawn } from "node:child_process";
import { WebSocketServer } from "ws";

/** Default model directory (override with $DSH_VOICE_MODELS_DIR). */
const DEFAULT_MODEL_DIR = "/Users/eastudio/Documents/DSWorkSpace/dsh-voice-models/zh-en-streaming";
const MODEL_DIR = process.env.DSH_VOICE_MODELS_DIR || DEFAULT_MODEL_DIR;

/** Recognition hotwords (domain terms to bias the recognizer toward). */
const HOTWORDS_FILE = process.env.DSH_VOICE_HOTWORDS || "/Users/eastudio/Documents/DSWorkSpace/dsh-voice-models/hotwords.txt";

/** Piper TTS model (local, offline). */
const PIPER_MODEL = process.env.DSH_PIPER_MODEL || join(homedir(), "piper-models", "zh_CN-huayan-medium.onnx");
const PIPER_CONFIG = process.env.DSH_PIPER_CONFIG || PIPER_MODEL + ".json";

/** Diagnostic log (workspace-writable) to observe activation from inside the server process. */
const DEBUG_LOG = "/Users/eastudio/Documents/DSWorkSpace/dsh-voice-server-debug.log";
function dbg(msg) {
  try { appendFileSync(DEBUG_LOG, new Date().toISOString() + " " + msg + "\n"); } catch { /* noop */ }
}
dbg("module evaluated");

let wss = null;

/** Lazy-load sherpa-onnx (native module) and build one recognizer. */
let recognizerPromise = null;
function loadRecognizer() {
  if (recognizerPromise !== null) return recognizerPromise;
  recognizerPromise = (async () => {
    const sherpa = await import("sherpa-onnx-node");
    const encoder = join(MODEL_DIR, "encoder-epoch-99-avg-1.onnx");
    const decoder = join(MODEL_DIR, "decoder-epoch-99-avg-1.onnx");
    const joiner = join(MODEL_DIR, "joiner-epoch-99-avg-1.onnx");
    const tokens = join(MODEL_DIR, "tokens.txt");
    const missing = [encoder, decoder, joiner, tokens].filter((p) => !existsSync(p));
    if (missing.length > 0) {
      throw new Error(`dsh-voice-server: missing model files in ${MODEL_DIR}: ${missing.join(", ")}`);
    }
    const rec = new sherpa.OnlineRecognizer({
      featConfig: { sampleRate: 16000, featureDim: 80 },
      modelConfig: {
        transducer: { encoder, decoder, joiner },
        tokens,
        numThreads: 2,
        provider: "cpu",
        debug: 0
      },
      enableEndpoint: true,
      decodingMethod: "modified_beam_search",
      ...(existsSync(HOTWORDS_FILE) ? { hotwordsFile: HOTWORDS_FILE, hotwordsScore: 1.5 } : {})
    });
    if (rec === null || rec === undefined) throw new Error("dsh-voice-server: sherpa-onnx failed to create the recognizer");
    return rec;
  })();
  recognizerPromise.catch(() => { recognizerPromise = null; }); // allow retry after failure
  return recognizerPromise;
}

function send(socket, payload) {
  if (socket.readyState === socket.OPEN) socket.send(JSON.stringify(payload));
}

/** One active connection: its online stream + accumulated text. */
function createSession(rec) {
  const stream = rec.createStream();
  // Trailing silence flushes the last syllable the streaming model otherwise drops.
  const TAIL_SILENCE = new Float32Array(9600); // 0.6s @ 16 kHz
  return {
    stream,
    closed: false,
    endpointSent: false,
    feed(pcm16 /* Int16Array */) {
      const samples = new Float32Array(pcm16.length);
      for (let i = 0; i < pcm16.length; i++) samples[i] = pcm16[i] / 32768;
      stream.acceptWaveform({ samples, sampleRate: 16000 });
      while (rec.isReady(stream)) rec.decode(stream);
      if (!this.endpointSent && rec.isEndpoint(stream)) {
        // Speech ended (trailing silence detected) — signal the client once.
        this.endpointSent = true;
        return { type: "endpoint", text: (rec.getResult(stream).text || "").trim() };
      }
      const partial = (rec.getResult(stream).text || "").trim();
      if (partial) return { type: "partial", text: partial };
      return null;
    },
    finalize() {
      stream.acceptWaveform({ samples: TAIL_SILENCE, sampleRate: 16000 });
      try { stream.inputFinished(); } catch { /* noop */ }
      while (rec.isReady(stream)) rec.decode(stream);
      this.closed = true;
      return (rec.getResult(stream).text || "").trim();
    }
  };
}

function attachSession(socket) {
  loadRecognizer()
    .then((rec) => {
      if (socket.readyState !== socket.OPEN) return;
      const session = createSession(rec);
      socket.on("message", (data, isBinary) => {
        if (session.closed) return;
        if (isBinary) {
          try {
            const pcm16 = new Int16Array(data.buffer, data.byteOffset, data.byteLength >> 1);
            const ev = session.feed(pcm16);
            if (ev) send(socket, ev);
          } catch (err) {
            send(socket, { type: "error", message: "decode error: " + String(err && err.message || err) });
          }
          return;
        }
        let msg;
        try { msg = JSON.parse(data.toString()); } catch { return; }
        if (msg && msg.type === "stop") {
          const final = session.finalize();
          if (final.trim()) send(socket, { type: "final", text: final });
          session.closed = true;
          try { socket.close(); } catch { /* noop */ }
        }
      });
      socket.on("close", () => { /* stream has no explicit free in 1.x; GC reclaims */ });
    })
    .catch((err) => {
      dbg("recognizer error: " + String(err && err.message || err));
      send(socket, { type: "error", message: String(err && err.message || err) });
      try { socket.close(); } catch { /* noop */ }
    });
}

/**
 * Synthesize speech via edge-tts (online, natural) or piper (offline).
 * @returns {Promise<{data: Buffer, contentType: string}>}
 */
function synthesize({ text, engine, voice, rate }) {
  return new Promise((resolve, reject) => {
    const dir = mkdtempSync(join(tmpdir(), "dsh-tts-"));
    const finish = (outPath, contentType) => {
      try {
        const data = readFileSync(outPath);
        resolve({ data, contentType });
      } catch (err) {
        reject(err);
      } finally {
        try { rmSync(dir, { recursive: true, force: true }); } catch { /* noop */ }
      }
    };

    let child;
    if (engine === "piper") {
      const outPath = join(dir, "out.wav");
      const args = ["-m", "piper", "-m", PIPER_MODEL, "-c", PIPER_CONFIG, "-f", outPath];
      if (rate && rate > 0) args.push("--length-scale", String(Math.max(0.4, Math.min(2, 1 / rate))));
      child = spawn("python3", args);
      let stderr = "";
      child.stderr.on("data", (d) => { stderr += d; });
      child.on("error", (err) => reject(err));
      child.on("close", (code) => {
        if (code !== 0) return reject(new Error("piper 合成失败: " + (stderr.trim() || code)));
        finish(outPath, "audio/wav");
      });
      child.stdin.write(text, "utf8");
      child.stdin.end();
      return;
    }

    // edge-tts (default)
    const outPath = join(dir, "out.mp3");
    const v = voice || "zh-CN-XiaoxiaoNeural";
    const args = ["-m", "edge_tts", "--voice", v, "--text", text, "--write-media", outPath];
    if (rate && rate > 0) {
      const pct = Math.round((rate - 1) * 100);
      if (pct !== 0) args.push("--rate", (pct > 0 ? "+" : "") + pct + "%");
    }
    child = spawn("python3", args);
    let stderr = "";
    child.stderr.on("data", (d) => { stderr += d; });
    child.on("error", (err) => reject(err));
    child.on("close", (code) => {
      if (code !== 0) return reject(new Error("edge-tts 合成失败: " + (stderr.trim() || code)));
      finish(outPath, "audio/mpeg");
    });
  });
}

/** Read a JSON request body. */
async function readJson(req) {
  let body = "";
  for await (const chunk of req) body += chunk;
  if (!body) return {};
  try { return JSON.parse(body); } catch { return null; }
}

export const inject = ["webServer"];

export function apply(ctx) {
  dbg("apply called");
  if (wss === null) wss = new WebSocketServer({ noServer: true });
  ctx.effect(() => {
    const disposeUpgrade = ctx.webServer.registerUpgrade({
      path: "/api/voice/asr",
      handler: (req, socket, head) => {
        wss.handleUpgrade(req, socket, head, (ws) => {
          attachSession(ws);
        });
      }
    });
    const disposePing = ctx.webServer.register({
      kind: "exact",
      path: "/api/voice/ping",
      handler: (req, res) => {
        res.writeHead(200, { "content-type": "application/json; charset=utf-8" });
        res.end(JSON.stringify({ ok: true, service: "dsh-voice-server" }));
      }
    });
    const disposeTts = ctx.webServer.register({
      kind: "exact",
      path: "/api/voice/tts",
      handler: async (req, res) => {
        if (req.method !== "POST") {
          res.writeHead(405, { "content-type": "application/json; charset=utf-8" });
          res.end(JSON.stringify({ error: "method not allowed" }));
          return;
        }
        const parsed = await readJson(req);
        if (parsed === null || typeof parsed.text !== "string" || !parsed.text.trim()) {
          res.writeHead(400, { "content-type": "application/json; charset=utf-8" });
          res.end(JSON.stringify({ error: "missing text" }));
          return;
        }
        try {
          const { data, contentType } = await synthesize({
            text: parsed.text,
            engine: parsed.engine === "piper" ? "piper" : "edge",
            voice: parsed.voice,
            rate: parsed.rate
          });
          res.writeHead(200, { "content-type": contentType, "content-length": data.length, "cache-control": "no-cache" });
          res.end(data);
        } catch (err) {
          dbg("tts error: " + String(err && err.message || err));
          res.writeHead(500, { "content-type": "application/json; charset=utf-8" });
          res.end(JSON.stringify({ error: String(err && err.message || err) }));
        }
      }
    });
    dbg("routes registered");
    return () => {
      dbg("effect disposed — routes torn down");
      disposeUpgrade();
      disposePing();
      disposeTts();
    };
  }, "dsh-voice-server: voice routes");
}