# nginx config — mandatory for SSE streaming
location /api/stream {
    proxy_pass http://gateway:3000;

    # Disable ALL response buffering for SSE routes
    proxy_buffering          off;
    proxy_cache              off;
    proxy_set_header         X-Accel-Buffering no;

    # Keep connection alive for long-running streams
    proxy_read_timeout       600s;
    proxy_connect_timeout    10s;

    # HTTP/1.1 required for chunked transfer encoding
    proxy_http_version       1.1;
    proxy_set_header         Connection '';
}

// ✅ Fastify SSE response — correct headers in the gateway
// file: src/routes/stream.ts

import type { FastifyRequest, FastifyReply } from 'fastify'

export async function streamRoute(req: FastifyRequest, reply: FastifyReply) {
  reply.raw.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache, no-transform',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',   // Disables nginx buffering
    'X-Content-Type-Options': 'nosniff',
  })
  reply.raw.flushHeaders()        // Immediately flush — critical for TTFT
}

data: {"token":"Hello"}\n\n
id: 1\n
data: {"token":"world"}\n\n
id: 2\n
data: {"token":"!"}\n\n
id: 3\n
data: [DONE]\n\n

// ✅ Production SSE gateway with AbortController cancellation
// file: src/routes/chat-stream.ts

import type { FastifyRequest, FastifyReply } from 'fastify'
import { AiGateway } from '../gateway/AiGateway'
import { ChatRequestSchema } from '@workspace/ai-schemas'

interface StreamQuery {
  lastEventId?: string  // For reconnection resume
}

export async function chatStreamRoute(
  req: FastifyRequest<{ Querystring: StreamQuery; Body: unknown }>,
  reply: FastifyReply,
) {
  // Validate incoming request before opening the stream
  const body = ChatRequestSchema.parse(req.body)

  // Set up AbortController wired to client disconnect — prevents token billing leak
  const controller = new AbortController()
  req.raw.on('close', () => controller.abort())

  reply.raw.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache, no-transform',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',
  })
  reply.raw.flushHeaders()

  let eventId = parseInt(req.query.lastEventId ?? '0', 10)

  try {
    const gateway = req.diScope.resolve<AiGateway>('aiGateway')

    for await (const action of gateway.streamResponse(body, controller.signal)) {
      if (controller.signal.aborted) break

      eventId++
      const payload = JSON.stringify(action)

      // SSE wire format: id, then data, then double newline
      reply.raw.write(`id: ${eventId}\ndata: ${payload}\n\n`)
    }

    reply.raw.write(`data: [DONE]\n\n`)
  } catch (error) {
    if (!controller.signal.aborted) {
      reply.raw.write(`event: error\ndata: ${JSON.stringify({ message: 'Stream error' })}\n\n`)
    }
  } finally {
    reply.raw.end()
  }
}

// ✅ Fetch-based SSE client with deterministic rehydration
// file: apps/web/src/lib/sseClient.ts

export interface SSEEvent {
  id: string
  data: string
}

export async function* fetchSSE(
  url: string,
  body: unknown,
  signal: AbortSignal,
  lastEventId?: string,
): AsyncGenerator<SSEEvent> {
  const headers: Record<string, string> = {
    'Content-Type': 'application/json',
    'Accept': 'text/event-stream',
  }
  if (lastEventId) headers['Last-Event-ID'] = lastEventId

  const response = await fetch(url, {
    method: 'POST',
    headers,
    body: JSON.stringify(body),
    signal,
  })

  if (!response.ok) throw new Error(`SSE error: ${response.status}`)
  if (!response.body) throw new Error('No response body')

  const reader = response.body.getReader()
  const decoder = new TextDecoder()
  let buffer = ''
  let currentId = lastEventId ?? '0'

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

      // TCP can split SSE events across chunks — buffer until double newline
      buffer += decoder.decode(value, { stream: true })

      // Split on event boundaries (double newline)
      const events = buffer.split('\n\n')
      buffer = events.pop() ?? '' // Last item may be incomplete — keep in buffer

      for (const rawEvent of events) {
        if (!rawEvent.trim()) continue

        // Parse SSE fields
        const lines = rawEvent.split('\n')
        let id = currentId
        let data = ''

        for (const line of lines) {
          if (line.startsWith('id: ')) id = line.slice(4)
          else if (line.startsWith('data: ')) data = line.slice(6)
        }

        if (data === '[DONE]') return
        if (data) {
          currentId = id
          yield { id, data }
        }
      }
    }
  } finally {
    reader.cancel()
  }
}

// ✅ Auto-reconnecting SSE hook with persistent state
// file: apps/web/src/hooks/useStreamingChat.ts

import { useCallback, useRef, useState } from 'react'
import { fetchSSE } from '../lib/sseClient'
import { GatewayActionSchema } from '@workspace/ai-schemas'

export function useStreamingChat() {
  const [tokens, setTokens] = useState<string[]>([])
  const [status, setStatus] = useState<'idle' | 'streaming' | 'done' | 'error'>('idle')
  const lastEventIdRef = useRef<string>()
  const abortRef = useRef<AbortController>()

  const send = useCallback(async (message: string, history: ChatMessage[]) => {
    abortRef.current?.abort()
    abortRef.current = new AbortController()
    setTokens([])
    setStatus('streaming')

    const MAX_RETRIES = 3
    let attempt = 0

    while (attempt < MAX_RETRIES) {
      try {
        for await (const event of fetchSSE(
          '/api/chat/stream',
          { message, history },
          abortRef.current.signal,
          lastEventIdRef.current,
        )) {
          lastEventIdRef.current = event.id
          const action = GatewayActionSchema.safeParse(JSON.parse(event.data))
          if (action.success && action.data.type === 'text_reply') {
            setTokens(prev => [...prev, action.data.payload.content])
          }
        }
        setStatus('done')
        break
      } catch (error) {
        if (abortRef.current.signal.aborted) break // User cancelled
        attempt++
        if (attempt >= MAX_RETRIES) { setStatus('error'); break }
        await new Promise(r => setTimeout(r, 500 * attempt)) // Exponential backoff
      }
    }
  }, [])

  const cancel = useCallback(() => abortRef.current?.abort(), [])

  return { tokens, status, send, cancel }
}

// ✅ Raw fetch ReadableStream for structured streaming
// Used internally by the gateway — not directly consumed by React components

export async function* streamRawChunks(
  response: Response,
  signal: AbortSignal,
): AsyncGenerator<string> {
  if (!response.body) throw new Error('No body')

  const reader = response.body.getReader()
  const decoder = new TextDecoder()

  try {
    while (!signal.aborted) {
      const { done, value } = await reader.read()
      if (done) break
      yield decoder.decode(value, { stream: true })
    }
  } finally {
    reader.cancel()
  }
}