igneum/tools/jobs.mjs

#!/usr/bin/env node
// Mac side of the remote jobs (app/igneum-app/src/jobs.rs, published by packaging/ota/publish-jobs.sh).
//   node tools/jobs.mjs                 the published jobs file: the signed envelope (or the pair) fetched from the
//                                       downloads host, signature checked
//   node tools/jobs.mjs status          per machine, from the log intake: the latest job run and its SUMMARY line,
//                                       plus the woken latency (job started_at minus the publish that added it,
//                                       from relay_wake, written by publish-jobs.sh --deploy since 0.3.6)
//   node tools/jobs.mjs <job id>        the result: the newest upload per label under run_id job-<id>-<machine>
//   node tools/jobs.mjs <job id> --all  every upload, oldest first (the 5-minute progress reports of a long job)
//   node tools/jobs.mjs watch <job id>  poll the intake every 30 s until every reporting machine is final
// Reads ~/.config/igneum/env (DATABASE_URL, the same Neon HTTP SQL as tools/logs.mjs), dl-token and
// ota-signing-key.pub. No dependencies.
import { readFileSync } from 'node:fs';
import { homedir } from 'node:os';
import { createPublicKey, verify } from 'node:crypto';

process.stdout.on('error', e => { if (e.code === 'EPIPE') process.exit(0); throw e; });
const cfg = n => { try { return readFileSync(`${homedir()}/.config/igneum/${n}`, 'utf8').trim(); } catch { return ''; } };

