// ❌ Component-scoped socket — creates reconnection storms
function FleetMap() {
  const [socket, setSocket] = useState<WebSocket | null>(null)

  useEffect(() => {
    const ws = new WebSocket('wss://api.fleettracker.io/ws/fleet')
    setSocket(ws)

    return () => ws.close() // ← Closes on every re-render!
  }, [])

  // Problems:
  // 1. Each mount of FleetMap creates a new WebSocket connection
  // 2. When FleetMap unmounts (route change, conditional render),
  //    the socket closes — even if another component needs it
  // 3. React 18 StrictMode double-invokes useEffect → 2 connections opened
  // 4. Navigation away and back → old socket closed, new one opened
}

// client/lib/SocketManager.ts
import { getTicket } from './auth'

type EventHandler = (data: unknown) => void

class SocketManager {
  private socket: WebSocket | null = null
  private subscriptions = new Map<string, Set<EventHandler>>()
  private refCount = 0
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null
  private reconnectAttempts = 0

  /** Increment ref count — called when a component subscribes */
  async acquire(): Promise<void> {
    this.refCount++
    if (this.refCount === 1) {
      await this.connect()
    }
  }

  /** Decrement ref count — called when a component unsubscribes */
  release(): void {
    this.refCount = Math.max(0, this.refCount - 1)
    if (this.refCount === 0) {
      this.disconnect()
    }
  }

  on(event: string, handler: EventHandler): () => void {
    if (!this.subscriptions.has(event)) {
      this.subscriptions.set(event, new Set())
    }
    this.subscriptions.get(event)!.add(handler)

    // Return unsubscribe function
    return () => {
      this.subscriptions.get(event)?.delete(handler)
    }
  }

  private async connect(): Promise<void> {
    // Fetch a short-lived ticket for the WebSocket handshake
    const { ticket } = await getTicket()
    const url = `wss://api.fleettracker.io/ws/fleet?ticket=${ticket}`

    this.socket = new WebSocket(url)

    this.socket.onmessage = (e) => {
      const msg = JSON.parse(e.data) as { type: string; data: unknown }
      this.subscriptions.get(msg.type)?.forEach((handler) => handler(msg.data))
    }

    this.socket.onclose = (e) => {
      if (e.code !== 1000 && e.code !== 1001) {
        // Abnormal close — reconnect with exponential backoff + jitter
        const delay = Math.random() * Math.min(30_000, 1_000 * Math.pow(2, this.reconnectAttempts++))
        this.reconnectTimer = setTimeout(() => this.connect(), delay)
      }
    }

    this.socket.onerror = () => {
      this.socket?.close()
    }
  }

  private disconnect(): void {
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
    this.socket?.close(1000, 'Client unmounted')
    this.socket = null
    this.reconnectAttempts = 0
  }
}

// Module-level singleton — one instance for the entire application
export const socketManager = new SocketManager()

Mount #1  → acquire() → refCount = 1 → connect() (WebSocket opens)
Unmount   → release() → refCount = 0 → disconnect() (WebSocket closes)
Mount #2  → acquire() → refCount = 1 → connect() (WebSocket opens again)

Component A mounts → acquire() → refCount = 1 → connect()
Component B mounts → acquire() → refCount = 2 → (no-op, already connected)
Component A unmounts → release() → refCount = 1 → (no-op, B still needs it)
Component B unmounts → release() → refCount = 0 → disconnect()

// client/hooks/useRealtimeEvent.ts
import { useEffect, useRef } from 'react'
import { socketManager } from '../lib/SocketManager'

export function useRealtimeEvent<T>(
  eventType: string,
  handler: (data: T) => void,
  enabled = true,
): void {
  // Stable ref to avoid re-subscribing on every render
  const handlerRef = useRef(handler)
  handlerRef.current = handler

  useEffect(() => {
    if (!enabled) return

    let unsubscribe: (() => void) | null = null

    socketManager.acquire().then(() => {
      unsubscribe = socketManager.on(eventType, (data) => {
        handlerRef.current(data as T)
      })
    })

    return () => {
      unsubscribe?.()
      socketManager.release()
    }
  }, [eventType, enabled])
}

