feat(streaming): proxy /stream/ through the app

The app forwards /stream/ to the session manager, WebSocket upgrades
and the save-sync event stream included, ahead of helmet and
compression. A reverse proxy in front now needs no route of its own
for streams, and the session manager publishes no ports.

backend/src/config.ts

@@ -173,6 +173,8 @@ export const config = {
   // Java applets (flashpoint-ultimate-server). Streaming is off unless both are set.
   streamingApiUrl: process.env.STREAMING_API_URL || null,
   streamingApiToken: process.env.STREAMING_API_TOKEN || null,
+  // Where /stream/ is proxied to: the session manager's stream port
+  streamingStreamUrl: process.env.STREAMING_STREAM_URL || null,
 
   // Sign users in from a header set by an authenticating reverse proxy, such as
   // Authentik forward auth. Off unless the header name and the proxy addresses

backend/src/routes/streamProxy.test.ts

@@ -0,0 +1,87 @@
+import { describe, it, expect, beforeAll, afterAll, vi } from 'vitest';
+import http, { type Server } from 'http';
+import type { AddressInfo } from 'net';
+import express from 'express';
+import { mountStreamProxy } from './streamProxy';
+
+vi.mock('../utils/logger', () => ({
+  logger: { info: vi.fn(), error: vi.fn(), debug: vi.fn(), warn: vi.fn() },
+}));
+
+let upstream: Server;
+let app: Server;
+let base: string;
+
+function listen(server: Server): Promise<number> {
+  return new Promise((resolve) =>
+    server.listen(0, '127.0.0.1', () => resolve((server.address() as AddressInfo).port))
+  );
+}
+
+beforeAll(async () => {
+  // A stand-in session manager: echoes the path, and accepts WebSocket upgrades
+  upstream = http.createServer((req, res) => {
+    res.setHeader('Content-Type', 'text/plain');
+    res.end(`upstream saw ${req.method} ${req.url}`);
+  });
+  upstream.on('upgrade', (req, socket) => {
+    socket.write(
+      'HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n'
+    );
+    socket.write(`ws ${req.url}`);
+    socket.end();
+  });
+  const upstreamPort = await listen(upstream);
+
+  const expressApp = express();
+  const attach = mountStreamProxy(expressApp, `http://127.0.0.1:${upstreamPort}`);
+  expressApp.get('/other', (_req, res) => {
+    res.send('app');
+  });
+  app = http.createServer(expressApp);
+  attach(app);
+  base = `http://127.0.0.1:${await listen(app)}`;
+});
+
+afterAll(() => {
+  app.close();
+  upstream.close();
+});
+
+describe('stream proxy', () => {
+  it('forwards /stream/ requests with their full path', async () => {
+    const res = await fetch(`${base}/stream/abc/_fp/start`, { method: 'POST' });
+    expect(await res.text()).toBe('upstream saw POST /stream/abc/_fp/start');
+  });
+
+  it('leaves other routes to the app', async () => {
+    expect(await (await fetch(`${base}/other`)).text()).toBe('app');
+  });
+
+  it('forwards WebSocket upgrades', async () => {
+    const received = await new Promise<string>((resolve, reject) => {
+      const req = http.request(`${base}/stream/abc/websocket`, {
+        headers: { Connection: 'Upgrade', Upgrade: 'websocket' },
+      });
+      req.on('upgrade', (res, socket, head) => {
+        expect(res.statusCode).toBe(101);
+        let data = head.toString();
+        socket.on('data', (chunk) => (data += chunk.toString()));
+        socket.on('end', () => resolve(data));
+      });
+      req.on('error', reject);
+      req.end();
+    });
+    expect(received).toBe('ws /stream/abc/websocket');
+  });
+
+  it('answers 502 when the session manager is down', async () => {
+    const down = express();
+    mountStreamProxy(down, 'http://127.0.0.1:9');
+    const server = http.createServer(down);
+    const port = await listen(server);
+    const res = await fetch(`http://127.0.0.1:${port}/stream/abc/`);
+    server.close();
+    expect(res.status).toBe(502);
+  });
+});

backend/src/routes/streamProxy.ts

@@ -0,0 +1,40 @@
+import { ServerResponse, type IncomingMessage, type Server } from 'http';
+import type { Socket } from 'net';
+import type { Express } from 'express';
+import { createProxyMiddleware } from 'http-proxy-middleware';
+import { logger } from '../utils/logger';
+
+/**
+ * Proxies /stream/ to the session manager, WebSockets included, so a reverse
+ * proxy in front of the app needs no route of its own for streams. Returns a
+ * function that wires WebSocket upgrades into the HTTP server once it exists.
+ */
+export function mountStreamProxy(app: Express, target: string): (server: Server) => void {
+  const proxy = createProxyMiddleware('/stream/', {
+    target,
+    ws: true,
+    logLevel: 'silent',
+    // Save sync is a long-lived event stream; don't let the server's idle timeout cut it
+    onProxyReq: (_proxyReq, req) => {
+      req.socket.setTimeout(0);
+    },
+    onError: (error, req, res) => {
+      logger.warn('[Streaming] Stream proxy error', { url: req.url, error: error.message });
+      // A failed WebSocket upgrade hands over the raw socket instead of a response
+      if (!(res instanceof ServerResponse)) {
+        (res as Socket).destroy();
+      } else if (!res.headersSent) {
+        res.writeHead(502).end();
+      }
+    },
+  });
+  app.use(proxy);
+
+  return (server) => {
+    server.on('upgrade', (req: IncomingMessage, socket: Socket, head: Buffer) => {
+      if (!req.url?.startsWith('/stream/')) return;
+      socket.setTimeout(0);
+      proxy.upgrade?.(req as Parameters<NonNullable<typeof proxy.upgrade>>[0], socket, head);
+    });
+  };
+}

backend/src/server.ts

@@ -44,6 +44,8 @@ import { zipManager } from './game/zip-manager';
 import { gameZipServer } from './game/gamezipserver';
 import gameProxyRouter from './routes/game-proxy';
 import gameZipRouter from './routes/game-zip';
+import { mountStreamProxy } from './routes/streamProxy';
+import { StreamingService } from './services/StreamingService';
 
 /**
  * Compute SHA-256 CSP hashes for inline <script> blocks (those without a src
@@ -112,6 +114,13 @@ async function startServer() {
   app.use('/game-proxy', gameCors, startupGate, gameProxyRouter);
   app.use('/game-zip', gameCors, startupGate, gameZipRouter);
 
+  // Streams are proxied before helmet and compression, which would break the
+  // streamer's page and its event stream
+  const attachStreamProxy =
+    StreamingService.isEnabled() && config.streamingStreamUrl
+      ? mountStreamProxy(app, config.streamingStreamUrl)
+      : null;
+
   // When the backend also serves the SPA, allow the app's own resources by
   // hash/keyword rather than the blanket 'unsafe-inline'/'unsafe-eval':
   //   - scriptSrc: 'self' for hashed bundles, sha256 hashes for the inline theme
@@ -230,6 +239,7 @@ async function startServer() {
     logger.info(`🎮 Game content: /game-proxy/* and /game-zip/* (integrated)`);
     logger.info(`🌍 Environment: ${config.nodeEnv}`);
   });
+  attachStreamProxy?.(server);
 
   // Prevent connection exhaustion - critical for handling game file requests
   server.keepAliveTimeout = 65000; // 65s - slightly above common load balancer timeout