ProdDash/modules/overflow-tap/server.js

'use strict';

/**
 * Overflow Tap — server entry.
 *
 * Does what Overflow's "ProPresenter Integration" helper app does, inside
 * ProdDash: watches the live slide in ProPresenter (the presentation layer
 * and the announcement layer), and when its notes carry one of the keywords
 * mapped in the Overflow Dashboard, tells Overflow — which points that
 * keyword's Tap groups at its destination, so what people reach when they
 * tap a disc follows what is on screen. A slide with no keyword changes
 * nothing; the last destination stays until the next keyworded slide.
 *
 * A switch Overflow refuses is retried three times, ten seconds apart, with
 * the tile's dot left alone meanwhile; only when those are spent does the
 * dot go red (and the alert fire, if switched on) — the booth hears about a
 * problem the module could not fix itself, not about every hiccup.
 *
 *   GET  /state      → the view-model every tile draws
 *   GET  /stream     → SSE, `state` events on every change (heartbeated)
 *   POST /refresh    → re-read the keyword mappings from the Dashboard now
 *   GET  /keywords   → the mappings as read, for a look while setting up
 *
 * ProPresenter is polled (GET /v1/status/slide for the presentation layer,
 * GET /v1/announcement/slide_index for announcements, once a second each —
 * small requests, nothing heavier) because its chunked update stream is the
 * thing that goes quiet without saying so. Overflow is the Dashboard's own
 * API, signed in with a Dashboard login — the calls, and where they come
 * from, are at the top of overflow.js. Keyword mappings are read on start,
 * every few minutes after that, on /refresh, and from every trigger's answer.
 *
 * OVERFLOW_TAP_API_BASE=http://127.0.0.1:24720 points the module at
 * tools/overflow-tap-mock.js, which also plays a ProPresenter on :1602.
 */

const crypto = require('crypto');
const { createOverflowClient, OverflowError, describeSession } = require('./overflow');
const { matchKeywords, sameSet, cleanNotes, norm } = require('./keywords');

const POLL_MS = 1000;                  // ProPresenter, while it answers
const RETRY_MS = 3000;                 // ProPresenter, while it does not
const PP_TIMEOUT_MS = 4000;
const HEARTBEAT_MS = 15000;
const KEYWORDS_REFRESH_MS = 5 * 60 * 1000;
const OVERFLOW_RETRY_MS = 30 * 1000;
const LOGIN_MIN_GAP_MS = 20 * 1000;    // a refused sign-in is not retried sooner (429 = "too many attempts")
const SESSION_MARGIN_MS = 60 * 1000;   // sign in again this long before the token expires
const SWITCH_RETRY_MS = 10 * 1000;
const SWITCH_RETRIES = 3;
const HISTORY_MAX = 8;                 // the tile shows four; a few more for the hover
const ALERT_SECONDS = 30;

/** Set by init(), read by routes() — both are rebuilt together on a config save. */
let current = null;

/**
 * The signed-in session, kept across re-inits for the same login: an admin
 * saving config five times in a minute must not become five sign-ins
 * (Overflow refuses that with 429). Keyed on a hash of the login, so a
 * changed email or password starts fresh; the token itself never leaves the
 * server process.
 */
let sessionCache = null; // { key, token, expiresAt, email, nonprofitId }

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

/** One line for a log or a tile; a fetch failure names its cause (ECONNREFUSED, ENOTFOUND…). */
function describe(err) {
  if (err instanceof OverflowError) return err.message;
  if (err && err.name === 'AbortError') return 'timed out';
  const msg = err && err.message ? err.message : String(err);
  const cause = err && err.cause && (err.cause.code || err.cause.message);
  return cause && !msg.includes(String(cause)) ? `${msg} (${cause})` : msg;
}

function unique(list) {
  const seen = new Set();
  const out = [];
  for (const v of list) {
    const k = norm(v);
    if (!k || seen.has(k)) continue;
    seen.add(k);
    out.push(v);
  }
  return out;
}

