igneum/relay/api/console.mjs

// Igneum console: one function, dispatched on ?fn=, reached through the rewrite /r/<token>/c/<fn>.
// Auth: the token in the path (or x-relay-token), or the intake key in x-igneum-key. Nothing else. No secret
// reaches the client: DL_TOKEN (the downloads folder) is read here from the project env.
// Every GET answer is cached 10 s in the function instance.
//   GET  machines                 one card per machine from the log intake (miner_logs): node, cards, telemetry, faults
//   GET  jobs                     igneum-jobs.json from the downloads host + per-machine status from run_id job-<id>-<machine>
//   GET  job?run_id=              the newest upload of that job run, in full
//   GET  builds                   OTA manifest, CI json, build posts, downloads listing (synced by tools/console.mjs sync-dl)
//   GET  chain                    igneum.network/api/live trimmed + the Hetzner results item
//   GET  log?limit=&since=        work-log items (kinds log, build, note), newest first
//   GET  results                  bench entries (synced from docs/bench-log.md) + the FUD ledger counts
//   POST post   {kind,title,body,who,key?,meta?}   one item; with key it upserts
//   POST sync   {items:[...]}                      bulk upsert by key
import { neon, authed, readJson, str, iso } from '../lib/relay.mjs';
import { kv, kvNum, lastMatch, FAULT, parseLabel, parseMinerTail, parseHeader, parseAppTail, STALE_S, markStale } from '../lib/parse.mjs';

const json = (res, status, obj) => { res.status(status).setHeader('Content-Type', 'application/json; charset=utf-8'); res.end(JSON.stringify(obj)); };
const CACHE_MS = 10_000;
const cache = new Map();
async function cached(key, fn) {
  const hit = cache.get(key);
  if (hit && Date.now() - hit.at < CACHE_MS) return hit.value;
  const value = await fn();
  cache.set(key, { at: Date.now(), value });
  return value;
}
const DL_BASE = () => process.env.DL_TOKEN ? `https://dl.igneum.network/dl/${process.env.DL_TOKEN}` : '';
const LIVE_URL = process.env.LIVE_URL || 'https://igneum.network/api/live';
const NAMES = { ae432dc7: 'PC 1', '1ccfe586': 'PC 2', '3a9bf309': "Sam's Mac" };
const NAMED = new Set(Object.values(NAMES).concat('Mac'));
const HIDE_AFTER_S = 1800; // ids other than the named machines drop off 30 min after their last upload (test runs)
const SILENT_S = 180;
const num = v => (v === null || v === undefined || v === '' ? null : Number(v));

async function fetchJson(url, ms = 6000) {
  const r = await fetch(url, { signal: AbortSignal.timeout(ms), headers: { accept: 'application/json' } });
  if (!r.ok) return null;
  return r.json();
}

