igneum/tools/relay.mjs

#!/usr/bin/env node
// Mac side of the Igneum relay (relay/ in this repo, https://relay.igneum.network).
//   node tools/relay.mjs                         the feed, newest first (last 50)
//   node tools/relay.mjs list [N] [--machine X]  more of the feed
//   node tools/relay.mjs read <id>               print an item; its file is downloaded to --out (default $TMPDIR/igneum-relay)
//   node tools/relay.mjs drop "<text>" | <file>  post a note or a file from the Mac   [--to PC1] [--title "..."] [--body "..." with a file]
//   node tools/relay.mjs task <machine> "title" [file] [--body "..."]       a task for a person or a Claude session on that PC
//   node tools/relay.mjs run <machine> "title" <script.ps1|.sh> [--elevated] [--reboot-continue]   a script the igneum-agent runs
//   node tools/relay.mjs watch [--since <id>]    poll every 10 s and print new items (results included)
//   node tools/relay.mjs inbox <machine> [--ack] what that machine has not read yet
//   node tools/relay.mjs machines | role <name> <miner|prover|bench|mac|phone> | name <hostname> <name>
//   node tools/relay.mjs ack <id> | done <id> | rm <id> | url
// Reads ~/.config/igneum/relay-token (the URL secret), log-intake-key (x-igneum-key), dl-token (for __DL_BASE__ in
// playbooks) and relay-url (optional, default https://relay.igneum.network). Zero dependencies.
import { readFileSync, writeFileSync, mkdirSync, existsSync, statSync } from 'node:fs';
import { homedir, tmpdir, hostname } from 'node:os';
import { basename, join, resolve } from 'node:path';

process.stdout.on('error', e => { if (e.code === 'EPIPE') process.exit(0); throw e; });
const cfg = n => { try { return readFileSync(join(homedir(), '.config', 'igneum', n), 'utf8').trim(); } catch { return ''; } };
const TOKEN = cfg('relay-token'); const KEY = cfg('log-intake-key'); const DL = cfg('dl-token');
const BASE = (cfg('relay-url') || 'https://relay.igneum.network').replace(/\/$/, '');
if (!TOKEN) { console.error('no ~/.config/igneum/relay-token'); process.exit(1); }
const API = `${BASE}/r/${TOKEN}/api/`;
const WEB = `${BASE}/r/${TOKEN}`;

const argv = process.argv.slice(2);
const flags = {}; const pos = [];
for (let i = 0; i < argv.length; i++) {
  const a = argv[i];
  if (a.startsWith('--')) { const k = a.slice(2); const next = argv[i + 1]; if (next !== undefined && !next.startsWith('--') && !['elevated', 'reboot-continue', 'ack', 'all'].includes(k)) { flags[k] = next; i++; } else flags[k] = true; }
  else pos.push(a);
}
const cmd = pos[0] && !/^\d+$/.test(pos[0]) ? pos[0] : (pos[0] ? 'read' : 'list');
if (cmd === 'read' && /^\d+$/.test(pos[0])) pos.unshift('read');

