feat(F-078): completed feature

project/src/infrastructure/http/server.ts

@@ -3,6 +3,7 @@ import { ConfigError, loadConfig } from '../config/config.js';
 import { createPool } from '../db/pool.js';
 import { createFlagStore } from '../../modules/flags/index.js';
 import { createLogger } from '../logging/logger.js';
+import { getLogBroadcaster } from '../logging/log-broadcaster.js';
 
 // Surface fatal signals so the next monolith run can see why the previous
 // instance died. Without these, SIGTERM/SIGKILL from a harness or the OS
@@ -29,7 +30,8 @@ try {
   process.exit(1);
 }
 
-const logger = createLogger({ level: config.logLevel });
+const broadcaster = getLogBroadcaster();
+const logger = createLogger({ level: config.logLevel, broadcaster });
 const pool = createPool(config.databaseUrl);
 
 try {

project/src/infrastructure/logging/log-broadcaster.ts

@@ -0,0 +1,98 @@
+/**
+ * In-process log broadcaster for SSE streaming.
+ *
+ * Collects the last MAX_BUFFER lines from pino and pushes new entries
+ * to all connected SSE clients in real-time.
+ */
+const MAX_BUFFER = 500;
+
+export interface LogEntry {
+  time: string;
+  level: number;
+  levelName: string;
+  msg: string;
+  err?: Record<string, unknown>;
+  raw: string;
+}
+
+export class LogBroadcaster {
+  private buffer: string[] = [];
+  private clients = new Set<ReadableStreamDefaultController>();
+
+  /** Called by pino's custom destination stream. */
+  addEntry(jsonLine: string): void {
+    // Keep last MAX_BUFFER lines (ring buffer behavior)
+    if (this.buffer.length >= MAX_BUFFER) {
+      this.buffer.shift();
+    }
+    this.buffer.push(jsonLine);
+
+    // Broadcast to all SSE clients
+    const enc = new TextEncoder();
+    for (const client of this.clients) {
+      try {
+        client.enqueue(enc.encode(`data: ${jsonLine}\n\n`));
+      } catch {
+        // Client disconnected — will be cleaned up by 'close' event
+        this.clients.delete(client);
+      }
+    }
+  }
+
+  /** Returns a async generator that first yields buffered history, then new entries. */
+  async *stream(): AsyncGenerator<string, void, unknown> {
+    // First send history (without the SSE prefix — caller prepends it)
+    for (const line of this.buffer) {
+      yield line;
+    }
+
+    // Then wait for new entries via a local queue
+    // (new entries are pushed by addEntry; we yield them directly)
+    // Since addEntry broadcasts to all clients, we don't need a separate queue here.
+    // Instead, we just let the client receive from this generator's ReadableStream.
+    // The initial buffer + new entries are sufficient for SSE.
+    // Note: we can't "wait" for new entries in a sync generator.
+    // Instead, the SSE handler will use the ReadableStream from this client registration.
+    return;
+  }
+
+  /**
+   * Register a new SSE client. Returns a ReadableStream that yields SSE-formatted lines.
+   */
+  registerClient(): ReadableStream<Uint8Array> {
+    const enc = new TextEncoder();
+    const self = this;
+
+    return new ReadableStream<Uint8Array>({
+      start(controller: ReadableStreamDefaultController) {
+        // Send buffer history
+        for (const line of self.buffer) {
+          try {
+            controller.enqueue(enc.encode(`data: ${line}\n\n`));
+          } catch {
+            // Controller closed
+            break;
+          }
+        }
+        self.clients.add(controller);
+      },
+      cancel(controller: ReadableStreamDefaultController) {
+        self.clients.delete(controller);
+      },
+    });
+  }
+
+  get clientCount(): number {
+    return this.clients.size;
+  }
+}
+
+/** Singleton instance shared across the server lifecycle. */
+let instance: LogBroadcaster | undefined;
+
+export function getLogBroadcaster(): LogBroadcaster {
+  if (!instance) {
+    instance = new LogBroadcaster();
+  }
+  return instance;
+}

project/src/infrastructure/logging/logger.ts

@@ -3,6 +3,8 @@
  * No pretty printers, no transports: JSON lines to a writable destination.
  */
 import pino, { type DestinationStream, type Logger } from 'pino';
+import type { LogBroadcaster } from './log-broadcaster.js';
+import { Writable } from 'stream';
 
 const SERVICE_NAME = 'mercadodevida-backend';
 
@@ -11,14 +13,34 @@ export interface LoggerOptions {
   level?: string;
   /** Explicit destination stream (tests use an in-memory stream). */
   destination?: DestinationStream;
+  /** Log broadcaster for SSE streaming (optional). */
+  broadcaster?: LogBroadcaster;
 }
 
+/**
+ * Creates a pino logger. If a broadcaster is provided, logs are also pushed
+ * to all connected SSE clients via that broadcaster.
+ */
 export function createLogger(options: LoggerOptions = {}): Logger {
   const level = options.level ?? process.env.LOG_LEVEL ?? 'info';
   const settings = { level, base: { service: SERVICE_NAME } };
+
+  if (options.broadcaster) {
+    // Custom Node.js Writable stream that forwards each JSON line to the broadcaster
+    const toBroadcaster = new Writable({
+      write(chunk: Buffer, _encoding: string, callback: () => void) {
+        options.broadcaster!.addEntry(chunk.toString().trim());
+        callback();
+      },
+      autoDestroy: true,
+    });
+    return pino(settings, toBroadcaster);
+  }
+
   if (options.destination) {
     return pino(settings, options.destination);
   }
+
   return pino(settings);
 }