// ---- machines ---------------------------------------------------------------------------------------------------
function parseNodeTail(tail) {
  const lines = tail.split('\n');
  const n = {};
  let m;
  if ((m = lastMatch(lines, /Processed (\d+) blocks and (\d+) headers in the last/))) { n.blocks_10s = Number(m[1]); n.headers_10s = Number(m[2]); }
  if ((m = lastMatch(lines, /Connection manager: has (\d+)\/(\d+) outgoing/))) { n.peers = Number(m[1]); n.peers_wanted = Number(m[2]); }
  if ((m = lastMatch(lines, /\(daa (\d+)/))) n.daa = Number(m[1]);
  if ((m = lastMatch(lines, /checkpoint (\d+) determined: block [0-9a-f]+ \(blue score (\d+), daa (\d+)\)/))) { n.checkpoint = Number(m[1]); n.blue_score = Number(m[2]); if (!n.daa || Number(m[3]) > n.daa) n.daa = Number(m[3]); }
  if ((m = lastMatch(lines, /IBD/))) n.ibd_seen = true;
  if ((m = lastMatch(lines, /^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d+[+-]\d{2}:\d{2})/))) n.last_line_at = iso(m[1].replace(' ', 'T'));
  if ((m = lastMatch(lines, /igneumd\/(\S+)/))) n.version = m[1];
  const errs = lines.filter(l => / \[ERROR\] | \[WARN \] /.test(l));
  n.warnings = errs.length;
  n.last_warning = errs.length ? errs[errs.length - 1].replace(/^\S+ \S+ /, '').slice(0, 200) : null;
  return n;
}
// The app uploads at most the last 256 KiB of its log, so an update line scrolls out of the tail in about half an hour
// on a PC that logs every block (4 October 2026: PC 1's "installing 0.3.4" left its card while the install was still
// stuck). The last OTA state seen per machine and app run is kept in console_ota_memo and read back, for the same run,
// when the tail has none; a new run (the app relaunched) starts clean.
async function otaMemo(sql, groups) {
  try {
    await sql(`CREATE TABLE IF NOT EXISTS console_ota_memo (id8 text PRIMARY KEY, run_id text NOT NULL, ota jsonb NOT NULL, updated_at timestamptz NOT NULL DEFAULT now())`);
    const live = [...groups.values()].filter(g => g.app && !g.id.startsWith('legacy:'));
    if (!live.length) return;
    const rows = await sql(`SELECT id8, run_id, ota FROM console_ota_memo WHERE id8 = ANY($1)`, [live.map(g => g.id)]);
    const memo = new Map(rows.map(r => [r.id8, r]));
    for (const g of live) {
      const m = memo.get(g.id);
      if (g.app.ota) {
        if (!m || m.run_id !== g.app.run_id || JSON.stringify(m.ota) !== JSON.stringify(g.app.ota)) {
          await sql(`INSERT INTO console_ota_memo (id8, run_id, ota, updated_at) VALUES ($1, $2, $3::jsonb, now())
                     ON CONFLICT (id8) DO UPDATE SET run_id = EXCLUDED.run_id, ota = EXCLUDED.ota, updated_at = now()`, [g.id, g.app.run_id, JSON.stringify(g.app.ota)]);
        }
      } else if (m && m.run_id === g.app.run_id && m.ota) {
        g.app.ota = { ...m.ota, remembered: true };
      }
    }
  } catch (e) { /* a convenience: the cards render without it */ }
}

