#!/usr/bin/env node
// Subscribes to HA conversation entity state changes, fetches latest
// pipeline_debug run, extracts STT transcript + final response, and
// writes a rolling window of last 5 exchanges to input_text.voice_h1..h5.

import { readFileSync } from "node:fs";

const HA_URL = process.env.HA_URL || "https://assistant.home.lolcat.dev";
const TOKEN = process.env.SUPERVISOR_TOKEN || process.env.HA_TOKEN || (() => {
  try { return readFileSync(new URL("./token", import.meta.url), "utf8").trim(); }
  catch { throw new Error("Set HA_TOKEN env var or write token to ./token"); }
})();
// Inside addon container, HA_URL = http://supervisor/core ; that hosts /api/websocket too
const WS_URL = (HA_URL.startsWith("http://supervisor")
  ? HA_URL.replace(/^http/, "ws")
  : HA_URL.replace(/^http/, "ws")) + "/api/websocket";
const PIPELINE_ID = process.env.PIPELINE_ID || "01krvc3027p3fxaemtfz512tfm";
const SLOTS = 5;

const seenRuns = new Set();
let ws;
let nextId = 1;
const pending = new Map();
const send = (msg) => {
  const id = nextId++;
  msg.id = id;
  ws.send(JSON.stringify(msg));
  return new Promise((resolve, reject) => pending.set(id, { resolve, reject }));
};

async function setInputText(entity_id, value) {
  const truncated = value.length > 255 ? value.slice(0, 252) + "…" : value;
  const res = await fetch(`${HA_URL}/api/services/input_text/set_value`, {
    method: "POST",
    headers: { Authorization: `Bearer ${TOKEN}`, "Content-Type": "application/json" },
    body: JSON.stringify({ entity_id, value: truncated }),
  });
  if (!res.ok) console.error(`set_value ${entity_id} failed:`, res.status, await res.text());
}

async function rotateAndPush(line) {
  for (let i = SLOTS; i > 1; i--) {
    const fromRes = await fetch(`${HA_URL}/api/states/input_text.voice_h${i - 1}`, {
      headers: { Authorization: `Bearer ${TOKEN}` },
    });
    const fromVal = (await fromRes.json()).state || "";
    await setInputText(`input_text.voice_h${i}`, fromVal);
  }
  await setInputText("input_text.voice_h1", line);
}

function summarizeRun(events) {
  let transcript = null;
  let response = null;
  let engine = null;
  let timestamp = null;
  let toolCalls = [];
  for (const e of events) {
    if (e.type === "run-start") timestamp = e.timestamp;
    if (e.type === "stt-end") transcript = e.data?.stt_output?.text || null;
    if (e.type === "intent-start") engine = e.data?.engine || null;
    if (e.type === "intent-progress") {
      const calls = e.data?.chat_log_delta?.tool_calls;
      if (Array.isArray(calls)) {
        for (const c of calls) {
          const name = c.tool_args?.name;
          if (name) toolCalls.push(name);
        }
      }
    }
    if (e.type === "intent-end") {
      response = e.data?.intent_output?.response?.speech?.plain?.speech || null;
    }
  }
  return { transcript, response, engine, timestamp, toolCalls };
}

async function processRun(runId) {
  if (seenRuns.has(runId)) return;
  seenRuns.add(runId);
  const res = await send({
    type: "assist_pipeline/pipeline_debug/get",
    pipeline_id: PIPELINE_ID,
    pipeline_run_id: runId,
  });
  const { transcript, response, engine, timestamp, toolCalls } = summarizeRun(res.events || []);
  if (!transcript && !response) return; // nothing useful
  const t = timestamp ? new Date(timestamp).toLocaleTimeString("sv-SE", { hour: "2-digit", minute: "2-digit" }) : "--:--";
  const agent = engine?.includes("openai") ? "gpt" : engine?.includes("home_assistant") ? "lokal" : (engine || "?");
  const tools = toolCalls.length ? ` [${toolCalls.join(", ")}]` : "";
  const u = transcript || "(intet)";
  const r = response || "(intet svar)";
  const line = `${t} ${agent}${tools} • ${u} → ${r}`;
  console.log(`[push] ${line}`);
  await rotateAndPush(line);
}

async function fetchLatestRuns() {
  const res = await send({
    type: "assist_pipeline/pipeline_debug/list",
    pipeline_id: PIPELINE_ID,
  });
  const runs = res.pipeline_runs || [];
  // newest last; process in chronological order
  for (const r of runs) {
    if (!seenRuns.has(r.pipeline_run_id)) {
      await processRun(r.pipeline_run_id);
    }
  }
}

function connect() {
  ws = new WebSocket(WS_URL);
  ws.onopen = () => console.log("[ws] connected");
  ws.onerror = (e) => console.error("[ws] error", e.message);
  ws.onclose = () => {
    console.error("[ws] disconnected, reconnecting in 5s…");
    setTimeout(connect, 5000);
  };
  ws.onmessage = async (e) => {
    const m = JSON.parse(e.data);
    if (m.type === "auth_required") {
      ws.send(JSON.stringify({ type: "auth", access_token: TOKEN }));
      return;
    }
    if (m.type === "auth_ok") {
      console.log("[ws] authed");
      // Seed known runs so we don't backfill on startup
      const res = await send({
        type: "assist_pipeline/pipeline_debug/list",
        pipeline_id: PIPELINE_ID,
      });
      for (const r of res.pipeline_runs || []) seenRuns.add(r.pipeline_run_id);
      console.log(`[seed] tracking ${seenRuns.size} existing runs; new ones from here on`);
      // Subscribe to conversation entity state changes as the wake signal
      await send({
        type: "subscribe_trigger",
        trigger: {
          platform: "state",
          entity_id: ["conversation.openai_conversation_2", "conversation.home_assistant"],
        },
      });
      console.log("[sub] listening for new conversation runs");
      return;
    }
    if (m.type === "result") {
      const slot = pending.get(m.id);
      if (slot) {
        pending.delete(m.id);
        if (m.success) slot.resolve(m.result);
        else slot.reject(new Error(JSON.stringify(m.error)));
      }
      return;
    }
    if (m.type === "event") {
      // Wait briefly so the pipeline_debug recorder has the events
      setTimeout(() => fetchLatestRuns().catch(err => console.error("fetch err:", err.message)), 1000);
    }
  };
}

connect();