Потоковые Route Handlers: Server-Sent Events (SSE), WebStreams API и Backpressure

+-----------------------------------------------------------------------------------+
|               АРХИТЕКТУРА WEB STREAMS & SERVER-SENT EVENTS (SSE)                  |
+-----------------------------------------------------------------------------------+
|  ИСТОЧНИК ДАННЫХ          NEXT.JS 16 ROUTE HANDLER               КЛИЕНТ (БРАУЗЕР) |
|                                                                                   |
|  [Redis Pub/Sub / ] ---> [ ReadableStreamSource        ]                         |
|  [LLM Inference / ]      |  controller.enqueue(chunk)  |                         |
|  [Postgres Cursor ]      |                             |                         |
|                          |    [ TransformStream ]      |                          |
|                          |    (SSE Text Encoder)       |                          |
|                          +--------------+--------------+                          |
|                                         | HTTP/2 Frame                            |
|                 Backpressure Signal     | "event: metric\n"                       |
|           <-----------------------------+ "data: {...}\n\n"                      |
|           (desiredSize <= 0: пауза)     |                                         |
|                                         v                                         |
|  [request.signal] <==================================== [ EventSource / fetch() ]|
|  (abort event)             Клиент закрыл вкладку        (getReader().read())      |
|  -> чистка таймеров/сокетов                                                       |
+-----------------------------------------------------------------------------------+

import { NextRequest } from 'next/server';

export const runtime = 'nodejs'; // Или 'edge' в зависимости от инфраструктуры
export const dynamic = 'force-dynamic';

interface TelemetryPayload {
  cpuUsage: number;
  memoryUsageMb: number;
  activeRequests: number;
  timestamp: number;
}

export async function GET(request: NextRequest) {
  const encoder = new TextEncoder();
  const abortSignal = request.signal;

  // Форматирование сообщения по спецификации SSE
  const formatSseEvent = (event: string, data: unknown, id?: string): string => {
    let message = '';
    if (id) message += `id: ${id}\n`;
    message += `event: ${event}\n`;
    message += `data: ${JSON.stringify(data)}\n\n`;
    return message;
  };

  const stream = new ReadableStream({
    start(controller) {
      let messageId = 0;
      let timer: NodeJS.Timeout | null = null;

      // Обработка принудительного разрыва соединения клиентом
      const cleanup = () => {
        if (timer) {
          clearInterval(timer);
          timer = null;
        }
        try {
          controller.close();
        } catch {
          // Игнорируем ошибку, если контроллер уже закрыт
        }
      };

      if (abortSignal.aborted) {
        cleanup();
        return;
      }

      abortSignal.addEventListener('abort', cleanup, { once: true });

      // Генератор телеметрии с контролем противодавления
      timer = setInterval(() => {
        // Проверяем сигнал отмены перед каждым тиком
        if (abortSignal.aborted) {
          cleanup();
          return;
        }

        // BACKPRESSURE CHECK: Если внутренний буфер переполнен, пропускаем такт
        if (controller.desiredSize !== null && controller.desiredSize <= 0) {
          console.warn('[SSE] Backpressure detected: client queue is full. Throttling generation.');
          return;
        }

        messageId++;
        const payload: TelemetryPayload = {
          cpuUsage: Math.round(Math.random() * 100),
          memoryUsageMb: Math.round(process.memoryUsage().heapUsed / 1024 / 1024),
          activeRequests: Math.floor(Math.random() * 500),
          timestamp: Date.now(),
        };

        const sseFormattedChunk = formatSseEvent('telemetry_tick', payload, String(messageId));

        try {
          controller.enqueue(encoder.encode(sseFormattedChunk));
        } catch (err) {
          console.error('[SSE] Failed to enqueue chunk, closing stream:', err);
          cleanup();
        }
      }, 1000);
    },
    cancel(reason) {
      console.log('[SSE] Stream cancelled by consumer:', reason);
    }
  });

  // Возврат стандартного ответа со специальными заголовками
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      'Connection': 'keep-alive',
      // Критично для Nginx / OpenResty: отключает внутреннюю буферизацию прокси!
      'X-Accel-Buffering': 'no',
    },
  });
}

'use client';

import { useEffect, useState, useCallback } from 'react';

export interface TelemetryData {
  cpuUsage: number;
  memoryUsageMb: number;
  activeRequests: number;
  timestamp: number;
}

export function useServerTelemetry() {
  const [data, setData] = useState<TelemetryData | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const [error, setError] = useState<string | null>(null);

  const connectStream = useCallback(() => {
    const controller = new AbortController();

    async function startFetchStream() {
      try {
        setIsConnected(true);
        setError(null);

        const response = await fetch('/api/telemetry/stream', {
          signal: controller.signal,
          headers: {
            Accept: 'text/event-stream',
          },
        });

        if (!response.ok || !response.body) {
          throw new Error(`HTTP Stream failed: ${response.status}`);
        }

        const reader = response.body.getReader();
        const decoder = new TextDecoder('utf-8');
        let buffer = '';

        while (true) {
          const { value, done } = await reader.read();
          if (done) break;

          buffer += decoder.decode(value, { stream: true });
          const parts = buffer.split('\n\n');
          buffer = parts.pop() ?? ''; // Сохраняем незавершенный кусок

          for (const rawMessage of parts) {
            const lines = rawMessage.split('\n');
            let eventType = 'message';
            let rawData = '';

            for (const line of lines) {
              if (line.startsWith('event:')) {
                eventType = line.replace('event:', '').trim();
              } else if (line.startsWith('data:')) {
                rawData = line.replace('data:', '').trim();
              }
            }

            if (eventType === 'telemetry_tick' && rawData) {
              const parsed: TelemetryData = JSON.parse(rawData);
              setData(parsed);
            }
          }
        }
      } catch (err: unknown) {
        if ((err as Error).name !== 'AbortError') {
          console.error('[SSE Client Error]:', err);
          setError((err as Error).message);
          setIsConnected(false);
        }
      }
    }

    startFetchStream();

    return () => {
      // При размонтировании компонента прерываем чтение, посылая сигнал abort на сервер
      controller.abort();
      setIsConnected(false);
    };
  }, []);

  useEffect(() => {
    const cleanup = connectStream();
    return cleanup;
  }, [connectStream]);

  return { data, isConnected, error };
}