id: 42\n
event: fleet:position\n
data: {"vehicleId":"B12","lat":37.7749,"lng":-122.4194,"speed":67}\n
retry: 3000\n
\n

data: {"type":"position","vehicles":[
data:   {"id":"B12","lat":37.7749},
data:   {"id":"C04","lat":37.7800}
data: ]}\n
\n

// ❌ Broken SSE endpoint — will silently fail in production
app.get('/api/stream/fleet', (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream')

  setInterval(() => {
    res.write(`data: ${JSON.stringify({ ts: Date.now() })}\n\n`)
  }, 1000)

  // Missing: req.on('close') cleanup → interval leaks forever
  // Missing: Cache-Control: no-cache → proxies may cache the stream
  // Missing: X-Accel-Buffering: no → Nginx buffers chunks until threshold
  // Missing: res.flushHeaders() → headers not sent until first write
})

// server/lib/SseConnection.ts
import type { Request, Response } from 'express'

export interface SseEvent {
  id: number
  type: string
  data: unknown
}

export class SseConnection {
  private lastEventId: number
  private heartbeatTimer: ReturnType<typeof setInterval>

  constructor(
    private readonly req: Request,
    private readonly res: Response,
  ) {
    // Resume from Last-Event-ID header (reconnection reconciliation)
    this.lastEventId = parseInt(req.headers['last-event-id'] as string ?? '0', 10)

    this.initHeaders()
    this.startHeartbeat()
    this.req.on('close', () => this.teardown())
  }

  private initHeaders(): void {
    this.res.writeHead(200, {
      'Content-Type':    'text/event-stream',
      'Cache-Control':   'no-cache, no-transform',
      'Connection':      'keep-alive',
      // Critical: Disable Nginx response buffering for this stream
      'X-Accel-Buffering': 'no',
    })
    // Flush headers immediately — client starts listening before first event
    this.res.flushHeaders()
  }

  /** Send a comment line — resets proxy idle timeout, fires no client event */
  private startHeartbeat(): void {
    this.heartbeatTimer = setInterval(() => {
      if (!this.res.writableEnded) {
        this.res.write(': keep-alive\n\n')
      }
    }, 20_000) // Every 20s — well within Nginx proxy_read_timeout 30s
  }

  send(event: SseEvent): void {
    if (this.res.writableEnded) return

    const payload = [
      `id: ${event.id}`,
      `event: ${event.type}`,
      `data: ${JSON.stringify(event.data)}`,
      `retry: 3000`,
      '', // blank line = event delimiter
    ].join('\n') + '\n'

    this.res.write(payload)
  }

  get resumeFromId(): number {
    return this.lastEventId
  }

  private teardown(): void {
    clearInterval(this.heartbeatTimer)
    if (!this.res.writableEnded) {
      this.res.end()
    }
  }
}

// server/routes/stream.ts
import { SseConnection } from '../lib/SseConnection'
import { subscribeToTenantEvents, unsubscribe } from '../lib/EventBridge'
import { replayMissedEvents } from '../lib/EventStore'

app.get('/api/stream/fleet', requireAuth, async (req, res) => {
  const tenantId = req.user!.tenantId
  const conn = new SseConnection(req, res)

  // Replay any events missed during the disconnect window
  const missed = await replayMissedEvents(tenantId, conn.resumeFromId)
  for (const event of missed) {
    conn.send(event)
  }

  // Subscribe to live events for this tenant
  const subscriptionId = subscribeToTenantEvents(tenantId, (event) => {
    conn.send(event)
  })

  // Clean up when client disconnects (tab close, navigation, network drop)
  req.on('close', () => {
    unsubscribe(subscriptionId)
  })
})

# /etc/nginx/sites-available/fleettracker.conf

upstream fleet_api {
    server 127.0.0.1:3000;
    keepalive 64; # Reuse upstream connections
}

server {
    listen 443 ssl http2; # HTTP/2 is mandatory for multi-tab SSE

    location /api/stream/ {
        proxy_pass http://fleet_api;
        proxy_http_version 1.1;

        # WebSocket/SSE upgrade headers
        proxy_set_header Connection '';  # empty string for HTTP/2 upstream
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;

        # Must exceed server heartbeat interval (20s) by a safe margin
        proxy_read_timeout 30s;

        # Disable Nginx response buffering for SSE streams
        proxy_buffering off;

        # Disable compression on streaming responses
        gzip off;

        # Signal to Express/upstream that buffering is disabled
        proxy_set_header X-Accel-Buffering no;
    }
}

// server/lib/EventStore.ts

interface StoredEvent {
  id: number
  tenantId: string
  type: string
  data: unknown
  createdAt: number
}

class InMemoryEventStore {
  private readonly BUFFER_SIZE = 1_000 // Last 1,000 events per tenant
  private readonly events = new Map<string, StoredEvent[]>()
  private idCounter = 0

  store(tenantId: string, type: string, data: unknown): StoredEvent {
    const event: StoredEvent = {
      id: ++this.idCounter,
      tenantId,
      type,
      data,
      createdAt: Date.now(),
    }

    if (!this.events.has(tenantId)) {
      this.events.set(tenantId, [])
    }

    const buffer = this.events.get(tenantId)!
    buffer.push(event)

    // Trim to BUFFER_SIZE — evict oldest events
    if (buffer.length > this.BUFFER_SIZE) {
      buffer.splice(0, buffer.length - this.BUFFER_SIZE)
    }

    return event
  }