async function machines(sql) {
  // node logs get a longer tail: the connection-manager (peers) line is rare on a busy node
  // the app log's tail is wide (400 KB, about three hours on a PC that logs every block) so a stuck update's lines do
  // not scroll out of the card: on 4 October 2026 PC 1's "installing 0.3.4" left the 60 KB window after 30 minutes
  const rows = await sql(`SELECT DISTINCT ON (label) label, machine, run_id, received_at, left(lines, 300) AS head,
        right(lines, CASE WHEN label LIKE 'nodelog-%' THEN 80000 WHEN label ~ '^(win|mac|linux)-[0-9a-f]{8}$' THEN 400000 ELSE 20000 END) AS tail
      FROM miner_logs WHERE received_at > now() - interval '36 hours' ORDER BY label, received_at DESC`);
  const groups = new Map();
  for (const r of rows) {
    const p = parseLabel(r.label, r.machine);
    if (!p) continue;
    const g = groups.get(p.id) || { id: p.id, platform: p.platform, hosts: new Set(), runs: new Set(), last_seen: null, node: {}, cards: [], app: null, labels: [], header: null };
    if (p.stream !== 'miner' && p.platform !== g.platform) g.platform = p.platform;   // a miner label carries no os; the node or app log decides (a Linux rig)
    const hd = parseHeader(r.head) || parseHeader(r.tail); if (hd) g.header = hd;
    g.hosts.add(r.machine.replace(/-[0-9a-f]{8}$/, ''));
    g.runs.add(r.run_id);
    const at = iso(r.received_at);
    if (!g.last_seen || at > g.last_seen) g.last_seen = at;
    g.labels.push({ label: r.label, at, run_id: r.run_id });
    if (p.stream === 'node') { g.node = { ...parseNodeTail(r.tail), at, run_id: r.run_id }; }
    else if (p.stream === 'miner') g.cards.push({ label: r.label.replace(/^miner-/, ''), vendor: p.vendor, card: p.card, at, ...parseMinerTail(r.tail) });
    else if (p.stream === 'app') g.app = { at, run_id: r.run_id, ...parseAppTail(r.tail) };
    groups.set(p.id, g);
  }
  // node version lives in the first upload of a run (the tail may have lost the banner): one cheap query for the current runs
  const runIds = [...new Set([...groups.values()].flatMap(g => [...g.runs]))];
  if (runIds.length) {
    const heads = await sql(`SELECT DISTINCT ON (run_id, label) run_id, label, left(lines, 3000) AS head FROM miner_logs WHERE run_id = ANY($1) ORDER BY run_id, label, received_at ASC`, [runIds]);
    for (const h of heads) {
      const g = [...groups.values()].find(x => x.runs.has(h.run_id)); if (!g) continue;
      const m = /igneumd\/(\S+)/.exec(h.head); if (m && h.label.startsWith('nodelog-') && !g.node.version) g.node.version = m[1];
      const hd = parseHeader(h.head); if (hd && !g.header) g.header = hd;
    }
  }
  await otaMemo(sql, groups);
  const now = Date.now();
  const out = [...groups.values()].map(g => {
    const id8 = g.id.startsWith('legacy:') ? null : g.id;
    const cards = markStale(g.cards.sort((a, b) => a.card - b.card), now);
    const silent_s = g.last_seen ? Math.round((now - new Date(g.last_seen).getTime()) / 1000) : null;
    return {
      id: g.id, id8, platform: g.platform,
      name: id8 ? (NAMES[id8] || (g.platform === 'mac' ? 'Mac' : g.platform === 'linux' ? (([...g.hosts][0] || 'Rig') + ' ' + id8) : 'PC ' + id8)) : g.id.slice(7),
      host: [...g.hosts][0] || null, legacy: !id8,
      last_seen: g.last_seen, silent_s, silent: silent_s === null || silent_s > SILENT_S,
      stopped: (silent_s === null || silent_s > SILENT_S) && g.app && g.app.stopped ? g.app.stopped : null,
      app_version: (g.app && g.app.version) || (g.header && g.header.version) || null,
      app: g.app ? { at: g.app.at, status: g.app.status || null, ota: g.app.ota || null, power_cap: g.app.power_cap || null, jobs_file: g.app.jobs_file || null, jobs: g.app.jobs || [] } : null,
      node: { version: g.node.version || (g.header && g.header.node) || (g.app && g.app.node_version) || null, daa: g.node.daa ?? null, blue_score: g.node.blue_score ?? null, checkpoint: g.node.checkpoint ?? null,
        peers: g.node.peers ?? (g.app && g.app.status && g.app.status.node_peers != null ? g.app.status.node_peers : null), peers_wanted: g.node.peers_wanted ?? null, peers_from: g.node.peers != null ? 'node log' : (g.app && g.app.status ? 'app status line' : null),
        blocks: g.app && g.app.status && g.app.status.node_blocks != null ? g.app.status.node_blocks : null,
        blocks_10s: g.node.blocks_10s ?? null, synced: cards.some(c => c.synced === true) || (g.node.blocks_10s != null && !g.node.ibd_seen) || !!(g.app && g.app.status && g.app.status.node_state === 'synced'), ibd: !!g.node.ibd_seen, last_line_at: g.node.last_line_at || null, warnings: g.node.warnings || 0, last_warning: g.node.last_warning || null, at: g.node.at || (g.app && g.app.status ? g.app.status.at : null) },
      cards, hash_total: Math.round(cards.reduce((a, c) => a + (c.stale ? 0 : c.hash_now || 0), 0) * 100) / 100, stale_after_s: STALE_S,
      accepted_total: cards.reduce((a, c) => a + (c.accepted || 0), 0) || (g.app && g.app.status && g.app.status.accepted_run) || 0,
      accepted_lifetime: g.app && g.app.status ? g.app.status.accepted_total ?? null : null,
      faults: cards.reduce((a, c) => a + (c.faults || 0), 0),
      telemetry: g.app ? g.app.telemetry : [], events: g.app ? g.app.events : [],
      labels: g.labels.sort((a, b) => (b.at > a.at ? 1 : -1)), runs: [...g.runs],
    };
  });
  // the three named machines first (always shown), then anything seen in the last 24 h, then the rest
  const order = m => m.name === 'Mac' ? 0 : m.name === 'PC 1' ? 1 : m.name === 'PC 2' ? 2 : m.name === "Sam's Mac" ? 3 : m.legacy ? 9 : 5;
  out.sort((a, b) => order(a) - order(b) || (b.last_seen || '').localeCompare(a.last_seen || ''));
  // several Mac installs (test runs): the newest is "Mac", the others keep their id
  let macSeen = false;
  for (const m of out) if (m.platform === 'mac' && m.name === 'Mac') { if (macSeen) m.name = `Mac ${m.id8}`; macSeen = true; }
  for (const n of ['Mac', 'PC 1', 'PC 2', "Sam's Mac"]) if (!out.some(m => m.name === n)) out.push({ id: n, id8: Object.keys(NAMES).find(k => NAMES[k] === n) || null, platform: n.endsWith('Mac') ? 'mac' : 'win', name: n, host: null, legacy: false, last_seen: null, silent_s: null, silent: true, app_version: null, app: null, node: {}, cards: [], hash_total: 0, accepted_total: 0, faults: 0, telemetry: [], events: [], labels: [], runs: [], never: true });
  out.sort((a, b) => order(a) - order(b) || (b.last_seen || '').localeCompare(a.last_seen || ''));
  return { machines: out.filter(m => NAMED.has(m.name) || (m.silent_s !== null && m.silent_s < HIDE_AFTER_S)), silent_after_s: SILENT_S, hide_after_s: HIDE_AFTER_S };
}