// Usage in any component — no socket management code
function VehicleTracker({ vehicleId }: { vehicleId: string }) {
  const [position, setPosition] = useState<VehiclePosition | null>(null)

  useRealtimeEvent<VehiclePosition>(
    'fleet:position',
    (data) => {
      if (data.vehicleId === vehicleId) setPosition(data)
    },
  )

  return <MapPin position={position} />
}

// ❌ Passing a long-lived JWT in the WebSocket URL — multiple failures
const ws = new WebSocket(`wss://api.example.com/ws?token=${jwtToken}`)

// Problems:
// 1. The full URL (including query string) is logged in Nginx access logs,
//    CloudFront access logs, and application server logs — token is plaintext
// 2. Browser history / referrer headers may expose the token
// 3. The JWT is valid for its full lifetime (hours) — replay attack window
// 4. Cannot be rotated without closing the WebSocket connection

// server/routes/auth.ts

import { randomBytes, createHmac } from 'node:crypto'

const TICKET_SECRET = process.env.WS_TICKET_SECRET! // 32-byte random secret
const pendingTickets = new Map<string, { userId: string; tenantId: string; expiresAt: number }>()

// Step 1: Client calls this REST endpoint with a valid Bearer token
app.post('/api/auth/ws-ticket', requireAuth, (req, res) => {
  const ticketId = randomBytes(32).toString('hex')
  const expiresAt = Date.now() + 30_000 // 30-second validity window

  // Sign ticket with HMAC-SHA256
  const signature = createHmac('sha256', TICKET_SECRET)
    .update(`${ticketId}:${expiresAt}:${req.user!.userId}`)
    .digest('hex')

  const ticket = `${ticketId}.${expiresAt}.${signature}`

  // Store pending ticket (in Redis for multi-node: SETEX ticket 30 userId)
  pendingTickets.set(ticketId, {
    userId: req.user!.userId,
    tenantId: req.user!.tenantId,
    expiresAt,
  })

  // Auto-expire if never used
  setTimeout(() => pendingTickets.delete(ticketId), 31_000)

  res.json({ ticket })
})

// server/ws/FleetWebSocketServer.ts
import { createHmac } from 'node:crypto'

function validateTicket(rawTicket: string | null): {
  userId: string
  tenantId: string
} | null {
  if (!rawTicket) return null

  const parts = rawTicket.split('.')
  if (parts.length !== 3) return null

  const [ticketId, expiresAtStr, signature] = parts as [string, string, string]
  const expiresAt = parseInt(expiresAtStr, 10)

  // Check expiry
  if (Date.now() > expiresAt) return null

  // Verify HMAC signature
  const pending = pendingTickets.get(ticketId)
  if (!pending) return null // Already consumed or expired

  const expectedSig = createHmac('sha256', TICKET_SECRET)
    .update(`${ticketId}:${expiresAt}:${pending.userId}`)
    .digest('hex')

  if (signature !== expectedSig) return null

  // Consume ticket — single-use enforcement
  pendingTickets.delete(ticketId)

  return { userId: pending.userId, tenantId: pending.tenantId }
}

wss.on('connection', (socket, req) => {
  const url = new URL(req.url!, 'http://localhost')
  const rawTicket = url.searchParams.get('ticket')
  const session = validateTicket(rawTicket)

  if (!session) {
    // Close with RFC 6455 Policy Violation code
    socket.close(1008, 'Invalid or expired ticket')
    return
  }

  connectionManager.register(socket, session)
})

// server/lib/ConnectionManager.ts
import { WebSocket } from 'ws'
import { randomUUID } from 'node:crypto'

interface Connection {
  id: string
  socket: WebSocket
  userId: string
  tenantId: string
  connectedAt: number
  lastPingAt: number
}

class ConnectionManager {
  private readonly connections = new Map<string, Connection>()
  private readonly byTenant   = new Map<string, Set<string>>() // tenantId → Set<connId>
  private readonly byUser     = new Map<string, Set<string>>() // userId → Set<connId>

  register(socket: WebSocket, session: { userId: string; tenantId: string }): string {
    const id = randomUUID()
    const conn: Connection = {
      id,
      socket,
      userId: session.userId,
      tenantId: session.tenantId,
      connectedAt: Date.now(),
      lastPingAt: Date.now(),
    }

    this.connections.set(id, conn)
    this.addToIndex(this.byTenant, session.tenantId, id)
    this.addToIndex(this.byUser, session.userId, id)

    return id
  }

