ProdDash/modules/prodcom-transcript/server.js

'use strict';

/**
 * ProdCom Transcript — server part.
 *
 * One connection to ProdCom for the whole server, however many dashboards
 * show the transcript. The module holds the channel list, the groups and
 * the last entries, follows ProdCom's SSE stream, and hands every browser a
 * snapshot on connect plus each change as it happens. (It used to proxy each
 * browser's stream to ProdCom one to one — N screens meant N upstream
 * connections; ProdCom now sees one.)
 *
 * Routes (under /api/modules/prodcom-transcript):
 *   GET  /state           → { status, channels, groups, entries }   (entries by date)
 *   GET  /stream          → SSE: `state` (the snapshot, first), then `entry` for each
 *                           changed entry, `clear`, and `status` when the link changes
 *   GET  /messaging       → what the Send Message module may send as: every
 *                           channel that accepts typed text (typingEnabled), as
 *                           an identity — typed text goes out *as* that channel
 *                           (the module guide's contract)
 *   POST /messaging/send  → { destination, text } with the Send Message unlock
 *                           as a Bearer token — checked with that module over
 *                           loopback, then POST /api/v1/channels/{id}/send, so a
 *                           dashboard's typed text goes out on a ProdCom channel
 *                           over this one connection. This module has no PIN and
 *                           no send tile of its own.
 *
 * ProdCom's API, as the reference listener speaks it:
 *   GET /api/v1/channels, /api/v1/groups, /api/v1/transcript?limit=N and the
 *   SSE /api/v1/transcript/stream, whose frames carry an entry, or a
 *   { type, entry | data } wrapper, in a few historical shapes (see
 *   extractEntries). POST /api/v1/channels/{id}/send with { text } puts typed
 *   text on a channel (it comes back on the stream with source "typed"). A
 *   bearer API key is sent when one is configured.
 */

const http = require('http');

/** The module that owns the PIN and the unlock tokens — see verifyUnlock(). */
const SENDER_ID = 'send-message';
const MAX_TEXT_CHARS = 2000;
const BODY_LIMIT_BYTES = 64 * 1024;

const HISTORY_LIMIT = 100;            // entries asked for on (re)connect
const ENTRIES_MAX = 300;              // entries kept in memory
const RETRY_MIN_MS = 2000;
const RETRY_MAX_MS = 15000;
const REQUEST_TIMEOUT_MS = 8000;
const REFRESH_LISTS_MS = 5 * 60 * 1000; // channels and groups re-read
const HEARTBEAT_MS = 15000;

/** The ProdCom base URL from the endpoint config. A legacy prodcomUrl (saved
    before the endpoint field existed) can only still be present until the
    admin form is re-saved, so when it exists it wins over the endpoint's
    schema default. */
function resolveTarget(config) {
  if (config.prodcomUrl) return new URL(config.prodcomUrl);
  const ep = config.prodcom;
  if (ep && ep.host && Number(ep.port)) return new URL(`http://${ep.host}:${Number(ep.port)}`);
  return null;
}

/** Entries inside whatever ProdCom sent: a bare entry, or one wrapped in data / payload / entry / entries / transcript. */
function extractEntries(obj) {
  if (!obj || typeof obj !== 'object') return [];
  if (Array.isArray(obj)) return obj.flatMap(extractEntries);
  if (typeof obj.text === 'string' && (obj.id || obj.channelId)) return [obj];
  for (const key of ['data', 'payload', 'entry', 'entries', 'transcript']) {
    if (obj[key] && typeof obj[key] === 'object') {
      const found = extractEntries(obj[key]);
      if (found.length) return found;
    }
  }
  return [];
}

/** One entry as the tiles read it; an entry without an id is keyed by channel and time. */
function normalizeEntry(e) {
  const id = e.id !== undefined && e.id !== null && e.id !== '' ? String(e.id) : `${e.channelId || ''}|${e.date || ''}`;
  return {
    id,
    channelId: e.channelId === undefined || e.channelId === null ? '' : String(e.channelId),
    channelName: e.channelName ? String(e.channelName) : '',
    text: String(e.text ?? ''),
    date: e.date ? String(e.date) : new Date().toISOString(),
    inProgress: Boolean(e.inProgress),
    // audio (speech), typed (a dashboard or ProdCom's own box), automation
    source: e.source === 'typed' || e.source === 'automation' ? String(e.source) : 'audio',
  };
}

