printhive/lib/printerLive.js

// Live, read-only connections to your printers.
//
// One connection per printer stays open, so status arrives as the printer sends
// it (every second or two while printing) instead of on a timer. The printer is
// only ever asked for a full report — never sent anything that changes what it's
// doing. Access codes are held in memory for the connection and never leave
// this module.
const mqtt = require('mqtt');
const { describeAms, describeTray, friendlyError, readStatus, PORT } = require('./printer');

const RECONNECT_MS = 15 * 1000;
const FULL_REPORT_MS = 10 * 60 * 1000; // a fresh full report now and then, in case a change was missed
const TRAYS_SETTLE_MS = 5 * 1000;      // let a spool swap finish reporting before syncing it

const connections = new Map(); // printer id → connection
let onTrays = async () => null;
let connect = (url, options) => mqtt.connect(url, options); // replaceable in tests

// What's loaded, reduced to what decides whether spools need syncing.
const traysKey = (trays) => JSON.stringify((trays || [])
  .map((t) => [t.unit, t.slot, t.empty, t.material, t.colour, t.uuid]));

function allTrays(c) {
  return [...(c.ams || []), ...(c.external ? [c.external] : [])];
}

function requestFullReport(c) {
  if (!c.client.connected) return;
  c.client.publish(`device/${c.serial}/request`,
    JSON.stringify({ pushing: { sequence_id: String(Date.now()), command: 'pushall' } }));
}

// Syncs spools with what's loaded now, and hands back what that did.
async function flushTrays(c) {
  clearTimeout(c.traysTimer);
  c.traysTimer = null;
  c.pendingKey = null;
  const trays = allTrays(c);
  if (!c.ams && !c.external) return null;
  c.syncedKey = traysKey(trays);
  try {
    c.lastSync = await onTrays(c.id, trays);
    return c.lastSync;
  } catch (err) {
    console.warn(`Printer ${c.id} spool sync failed:`, err.message);
    return null;
  }
}

function handleMessage(c, payload) {
  let msg;
  try { msg = JSON.parse(payload.toString()); } catch { return; }
  const p = msg && msg.print;
  if (!p) return;
  c.reports += 1;
  c.updatedAt = new Date();
  Object.assign(c.status, readStatus(p));

  // AMS details come whole when they change; a report without them changes nothing.
  const units = p.ams && p.ams.ams;
  if (Array.isArray(units) && units.every((u) => Array.isArray(u.tray))) c.ams = describeAms(units);
  if (p.vt_tray) c.external = describeTray(p.vt_tray, null);

  // A printer mid-print reports every second or so, so only a change in what's
  // loaded restarts the wait — otherwise the sync would never come round.
  const key = (c.ams || c.external) ? traysKey(allTrays(c)) : null;
  if (key && key !== c.syncedKey && key !== c.pendingKey) {
    c.pendingKey = key;
    clearTimeout(c.traysTimer);
    c.traysTimer = setTimeout(() => flushTrays(c), TRAYS_SETTLE_MS);
  }
  // Someone waiting for a full report is answered by one with the trays in it.
  if (Array.isArray(units) || p.vt_tray) for (const resolve of c.waiters.splice(0)) resolve();
}