// the same intake reader as tools/logs.mjs (Neon HTTP SQL over fetch)
function db() {
  const m = /^DATABASE_URL=(.*)$/m.exec(cfg('env'));
  if (!m) { console.error('DATABASE_URL not found in ~/.config/igneum/env'); process.exit(1); }
  const url = m[1].trim().replace(/^['"]|['"]$/g, '');
  const host = new URL(url).hostname.replace('-pooler', '');
  return async (query, params = []) => {
    const r = await fetch(`https://${host}/sql`, { method: 'POST', headers: { 'Neon-Connection-String': url, 'Content-Type': 'application/json' }, body: JSON.stringify({ query, params }) });
    const j = await r.json();
    if (!r.ok) throw new Error(j.message || JSON.stringify(j));
    return j.rows;
  };
}
const when = ts => new Date(ts).toISOString().replace('T', ' ').slice(0, 19) + ' UTC';
// The SUMMARY line is the first line of a progress upload but the app's closing report starts with its IGNEUM-APP
// header (4 October 2026: three watchers never saw a job finish). Scan the first lines for it, and read the closing
// 'job <id>: <status> (exit N)' line as final too.
const summaryOf = lines => {
  const ls = (lines || '').split('\n');
  let s = null;
  for (const l of ls.slice(0, 12)) { if (l.startsWith('SUMMARY ')) { try { s = JSON.parse(l.slice(8)); } catch { s = null; } break; } }
  const close = ls.slice(-40).map(l => /^job \S+: (done|failed|timeout|aborted) \(exit (-?\d+)\) after (\d+) s: (.*)$/.exec(l)).filter(Boolean).pop();
  if (close) { s = s || {}; s.status = close[1]; s.exit = Number(close[2]); s.duration_s = Number(close[3]); s.summary = close[4]; s.finished_at = s.finished_at || 'closed'; }
  else if (s && !s.finished_at) s.status = 'running';
  if (s) s.errors = ls.filter(l => /^(Error:|BUILD FAILED|GATE FAILED)|\bFAILED\b|panicked at/.test(l)).slice(0, 5);
  return s;
};

const [a, b] = process.argv.slice(2);

if (!a) {
  const tok = cfg('dl-token');
  if (!tok) { console.error('no ~/.config/igneum/dl-token'); process.exit(1); }
  const base = `https://dl.igneum.network/dl/${tok}`;
  // the envelope first (one object: file text and signature together, what the 0.3.9 apps read), the pair when
  // no envelope is published yet
  let bytes, sig, form = 'envelope';
  const e = await fetch(`${base}/igneum-jobs.signed.json`, { headers: { 'Cache-Control': 'no-cache' } });
  if (e.ok) {
    const env = JSON.parse(await e.text());
    if (env.format !== 'igneum-jobs-signed-1' || typeof env.file !== 'string' || typeof env.sig !== 'string') { console.error('the signed jobs file has another shape'); process.exit(1); }
    bytes = Buffer.from(env.file, 'utf8'); sig = env.sig.trim();
  } else {
    form = 'pair';
    const r = await fetch(`${base}/igneum-jobs.json`, { headers: { 'Cache-Control': 'no-cache' } });
    if (r.status === 404) { console.log('no jobs file published'); process.exit(0); }
    if (!r.ok) { console.error(`jobs file: http ${r.status}`); process.exit(1); }
    bytes = Buffer.from(await r.arrayBuffer());
    sig = (await (await fetch(`${base}/igneum-jobs.json.sig`)).text()).trim();
  }
  const pub = cfg('ota-signing-key.pub');
  let verified = 'not checked (no ~/.config/igneum/ota-signing-key.pub)';
  if (pub) {
    const key = createPublicKey({ key: Buffer.concat([Buffer.from('302a300506032b6570032100', 'hex'), Buffer.from(pub, 'hex')]), format: 'der', type: 'spki' });
    verified = verify(null, bytes, key, Buffer.from(sig, 'hex')) ? 'signature OK' : 'SIGNATURE DOES NOT VERIFY (the apps refuse this file)';
  }
  const f = JSON.parse(bytes.toString('utf8'));
  console.log(`${base.replace(tok, '<token>')}/${form === 'envelope' ? 'igneum-jobs.signed.json' : 'igneum-jobs.json (no envelope published)'}: published ${f.published_at}, ${f.jobs.length} job(s), ${verified}`);
  const now = Date.now();
  for (const j of f.jobs) {
    const exp = Date.parse(j.expires_at);
    const t = j.target || {};
    console.log(`  ${exp < now ? 'EXPIRED ' : ''}${j.id}  ${j.kind}  to ${Array.isArray(t.machine_ids) ? t.machine_ids.join(',') : t.machine_ids} on ${t.platform || 'any'}${t.requires && t.requires.length ? ' needs ' + t.requires.join(',') : ''}  until ${j.expires_at}  ${j.title || ''}`);
    const p = JSON.stringify(j.params || {});
    console.log(`      ${p.length > 180 ? p.slice(0, 180) + '...' : p}`);
  }
  process.exit(0);
}

const sql = db();

if (a === 'status') {
  const rows = await sql(`
    SELECT DISTINCT ON (machine) machine, run_id, label, received_at, lines FROM miner_logs
    WHERE run_id LIKE 'job-%' AND label LIKE 'job-%' ORDER BY machine, received_at DESC`);
  if (!rows.length) { console.log('no job reports in the intake yet'); process.exit(0); }
  // the woken latency: each publish is one relay_wake row (stamp, at, meta.added = the ids it added); a machine's
  // latest job that a publish added shows its started_at minus that publish's at. No table yet: nothing shown.
  let publishes = [];
  try { publishes = await sql('SELECT stamp, at, meta FROM relay_wake ORDER BY id DESC LIMIT 200'); } catch { publishes = []; }
  const tsOf = v => Date.parse(String(v || '').replace(' ', 'T').replace(/([+-]\d\d)$/, '$1:00'));
  const publishedAt = id => { for (const p of publishes) { const a = p.meta && Array.isArray(p.meta.added) ? p.meta.added : []; if (a.includes(id)) return tsOf(p.at); } return NaN; };
  const woken = s => {
    if (!s || !s.job || !s.started_at) return '';
    const d = Math.round((tsOf(s.started_at) - publishedAt(s.job)) / 1000);
    return Number.isFinite(d) && d >= 0 && d < 86400 ? `  woken +${d} s after the publish` : '';
  };
  for (const r of rows) {
    const s = summaryOf(r.lines);
    console.log(`${r.machine.padEnd(28)} ${r.run_id.padEnd(44)} ${when(r.received_at)}  ${s ? `${s.status} exit ${s.exit} after ${s.duration_s} s: ${s.summary}` : '(no SUMMARY line)'}${woken(s)}`);
    if (s && s.errors && s.errors.length) for (const e of s.errors) console.log(`${''.padEnd(28)} ERROR ${e.slice(0, 160)}`);
  }
  process.exit(0);
}

async function show(id, all) {
  const rows = all
    ? await sql('SELECT id, run_id, received_at, label, machine, bytes, lines FROM miner_logs WHERE run_id LIKE $1 ORDER BY received_at ASC', [`job-${id}-%`])
    : await sql('SELECT DISTINCT ON (run_id, label) id, run_id, received_at, label, machine, bytes, lines FROM miner_logs WHERE run_id LIKE $1 ORDER BY run_id, label, received_at DESC', [`job-${id}-%`]);
  if (!rows.length) return null;
  const final = [];
  for (const r of rows) {
    console.log(`===== id ${r.id} | ${r.run_id} | ${r.label} | ${r.machine} | ${when(r.received_at)} | ${r.bytes} bytes =====`);
    const s = summaryOf(r.lines);
    if (s) {
      console.log(`SUMMARY: ${s.status} exit ${s.exit}, started ${s.started_at}${s.finished_at ? ', finished ' + s.finished_at : ''}, ${s.duration_s} s: ${s.summary}`)
      if (s.errors && s.errors.length) for (const e of s.errors) console.log(`ERROR: ${e.slice(0, 200)}`);;
      if (s.results && s.results.length) for (const l of s.results) console.log(`  ${l}`);
      final.push(['done', 'failed', 'timeout', 'aborted'].includes(s.status));
      if (!all) { const rest = r.lines.split('\n').slice(1); const tail = rest.slice(-30); console.log(tail.join('\n')); if (rest.length > 30) console.log(`  (... ${rest.length - 30} more lines; --all prints every upload)`); }
      else process.stdout.write(r.lines.endsWith('\n') ? r.lines : r.lines + '\n');
    } else {
      process.stdout.write(r.lines.endsWith('\n') ? r.lines : r.lines + '\n');
    }
  }
  return final;
}

if (a === 'watch') {
  if (!b) { console.error('watch <job id>'); process.exit(1); }
  for (;;) {
    const f = await show(b, false);
    if (f && f.length && f.every(Boolean)) { console.log('final'); process.exit(0); }
    console.log(`${f ? 'still running' : 'nothing yet'}; again in 30 s (Ctrl+C to stop)`);
    await new Promise(r => setTimeout(r, 30000));
  }
}

const f = await show(a, b === '--all');
if (!f) { console.error(`no uploads for job ${a} (run_id job-${a}-<machine>); the app reports when the job starts and every 5 minutes`); process.exit(1); }