/** The destination and group names Overflow attached to a keyword mapping, whatever the field names turn out to be. */
function destinationOf(entry) {
  return String((entry && (entry.destinationName || (entry.destination && entry.destination.name))) || '');
}
function groupsOf(entry) {
  const list = entry && Array.isArray(entry.tapGroups) ? entry.tapGroups : [];
  return list.map((g) => String((g && (g.tapGroupName || g.name || g.id)) || '')).filter(Boolean);
}

module.exports = {
  init({ config, log, shell }) {
    /* ── config ────────────────────────────────────────────────────── */
    const ep = config.propresenter && typeof config.propresenter === 'object' ? config.propresenter : {};
    const host = String(ep.host || '').trim();
    const port = Number(ep.port) || 1025;
    const ppPassword = String(config.password || '');
    const email = String(config.overflowEmail || '').trim();
    const ofPassword = String(config.overflowPassword || '');
    const staffNonprofit = String(config.nonprofitId || '').trim();
    const configured = Boolean(email && ofPassword);
    const alertEnabled = Boolean(config.alertEnabled);
    const alertChannel = Math.min(6, Math.max(1, Math.round(Number(config.alertChannel) || 1)));
    const sessionKey = crypto.createHash('sha256').update(`${email}\n${ofPassword}`).digest('hex');

    const overflow = createOverflowClient();

    /* ── state ─────────────────────────────────────────────────────── */
    const streams = new Set();
    const timers = new Set();
    let stopped = false;
    let lastFrame = '';

    let keywordCache = new Map(); // norm(keyword) → the Dashboard's entry
    let session = null;           // { token, expiresAt, email, nonprofitId }
    let lastLoginAt = 0;
    let lastLoginError = null;
    let wanted = null;            // the keywords Tap should be on: { keywords, keyword, destination, groups }
    let inFlight = false;
    let retriesLeft = 0;
    let retryTimer = null;
    let alertRaised = false;
    /**
     * Refusals in a row, across keywords. The retry budget is per keyword, but
     * a service keeps bringing new keywords: if Overflow refuses every one of
     * them, each would start a fresh budget and the dot would never go red.
     * So the budget is also spent by refusals as a whole, and a success
     * clears it.
     */
    let consecutiveFailures = 0;

    // Presentation layer: the slide /v1/status/slide reports.
    let lastSlideKey = null;
    let lastSlide = null;
    let lastPresKeywords = [];
    // Announcement layer: the slide /v1/announcement/slide_index points at, read from /v1/announcement/active.
    let annSupported = true;
    let lastAnnKey = null;
    let annNotes = '';
    let annLive = false;
    let lastAnnKeywords = [];

    const state = {
      status: 'connecting',
      message: '',
      propresenter: { configured: Boolean(host), reachable: false, error: '', host, port },
      overflow: { configured, reachable: false, error: '', keywords: 0, loadedAt: 0 },
      keywords: [],
      slide: { live: false, keywords: [], hasNotes: false },
      announcement: { live: false, keywords: [] },
      current: null,
      history: [],
      updatedAt: 0,
    };

    /* ── timers that die with the handle ───────────────────────────── */

    function later(fn, ms) {
      const t = setTimeout(() => {
        timers.delete(t);
        if (!stopped) fn();
      }, ms);
      t.unref?.();
      timers.add(t);
      return t;
    }

    /* ── health, shared by the admin line and the tiles ────────────── */

    function assess(forAdmin = false) {
      const pp = state.propresenter;
      const of = state.overflow;
      if (!pp.configured) return ['error', 'No ProPresenter host configured — set one in /admin'];
      if (!of.configured) return ['error', 'Overflow login not set — add the email and password in /admin'];
      if (!of.reachable) return ['error', of.error || 'Signing in to Overflow…'];
      if (!pp.reachable) return ['error', pp.error ? `ProPresenter unreachable: ${pp.error}` : 'Reaching ProPresenter…'];
      if (!state.keywords.length) return ['error', 'No keyword mappings in the Overflow Dashboard (Integrations › ProPresenter)'];
      // A refused switch the module is still retrying on its own is amber —
      // being worked on; one it has given up on is red — the booth's problem.
      const cur = state.current;
      if (cur && !cur.ok && !cur.pending && !cur.retrying) {
        return ['error', `Tap not switched to ${cur.keyword} — ${cur.error}`];
      }
      if (cur && !cur.ok && (cur.retrying || consecutiveFailures > 0)) {
        return ['connecting', `Tap not switched to ${cur.keyword} yet — retrying (${consecutiveFailures} of ${SWITCH_RETRIES + 1} attempts failed): ${cur.error || 'in flight'}`];
      }
      const n = state.keywords.length;
      const who = forAdmin && session && session.email ? `${session.email}, ` : '';
      return ['ok', `Following ${host}:${port} · Overflow: ${who}${n} keyword${n === 1 ? '' : 's'}`];
    }

    function publish() {
      if (stopped) return;
      [state.status, state.message] = assess(false);
      const { updatedAt, ...rest } = state;
      let frame;
      try {
        frame = JSON.stringify(rest);
      } catch {
        return;
      }
      if (frame === lastFrame) return;
      lastFrame = frame;
      state.updatedAt = Date.now();
      const payload = `event: state\ndata: ${JSON.stringify(state)}\n\n`;
      for (const res of streams) {
        try {
          res.write(payload);
        } catch {
          streams.delete(res);
        }
      }
    }

    /* ── the Overflow session ──────────────────────────────────────── */

    async function ensureSession() {
      const now = Date.now();
      if (session && now < session.expiresAt - SESSION_MARGIN_MS) return session;
      if (sessionCache && sessionCache.key === sessionKey && now < sessionCache.expiresAt - SESSION_MARGIN_MS) {
        session = sessionCache;
        return session;
      }
      if (lastLoginError && now - lastLoginAt < LOGIN_MIN_GAP_MS) throw lastLoginError;
      lastLoginAt = now;
      let token;
      try {
        token = await overflow.login(email, ofPassword);
      } catch (err) {
        lastLoginError = err;
        throw err;
      }
      lastLoginError = null;
      const info = describeSession(token);
      if (!info.hasTap) throw new OverflowError('This Overflow login has no Tap permission — use one with the Organization Admin or Tap Only role');
      let nonprofitId = info.nonprofitId || (info.nonprofits.length === 1 ? info.nonprofits[0].id : '');
      if (info.staff) {
        if (!staffNonprofit) throw new OverflowError('An @overflow.co login has no nonprofit of its own — enter the Nonprofit ID in /admin');
        await overflow.selectNonprofit(token, staffNonprofit);
        nonprofitId = staffNonprofit;
      } else if (!nonprofitId) {
        nonprofitId = staffNonprofit;
      }
      if (!nonprofitId) throw new OverflowError('This Overflow login carries no nonprofit — enter the Nonprofit ID in /admin');
      session = { token, expiresAt: info.expiresAt || now + 12 * 3600 * 1000, email: info.email || email, nonprofitId };
      sessionCache = { key: sessionKey, ...session };
      log(`signed in to Overflow as ${session.email} (nonprofit ${nonprofitId})`);
      return session;
    }

    function dropSession() {
      session = null;
      if (sessionCache && sessionCache.key === sessionKey) sessionCache = null;
    }

    /** Run a signed-in call; when the session is refused, sign in again once and retry. */
    async function withSession(fn) {
      const s = await ensureSession();
      try {
        return await fn(s);
      } catch (err) {
        if (!(err instanceof OverflowError) || !err.auth) throw err;
        dropSession();
        const again = await ensureSession();
        return fn(again);
      }
    }

    /* ── the keyword mappings ──────────────────────────────────────── */

    function rebuildKeywordView() {
      state.keywords = [...keywordCache.values()].map((e) => ({
        keyword: String(e.keyword).trim(),
        destination: destinationOf(e),
        groups: groupsOf(e),
      }));
      state.overflow.keywords = state.keywords.length;
    }

    /** Replace the cache with the Dashboard's list (`merge` folds a partial answer into it instead). */
    function applyKeywordList(list, merge = false) {
      const next = merge ? new Map(keywordCache) : new Map();
      for (const entry of list || []) {
        if (!entry || typeof entry !== 'object') continue;
        const k = norm(entry.keyword);
        if (!k) continue;
        if (entry.active === false || entry.enabled === false || entry.isActive === false) {
          next.delete(k);
          continue;
        }
        next.set(k, entry);
      }
      keywordCache = next;
      rebuildKeywordView();
    }

    let loading = null;
    let refreshTimer = null;

    async function loadKeywords() {
      if (stopped || !configured) return;
      if (loading) return loading;
      loading = (async () => {
        try {
          const list = await withSession((s) => overflow.listKeywords(s.token, s.nonprofitId));
          applyKeywordList(list);
          state.overflow.reachable = true;
          state.overflow.error = '';
          state.overflow.loadedAt = Date.now();
          reconsider(); // a slide already up when the mappings arrive takes effect now
        } catch (err) {
          state.overflow.reachable = false;
          state.overflow.error = describe(err);
          log(`Overflow: ${state.overflow.error}`);
        }
        publish();
      })();
      try {
        await loading;
      } finally {
        loading = null;
      }
      if (stopped) return;
      if (refreshTimer) {
        clearTimeout(refreshTimer);
        timers.delete(refreshTimer);
      }
      refreshTimer = later(loadKeywords, state.overflow.reachable ? KEYWORDS_REFRESH_MS : OVERFLOW_RETRY_MS);
    }

    /* ── the switch ────────────────────────────────────────────────── */

    function describeTarget(keys) {
      const entries = keys.map((k) => keywordCache.get(norm(k))).filter(Boolean);
      return {
        keywords: keys,
        keyword: keys.join(' + '),
        destination: unique(entries.map(destinationOf).filter(Boolean)).join(' + '),
        groups: unique(entries.flatMap(groupsOf)),
      };
    }

    function pushHistory(entry) {
      const { pending, ...rest } = entry;
      state.history.unshift(rest);
      if (state.history.length > HISTORY_MAX) state.history.length = HISTORY_MAX;
    }

    function raiseAlert(keyword) {
      if (!alertEnabled || !shell || typeof shell.alert !== 'function') return;
      alertRaised = true;
      Promise.resolve(shell.alert({ channel: alertChannel, level: 'medium', text: `Tap not switched: ${keyword}`, duration: ALERT_SECONDS }))
        .catch((err) => log(`alert refused: ${describe(err)}`));
    }

    function clearAlert() {
      if (!alertRaised || !shell || typeof shell.clearAlert !== 'function') return;
      alertRaised = false;
      Promise.resolve(shell.clearAlert(alertChannel)).catch(() => {});
    }

    function requestSwitch(target) {
      wanted = target;
      retriesLeft = SWITCH_RETRIES;
      if (retryTimer) {
        clearTimeout(retryTimer);
        timers.delete(retryTimer);
        retryTimer = null;
      }
      runSwitch();
    }

    async function runSwitch() {
      if (stopped || inFlight || !wanted) return;
      const target = wanted;
      const cur = state.current;
      const sameTarget = cur && sameSet(cur.keywords, target.keywords);
      if (sameTarget && cur.ok) return; // already there
      inFlight = true;
      let entry;
      if (sameTarget) {
        entry = cur; // another go at the same switch
      } else {
        if (cur) pushHistory(cur);
        entry = {
          keyword: target.keyword,
          keywords: target.keywords,
          destination: target.destination,
          groups: target.groups,
          at: Date.now(),
          ok: false,
          pending: true,
          retrying: false, // a retry is scheduled: the module is still trying to fix this itself
          error: '',
          attempts: 0,
        };
        state.current = entry;
      }
      entry.pending = true;
      entry.attempts += 1;
      publish();
      try {
        const answer = await withSession((s) => overflow.trigger(s.token, s.nonprofitId, target.keywords));
        if (answer) {
          applyKeywordList(answer, true);
          const fresh = describeTarget(target.keywords);
          entry.destination = fresh.destination || entry.destination;
          entry.groups = fresh.groups.length ? fresh.groups : entry.groups;
        }
        entry.ok = true;
        entry.error = '';
        entry.retrying = false;
        entry.at = Date.now();
        consecutiveFailures = 0;
        clearAlert();
        log(`Tap → ${entry.destination || '?'} (${target.keyword})${entry.groups.length ? ` on ${entry.groups.join(', ')}` : ''}`);
      } catch (err) {
        if (stopped) return;
        entry.ok = false;
        entry.error = describe(err);
        consecutiveFailures += 1;
        if (retriesLeft > 0 && consecutiveFailures <= SWITCH_RETRIES) {
          // Still the module's problem: try again shortly; the dot goes amber.
          retriesLeft -= 1;
          entry.retrying = true;
          log(`Tap not switched to ${target.keyword}: ${entry.error} — retrying in ${SWITCH_RETRY_MS / 1000} s`);
          retryTimer = later(() => {
            retryTimer = null;
            runSwitch();
          }, SWITCH_RETRY_MS);
        } else {
          // Out of retries: now it is the booth's problem — red dot, and the
          // alert if the operator asked for one (raised once per outage, not
          // again for every further keyword Overflow refuses).
          entry.retrying = false;
          log(`Tap not switched to ${target.keyword}: ${entry.error} — giving up after ${consecutiveFailures} refusals`);
          if (!alertRaised) raiseAlert(target.keyword);
        }
      } finally {
        entry.pending = false;
        inFlight = false;
        publish();
        if (!stopped && wanted !== target) runSwitch(); // a newer keyword arrived meanwhile
      }
    }

    /* ── what is on screen, and what to do about it ────────────────── */

    /**
     * The helper's rule, per layer: when the set of keywords the layer's
     * slide carries changes to a non-empty set, send it. Returning to a
     * keyword after a slide without one sends it again; staying on the same
     * keyword across slides sends nothing.
     */
    function reconsider() {
      const list = [...keywordCache.values()];
      const presNotes = lastSlide ? cleanNotes(lastSlide.notes) : '';
      const presKeys = lastSlide ? matchKeywords(presNotes, list) : [];
      state.slide = { live: Boolean(lastSlide), keywords: presKeys, hasNotes: Boolean(presNotes.trim()) };
      const presChanged = !sameSet(presKeys, lastPresKeywords);
      if (presChanged) lastPresKeywords = presKeys;

      const annKeys = annLive ? matchKeywords(annNotes, list) : [];
      state.announcement = { live: annLive, keywords: annKeys };
      const annChanged = !sameSet(annKeys, lastAnnKeywords);
      if (annChanged) lastAnnKeywords = annKeys;

      const batch = [];
      if (presChanged && presKeys.length) batch.push(...presKeys);
      if (annChanged && annKeys.length) batch.push(...annKeys);
      if (batch.length) requestSwitch(describeTarget(unique(batch)));
    }

    /* ── ProPresenter ──────────────────────────────────────────────── */

    function ppUrl(target) {
      const u = new URL(`http://${host}:${port}${target}`);
      if (ppPassword) u.searchParams.set('password', ppPassword);
      return u;
    }

    async function ppJson(target) {
      const ctl = new AbortController();
      const timer = setTimeout(() => ctl.abort(), PP_TIMEOUT_MS);
      try {
        const res = await fetch(ppUrl(target), { headers: { Accept: 'application/json' }, signal: ctl.signal });
        if (!res.ok) {
          const err = new Error(`ProPresenter ${res.status} ${res.statusText}`);
          err.status = res.status;
          throw err;
        }
        const text = await res.text();
        return text ? JSON.parse(text) : null;
      } finally {
        clearTimeout(timer);
      }
    }

    /** The presentation layer: /v1/status/slide names the live slide and carries its notes. */
    async function pollSlide() {
      const data = await ppJson('/v1/status/slide');
      const cur = data && data.current && typeof data.current === 'object' ? data.current : null;
      const key = cur ? `${cur.uuid || ''}\u0000${cur.text || ''}\u0000${cur.notes || ''}` : '';
      if (key === lastSlideKey) return false;
      lastSlideKey = key;
      lastSlide = cur;
      return true;
    }

    /**
     * The announcement layer: its slide index names the document and the
     * slide; the document's slides (every group's, in order, disabled ones
     * included — that is the index's space) carry the notes. An older
     * ProPresenter without this API answers 404 once and is left alone.
     */
    async function pollAnnouncement() {
      if (!annSupported) return false;
      let idx;
      try {
        idx = await ppJson('/v1/announcement/slide_index');
      } catch (err) {
        if (err && err.status === 404) {
          annSupported = false;
          log('this ProPresenter has no announcement layer API — following the presentation layer only');
          return false;
        }
        throw err;
      }
      const node = idx && idx.announcement_index;
      if (!node || !Number.isInteger(node.index)) {
        if (lastAnnKey === '') return false;
        lastAnnKey = '';
        annLive = false;
        annNotes = '';
        return true;
      }
      const id = node.presentation_id || {};
      const key = `${id.uuid || ''}\u0000${id.name || ''}\u0000${node.index}`;
      if (key === lastAnnKey) return false;
      const doc = await ppJson('/v1/announcement/active');
      const ann = doc && doc.announcement;
      const slides = [];
      for (const g of (ann && Array.isArray(ann.groups) ? ann.groups : [])) {
        for (const s of (g && Array.isArray(g.slides) ? g.slides : [])) slides.push(s);
      }
      const slide = slides[node.index];
      annNotes = slide ? cleanNotes(slide.notes) : '';
      annLive = true;
      lastAnnKey = key;
      return true;
    }

    async function pollTick() {
      if (stopped) return;
      let ok = false;
      let changed = false;
      try {
        changed = await pollSlide();
        state.propresenter.reachable = true;
        state.propresenter.error = '';
        ok = true;
        try {
          if (await pollAnnouncement()) changed = true;
        } catch {
          /* the presentation poll already said ProPresenter answers; a hiccup here waits for the next tick */
        }
      } catch (err) {
        state.propresenter.reachable = false;
        state.propresenter.error = err && err.name === 'AbortError' ? 'timed out' : describe(err);
      }
      if (changed) reconsider();
      publish();
      later(pollTick, ok ? POLL_MS : RETRY_MS);
    }

    /* ── go ────────────────────────────────────────────────────────── */

    if (host) {
      pollTick();
      log(`following ProPresenter at ${host}:${port}`);
    } else {
      log('no ProPresenter host configured — set one in /admin');
    }
    if (configured) {
      loadKeywords();
      log(`Overflow at ${overflow.base}`);
    } else {
      log('Overflow login not set — add the email and password in /admin');
    }
    publish();

    const heartbeat = setInterval(() => {
      for (const res of streams) {
        try {
          res.write(': ping\n\n');
        } catch {
          streams.delete(res);
        }
      }
    }, HEARTBEAT_MS);
    heartbeat.unref?.();

    const handle = {
      streams,
      configured,
      getState: () => state,
      refresh: () => loadKeywords(),
    };
    current = handle;

    return {
      stop() {
        stopped = true;
        clearInterval(heartbeat);
        for (const t of timers) clearTimeout(t);
        timers.clear();
        clearAlert();
        for (const res of streams) {
          try { res.end(); } catch { /* already gone */ }
        }
        streams.clear();
        if (current === handle) current = null;
      },
      health() {
        const [status, message] = assess(true);
        return { status, message };
      },
    };
  },

  routes() {
    return {
      'GET /state': (req, res) => {
        if (!current) return json(res, 503, { error: 'Overflow Tap is not running.' });
        json(res, 200, { state: current.getState() });
      },

      '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) return void res.end();
        res.write(`event: state\ndata: ${JSON.stringify(current.getState())}\n\n`);
        try { req.socket.setKeepAlive(true, 15000); } catch { /* gone */ }
        current.streams.add(res);
        req.on('close', () => current && current.streams.delete(res));
      },

      'POST /refresh': async (req, res) => {
        if (!current) return json(res, 503, { error: 'Overflow Tap is not running.' });
        if (!current.configured) return json(res, 400, { error: 'Overflow login is not set.' });
        await current.refresh();
        const st = current.getState();
        if (!st.overflow.reachable) return json(res, 502, { error: st.overflow.error || 'Overflow unreachable.' });
        json(res, 200, { ok: true, keywords: st.overflow.keywords });
      },

      'GET /keywords': (req, res) => {
        if (!current) return json(res, 503, { error: 'Overflow Tap is not running.' });
        const st = current.getState();
        json(res, 200, { keywords: st.keywords, loadedAt: st.overflow.loadedAt, error: st.overflow.error });
      },
    };
  },
};