bible-app/follower-link.js

/*
 * follower-link.js — the relay that lets a second screen mirror the stage TV exactly.
 *
 * WHY THIS EXISTS: the booth needs the Bible TV's picture in ProPresenter (and on to the ATEM)
 * without running a video cable out of the stage PC and without pushing video over Wi-Fi. So no
 * pixels are sent at all. The stage renderer publishes STATE — which tab, which passage, the scroll
 * position, and every pen stroke as vectors — and a browser on the booth machine re-renders the same
 * page natively at full resolution and frame rate. A few KB/s of JSON replaces ~250 Mbit/s of video,
 * and a lost packet costs a stale frame rather than a torn picture.
 *
 *   stage renderer (follower-publish.js)  ──WS /follower-ws?role=publisher──▶  here
 *                                                                              │ fan-out
 *   booth browser  (index.html?follower=1) ◀─WS /follower-ws?role=subscriber───┘
 *
 * Two message classes, deliberately separated:
 *   - `snapshot`  — the whole UI state (tab, passage, geometry, committed strokes). Retained, so a
 *                   follower that connects or reconnects mid-service is correct within one frame.
 *   - everything else (`stroke:*`, `scroll`) — high-rate deltas, relayed and forgotten.
 *
 * The retained snapshot is the reason a follower can be power-cycled mid-sermon and come back right.
 *
 * Only ONE publisher is honoured at a time (the newest wins): two publishers would interleave
 * contradictory states, and on a stage that shows up as text flickering between two passages. A
 * second copy of the app opened by accident is therefore harmless — it simply takes over, and the
 * real TV's next snapshot (published every 250ms) takes it back.
 */
import { WebSocketServer } from 'ws';

const PATH = '/follower-ws';

