From c5ca8d0efa883bd680c339e9b466918358ce66fb Mon Sep 17 00:00:00 2001 From: Brad Anderson Date: Tue, 13 Oct 2026 21:08:53 -0700 Subject: [PATCH 1/7] 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. --- .env.example | 1 - backend/src/config.ts | 2 + backend/src/routes/streamProxy.test.ts | 87 ++++++++++++++++++++++++++ backend/src/routes/streamProxy.ts | 40 ++++++++++++ backend/src/server.ts | 10 +++ docker-compose.yml | 7 +-- 6 files changed, 142 insertions(+), 5 deletions(-) create mode 100644 backend/src/routes/streamProxy.test.ts create mode 100644 backend/src/routes/streamProxy.ts diff --git a/.env.example b/.env.example index 154a306..05dcc5e 100644 --- a/.env.example +++ b/.env.example @@ -107,7 +107,6 @@ ENABLE_CACHE_PREWARM=true # and enable the profile. See docs/10-features/13-streaming.md. # COMPOSE_PROFILES=streaming # STREAMING_API_TOKEN= -# STREAM_PORT=8081 # MAX_SESSIONS=4 # Sign in users an authenticating reverse proxy (e.g. Authentik forward auth) # names in a header. Off unless TRUSTED_HEADER_USERNAME and TRUSTED_PROXIES are diff --git a/backend/src/config.ts b/backend/src/config.ts index 65726c2..3f32282 100644 --- a/backend/src/config.ts +++ b/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 diff --git a/backend/src/routes/streamProxy.test.ts b/backend/src/routes/streamProxy.test.ts new file mode 100644 index 0000000..02fb125 --- /dev/null +++ b/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 { + 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((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); + }); +}); diff --git a/backend/src/routes/streamProxy.ts b/backend/src/routes/streamProxy.ts new file mode 100644 index 0000000..39c391f --- /dev/null +++ b/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>[0], socket, head); + }); + }; +} diff --git a/backend/src/server.ts b/backend/src/server.ts index 4a0bfa4..2c6217d 100644 --- a/backend/src/server.ts +++ b/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