  replay(tenantId: string, sinceId: number): StoredEvent[] {
    return (this.events.get(tenantId) ?? [])
      .filter((e) => e.id > sinceId)
  }
}

export const eventStore = new InMemoryEventStore()

export async function replayMissedEvents(
  tenantId: string,
  sinceId: number,
): Promise<StoredEvent[]> {
  if (sinceId === 0) return [] // First connection — no replay needed

  const buffered = eventStore.replay(tenantId, sinceId)

  if (buffered.length > 0) return buffered

  // In-memory buffer evicted (server restart / too many events) —
  // fall back to database for events missed in the last 5 minutes
  return db.events.find({
    tenantId,
    id: { $gt: sinceId },
    createdAt: { $gt: Date.now() - 5 * 60 * 1_000 },
  }).sort({ id: 1 })
}

// Before server shutdown: send high retry to spread reconnect load
export function sendShutdownSignal(connections: Set<SseConnection>): void {
  for (const conn of connections) {
    // Instruct clients to wait 10–40 seconds before reconnecting
    const jitter = Math.floor(Math.random() * 30_000)
    conn.sendRaw(`retry: ${10_000 + jitter}\n\n`)
    conn.sendRaw('event: server:shutdown\ndata: {}\n\n')
  }
}

// Basic EventSource usage
const source = new EventSource('/api/stream/fleet', {
  withCredentials: true, // Sends httpOnly cookies cross-origin
})

source.addEventListener('fleet:position', (e: MessageEvent) => {
  const vehicle = JSON.parse(e.data) as VehiclePosition
  updateMap(vehicle)
})

source.addEventListener('error', (e) => {
  if (source.readyState === EventSource.CLOSED) {
    console.error('SSE connection permanently closed')
  }
  // If readyState === CONNECTING, browser is auto-reconnecting
})

// Cleanup
source.close()

// hooks/useSSE.ts
import { useEffect, useRef, useCallback, useState } from 'react'

type SseEventMap = Record<string, (data: unknown) => void>

interface UseSseOptions {
  url: string
  events: SseEventMap
  enabled?: boolean
}

export function useSSE({ url, events, enabled = true }: UseSseOptions) {
  const [status, setStatus] = useState<'connecting' | 'open' | 'closed'>('closed')
  const sourceRef = useRef<EventSource | null>(null)
  const cleanupRef = useRef<(() => void) | null>(null)

  const connect = useCallback(() => {
    if (!enabled) return

    // Ticket-based auth: fetch a short-lived ticket from the REST API
    fetch('/api/auth/sse-ticket', { method: 'POST' })
      .then((r) => r.json())
      .then(({ ticket }: { ticket: string }) => {
        const source = new EventSource(`${url}?ticket=${ticket}`)
        sourceRef.current = source
        setStatus('connecting')

        source.onopen = () => setStatus('open')
        source.onerror = () => setStatus('connecting') // Auto-reconnects

        // Register all event listeners
        const handlers = Object.entries(events).map(([type, handler]) => {
          const listener = (e: MessageEvent) => handler(JSON.parse(e.data))
          source.addEventListener(type, listener)
          return { type, listener }
        })

        cleanupRef.current = () => {
          handlers.forEach(({ type, listener }) => {
            source.removeEventListener(type, listener)
          })
          source.close()
          setStatus('closed')
        }
      })
  }, [url, enabled, events])

  useEffect(() => {
    // React 18/19 StrictMode: double-invoke guard
    let cancelled = false

    if (!cancelled) connect()

    return () => {
      cancelled = true
      cleanupRef.current?.()
      cleanupRef.current = null
    }
  }, [connect])

  // Pause streaming when tab is hidden — saves bandwidth and server resources
  useEffect(() => {
    const handleVisibility = () => {
      if (document.visibilityState === 'hidden') {
        cleanupRef.current?.()
      } else {
        connect()
      }
    }

    document.addEventListener('visibilitychange', handleVisibility)
    return () => document.removeEventListener('visibilitychange', handleVisibility)
  }, [connect])

  return { status }
}

// Usage in a component
function FleetMap() {
  const { status } = useSSE({
    url: '/api/stream/fleet',
    events: {
      'fleet:position': (data) => updateVehiclePosition(data as VehiclePosition),
      'fleet:alert':    (data) => showAlert(data as FleetAlert),
      'server:shutdown': () => showReconnectingBanner(),
    },
  })

  return (
    <div>
      <ConnectionBadge status={status} />
      <MapView />
    </div>
  )
}

// server: Handle reconnection with full-sync fallback
app.get('/api/stream/fleet', requireAuth, async (req, res) => {
  const tenantId = req.user!.tenantId
  const conn = new SseConnection(req, res)
  const sinceId = conn.resumeFromId

  const missed = await replayMissedEvents(tenantId, sinceId)

  if (sinceId > 0 && missed.length === 0) {
    // Buffer miss — force client to do a full state sync before resuming
    conn.sendRaw('event: sync:required\ndata: {}\n\n')
    // Client must call GET /api/state/full and re-render from scratch
  } else {
    for (const event of missed) conn.send(event)
  }

  // Continue streaming live events...
})