  unregister(connId: string): void {
    const conn = this.connections.get(connId)
    if (!conn) return

    this.connections.delete(connId)
    this.byTenant.get(conn.tenantId)?.delete(connId)
    this.byUser.get(conn.userId)?.delete(connId)
  }

  /** Broadcast an event to all connections for a specific tenant */
  broadcastToTenant(tenantId: string, message: unknown): void {
    const connIds = this.byTenant.get(tenantId)
    if (!connIds) return

    const payload = JSON.stringify(message)

    for (const connId of connIds) {
      const conn = this.connections.get(connId)
      if (conn?.socket.readyState === WebSocket.OPEN) {
        conn.socket.send(payload)
      } else {
        // Socket closed — clean up stale connection
        this.unregister(connId)
      }
    }
  }

  get activeCount(): number {
    return this.connections.size
  }

  private addToIndex(index: Map<string, Set<string>>, key: string, value: string): void {
    if (!index.has(key)) index.set(key, new Set())
    index.get(key)!.add(value)
  }
}

export const connectionManager = new ConnectionManager()

// ❌ Change Streams on a standalone MongoDB — throws immediately
const client = new MongoClient('mongodb://localhost:27017') // Standalone
await client.connect()

// This throws: MongoServerError: The $changeStream stage is only
// supported on replica sets and sharded clusters.
const stream = db.collection('vehicles').watch()

# Start a single-node replica set for local development
mongod --replSet rs0 --port 27017 --dbpath /data/db

# In mongosh: initialize the replica set
rs.initiate({
  _id: 'rs0',
  members: [{ _id: 0, host: 'localhost:27017' }]
})

// ❌ Opening a Change Stream cursor per connected WebSocket client
wss.on('connection', async (socket, req) => {
  const { tenantId } = validateTicket(/* ... */)

  // At 1,000 concurrent users: 1,000 open MongoDB cursors
  // Each cursor tails the oplog and transmits all events to the application
  // MongoDB Atlas free tier: 500 cursor limit — will throw at 501st user
  const stream = db.collection('vehicles').watch([
    { $match: { 'fullDocument.tenantId': tenantId } }
  ])

  stream.on('change', (change) => socket.send(JSON.stringify(change)))
  socket.on('close', () => stream.close())
})

// server/workers/ChangeStreamWorker.ts
import { MongoClient, ChangeStreamDocument } from 'mongodb'
import { connectionManager } from '../lib/ConnectionManager'
import { eventStore } from '../lib/EventStore'

const RECONNECT_DELAY_MS = 1_000
const MAX_RECONNECT_DELAY_MS = 30_000

class ChangeStreamWorker {
  private resumeToken: unknown = null
  private reconnectDelay = RECONNECT_DELAY_MS
  private running = true

  async start(client: MongoClient): Promise<void> {
    const db = client.db('fleettracker')

    while (this.running) {
      try {
        await this.runStream(db)
      } catch (err) {
        // ChangeStreamFatalError: resume token is stale (oplog rotated)
        if ((err as Error).message.includes('ChangeStreamFatalError') ||
            (err as Error).message.includes('resume token')) {
          console.error('Resume token stale — triggering Full State Sync')
          this.resumeToken = null
          await this.triggerFullStateSync(db)
        }

        // Reconnect with backoff
        await sleep(this.reconnectDelay)
        this.reconnectDelay = Math.min(MAX_RECONNECT_DELAY_MS, this.reconnectDelay * 2)
      }
    }
  }

  private async runStream(db: ReturnType<MongoClient['db']>): Promise<void> {
    const options = {
      fullDocument: 'updateLookup' as const, // Include the full document on updates
      ...(this.resumeToken ? { resumeAfter: this.resumeToken } : {}),
    }

    // Single cursor for the entire application
    const stream = db.collection('vehicles').watch(
      [{ $match: { operationType: { $in: ['insert', 'update', 'replace', 'delete'] } } }],
      options,
    )

    console.log('Change Stream worker connected, resumeToken:', this.resumeToken ?? 'none')

    for await (const change of stream) {
      this.reconnectDelay = RECONNECT_DELAY_MS // Reset backoff on successful events
      this.resumeToken = change._id // Checkpoint the resume token

      await this.processChange(change)
    }
  }