/* ── small helpers ──────────────────────────────────────────────────── */

function sendJson(res, code, body) {
  res.writeHead(code, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store' });
  res.end(JSON.stringify(body));
}

function readJson(req) {
  return new Promise((resolve, reject) => {
    let data = '';
    let size = 0;
    req.on('data', (chunk) => {
      size += chunk.length;
      if (size > BODY_LIMIT_BYTES) {
        reject(new Error('Request body too large.'));
        req.destroy();
        return;
      }
      data += chunk;
    });
    req.on('end', () => {
      if (!data.trim()) return resolve({});
      try {
        const body = JSON.parse(data);
        resolve(body && typeof body === 'object' ? body : {});
      } catch {
        reject(new Error('Malformed JSON.'));
      }
    });
    req.on('error', reject);
  });
}

/** Route handlers are async — the shell only catches synchronous throws. */
function guard(fn) {
  return (req, res) => {
    Promise.resolve()
      .then(() => fn(req, res))
      .catch((err) => {
        if (!res.headersSent) sendJson(res, 500, { error: err?.message || String(err) });
        else try { res.end(); } catch { /* gone */ }
      });
  };
}

function bearer(req) {
  const m = /^Bearer\s+([A-Za-z0-9]+)\s*$/.exec(String(req.headers.authorization || ''));
  return m ? m[1] : '';
}

/**
 * Is the unlock token on this request good? Only the Send Message module
 * knows — it owns the PIN — so ask it over loopback. Answers { ok: true,
 * expiresAt } or { ok: false, status, error, code } ready to send back.
 */
async function verifyUnlock(req, shellPort) {
  const token = bearer(req);
  if (!token) return { ok: false, status: 401, error: 'Locked — enter the PIN', code: 'locked' };
  if (!shellPort) return { ok: false, status: 503, error: 'This ProdDash cannot verify a PIN (needs 1.4.0 or newer)', code: 'no_shell_port' };
  let res;
  try {
    res = await fetch(`http://127.0.0.1:${shellPort}/api/modules/${SENDER_ID}/verify`, {
      headers: { Authorization: `Bearer ${token}`, Accept: 'application/json' },
      signal: AbortSignal.timeout(5000),
    });
  } catch (err) {
    return { ok: false, status: 502, error: `Cannot reach the Send Message module (${err.cause?.code || err.message})`, code: 'sender_unreachable' };
  }
  if (res.status === 404) return { ok: false, status: 503, error: 'Install the Send Message module to send from a dashboard', code: 'no_sender' };
  let body = {};
  try { body = await res.json(); } catch { /* a non-JSON refusal */ }
  if (res.status === 401 || body?.ok === false) return { ok: false, status: 401, error: body?.error || 'Locked — enter the PIN', code: body?.code || 'locked' };
  if (!res.ok) return { ok: false, status: 502, error: `Send Message module answered HTTP ${res.status}`, code: 'sender_error' };
  return { ok: true, expiresAt: Number(body?.expiresAt) || 0 };
}

let current = null; // the running instance, for routes()

module.exports = {
  init({ config, log, shell }) {
    const shellPort = Number(shell && shell.port) || 0;
    const target = resolveTarget(config);
    const apiKey = String(config.apiKey || '');
    const channels = [];
    const groups = [];
    const entries = new Map(); // id → entry, in arrival order
    const streams = new Set(); // browsers' SSE responses
    let status = { state: 'connecting', message: target ? 'Connecting to ProdCom…' : 'No ProdCom server configured' };
    let stopped = false;
    let upstream = null;
    let generation = 0;        // callbacks from a superseded connection are ignored
    let retryTimer = null;
    let retryMs = RETRY_MIN_MS;

    const authHeaders = () => (apiKey ? { authorization: `Bearer ${apiKey}` } : {});

    async function getJson(pathname) {
      const res = await fetch(new URL(pathname, target), {
        headers: { accept: 'application/json', ...authHeaders() },
        signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
      });
      if (!res.ok) throw new Error(`HTTP ${res.status} for ${pathname}`);
      return res.json();
    }

    function sortedEntries() {
      return [...entries.values()].sort((a, b) => new Date(a.date) - new Date(b.date));
    }

    function snapshot() {
      return { status, channels, groups, entries: sortedEntries() };
    }

    function broadcast(event, data) {
      if (!streams.size) return;
      const frame = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
      for (const res of streams) {
        try {
          res.write(frame);
        } catch {
          streams.delete(res);
        }
      }
    }

    function setStatus(state, message) {
      if (status.state === state && status.message === message) return;
      status = { state, message };
      broadcast('status', status);
    }

    function trim() {
      while (entries.size > ENTRIES_MAX) entries.delete(entries.keys().next().value);
    }

    /** A changed entry: kept, and told to every browser. */
    function upsert(raw) {
      const e = normalizeEntry(raw);
      const prev = entries.get(e.id);
      if (prev && prev.text === e.text && prev.inProgress === e.inProgress && prev.channelId === e.channelId) return;
      entries.set(e.id, e);
      trim();
      broadcast('entry', e);
    }

    function clearAll() {
      entries.clear();
      broadcast('clear', {});
    }

    async function loadLists() {
      const [ch, gr] = await Promise.all([
        getJson('/api/v1/channels'),
        getJson('/api/v1/groups').catch(() => ({ data: [] })), // groups are optional in ProdCom
      ]);
      channels.splice(0, channels.length, ...(ch.data || []).map((c) => ({
        id: String(c.id),
        name: String(c.name || c.id),
        color: c.color || '#8b98a5',
        // whether ProdCom lets text be typed onto this channel (an older
        // ProdCom that doesn't say is taken at its word when asked)
        typingEnabled: c.typingEnabled !== false,
      })));
      groups.splice(0, groups.length, ...(gr.data || []).map((g) => ({ id: String(g.id), name: String(g.name || g.id), channelIds: Array.isArray(g.channelIds) ? g.channelIds.map(String) : [] })));
    }

    async function loadHistory() {
      const body = await getJson(`/api/v1/transcript?limit=${HISTORY_LIMIT}`);
      for (const e of extractEntries(body).map(normalizeEntry)) entries.set(e.id, e);
      trim();
    }

    function handleFrame(block) {
      const data = [];
      for (const line of block.split('\n')) {
        if (line.startsWith('data:')) data.push(line.slice(5).replace(/^ /, ''));
      }
      if (!data.length) return;
      let obj;
      try {
        obj = JSON.parse(data.join('\n'));
      } catch {
        return;
      }
      const kind = String(obj?.type || obj?.event || '').toLowerCase();
      if (kind.includes('clear')) return clearAll();
      extractEntries(obj).forEach(upsert);
    }

    function scheduleRetry() {
      if (stopped) return;
      clearTimeout(retryTimer);
      retryTimer = setTimeout(connect, retryMs);
      retryMs = Math.min(RETRY_MAX_MS, retryMs * 2);
    }

    function fail(gen, why) {
      if (stopped || gen !== generation) return;
      generation += 1; // whatever else this connection reports is stale now
      try { upstream?.destroy(); } catch { /* already gone */ }
      upstream = null;
      setStatus('error', `ProdCom unreachable (${why}) — retrying`);
      log(`ProdCom ${target.origin}: ${why} — retrying in ${retryMs / 1000} s`);
      scheduleRetry();
    }

    /** Lists and history first (a fresh snapshot for the tiles), then the live stream. */
    async function connect() {
      if (stopped || !target) return;
      const gen = ++generation;
      try {
        await loadLists();
        await loadHistory();
      } catch (err) {
        if (gen !== generation || stopped) return;
        return fail(gen, err.cause?.code || err.code || err.message);
      }
      if (gen !== generation || stopped) return;
      broadcast('state', snapshot());
      const url = new URL('/api/v1/transcript/stream', target);
      const req = http.request({
        hostname: url.hostname,
        port: url.port || 80,
        path: url.pathname,
        method: 'GET',
        headers: { accept: 'text/event-stream', ...authHeaders() },
      }, (res) => {
        if (gen !== generation) return void res.destroy();
        if (res.statusCode !== 200) {
          res.resume();
          return fail(gen, `HTTP ${res.statusCode} from the transcript stream`);
        }
        retryMs = RETRY_MIN_MS;
        setStatus('ok', 'Live');
        log(`following ${target.origin} — ${channels.length} channels, ${entries.size} entries`);
        let buf = '';
        res.setEncoding('utf8');
        res.on('data', (chunk) => {
          if (gen !== generation) return;
          buf += chunk;
          let i;
          while ((i = buf.indexOf('\n\n')) >= 0) {
            handleFrame(buf.slice(0, i));
            buf = buf.slice(i + 2);
          }
          if (buf.length > 1e6) buf = ''; // a frame that never ends is dropped
        });
        res.on('end', () => fail(gen, 'stream ended'));
        res.on('error', (err) => fail(gen, err.code || err.message));
      });
      req.on('error', (err) => fail(gen, err.code || err.message));
      req.end();
      upstream = req;
    }

    const listsTimer = setInterval(() => {
      if (stopped || status.state !== 'ok') return;
      loadLists().then(() => broadcast('state', snapshot())).catch(() => { /* the next reconnect re-reads them */ });
    }, REFRESH_LISTS_MS);
    listsTimer.unref?.();
    const heartbeat = setInterval(() => {
      for (const res of streams) {
        try {
          res.write(': ping\n\n');
        } catch {
          streams.delete(res);
        }
      }
    }, HEARTBEAT_MS);
    heartbeat.unref?.();

    if (target) {
      log(`one connection to ProdCom at ${target.origin}, shared by every dashboard`);
      connect();
    } else {
      log('no ProdCom server configured — set the endpoint in /admin');
    }

    /** A channel by id, or by name (case-insensitive) — ProdCom accepts both, the tiles use ids. */
    function channelByRef(ref) {
      const s = String(ref ?? '').trim();
      if (!s) return null;
      return channels.find((c) => c.id === s) || channels.find((c) => c.name.toLowerCase() === s.toLowerCase()) || null;
    }

    /**
     * What the Send Message module may send as: the channels that accept
     * typed text. A ProdCom channel is an identity, not a recipient — text
     * typed onto it goes out *as* that channel — so the destinations are
     * declared as such, and a channel with typing off is left out.
     */
    function messagingDescriptor() {
      const h = current ? current.health() : { status: 'connecting', message: 'Restarting…' };
      return {
        label: 'ProdCom',
        identity: '',
        destinationRole: 'identity',
        status: { state: h.status, message: h.message },
        destinations: channels.filter((c) => c.typingEnabled).map((c) => ({ id: c.id, label: c.name, color: c.color || '' })),
        emptyMessage: channels.length
          ? 'No ProdCom channel accepts typed text — turn on typing for one in ProdCom'
          : 'No ProdCom channels yet',
        maxChars: MAX_TEXT_CHARS,
      };
    }

    /** Typed text onto a channel: the one ProdCom connection does the talking. */
    async function sendText(ch, text) {
      if (!target) return { status: 502, error: 'No ProdCom server configured — set the endpoint in Admin → ProdCom', code: 'no_target' };
      let res;
      try {
        res = await fetch(new URL(`/api/v1/channels/${encodeURIComponent(ch.id)}/send`, target), {
          method: 'POST',
          headers: { 'content-type': 'application/json; charset=utf-8', accept: 'application/json', ...authHeaders() },
          body: JSON.stringify({ text }),
          signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
        });
      } catch (err) {
        return { status: 502, error: `ProdCom unreachable (${err.cause?.code || err.code || err.message}) — not sent`, code: 'network' };
      }
      if (res.ok) {
        res.body?.cancel?.().catch?.(() => {});
        return { status: 200 };
      }
      let detail = '';
      let code = '';
      try {
        const body = await res.json();
        detail = String(body?.error?.message || '');
        code = String(body?.error?.code || '');
      } catch { /* no JSON */ }
      const tail = detail ? ` — ${detail}` : '';
      if (res.status === 401) return { status: 502, error: `ProdCom rejected the API key (${code || 'UNAUTHORIZED'}) — check Admin → ProdCom`, code: 'unauthorized' };
      if (res.status === 404) return { status: 404, error: `ProdCom has no channel "${ch.name}" any more (${code || 'NOT_FOUND'}) — pick another in Send as`, code: 'not_found' };
      if (res.status === 429) return { status: 429, error: `ProdCom is rate-limiting — try again in a moment${tail}`, code: 'ratelimited' };
      if (res.status === 400 || res.status === 422) return { status: 400, error: `ProdCom refused the message (${code || res.status})${tail}`, code: 'refused' };
      return { status: 502, error: `ProdCom answered HTTP ${res.status}${tail}`, code: code || `http_${res.status}` };
    }

    current = {
      snapshot,
      streams,
      shellPort,
      channelByRef,
      messagingDescriptor,
      sendText,
      log,
      health() {
        if (status.state === 'ok') return { status: 'ok', message: `Live — ${channels.length} channel${channels.length === 1 ? '' : 's'}, ${entries.size} entries` };
        // Nothing to connect to is a setting that is missing, not a link that
        // is coming up — it reads red, like the other modules say it.
        if (!target) return { status: 'error', message: status.message };
        return { status: status.state === 'error' ? 'error' : 'connecting', message: status.message };
      },
    };

    return {
      stop() {
        stopped = true;
        generation += 1;
        clearTimeout(retryTimer);
        clearInterval(listsTimer);
        clearInterval(heartbeat);
        try { upstream?.destroy(); } catch { /* already gone */ }
        for (const res of streams) {
          try { res.end(); } catch { /* gone */ }
        }
        streams.clear();
        if (current?.streams === streams) current = null;
      },
      health: () => (current ? current.health() : { status: 'connecting', message: 'Restarting…' }),
    };
  },

  routes() {
    const unmounted = (res) => sendJson(res, 503, { error: 'ProdCom module is restarting — try again' });
    return {
      'GET /state': (req, res) => {
        if (!current) return unmounted(res);
        sendJson(res, 200, current.snapshot());
      },

      // The Send Message module reads this over loopback: where it may send.
      'GET /messaging': (req, res) => {
        if (!current) return unmounted(res);
        sendJson(res, 200, current.messagingDescriptor());
      },

      // …and hands a message here, with the browser's unlock token. The token
      // is checked with the Send Message module first: no PIN entered there,
      // nothing goes out on a channel.
      'POST /messaging/send': guard(async (req, res) => {
        if (!current) return unmounted(res);
        const v = await verifyUnlock(req, current.shellPort);
        if (!v.ok) return sendJson(res, v.status, { error: v.error, code: v.code });
        let body;
        try {
          body = await readJson(req);
        } catch (err) {
          return sendJson(res, 400, { error: err.message });
        }
        const text = String(body.text ?? '').replace(/\r\n/g, '\n').trim();
        if (!text) return sendJson(res, 400, { error: 'Nothing to send' });
        if (text.length > MAX_TEXT_CHARS) return sendJson(res, 400, { error: `Too long — ${MAX_TEXT_CHARS} characters at most` });
        const ch = current.channelByRef(body.destination);
        if (!ch) {
          const s = String(body.destination ?? '').trim();
          return sendJson(res, 400, { error: s ? `No ProdCom channel "${s}" — pick another in Send as` : 'Which ProdCom channel?', code: 'bad_destination' });
        }
        if (!ch.typingEnabled) {
          return sendJson(res, 400, { error: `${ch.name} doesn't accept typed text — turn on typing for it in ProdCom`, code: 'typing_off' });
        }
        const out = await current.sendText(ch, text);
        if (out.status !== 200) {
          current.log(`send to ${ch.name} failed: ${out.error}`);
          return sendJson(res, out.status, { error: out.error, code: out.code });
        }
        const preview = text.replace(/\s+/g, ' ').slice(0, 60);
        current.log(`sent to ${ch.name}: "${preview}${text.length > 60 ? '…' : ''}"`);
        sendJson(res, 200, { ok: true, label: ch.name, expiresAt: v.expiresAt });
      }),
      'GET /stream': (req, res) => {
        res.writeHead(200, {
          'Content-Type': 'text/event-stream; charset=utf-8',
          'Cache-Control': 'no-cache, no-transform',
          Connection: 'keep-alive',
          'X-Accel-Buffering': 'no',
        });
        res.write('retry: 3000\n\n');
        if (!current) {
          res.write(`event: state\ndata: ${JSON.stringify({ status: { state: 'connecting', message: 'ProdCom module is restarting…' }, channels: [], groups: [], entries: [] })}\n\n`);
          return void res.end();
        }
        res.write(`event: state\ndata: ${JSON.stringify(current.snapshot())}\n\n`);
        try { req.socket.setKeepAlive(true, 15000); } catch { /* gone */ }
        current.streams.add(res);
        req.on('close', () => current && current.streams.delete(res));
      },
    };
  },
};