// ---- jobs ---------------------------------------------------------------------------------------------------------
function normaliseJobs(j) {
  const list = Array.isArray(j) ? j : Array.isArray(j.jobs) ? j.jobs : [];
  return list.map((x, i) => ({
    id: String(x.id ?? x.job_id ?? x.name ?? i), title: String(x.title ?? x.name ?? x.id ?? `job ${i + 1}`),
    created: x.created ?? x.created_at ?? x.ts ?? null, to: x.to ?? x.machines ?? x.machine ?? 'all',
    command: x.command ?? x.run ?? x.script ?? x.cmd ?? '', shell: x.shell ?? null, raw: x,
  }));
}
async function jobs(sql) {
  const base = DL_BASE();
  let file = null; let note = '';
  if (!base) note = 'DL_TOKEN is not set on the project';
  else { try { file = await fetchJson(`${base}/igneum-jobs.json`); if (!file) note = 'no igneum-jobs.json on the downloads host yet'; } catch (e) { note = `jobs file: ${e.message}`; } }
  const list = file ? normaliseJobs(file) : [];
  const rows = await sql(`SELECT DISTINCT ON (run_id) run_id, label, machine, received_at, right(lines, 6000) AS tail
      FROM miner_logs WHERE run_id LIKE 'job-%' AND received_at > now() - interval '14 days' ORDER BY run_id, received_at DESC`);
  const runs = rows.map(r => {
    const m = /^job-(.+)-([0-9a-f]{8,32})$/.exec(r.run_id);
    const lines = r.tail.split('\n').filter(l => l.trim());
    // Final only when the app's closing report is in: a SUMMARY line with finished_at set, or its
    // "job <id>: done|failed|timeout|aborted (exit N)" line. RESULT lines and a stage's own "exit N" echo are progress.
    let summaryJson = null;
    for (const l of lines) { const mm = /^SUMMARY\s+(\{.*\})\s*$/.exec(l); if (mm) { try { summaryJson = JSON.parse(mm[1]); } catch {} } }
    const finalLine = lastMatch(lines, /^job \S+: (?:done|failed|timeout|aborted) \(exit (-?\d+)\)/);
    const done = !!(summaryJson && summaryJson.finished_at) || !!finalLine;
    const ex = finalLine || (summaryJson && summaryJson.finished_at ? [null, String(summaryJson.exit)] : null);
    const sum = lastMatch(lines, /^(?:RESULT|SUMMARY)[:\s](.+)$/i);
    const machineId = m ? m[2] : r.machine;
    return { run_id: r.run_id, job_id: m ? m[1] : r.run_id, machine_id: machineId, machine: NAMES[machineId.slice(0, 8)] || r.machine, host: r.machine, at: iso(r.received_at), label: r.label,
      status: done ? 'done' : 'running', exit_code: ex ? Number(ex[1]) : null, summary: (sum ? sum[1] : lines[lines.length - 1] || '').slice(0, 300) };
  });
  const seen = new Set(runs.map(r => r.job_id));
  const out = list.map(j => ({ ...j, runs: runs.filter(r => r.job_id === j.id) }));
  for (const id of seen) if (!out.some(j => j.id === id)) out.push({ id, title: id, created: null, to: null, command: '', runs: runs.filter(r => r.job_id === id), unlisted: true });
  out.sort((a, b) => String(b.created || b.runs[0]?.at || '').localeCompare(String(a.created || a.runs[0]?.at || '')));
  return { jobs: out, note, file_seen: !!file, generated: file && !Array.isArray(file) ? file.generated ?? file.updated ?? null : null };
}

