Files
sencho/backend/src/websocket/logs.ts
T
Anso dc3699189d refactor(backend): extract remote proxy, WebSocket upgrade handler, and server factory (phase 3) (#733)
Phase 3 of the index.ts refactor. Pulls the remote HTTP/WS proxy plumbing,
the WebSocket upgrade dispatcher, and the http/WSS construction out of the
monolith. index.ts drops roughly 620 lines.

New modules:
- proxy/websocketProxy.ts: shared httpProxy.createProxyServer singleton
  (used by both the HTTP proxy middleware and the remote WS forwarder)
- proxy/remoteNodeProxy.ts: createRemoteProxyMiddleware() factory; consumes
  the isProxyExemptPath helper instead of open-coding the prefix list
- server.ts: createServer(app) returns { server, wss, pilotTunnelWss }
- services/FleetUpdateTrackerService.ts: singleton wrapping the in-flight
  fleet update tracker Map with create()/resolve() helpers
- helpers/consoleSession.ts: mintConsoleSession(), isConsoleSessionScope()
- websocket/upgradeHandler.ts: attachUpgrade(server, deps) dispatcher that
  runs the manual cookie/JWT verify and delegates to sub-handlers
- websocket/pilotTunnel.ts: handlePilotTunnel (pilot_enroll consumption and
  pilot_tunnel registration)
- websocket/notifications.ts: /ws/notifications local subscriber
- websocket/remoteForwarder.ts: remote-node WS proxy with console_session
  token exchange for interactive paths
- websocket/logs.ts: /api/stacks/:name/logs supervisor stream
- websocket/hostConsole.ts: /api/system/host-console PTY, Admiral-gated
- websocket/generic.ts: /ws exec + streamStats action dispatch, owns the
  terminalWs single-instance reference
- websocket/reject.ts: shared rejectUpgrade helper (replaces five copies)

Service extension:
- NotificationService: setBroadcaster(fn) replaced by subscribe(ws) that
  returns an unsubscriber; broadcastToSubscribers is now internal. Subscriber
  set lives on the service rather than in index.ts.

Wiring in index.ts:
- const app = createApp() already in place from Phase 2
- const { server, wss, pilotTunnelWss } = createServer(app)
- attachUpgrade(server, { wss, pilotTunnelWss })
- app.use('/api/', createRemoteProxyMiddleware())
- /api/system/console-token route now uses mintConsoleSession()
- deploy/down/update routes read the streaming target via getTerminalWs()
  (return type is WebSocket | undefined so the || undefined fallback is gone)

Code review fixes: five duplicated reject helpers collapsed into
websocket/reject.ts; dropped the createTracker/resolveTracker bind
aliases in index.ts so call sites go through the service directly;
removed em dashes; replaced req.url! with req.url || '/'.
2026-04-23 19:31:16 -04:00

45 lines
1.7 KiB
TypeScript

import type { IncomingMessage } from 'http';
import type { Duplex } from 'stream';
import WebSocket, { WebSocketServer } from 'ws';
import { ComposeService } from '../services/ComposeService';
import { isValidStackName } from '../utils/validation';
import { isDebugEnabled } from '../utils/debug';
import { getErrorMessage } from '../utils/errors';
/**
* Handle `/api/stacks/:stackName/logs` WebSocket upgrades. Streams the
* supervisor log loop for the given stack on the caller's node context.
*/
export function handleLogsWs(
req: IncomingMessage,
socket: Duplex,
head: Buffer,
opts: { nodeId: number; stackName: string },
): void {
const { nodeId, stackName } = opts;
const logsWss = new WebSocketServer({ noServer: true });
logsWss.handleUpgrade(req, socket, head, (ws) => {
// Close the per-connection server immediately after the upgrade completes.
// The wss instance is only needed to negotiate the handshake; keeping it
// open accumulates listeners and allocates memory for every connection.
logsWss.close();
if (!isValidStackName(stackName)) {
ws.send('Error: Invalid stack name\r\n');
ws.close();
return;
}
try {
if (isDebugEnabled()) console.debug('[Stacks:debug] WS log stream opened', { stackName, nodeId });
ws.on('close', () => {
if (isDebugEnabled()) console.debug('[Stacks:debug] WS log stream closed', { stackName, nodeId });
});
ComposeService.getInstance(nodeId).streamLogs(stackName, ws);
} catch (error) {
console.error('[Stacks] Failed to stream logs:', error);
if (ws.readyState === WebSocket.OPEN) {
ws.send(`Error streaming logs: ${getErrorMessage(error, 'unknown')}\n`);
}
}
});
}