  private async processChange(change: ChangeStreamDocument): Promise<void> {
    const doc = (change as any).fullDocument
    if (!doc?.tenantId) return

    const event = {
      type: `fleet:${change.operationType}`,
      data: doc,
    }

    // Store in event buffer for SSE/long-poll replay
    const stored = eventStore.store(doc.tenantId, event.type, event.data)

    // Fan out to all active WebSocket connections for this tenant
    connectionManager.broadcastToTenant(doc.tenantId, {
      type: stored.type,
      id: stored.id,
      data: stored.data,
    })
  }

  private async triggerFullStateSync(db: ReturnType<MongoClient['db']>): Promise<void> {
    // Notify all connected clients that they must re-fetch full state
    // In practice: iterate all tenants with active connections and broadcast
    const allVehicles = await db.collection('vehicles').find({}).toArray()

    const vehiclesByTenant = new Map<string, unknown[]>()
    for (const v of allVehicles) {
      if (!vehiclesByTenant.has(v.tenantId)) vehiclesByTenant.set(v.tenantId, [])
      vehiclesByTenant.get(v.tenantId)!.push(v)
    }

    for (const [tenantId, vehicles] of vehiclesByTenant) {
      connectionManager.broadcastToTenant(tenantId, {
        type: 'sync:full',
        data: { vehicles },
      })
    }
  }
}

export const changeStreamWorker = new ChangeStreamWorker()

// mongosh: Check current oplog size and estimated coverage window
use local
db.oplog.rs.stats().maxSize      // e.g., 2GB
db.oplog.rs.find().sort({ ts: 1 }).limit(1).next().ts  // oldest entry
db.oplog.rs.find().sort({ ts: -1 }).limit(1).next().ts // newest entry

// Calculate window (seconds)
const oldest = db.oplog.rs.find().sort({ ts: 1 }).limit(1).next().ts
const newest = db.oplog.rs.find().sort({ ts: -1 }).limit(1).next().ts
const windowSeconds = newest.t - oldest.t
print(`Oplog window: ${windowSeconds / 3600} hours`)

// Set oplog size (requires replica set reconfiguration)
// In mongod.conf:
replication:
  replSetName: rs0
  oplogSizeMB: 10240  # 10GB — target 48+ hours of coverage at peak write throughput

// Server: configure write concern for Change Stream compatibility
const db = client.db('fleettracker', {
  writeConcern: { w: 'majority', j: true },
  readConcern: { level: 'majority' },
})

// Change Streams with readConcern: majority guarantee:
// - Events are emitted ONLY for writes that have been committed to a majority of replica set members
// - If the primary fails and a new one is elected, rolled-back writes are NEVER emitted to the stream
// - This enforces CP semantics: clients always see a consistent, committed view of the database

// client/hooks/useOptimisticUpdate.ts
import { useState, useCallback } from 'react'
import { socketManager } from '../lib/SocketManager'

interface OptimisticState<T> {
  data: T
  pending: boolean
  error: string | null
}

export function useOptimisticUpdate<T>(
  initialData: T,
  eventType: string,
) {
  const [state, setState] = useState<OptimisticState<T>>({
    data: initialData,
    pending: false,
    error: null,
  })

  const mutate = useCallback(async (optimisticData: T, serverFn: () => Promise<T>) => {
    // Render optimistic state immediately
    setState({ data: optimisticData, pending: true, error: null })

    try {
      const confirmed = await serverFn()
      // Server confirmed — apply authoritative data
      setState({ data: confirmed, pending: false, error: null })
    } catch (err) {
      // Server rejected — roll back to pre-mutation state
      setState({ data: initialData, pending: false, error: (err as Error).message })
    }
  }, [initialData])

  // Listen for server-side updates (e.g., another user changed the same record)
  useRealtimeEvent<T>(eventType, (serverData) => {
    setState((s) => s.pending ? s : { data: serverData, pending: false, error: null })
  })

  return { state, mutate }
}