// ---- builds, chain, results, log --------------------------------------------------------------------------------
const itemOut = r => ({ id: Number(r.id), ts: iso(r.ts), kind: r.kind, key: r.key, who: r.who, title: r.title, body: r.body, meta: r.meta || {} });
async function builds(sql) {
  const base = DL_BASE();
  const [manifest, ci, posts, dl] = await Promise.all([
    base ? fetchJson(`${base}/igneum-app-latest.json`).catch(() => null) : null,
    base ? fetchJson(`${base}/igneum-windows-ci.json`).catch(() => null) : null,
    sql(`SELECT * FROM console_items WHERE kind = 'build' ORDER BY ts DESC LIMIT 40`),
    sql(`SELECT * FROM console_items WHERE key = 'dl' LIMIT 1`),
  ]);
  return { manifest, ci, builds: posts.map(itemOut), dl: dl.length ? itemOut(dl[0]) : null, dl_set: !!base };
}
async function chain(sql) {
  const [live, het] = await Promise.all([fetchJson(LIVE_URL, 8000).catch(e => ({ error: e.message })), sql(`SELECT * FROM console_items WHERE key = 'hetzner' LIMIT 1`)]);
  if (!live || live.error || live.ok === false) return { ok: true, live: false, error: live && live.error || 'live feed unreachable', state: null, finality: {}, events: [], hetzner: het.length ? itemOut(het[0]) : null };
  const f = live.finality || {};
  return {
    live: true, now: live.now, state: live.state,
    finality: { active: !!f.active, next_index: f.next_index, latest_locked_index: f.latest_locked_index, latest_locked_blue_score: f.latest_locked_blue_score, latest_locked_hash: f.latest_locked_hash, voters: f.weights && f.weights.keys ? f.weights.keys.filter(k => k.voter).length : null, keys: f.weights && f.weights.keys ? f.weights.keys.length : null, checkpoints: (f.checkpoints || []).slice(-8).map(c => ({ index: c.index, state: c.state, fraction_total: c.fraction_total, locked_at: c.locked_at })) },
    miners: (live.miners || []).slice(0, 12), events: (live.events || []).slice(0, 12),
    hetzner: het.length ? itemOut(het[0]) : null,
  };
}
async function results(sql) {
  const [bench, ledger] = await Promise.all([
    sql(`SELECT * FROM console_items WHERE kind = 'bench' ORDER BY (meta->>'date') DESC NULLS LAST, (meta->>'pos')::int DESC LIMIT 60`),
    sql(`SELECT * FROM console_items WHERE key = 'ledger' LIMIT 1`),
  ]);
  return { bench: bench.map(itemOut), ledger: ledger.length ? itemOut(ledger[0]) : null };
}
async function upsert(sql, o) {
  const kind = str(o.kind, 20) || 'log';
  const meta = o.meta && typeof o.meta === 'object' ? o.meta : {};
  if (o.key) {
    const r = await sql(`INSERT INTO console_items (kind, key, who, title, body, meta) VALUES ($1,$2,$3,$4,$5,$6::jsonb)
        ON CONFLICT (key) DO UPDATE SET kind = EXCLUDED.kind, who = EXCLUDED.who, title = EXCLUDED.title, body = EXCLUDED.body, meta = EXCLUDED.meta, ts = CASE WHEN console_items.title = EXCLUDED.title AND console_items.body = EXCLUDED.body THEN console_items.ts ELSE now() END RETURNING id`,
      [kind, str(o.key, 200), str(o.who, 80) || 'Mac', str(o.title, 400), str(o.body, 64000), JSON.stringify(meta)]);
    return Number(r[0].id);
  }
  const r = await sql(`INSERT INTO console_items (kind, who, title, body, meta) VALUES ($1,$2,$3,$4,$5::jsonb) RETURNING id`,
    [kind, str(o.who, 80) || 'Mac', str(o.title, 400), str(o.body, 64000), JSON.stringify(meta)]);
  return Number(r[0].id);
}

