Real-Time Log Streaming — Streaming Logs for Live Debugging

// WebSocket log stream
const WebSocket = require('ws');

class LogStreamServer {
  constructor(server) {
    this.wss = new WebSocket.Server({ server, path: '/logs/stream' });
    this.clients = new Set();
    this.filters = new Map(); // client -> filter rules

    this.wss.on('connection', (ws, req) => {
      const clientId = uuidv4();
      this.clients.add(ws);
      this.filters.set(clientId, {
        levels: ['ERROR', 'WARN', 'INFO'],
        services: [],
        correlationId: null
      });

      ws.on('message', (data) => {
        try {
          const msg = JSON.parse(data);
          if (msg.type === 'setFilter') {
            this.filters.set(clientId, { ...this.filters.get(clientId), ...msg.filter });
          }
        } catch {}
      });

      ws.on('close', () => {
        this.clients.delete(ws);
        this.filters.delete(clientId);
      });
    });
  }

  broadcast(logEntry) {
    for (const ws of this.clients) {
      if (ws.readyState === WebSocket.OPEN) {
        const clientFilters = this.filters.get(ws.clientId);
        if (this.matchesFilter(logEntry, clientFilters)) {
          ws.send(JSON.stringify(logEntry));
        }
      }
    }
  }

  matchesFilter(entry, filter) {
    if (!filter) return true;
    if (filter.levels && !filter.levels.includes(entry.level)) return false;
    if (filter.services?.length && !filter.services.includes(entry.service?.name)) return false;
    if (filter.correlationId && entry.correlationId !== filter.correlationId) return false;
    if (filter.search && !JSON.stringify(entry).toLowerCase().includes(filter.search.toLowerCase())) return false;
    return true;
  }
}

// Server-Sent Events stream
app.get('/logs/sse', (req, res) => {
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no'
  });

  const filter = { levels: req.query.levels?.split(',') || ['ERROR'] };

  const listener = (entry) => {
    if (filter.levels.includes(entry.level)) {
      res.write(`data: ${JSON.stringify(entry)}\n\n`);
    }
  };

  logEmitter.on('log', listener);
  req.on('close', () => logEmitter.off('log', listener));
});