async function api(fn, { q, body } = {}) {
  const r = await fetch(API + fn + (q ? '?' + new URLSearchParams(q) : ''), body === undefined ? { headers: { 'x-igneum-key': KEY } }
    : { method: 'POST', headers: { 'Content-Type': 'application/json', 'x-igneum-key': KEY }, body: JSON.stringify(body) });
  const j = await r.json().catch(() => ({ ok: false, error: `http ${r.status}` }));
  if (!r.ok || j.ok === false) throw new Error(j.error || `http ${r.status}`);
  return j;
}
async function uploadFile(path) {
  const buf = readFileSync(path); const name = basename(path);
  const t = await api('upload', { body: { name, size: buf.length } });
  if (buf.length > t.max) throw new Error(`${name} is ${buf.length} bytes, over the ${t.max} byte cap`);
  const r = await fetch(t.put_url, { method: 'PUT', body: buf, headers: { authorization: `Bearer ${t.token}`, 'x-api-version': t.api_version, 'x-add-random-suffix': '1', 'x-content-type': 'application/octet-stream' } });
  const j = await r.json().catch(() => ({}));
  if (!r.ok || !j.url) throw new Error(`blob upload failed: ${r.status} ${JSON.stringify(j).slice(0, 200)}`);
  return { file_name: name, file_url: j.url, size: buf.length };
}
const fmtSize = n => n < 1024 ? `${n} B` : n < 1048576 ? `${(n / 1024).toFixed(1)} KB` : `${(n / 1048576).toFixed(1)} MB`;
const when = ts => new Date(ts).toISOString().replace('T', ' ').slice(5, 16);
function line(it) {
  const route = it.to === 'all' ? it.from : `${it.from} > ${it.to}`;
  const st = it.done ? ' done' : it.read && (it.kind === 'task' || it.kind === 'run') ? ' read' : (it.kind === 'task' || it.kind === 'run') ? ' NEW' : '';
  const first = (it.body || '').split('\n').find(l => l.trim()) || '';
  const file = it.has_file ? ` [${it.file_name} ${fmtSize(it.size)}]` : '';
  return `${('#' + it.id).padStart(5)} ${when(it.ts)} ${it.kind.padEnd(6)} ${route.padEnd(12)} ${it.title ? it.title + (first ? ': ' : '') : ''}${first.slice(0, 90)}${file}${st}`;
}
function printItem(it) {
  console.log(`===== #${it.id} ${it.kind} from ${it.from} to ${it.to} at ${it.ts}${it.title ? ` | ${it.title}` : ''} =====`);
  const fl = Object.entries(it.flags || {}).filter(([, v]) => v !== false).map(([k, v]) => v === true ? k : `${k}=${v}`).join(' ');
  if (fl || it.task_id) console.log(`[${[fl, it.task_id ? `task #${it.task_id}` : '', it.done ? 'done' : it.read ? 'read' : ''].filter(Boolean).join(' | ')}]`);
  if (it.body) process.stdout.write(it.body.endsWith('\n') ? it.body : it.body + '\n');
}
const outDir = () => { const d = resolve(flags.out || process.env.RELAY_DOWNLOAD_DIR || join(tmpdir(), 'igneum-relay')); mkdirSync(d, { recursive: true }); return d; };
async function download(it) {
  const r = await fetch(`${API}file?id=${it.id}`, { headers: { 'x-igneum-key': KEY }, redirect: 'follow' });
  if (!r.ok) throw new Error(`download http ${r.status}`);
  const buf = Buffer.from(await r.arrayBuffer());
  const p = join(outDir(), `${it.id}-${it.file_name || 'file'}`);
  writeFileSync(p, buf); return p;
}
function playbook(path) {
  let s = readFileSync(path, 'utf8');
  s = s.replace(/__DL_BASE__/g, DL ? `https://dl.igneum.network/dl/${DL}` : 'https://dl.igneum.network/dl/MISSING-DL-TOKEN');
  return s;
}

try {
  if (cmd === 'url') { console.log(WEB); }
  else if (cmd === 'list') {
    const n = Number(pos[1]) || 50; const q = { limit: n }; if (flags.machine) q.machine = flags.machine;
    const j = await api('feed', { q });
    const waiting = j.machines.filter(m => m.unread).map(m => `${m.name} ${m.unread}`).join(', ');
    console.log(`${WEB}\nmachines: ${j.machines.map(m => `${m.name}${m.role ? '/' + m.role : ''}${m.named === false ? ' (unnamed, hostname ' + m.hostname + ')' : ''}${m.last_seen ? ' seen ' + when(m.last_seen) : ''}`).join(' | ')}${waiting ? `\nwaiting: ${waiting}` : ''}`);
    if (!j.items.length) console.log('no items yet');
    for (const it of j.items) console.log(line(it));
  }
  else if (cmd === 'read') {
    const it = (await api('item', { q: { id: pos[1] } })).item; printItem(it);
    if (it.has_file) console.log(`file: ${await download(it)}`);
  }
  else if (cmd === 'drop') {
    const what = pos[1]; if (!what) throw new Error('drop "<text>" or drop <file>');
    const o = { from: flags.from || 'Mac', to: flags.to || 'all', title: flags.title || '' };
    if (existsSync(what) && statSync(what).isFile()) Object.assign(o, await uploadFile(what), { kind: 'file', body: typeof flags.body === 'string' ? flags.body : '' }); else { o.body = what; o.kind = flags.kind || 'text'; }
    const r = await api('drop', { body: o }); console.log(`sent #${r.id} (${r.kind})`);
  }
  else if (cmd === 'task' || cmd === 'run') {
    const [, to, title, file] = pos;
    if (!to || !title) throw new Error(`${cmd} <machine> "title" ${cmd === 'run' ? '<script>' : '[file]'}`);
    const o = { from: flags.from || 'Mac', to, title, kind: cmd, body: flags.body || '', flags: {} };
    if (cmd === 'run') {
      if (!file || !existsSync(file)) throw new Error('run needs a script file (.ps1 for Windows, .sh for the Mac)');
      o.body = playbook(file); o.flags = { elevated: !!flags.elevated, reboot_continue: !!flags['reboot-continue'], shell: file.endsWith('.sh') ? 'bash' : 'powershell', script: basename(file) };
    } else if (file) { if (!existsSync(file)) throw new Error(`no such file ${file}`); Object.assign(o, await uploadFile(file)); }
    const r = await api('task', { body: o }); console.log(`queued #${r.id} ${cmd} for ${to}: ${title}`);
  }
  else if (cmd === 'inbox') {
    const machine = pos[1]; if (!machine) throw new Error('inbox <machine>');
    const j = await api('inbox', { q: { machine, ...(flags.ack ? { ack: 1 } : {}) } });
    if (!j.items.length) console.log(`nothing waiting for ${machine}`); for (const it of j.items) printItem(it);
  }
  else if (cmd === 'machines') { for (const m of (await api('machines')).machines) console.log(`${m.name.padEnd(10)} ${(m.role || '-').padEnd(8)} ${(m.hostname || '-').padEnd(18)} ${m.named === false ? 'UNNAMED ' : ''}${m.last_seen ? 'seen ' + when(m.last_seen) : 'never seen'}${m.info && m.info.gpus ? ' gpu ' + [].concat(m.info.gpus).join(', ') : ''}${m.info && m.info.wsl ? ' wsl ' + m.info.wsl : ''}`); }
  else if (cmd === 'role') { const r = await api('role', { body: { name: pos[1], role: pos[2] } }); console.log(`${r.name} is now ${r.role}`); }
  else if (cmd === 'name') { const r = await api('name', { body: { hostname: pos[1], name: pos[2] } }); console.log(`${r.hostname} is now ${r.name}`); }
  else if (cmd === 'ack') { await api('ack', { body: { ids: [Number(pos[1])] } }); console.log(`#${pos[1]} read`); }
  else if (cmd === 'done') { await api('done', { body: { id: Number(pos[1]) } }); console.log(`#${pos[1]} done`); }
  else if (cmd === 'rm') { await api('delete', { body: { id: Number(pos[1]) } }); console.log(`#${pos[1]} deleted`); }
  else if (cmd === 'watch') {
    let since = Number(flags.since) || (await api('feed', { q: { limit: 1 } })).items[0]?.id || 0;
    console.log(`watching ${WEB} from #${since} (every 10 s, Ctrl+C to stop)`);
    for (;;) {
      try {
        const j = await api('feed', { q: { since } });
        for (const it of j.items.reverse()) {
          console.log(line(it));
          if (it.kind === 'result' || it.kind === 'text') { const tail = (it.body || '').split('\n').slice(-12).join('\n'); if (tail.trim() && it.body.split('\n').length > 1) console.log(tail.replace(/^/gm, '    ')); }
          if (it.has_file && (flags.download || it.kind === 'result')) { try { console.log(`    file: ${await download(it)}`); } catch (e) { console.log(`    file download failed: ${e.message}`); } }
          since = Math.max(since, it.id);
        }
      } catch (e) { console.log(`poll failed: ${e.message}`); }
      await new Promise(r => setTimeout(r, 10000));
    }
  }
  else throw new Error(`unknown command ${cmd}`);
} catch (e) { console.error(e.message); process.exit(1); }