Statum/js/statum-worker.js

/*!
 * Statum realtime channel worker.
 *
 * Shared across every tab of an origin via `SharedWorker`. It owns the single
 * realtime connection to `GET /_statum/channel`, reads the channel id from the
 * stream's first `event: channel`, reference-counts slot subscriptions across
 * all connected tabs, manages them with `PATCH /_statum/channel/{id}`, and
 * relays incoming `set`/`clear` events to every tab through the cross-tab
 * `BroadcastChannel` named "statum" (which the main library already listens
 * on). The worker never touches Web Storage or the DOM; tabs apply events and
 * persist frames themselves.
 *
 * Distribution: a standalone script (served at `/_statum/worker.js` by
 * default, overridable with `stm-worker`). It has no dependencies.
 */
(function () {
  'use strict';

  var CHANNEL_NAME = 'statum';
  var channelUrl = '/_statum/channel'; // default; overridden by a tab's `init` message

  var bc = null;       // BroadcastChannel to the tabs
  var es = null;       // EventSource to /_statum/channel
  var channelId = null;
  var refs = new Map();      // slotKey -> subscription count (across all ports)
  var portKeys = new Map();  // MessagePort -> Set(slotKey)

  var pending = { subscribe: [], unsubscribe: [] };
  var patchScheduled = false;

  function ensureBroadcast() {
    if (bc || typeof BroadcastChannel === 'undefined') return;
    try { bc = new BroadcastChannel(CHANNEL_NAME); } catch (e) { bc = null; }
  }

  function postToTabs(message) {
    ensureBroadcast();
    if (bc) { try { bc.postMessage(message); } catch (e) {} }
  }

  function openStream() {
    if (es) return;
    try {
      es = new EventSource(channelUrl, { withCredentials: true });
    } catch (e) { es = null; return; }

    es.addEventListener('channel', function (ev) {
      channelId = ev.data;
      flushPatch();
    });
    es.addEventListener('set', function (ev) {
      var frame;
      try { frame = JSON.parse(ev.data); } catch (e) { return; }
      postToTabs({ kind: 'set', frame: frame });
    });
    es.addEventListener('clear', function (ev) {
      postToTabs({ kind: 'clear', key: ev.data });
    });
    es.onerror = function () { /* EventSource reconnects automatically */ };
  }

  function closeStream() {
    if (es) { try { es.close(); } catch (e) {} es = null; }
    channelId = null;
  }

  function schedulePatch() {
    if (patchScheduled) return;
    patchScheduled = true;
    Promise.resolve().then(function () { patchScheduled = false; flushPatch(); });
  }

  function flushPatch() {
    if (!channelId) return;
    var sub = pending.subscribe, uns = pending.unsubscribe;
    if (!sub.length && !uns.length) return;
    pending = { subscribe: [], unsubscribe: [] };
    var body = {};
    if (sub.length) body.subscribe = sub;
    if (uns.length) body.unsubscribe = uns;
    try {
      fetch(channelUrl + '/' + encodeURIComponent(channelId), {
        method: 'PATCH',
        headers: { 'Content-Type': 'application/json' },
        credentials: 'same-origin',
        body: JSON.stringify(body)
      }).catch(function () {});
    } catch (e) {}
  }

  function addSub(port, key) {
    var keys = portKeys.get(port);
    if (!keys) { keys = new Set(); portKeys.set(port, keys); }
    if (keys.has(key)) return; // this port already subscribed
    keys.add(key);
    var c = refs.get(key) || 0;
    refs.set(key, c + 1);
    if (c === 0) {
      openStream();
      pending.subscribe.push(key);
      schedulePatch();
    }
  }

  function removeSub(port, key) {
    var keys = portKeys.get(port);
    if (!keys || !keys.has(key)) return;
    keys.delete(key);
    var c = refs.get(key) || 0;
    if (c <= 1) {
      refs.delete(key);
      pending.unsubscribe.push(key);
      schedulePatch();
      if (refs.size === 0) setTimeout(function () { if (refs.size === 0) closeStream(); }, 0);
    } else {
      refs.set(key, c - 1);
    }
  }

  function disconnectPort(port) {
    var keys = portKeys.get(port);
    if (keys) keys.forEach(function (key) { removeSub(port, key); });
    portKeys.delete(port);
  }

  function handlePortMessage(port, ev) {
    var msg = ev.data;
    if (!msg) return;
    if (msg.url) channelUrl = msg.url;        // {op:'init', url}
    if (msg.op === 'subscribe' && msg.key != null) addSub(port, msg.key);
    else if (msg.op === 'unsubscribe' && msg.key != null) removeSub(port, msg.key);
    else if (msg.op === 'disconnect') disconnectPort(port);
  }

  // SharedWorker entry point: each connecting tab gets its own MessagePort.
  self.onconnect = function (e) {
    var port = e.ports[0];
    portKeys.set(port, new Set());
    port.onmessage = function (ev) { handlePortMessage(port, ev); };
    port.start();
    // Port closure (e.g. tab crash) is not reliably signalled, so refcounts
    // for an abruptly-closed tab are reclaimed when the worker itself is torn
    // down; a graceful close sends an explicit `disconnect` message first.
  };
})();