export function attachFollowerLink(server, opts = {}) {
    const log = opts.log || ((...a) => console.log('[FOLLOWER]', ...a));
    // Settings › Follower Screen switch. Off = no screen may subscribe; anyone connected is dropped.
    const isEnabled = typeof opts.isEnabled === 'function' ? opts.isEnabled : () => true;
    const wss = new WebSocketServer({ noServer: true });

    let publisher = null;          // the stage renderer
    const subscribers = new Set(); // booth screens
    let snapshot = null;           // last full state, replayed to anyone who joins
    let seq = 0;

    const stats = { published: 0, relayed: 0, dropped: 0, subscribersPeak: 0, startedAt: Date.now() };

    function sendTo(ws, obj) {
        // readyState 1 = OPEN. Sending to a socket that is closing throws and would kill the relay.
        if (!ws || ws.readyState !== 1) return false;
        try { ws.send(typeof obj === 'string' ? obj : JSON.stringify(obj)); return true; } catch (_) { return false; }
    }

    function fanOut(raw) {
        stats.relayed++;
        for (const ws of subscribers) {
            /*
             * Never queue behind a slow screen. A booth machine that stalls (or a Wi-Fi hiccup) would
             * otherwise build an unbounded backlog and then replay a burst of stale stroke points —
             * the line would visibly redraw itself late. Skipping while the buffer is deep keeps the
             * follower in the present; the next snapshot repairs whatever it missed.
             */
            if (ws.bufferedAmount > 1 << 20) { stats.dropped++; ws._missed = true; continue; }
            /*
             * It missed something. A snapshot only CARRIES the stroke list when it changed, so a skipped
             * clear-all or a skipped restored list would otherwise never be repaired while the socket
             * stays open. Hand it the completed retained snapshot (which always carries the list) before
             * the current message.
             */
            if (ws._missed) { ws._missed = false; if (snapshot) sendTo(ws, snapshot); }
            sendTo(ws, raw);
        }
    }

    /*
     * Last known committed strokes per passage. A snapshot only CARRIES the stroke list when it changed
     * (bandwidth), so the retained snapshot handed to a late joiner could otherwise lack it — and a
     * follower that reconnected after a Wi-Fi blip would show the chapter with no ink until the next
     * stroke. The retained copy is completed from here before it is stored.
     */
    const lastStrokes = new Map();
    function remember(msg) {
        if (msg && msg.passageKey && Array.isArray(msg.strokes)) lastStrokes.set(msg.passageKey, msg.strokes);
    }
    function complete(msg) {
        if (msg && msg.passageKey && !Array.isArray(msg.strokes) && lastStrokes.has(msg.passageKey)) {
            msg.strokes = lastStrokes.get(msg.passageKey);
        }
        // The Browser tab's page (tens of KB of HTML) rides along only when it changed, exactly like a
        // stroke list — so the retained copy gets it back here, or a screen joining mid-service would show
        // the Browser tab blank until the page next changed.
        if (msg && msg.browser && typeof msg.browser.html !== 'string' && lastBrowser && lastBrowser.ver === msg.browser.domVer) {
            msg.browser.html = lastBrowser.html;
        }
        return msg;
    }
    let lastBrowser = null;   // { ver, html } — the newest Browser-tab page the stage published
    function rememberBrowser(msg) {
        if (msg && msg.browser && typeof msg.browser.html === 'string' && msg.browser.domVer) {
            lastBrowser = { ver: msg.browser.domVer, html: msg.browser.html };
        }
    }

    /*
     * Only the machine this server runs on may publish. The stage renderer connects to its own server
     * over loopback; anything arriving from a LAN address is some other copy of the app — a browser
     * peeking at the stage URL, a booth screen, the admin's laptop. Letting it publish would hijack the
     * program feed: two publishers alternate as each reconnects and the follower flips between their
     * states. (Exactly what happened once with a forgotten test tab.) Refused here regardless of what
     * the client asks for, so no client-side mistake can ever cause it again.
     */
    function isLoopback(req) {
        const a = String((req.socket && req.socket.remoteAddress) || '');
        return a === '127.0.0.1' || a === '::1' || a === '::ffff:127.0.0.1' || a.startsWith('127.');
    }

    /*
     * Second line of defence, for clients this server did not ship: `stage=1` is added to the URL only
     * by the Electron stage window (follower-publish.js). A browser tab on the stage machine running an
     * OLD cached copy of the app is loopback AND still publishes (it predates the client-side gate) —
     * it did exactly that once, alternating with the real stage 229 times in eight minutes and making
     * the booth feed flip between two states. So: while a marked stage publisher is connected, every
     * unmarked publisher is refused; a marked one always displaces an unmarked one; and only another
     * marked publisher (a second Electron window — should not happen) can displace a marked one.
     */
    let publisherIsStage = false;

    wss.on('connection', (ws, req) => {
        let role = 'subscriber', marked = false;
        try {
            const q = new URL(req.url, 'http://x').searchParams;
            role = q.get('role') === 'publisher' ? 'publisher' : 'subscriber';
            marked = q.get('stage') === '1';
        } catch (_) {}

        if (role === 'publisher' && !isLoopback(req)) {
            log(`refused publisher from ${req.socket && req.socket.remoteAddress} — only the stage machine may publish`);
            stats.refusedPublishers = (stats.refusedPublishers || 0) + 1;
            try { ws.close(4003, 'publisher must be local'); } catch (_) {}
            return;
        }

        if (role === 'publisher') {
            if (publisher && publisher !== ws) {
                if (publisherIsStage && !marked) {
                    log('refused an unmarked local publisher (an old browser copy of the app?) — the stage window is already publishing');
                    stats.refusedPublishers = (stats.refusedPublishers || 0) + 1;
                    try { ws.close(4004, 'stage window is publishing'); } catch (_) {}
                    return;
                }
                log(`WARNING: a ${marked ? 'stage' : 'local'} publisher replaced the previous ${publisherIsStage ? 'stage' : 'local'} one`);
                stats.publisherReplacements = (stats.publisherReplacements || 0) + 1;
                try { publisher.close(4000, 'replaced'); } catch (_) {}
            }
            publisher = ws;
            publisherIsStage = marked;
            // A new stage session: nothing the previous one published (its ink, its passage) may be handed to
            // a screen that joins now. The new publisher's first snapshot arrives within a tick.
            lastStrokes.clear();
            lastBrowser = null;
            snapshot = null;
            log(`${marked ? 'stage renderer' : 'local publisher (unmarked)'} connected`);
        } else {
            if (!isEnabled()) {
                stats.refusedSubscribers = (stats.refusedSubscribers || 0) + 1;
                try { ws.close(4005, 'follower screen is turned off in Settings'); } catch (_) {}
                return;
            }
            subscribers.add(ws);
            stats.subscribersPeak = Math.max(stats.subscribersPeak, subscribers.size);
            log(`follower connected (${subscribers.size} watching)`);
            // Make the new screen correct immediately instead of at the next state change.
            if (snapshot) sendTo(ws, snapshot);
            else sendTo(ws, JSON.stringify({ type: 'waiting' }));
        }

        ws.on('message', (data, isBinary) => {
            if (isBinary || ws !== publisher) return;   // only the stage renderer may publish
            const raw = data.toString();
            stats.published++;
            // Peek at the type without parsing the whole payload on the hot path: stroke points are
            // relayed byte-for-byte, and only a snapshot needs to be retained.
            if (raw.indexOf('"snapshot"') !== -1 || raw.indexOf('"commit"') !== -1 || raw.indexOf('"clearAll"') !== -1) {
                try {
                    const msg = JSON.parse(raw);
                    if (msg && msg.type === 'clearAll') {
                        // The stage wiped every passage. Forget every remembered list, and strip the ink
                        // from the retained snapshot so a screen joining right now gets none of the old marks.
                        lastStrokes.clear();
                        if (snapshot) { try { const s = JSON.parse(snapshot); s.strokes = []; snapshot = JSON.stringify(s); } catch (_) {} }
                        fanOut(raw);
                        return;
                    }
                    if (msg && msg.type === 'snapshot') {
                        msg.seq = ++seq;
                        remember(msg);
                        rememberBrowser(msg);
                        // Live subscribers already hold the strokes; only the RETAINED copy is completed.
                        fanOut(JSON.stringify(msg));
                        snapshot = JSON.stringify(complete(msg));
                        return;
                    }
                    if (msg && msg.type === 'commit') { remember(msg); fanOut(raw); return; }
                } catch (_) { /* fall through and relay as-is */ }
            }
            fanOut(raw);
        });

        ws.on('close', () => {
            if (ws === publisher) { publisher = null; publisherIsStage = false; log('publisher disconnected'); }
            else if (subscribers.delete(ws)) log(`follower disconnected (${subscribers.size} watching)`);
        });
        ws.on('error', () => { try { ws.close(); } catch (_) {} });
    });

    server.on('upgrade', (req, socket, head) => {
        let pathname = '';
        try { pathname = new URL(req.url, 'http://x').pathname; } catch (_) {}
        if (pathname !== PATH) return;   // leave every other upgrade alone
        wss.handleUpgrade(req, socket, head, (ws) => wss.emit('connection', ws, req));
    });

    /*
     * Heartbeat. A booth screen that loses Wi-Fi leaves a half-open socket that never fires 'close',
     * so without this the relay would keep "successfully" writing stroke points into a dead pipe and
     * report a follower that is not there.
     */
    const beat = setInterval(() => {
        for (const ws of [publisher, ...subscribers]) { if (ws) { try { ws.ping(); } catch (_) {} } }
    }, 15000);
    try { beat.unref(); } catch (_) {}

    return {
        // Called when the operator flips the Settings switch. Turning it off drops every screen at once,
        // so the program feed goes black immediately rather than at the next reconnect.
        applyEnabled() {
            if (isEnabled()) return;
            for (const ws of subscribers) { try { ws.close(4005, 'follower screen is turned off in Settings'); } catch (_) {} }
            subscribers.clear();
        },
        status() {
            return {
                enabled: isEnabled(),
                publisher: !!publisher,
                publisherIsStage,
                followers: subscribers.size,
                hasSnapshot: !!snapshot,
                seq,
                ...stats,
                uptimeMs: Date.now() - stats.startedAt
            };
        }
    };
}