pulse-signage/src/web/bootstrap.js

// Web-to-player websocket bridge and dashboard state synchronizer.

const { WebSocketServer, WebSocket } = require('ws');
const { createDashboardStateService } = require('./lib/dashboard-state');
const { createUploadSyncService } = require('./lib/media');
const { createRequestAuthHeaders } = require('#src/request-auth');

function createWebBootstrap(options) {
  const pool = options && options.pool;
  const common = options && options.common;
  const configuredPlayerInternalBaseUrl = String(options && options.playerInternalBaseUrl || '').replace(/\/$/, '');
  const configuredThinClientBaseUrl = String(options && options.thinClientBaseUrl || process.env.THIN_CLIENT_BASE_URL || '').trim().replace(/\/$/, '');
  const uploadDir = String(options && options.uploadDir || '').trim();
  const dashboardRefreshIntervalMs = 5000;
  const formatDashboardDate = options && options.formatDashboardDate;
  const notifyPlayerScreens = options && options.notifyPlayerScreens;
  const backgroundTaskQueue = options && options.backgroundTaskQueue;

  if (!pool || !common || !uploadDir || typeof formatDashboardDate !== 'function' || typeof notifyPlayerScreens !== 'function') {
    throw new Error('createWebBootstrap requires the web bootstrap dependencies.');
  }

  const dashboardWs = new WebSocketServer({ noServer: true });
  const dashboardClients = new Set();
  const playerSnapshotCache = options && options.playerSnapshotCache ? options.playerSnapshotCache : new Map();
  const playerSnapshotSockets = options && options.playerSnapshotSockets ? options.playerSnapshotSockets : new Map();
  let dashboardRefreshInFlight = null;
  let broadcastDashboardState = null;
  function getPlayerSnapshotSocketUrl(slug) {
    const resolvedPlayerInternalBaseUrl = configuredThinClientBaseUrl || configuredPlayerInternalBaseUrl;
    if (!resolvedPlayerInternalBaseUrl) {
      throw new Error('Unable to resolve the player internal base URL.');
    }

    const url = new URL(resolvedPlayerInternalBaseUrl.replace(/^http/, 'ws'));
    url.pathname = `/ws/screens/${encodeURIComponent(slug)}/events`;
    url.search = '';
    return url.toString();
  }

  function clearPlayerSnapshotSocket(slug) {
    const key = String(slug || '').trim();
    playerSnapshotSockets.delete(key);
  }

  function ensurePlayerSnapshotSubscription(slug) {
    if (options && typeof options.ensurePlayerSnapshotSubscription === 'function' && options.ensurePlayerSnapshotSubscription !== ensurePlayerSnapshotSubscription) {
      return options.ensurePlayerSnapshotSubscription(slug);
    }

    const key = String(slug || '').trim();
    if (!key || playerSnapshotSockets.has(key)) {
      return;
    }

    playerSnapshotSockets.set(key, null);
    const authHeaders = createRequestAuthHeaders({
      method: 'GET',
      pathname: `/ws/screens/${encodeURIComponent(key)}/events`
    });

    Promise.resolve(getPlayerSnapshotSocketUrl(key)).then(function (socketUrl) {
      const socket = new WebSocket(socketUrl, {
        headers: authHeaders
      });
      playerSnapshotSockets.set(key, socket);

      socket.onmessage = function (event) {
        try {
          const payload = JSON.parse(String(event.data || '{}'));
          if (!payload || payload.type !== 'snapshot' || payload.slug !== key) {
            return;
          }
          playerSnapshotCache.set(key, {
            slug: key,
            count: Array.isArray(payload.connections) ? payload.connections.length : 0,
            connections: Array.isArray(payload.connections) ? payload.connections : []
          });
          if (broadcastDashboardState) {
            broadcastDashboardState().catch(function (error) {
              console.error(error);
            });
          }
        } catch (_error) {
          // Ignore malformed player snapshot payloads.
        }
      };

      socket.onclose = function () {
        clearPlayerSnapshotSocket(key);
        setTimeout(function () {
          ensurePlayerSnapshotSubscription(key);
        }, 2000);
      };

      socket.onerror = function () {
        try {
          socket.close();
        } catch (_error) {
          // ignore close errors
        }
      };
    }).catch(function (error) {
      clearPlayerSnapshotSocket(key);
      console.error(error);
    });
  }

  const dashboardStateService = createDashboardStateService({
    pool: pool,
    common: common,
    thinClientBaseUrl: configuredThinClientBaseUrl,
    playerSnapshotCache: playerSnapshotCache,
    playerSnapshotSockets: playerSnapshotSockets,
    ensurePlayerSnapshotSubscription: ensurePlayerSnapshotSubscription,
    formatDashboardDate: formatDashboardDate
  });
  const buildDashboardState = dashboardStateService.buildDashboardState;

  const uploadSyncService = createUploadSyncService({
    pool: pool,
    common: common,
    playerInternalBaseUrl: configuredPlayerInternalBaseUrl,
    playerSnapshotCache: playerSnapshotCache,
    notifyPlayerScreens: notifyPlayerScreens,
    backgroundTaskQueue: backgroundTaskQueue
  });

  const upload = uploadSyncService.createUploadMiddleware(uploadDir);
  const collectUploadReferencesFromSlide = uploadSyncService.collectUploadReferencesFromSlide;
  const collectUploadReferencesFromTemplate = uploadSyncService.collectUploadReferencesFromTemplate;
  const collectUploadReferencesFromPayload = uploadSyncService.collectUploadReferencesFromPayload;
  const removeUnusedUploadFiles = uploadSyncService.removeUnusedUploadFiles;
  const collectUploadPathsFromDirectory = uploadSyncService.collectUploadPathsFromDirectory;
  const syncPlaylistUploadsOnChange = uploadSyncService.syncPlaylistUploadsOnChange;
  const runMediaSyncTask = uploadSyncService.runMediaSyncTask;
  let lastDashboardState = null;

  function getFallbackDashboardState() {
    return lastDashboardState || {
      playlists: [],
      screens: [],
      clients: [],
      kioskPlayers: [],
      slides: [],
      playerServiceConnected: false,
      connectedPlayersCount: 0,
      connectedClientsCount: 0
    };
  }

  async function resolveDashboardState() {
    try {
      const state = await buildDashboardState();
      lastDashboardState = state;
      return state;
    } catch (error) {
      if (lastDashboardState) {
        console.error(error);
        return lastDashboardState;
      }

      throw error;
    }
  }

  async function sendDashboardStateToSocket(socket) {
    if (!socket || socket.readyState !== WebSocket.OPEN) {
      return;
    }
    const state = await resolveDashboardState().catch(function (error) {
      console.error(error);
      return getFallbackDashboardState();
    });
    socket.send(JSON.stringify({ type: 'dashboard-state', state: state }));
  }

  broadcastDashboardState = async function () {
    if (dashboardRefreshInFlight) {
      return dashboardRefreshInFlight;
    }

    dashboardRefreshInFlight = (async function () {
      const state = await resolveDashboardState().catch(function (error) {
        console.error(error);
        return getFallbackDashboardState();
      });
      const payload = JSON.stringify({ type: 'dashboard-state', state: state });
      for (const socket of dashboardClients) {
        if (socket && socket.readyState === WebSocket.OPEN) {
          socket.send(payload);
        }
      }
      return state;
    })().finally(function () {
      dashboardRefreshInFlight = null;
    });

    return dashboardRefreshInFlight;
  };

  function installDashboardWebsocket(server, loadCurrentUser) {
    server.on('upgrade', async function (request, socket, head) {
      let pathname = '';
      try {
        pathname = new URL(request.url, 'http://localhost').pathname;
      } catch (_error) {
        socket.destroy();
        return;
      }

      if (pathname !== '/ws/dashboard') {
        socket.destroy();
        return;
      }

      try {
        const currentUser = await loadCurrentUser(pool, request);
        if (!currentUser) {
          socket.destroy();
          return;
        }
      } catch (_error) {
        socket.destroy();
        return;
      }

      dashboardWs.handleUpgrade(request, socket, head, function (ws) {
        dashboardWs.emit('connection', ws, request);
      });
    });

    dashboardWs.on('connection', function (socket) {
      dashboardClients.add(socket);
      sendDashboardStateToSocket(socket);

      socket.on('close', function () {
        dashboardClients.delete(socket);
      });

      socket.on('error', function () {
        dashboardClients.delete(socket);
      });
    });

    setInterval(function () {
      broadcastDashboardState().catch(function (error) {
        console.error(error);
      });
    }, dashboardRefreshIntervalMs);
  }

  return {
    upload: upload,
    uploadSyncService: uploadSyncService,
    playerInternalBaseUrl: configuredPlayerInternalBaseUrl || null,
    buildDashboardState: buildDashboardState,
    collectUploadReferencesFromSlide: collectUploadReferencesFromSlide,
    collectUploadReferencesFromTemplate: collectUploadReferencesFromTemplate,
    collectUploadReferencesFromPayload: collectUploadReferencesFromPayload,
    removeUnusedUploadFiles: removeUnusedUploadFiles,
    collectUploadPathsFromDirectory: collectUploadPathsFromDirectory,
    syncPlaylistUploadsOnChange: syncPlaylistUploadsOnChange,
    runMediaSyncTask: runMediaSyncTask,
    broadcastDashboardState: broadcastDashboardState,
    installDashboardWebsocket: installDashboardWebsocket
  };
}

module.exports = { createWebBootstrap };