function open(p) {
  const c = {
    id: p.id,
    serial: p.serial,
    key: [p.host, p.serial, p.access_code].join('\n'),
    connected: false,
    error: null,
    reports: 0,
    updatedAt: null,
    connectedAt: null,
    status: {},
    ams: null,
    external: null,
    syncedKey: null,
    pendingKey: null,
    lastSync: null,
    traysTimer: null,
    waiters: [],
  };
  c.client = connect(`mqtts://${p.host}:${PORT}`, {
    username: 'bblp',
    password: p.access_code,
    rejectUnauthorized: false, // the printer's certificate is self-signed
    connectTimeout: 10 * 1000,
    reconnectPeriod: RECONNECT_MS,
    keepalive: 30,
    clientId: `printhive-${p.id}-${Math.random().toString(16).slice(2, 8)}`,
  });
  c.client.on('connect', () => {
    c.connected = true;
    c.error = null;
    c.connectedAt = new Date();
    c.client.subscribe(`device/${c.serial}/report`, (err) => {
      if (err) { c.error = `The printer wouldn’t let Printhive listen: ${err.message}`; return; }
      requestFullReport(c);
    });
  });
  c.client.on('message', (_topic, payload) => handleMessage(c, payload));
  c.client.on('error', (err) => { c.error = friendlyError(err); });
  c.client.on('close', () => { c.connected = false; });
  c.client.on('offline', () => { c.connected = false; });
  c.fullReportTimer = setInterval(() => requestFullReport(c), FULL_REPORT_MS);
  c.fullReportTimer.unref();
  return c;
}

function close(c) {
  clearInterval(c.fullReportTimer);
  clearTimeout(c.traysTimer);
  for (const resolve of c.waiters.splice(0)) resolve();
  c.client.end(true);
}

// Brings the connections in line with the printers you have: new ones opened,
// removed ones closed, and any whose address, serial or code changed reopened.
function reconcile(printerRows) {
  const wanted = new Map();
  for (const p of printerRows) {
    if (p.host && p.serial && p.access_code) wanted.set(p.id, p);
  }
  for (const [id, c] of connections) {
    const p = wanted.get(id);
    if (!p || [p.host, p.serial, p.access_code].join('\n') !== c.key) {
      close(c);
      connections.delete(id);
    }
  }
  for (const [id, p] of wanted) {
    if (!connections.has(id)) connections.set(id, open(p));
  }
}

// What one printer is doing, as last reported. Nothing here identifies a spool
// or carries a credential.
function snapshot(id) {
  const c = connections.get(id);
  if (!c) return { live: false };
  const loaded = allTrays(c).map((t) => ({
    unit: t.unit, slot: t.slot, empty: t.empty, material: t.material, colour: t.colour, tagged: Boolean(t.uuid),
  }));
  return {
    live: true,
    connected: c.connected,
    error: c.connected ? null : c.error,
    updatedAt: c.updatedAt,
    status: c.status,
    loaded: c.ams || c.external ? loaded : null,
  };
}

function snapshots() {
  const out = {};
  for (const id of connections.keys()) out[id] = snapshot(id);
  return out;
}

// Asks for a full report now, waits for it, and syncs spools from it — what
// "Test & sync" does. Uses the open connection rather than making another.
async function refresh(id, waitMs = 8000) {
  const c = connections.get(id);
  if (!c) return null;
  const started = Date.now();
  const before = c.reports;
  if (!c.connected && !c.client.reconnecting) c.client.reconnect();
  await new Promise((resolve) => {
    const timer = setTimeout(resolve, waitMs);
    const ask = () => {
      if (c.connected) { requestFullReport(c); return; }
      if (Date.now() - started < waitMs) setTimeout(ask, 250);
    };
    c.waiters.push(() => { clearTimeout(timer); setTimeout(resolve, 1000); }); // let the rest of the report arrive
    ask();
  });
  const sync = c.reports > before ? await flushTrays(c) : null;
  return {
    connected: c.connected,
    reports: c.reports - before,
    printer: { state: c.status.state, progress: c.status.progress },
    ams: c.ams,
    external: c.external,
    error: c.connected ? null : (c.error || 'Couldn’t reach the printer.'),
    sync,
    ms: Date.now() - started,
  };
}

function setTraysHandler(fn) {
  onTrays = fn;
}

function stopAll() {
  for (const c of connections.values()) close(c);
  connections.clear();
}

// For tests: connect with a stand-in instead of the network.
function setConnector(fn) {
  connect = fn;
}

module.exports = { reconcile, snapshot, snapshots, refresh, setTraysHandler, stopAll, setConnector };