AkurAI-Monitor / scripts/akurai-error-forwarder.ts

Latest change 58c391dfbe412723839c809952b9554310439f0e - scripts: add akurai-error-forwarder (Monitor error/warn -> pi-bun #errors WS) by agent

#!/usr/bin/env bun
// akurai-error-forwarder — bridges AkurAI-Monitor error/warning logs into the
// pi-bun Workbench #errors channel via its token-authed /ws/ingest WebSocket.
//
// Display-only: the WS endpoint stores frames under the "Error" author, so they
// render as an error component and never trigger a channel agent reply. Runs as
// a systemd oneshot on a timer; each run forwards logs newer than the last id.
//
// Config (env, from /etc/akurai-error-forwarder.env):
//   MONITOR_URL         default https://monitor.olibuijr.com
//   MONITOR_API_TOKEN   bearer for Monitor /api/logs (PassVault entry 25)
//   PIBUN_WS_URL        default ws://192.168.1.10:4173/ws/ingest
//   WORKBENCH_TOKEN     pi-bun operator token (PassVault entry 23)
//   ERROR_CHANNEL       default "errors"
//   STATE_DIRECTORY     systemd StateDirectory (holds the last-id cursor)

const monitor = (Bun.env.MONITOR_URL ?? "https://monitor.olibuijr.com").replace(/\/$/, "");
const monitorToken = Bun.env.MONITOR_API_TOKEN ?? "";
const pibunWs = Bun.env.PIBUN_WS_URL ?? "ws://192.168.1.10:4173/ws/ingest";
const pibunToken = Bun.env.WORKBENCH_TOKEN ?? "";
const channel = Bun.env.ERROR_CHANNEL ?? "errors";
const stateFile = `${Bun.env.STATE_DIRECTORY ?? "/var/lib/akurai-error-forwarder"}/last-id`;

if (!monitorToken || !pibunToken) {
  throw new Error("akurai-error-forwarder: MONITOR_API_TOKEN and WORKBENCH_TOKEN are required");
}

type LogRow = { id: number; source: string; line: string; level: string | null };

async function fetchLevel(level: "error" | "warn"): Promise<LogRow[]> {
  const res = await fetch(`${monitor}/api/logs?level=${level}&hours=2&limit=500`, {
    headers: { Authorization: `Bearer ${monitorToken}` },
  });
  if (!res.ok) throw new Error(`monitor /api/logs ${level} -> ${res.status}`);
  const data: unknown = await res.json();
  if (!data || typeof data !== "object" || !("logs" in data) || !Array.isArray(data.logs)) return [];
  // Monitor's /api/logs rows; individual fields are read defensively below.
  const logs = data.logs as LogRow[];
  return logs;
}
const lastId = Number((await Bun.file(stateFile).text().catch(() => "0")).trim()) || 0;
const rows = [...(await fetchLevel("error")), ...(await fetchLevel("warn"))]
  .filter((row) => row.id > lastId)
  .sort((a, b) => a.id - b.id);

if (rows.length === 0) {
  console.log("akurai-error-forwarder: no new error/warn logs");
} else {
  const ws = new WebSocket(`${pibunWs}?token=${encodeURIComponent(pibunToken)}`);
  const opened = Promise.withResolvers<void>();
  ws.onopen = () => opened.resolve();
  ws.onerror = () => opened.reject(new Error("pi-bun ws connect failed"));
  const openTimer = setTimeout(() => opened.reject(new Error("pi-bun ws open timeout")), 8000);
  await opened.promise;
  clearTimeout(openTimer);

  let acked = 0;
  ws.onmessage = (message) => {
    try {
      if (JSON.parse(String(message.data)).ok) acked += 1;
    } catch {
      /* ignore malformed ack */
    }
  };

  let maxId = lastId;
  for (const row of rows) {
    ws.send(JSON.stringify({ channel, level: row.level ?? "error", source: row.source, body: row.line.slice(0, 4000) }));
    if (row.id > maxId) maxId = row.id;
  }

  await Bun.sleep(1000); // let acks arrive before closing
  ws.close();
  await Bun.write(stateFile, String(maxId));
  console.log(`akurai-error-forwarder: forwarded ${rows.length} log(s), acked ${acked}, cursor ${maxId}`);
}