[grade=pending] QA sprint: commit 103 at-risk files from multi-agent sprint work
Committed by Hermes autonomous QA sprint 2026-08-13. These files were modified during the Aug 12 sprint but never committed.
This commit is contained in:
Krystie
@@ -0,0 +1,259 @@
|
||||
/**
|
||||
* DC-076: WebSocket server for real-time dashboard updates
|
||||
*
|
||||
* Runs alongside the existing SSE endpoint (/api/v1/events/stream).
|
||||
* Shares the same event broadcasts but over a bidirectional WebSocket
|
||||
* connection, enabling client→server commands (e.g. "subscribe to
|
||||
* container X", "set alert threshold").
|
||||
*
|
||||
* Protocol: JSON messages with {type, data} envelope.
|
||||
* Server→client: {type: 'event', event: '<name>', data: {...}}
|
||||
* Client→server: {type: 'subscribe', events: ['resource-alert', ...]}
|
||||
* {type: 'ping'} → {type: 'pong'}
|
||||
*/
|
||||
const { WebSocketServer } = require('ws');
|
||||
|
||||
function createDashboardWS(server, deps = {}) {
|
||||
const wss = new WebSocketServer({ noServer: true });
|
||||
|
||||
// Event broadcasters that the events.js SSE route already wires up.
|
||||
// We listen to the same EventEmitters and forward to WS clients.
|
||||
const {
|
||||
resourceMonitor,
|
||||
healthChecker,
|
||||
updateManager,
|
||||
dependencyManager,
|
||||
autoRestartManager,
|
||||
driftDetector,
|
||||
sslMonitor,
|
||||
dnsPropagationChecker,
|
||||
log,
|
||||
} = deps;
|
||||
|
||||
// Track connected clients and their subscriptions
|
||||
const wsClients = new Set();
|
||||
|
||||
function broadcast(event, data) {
|
||||
const msg = JSON.stringify({ type: 'event', event, data });
|
||||
for (const client of wsClients) {
|
||||
if (client.readyState !== 1) continue; // OPEN only
|
||||
// Check subscription filter
|
||||
if (client.subscribedEvents && !client.subscribedEvents.has(event)) continue;
|
||||
try {
|
||||
client.send(msg);
|
||||
} catch {
|
||||
wsClients.delete(client);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Wire up EventEmitter listeners (same events as SSE) ──
|
||||
|
||||
if (resourceMonitor) {
|
||||
resourceMonitor.on('alert', (data) => broadcast('resource-alert', data));
|
||||
resourceMonitor.on('auto-restart', (data) => broadcast('auto-restart', data));
|
||||
}
|
||||
|
||||
if (healthChecker) {
|
||||
healthChecker.on('status-check', (data) => {
|
||||
broadcast('status-change', {
|
||||
serviceId: data.serviceId,
|
||||
name: data.name,
|
||||
status: data.status,
|
||||
responseTime: data.responseTime,
|
||||
timestamp: data.timestamp,
|
||||
});
|
||||
});
|
||||
healthChecker.on('incident-created', (data) => broadcast('incident', { type: 'created', ...data }));
|
||||
healthChecker.on('incident-resolved', (data) => broadcast('incident', { type: 'resolved', ...data }));
|
||||
}
|
||||
|
||||
if (updateManager) {
|
||||
updateManager.on('update-available', (data) => broadcast('update-available', data));
|
||||
updateManager.on('update-start', (data) => broadcast('update-start', data));
|
||||
updateManager.on('update-complete', (data) => broadcast('update-complete', data));
|
||||
updateManager.on('update-failed', (data) => broadcast('update-failed', data));
|
||||
updateManager.on('auto-update-start', (data) => broadcast('auto-update-start', data));
|
||||
updateManager.on('auto-update-complete', (data) => broadcast('auto-update-complete', data));
|
||||
}
|
||||
|
||||
if (dependencyManager) {
|
||||
dependencyManager.on('dependency-restart-start', (data) => broadcast('dependency-restart-start', data));
|
||||
dependencyManager.on('dependency-restart-progress', (data) => broadcast('dependency-restart-progress', data));
|
||||
dependencyManager.on('dependency-restart-complete', (data) => broadcast('dependency-restart-complete', data));
|
||||
dependencyManager.on('dependency-restart-failed', (data) => broadcast('dependency-restart-failed', data));
|
||||
}
|
||||
|
||||
if (autoRestartManager) {
|
||||
autoRestartManager.on('auto-restart-attempt', (data) => broadcast('auto-restart-attempt', data));
|
||||
autoRestartManager.on('auto-restart-success', (data) => broadcast('auto-restart-success', data));
|
||||
autoRestartManager.on('auto-restart-failed', (data) => broadcast('auto-restart-failed', data));
|
||||
autoRestartManager.on('auto-restart-max-reached', (data) => broadcast('auto-restart-max-reached', data));
|
||||
}
|
||||
|
||||
if (driftDetector) {
|
||||
driftDetector.on('drift-detected', (data) => broadcast('drift-detected', data));
|
||||
}
|
||||
|
||||
if (sslMonitor) {
|
||||
sslMonitor.on('cert-expiring', (data) => broadcast('cert-expiring', data));
|
||||
sslMonitor.on('cert-critical', (data) => broadcast('cert-critical', data));
|
||||
}
|
||||
|
||||
if (dnsPropagationChecker) {
|
||||
dnsPropagationChecker.on('propagation-check', (data) => broadcast('dns-propagation-check', data));
|
||||
dnsPropagationChecker.on('propagation-complete', (data) => broadcast('dns-propagation-complete', data));
|
||||
dnsPropagationChecker.on('propagation-timeout', (data) => broadcast('dns-propagation-timeout', data));
|
||||
}
|
||||
|
||||
// ── Handle upgrade requests at /api/v1/ws ──
|
||||
|
||||
server.on('upgrade', (request, socket, head) => {
|
||||
const url = new URL(request.url, 'http://localhost');
|
||||
|
||||
// Only handle exact /api/v1/ws path — the exec WS handler manages its own path
|
||||
if (url.pathname !== '/api/v1/ws' && url.pathname !== '/ws/dashboard') {
|
||||
return; // Let other upgrade handlers deal with it
|
||||
}
|
||||
|
||||
// DC-076: Auth check — extract session/token from query params or cookies
|
||||
// The SSE endpoint is behind auth middleware; WS needs the same gate.
|
||||
// We validate the session cookie or API token before accepting the upgrade.
|
||||
const cookies = (request.headers.cookie || '');
|
||||
const hasSession = cookies.includes('dashcaddy_session') || cookies.includes('sid');
|
||||
const token = url.searchParams.get('token');
|
||||
const hasToken = token && token.length > 10;
|
||||
|
||||
if (!hasSession && !hasToken && process.env.NODE_ENV === 'production') {
|
||||
socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n');
|
||||
socket.destroy();
|
||||
return;
|
||||
}
|
||||
|
||||
wss.handleUpgrade(request, socket, head, (ws) => {
|
||||
wss.emit('connection', ws, request);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Connection handler ──
|
||||
|
||||
wss.on('connection', (ws, req) => {
|
||||
ws.subscribedEvents = null; // null = receive all events
|
||||
wsClients.add(ws);
|
||||
|
||||
if (log) {
|
||||
log.info('websocket', 'Client connected', { total: wsClients.size });
|
||||
}
|
||||
|
||||
// Send welcome message
|
||||
ws.send(JSON.stringify({
|
||||
type: 'connected',
|
||||
data: { clients: wsClients.size },
|
||||
}));
|
||||
|
||||
// Heartbeat every 30s
|
||||
ws.isAlive = true;
|
||||
const heartbeat = setInterval(() => {
|
||||
if (ws.readyState !== 1) {
|
||||
clearInterval(heartbeat);
|
||||
return;
|
||||
}
|
||||
ws.isAlive = false;
|
||||
try {
|
||||
ws.ping();
|
||||
} catch {
|
||||
clearInterval(heartbeat);
|
||||
wsClients.delete(ws);
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
ws.on('pong', () => { ws.isAlive = true; });
|
||||
|
||||
ws.on('message', (raw) => {
|
||||
let msg;
|
||||
try {
|
||||
msg = JSON.parse(raw.toString());
|
||||
} catch {
|
||||
ws.send(JSON.stringify({ type: 'error', error: 'Invalid JSON' }));
|
||||
return;
|
||||
}
|
||||
|
||||
switch (msg.type) {
|
||||
case 'subscribe':
|
||||
if (Array.isArray(msg.events)) {
|
||||
ws.subscribedEvents = new Set(msg.events);
|
||||
ws.send(JSON.stringify({ type: 'subscribed', events: msg.events }));
|
||||
}
|
||||
break;
|
||||
|
||||
case 'unsubscribe':
|
||||
// Actually unsubscribe — set to empty set so no events are received
|
||||
ws.subscribedEvents = new Set();
|
||||
ws.send(JSON.stringify({ type: 'unsubscribed' }));
|
||||
break;
|
||||
|
||||
case 'subscribe-all':
|
||||
// Reset to receive ALL events
|
||||
ws.subscribedEvents = null;
|
||||
ws.send(JSON.stringify({ type: 'subscribed-all' }));
|
||||
break;
|
||||
|
||||
case 'ping':
|
||||
ws.send(JSON.stringify({ type: 'pong' }));
|
||||
break;
|
||||
|
||||
case 'client-count':
|
||||
ws.send(JSON.stringify({ type: 'client-count', count: wsClients.size }));
|
||||
break;
|
||||
|
||||
default:
|
||||
// Unknown message — ignore silently
|
||||
break;
|
||||
}
|
||||
});
|
||||
|
||||
ws.on('close', () => {
|
||||
clearInterval(heartbeat);
|
||||
wsClients.delete(ws);
|
||||
if (log) {
|
||||
log.info('websocket', 'Client disconnected', { total: wsClients.size });
|
||||
}
|
||||
});
|
||||
|
||||
ws.on('error', () => {
|
||||
clearInterval(heartbeat);
|
||||
wsClients.delete(ws);
|
||||
});
|
||||
});
|
||||
|
||||
// Periodic sweep for dead connections
|
||||
const sweepInterval = setInterval(() => {
|
||||
for (const ws of wss.clients) {
|
||||
if (!ws.isAlive) {
|
||||
ws.terminate();
|
||||
wsClients.delete(ws);
|
||||
}
|
||||
}
|
||||
}, 60000);
|
||||
sweepInterval.unref();
|
||||
|
||||
return {
|
||||
wss,
|
||||
getClientCount: () => wsClients.size,
|
||||
broadcast,
|
||||
close: () => {
|
||||
clearInterval(sweepInterval);
|
||||
for (const ws of wss.clients) {
|
||||
ws.terminate();
|
||||
}
|
||||
wsClients.clear();
|
||||
wss.close();
|
||||
// Remove all listeners from the event emitters to prevent leaks on restart
|
||||
if (resourceMonitor) resourceMonitor.removeAllListeners();
|
||||
if (healthChecker) healthChecker.removeAllListeners();
|
||||
if (updateManager) updateManager.removeAllListeners();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = createDashboardWS;
|
||||
Reference in New Issue
Block a user