export default async function handler(req, res) {
  res.setHeader('Cache-Control', 'no-store');
  if (!authed(req)) return json(res, 401, { ok: false, error: 'no token' });
  const fn = String(req.query.fn || '');
  const q = req.query;
  let sql;
  try { sql = neon(); } catch (e) { return json(res, 500, { ok: false, error: e.message }); }
  try {
    if (req.method === 'GET') {
      if (fn === 'machines') return json(res, 200, { ok: true, now: new Date().toISOString(), ...(await cached('machines', () => machines(sql))) });
      if (fn === 'jobs') return json(res, 200, { ok: true, now: new Date().toISOString(), ...(await cached('jobs', () => jobs(sql))) });
      if (fn === 'job') {
        const rows = await sql(`SELECT id, label, machine, run_id, received_at, lines FROM miner_logs WHERE run_id = $1 ORDER BY received_at DESC LIMIT 1`, [str(q.run_id, 200)]);
        if (!rows.length) return json(res, 404, { ok: false, error: 'no upload for that run' });
        return json(res, 200, { ok: true, run_id: rows[0].run_id, label: rows[0].label, machine: rows[0].machine, at: iso(rows[0].received_at), lines: rows[0].lines });
      }
      if (fn === 'builds') return json(res, 200, { ok: true, now: new Date().toISOString(), ...(await cached('builds', () => builds(sql))) });
      if (fn === 'chain') return json(res, 200, { ok: true, ...(await cached('chain', () => chain(sql))) });
      if (fn === 'results') return json(res, 200, { ok: true, ...(await cached('results', () => results(sql))) });
      if (fn === 'log') {
        const limit = Math.min(300, Math.max(1, Number(q.limit) || 100));
        const params = []; let where = `kind IN ('log','build','note')`;
        if (q.since) { params.push(Number(q.since)); where += ` AND id > $${params.length}`; }
        const rows = await sql(`SELECT * FROM console_items WHERE ${where} ORDER BY id DESC LIMIT ${limit}`, params);
        return json(res, 200, { ok: true, items: rows.map(itemOut), now: new Date().toISOString() });
      }
      return json(res, 404, { ok: false, error: `unknown fn ${fn}` });
    }
    if (req.method !== 'POST') return json(res, 405, { ok: false, error: 'method' });
    let body;
    try { body = await readJson(req); } catch { return json(res, 400, { ok: false, error: 'bad json' }); }
    if (fn === 'post') {
      if (!body.title && !body.body) return json(res, 400, { ok: false, error: 'title or body required' });
      const id = await upsert(sql, body);
      cache.clear();
      return json(res, 200, { ok: true, id });
    }
    if (fn === 'sync') {
      const items = Array.isArray(body.items) ? body.items : [];
      let count = 0;
      for (const it of items) { if (!it || !it.key) continue; await upsert(sql, it); count++; }
      cache.clear();
      return json(res, 200, { ok: true, count });
    }
    if (fn === 'delete') {
      const id = Number(body.id); if (!id) return json(res, 400, { ok: false, error: 'id required' });
      await sql(`DELETE FROM console_items WHERE id = $1`, [id]);
      cache.clear();
      return json(res, 200, { ok: true, id });
    }
    return json(res, 404, { ok: false, error: `unknown fn ${fn}` });
  } catch (e) {
    return json(res, 500, { ok: false, error: String(e.message || e) });
  }
}