/**
 * CDC SSE (Server-Sent Events) Transport for postgres.do
 *
 * Provides HTTP-based change data capture streaming via SSE:
 * - Works in environments where WebSocket is blocked
 * - Uses standard EventSource/fetch with streaming body
 * - Automatic reconnection with Last-Event-ID header
 * - LSN-based resumption for reliable delivery
 *
 * Performance optimizations:
 * - Event batching for high-throughput scenarios
 * - Adaptive heartbeat intervals based on activity
 * - Reusable TextDecoder for memory efficiency
 * - Exponential backoff with jitter for reconnection
 * - Compression support via Accept-Encoding
 */

import type {
  Row,
  CDCChangeEvent,
  CDCTransport,
  CDCSubscriptionOptions,
  CDCSseTransportConfig,
  CDCOperation,
} from '../types'
import { ConnectionError } from '../types'

/**
 * SSE event types for CDC
 */
type SseEventType = 'insert' | 'update' | 'delete' | 'truncate' | 'begin' | 'commit' | 'keepalive' | 'error' | 'subscribe_result' | 'unsubscribe_result'

/**
 * SSE event data structure
 */
interface SseEventData {
  type: SseEventType
  subscriptionId?: string
  table?: string
  schema?: string
  data?: Row
  old?: Row
  new?: Row
  changedColumns?: string[]
  xid?: number
  isLastInTransaction?: boolean
  success?: boolean
  error?: string
  message?: string
  code?: string
}

/**
 * Batched event for high-throughput delivery
 */
interface BatchedEvent {
  subscriptionId: string
  event: CDCChangeEvent
  timestamp: number
}

/**
 * Update event data with old and new row values
 */
interface UpdateEventData {
  old?: Row
  new?: Row
}

/**
 * Default configuration values
 */
const DEFAULTS = {
  retryMs: 3000,
  connectTimeout: 10000,
  maxReconnectAttempts: 10,
  // Batching defaults
  batchSize: 100,
  batchTimeoutMs: 50,
  // Adaptive heartbeat defaults
  minHeartbeatMs: 5000,
  maxHeartbeatMs: 60000,
  heartbeatScaleFactor: 1.5,
  // Memory optimization
  maxBufferSize: 64 * 1024, // 64KB buffer limit
}

/**
 * Extended SSE transport configuration with optimization options
 */
export interface CDCSseTransportOptimizedConfig extends CDCSseTransportConfig {
  /** Enable event batching for high-throughput scenarios */
  enableBatching?: boolean
  /** Maximum events to batch before flushing */
  batchSize?: number
  /** Maximum time to wait before flushing batch (ms) */
  batchTimeoutMs?: number
  /** Enable compression (gzip/deflate) if server supports it */
  enableCompression?: boolean
  /** Enable adaptive heartbeat intervals */
  adaptiveHeartbeat?: boolean
  /** Minimum heartbeat interval (ms) */
  minHeartbeatMs?: number
  /** Maximum heartbeat interval (ms) */
  maxHeartbeatMs?: number
  /** Maximum buffer size before applying backpressure (bytes) */
  maxBufferSize?: number
}

/**
 * CDC SSE Transport implementation with streaming optimizations
 */
export class CDCSseTransport implements CDCTransport {
  private config: {
    url: string
    apiKey?: string
    resumeFrom?: string
    retryMs: number
    fetch?: typeof fetch
    headers?: Record<string, string>
  }
  private fetchImpl: typeof fetch
  private connected = false
  private lastEventId: string | null = null
  private shouldReconnect = true
  private reconnectAttempt = 0
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null
  private abortController: AbortController | null = null

  // Event handlers
  private eventHandler: ((subscriptionId: string, event: CDCChangeEvent) => void) | null = null
  private errorHandler: ((subscriptionId: string, error: Error) => void) | null = null
  private reconnectHandler: ((attempt: number, lastLsn: string | undefined) => void) | null = null
  private closeHandler: (() => void) | null = null

  // Active subscriptions for URL construction
  private activeSubscriptions = new Map<string, {
    table: string
    schema: string
    options: CDCSubscriptionOptions
  }>()

  // Pending subscription/unsubscription requests
  private pendingSubscriptions = new Map<string, {
    resolve: () => void
    reject: (error: Error) => void
  }>()
  private pendingUnsubscriptions = new Map<string, {
    resolve: () => void
    reject: (error: Error) => void
  }>()

  // ========== Streaming Optimizations ==========

  // Reusable TextDecoder for memory efficiency
  private readonly decoder = new TextDecoder()

  // Event batching state
  private enableBatching: boolean
  private batchSize: number
  private batchTimeoutMs: number
  private eventBatch: BatchedEvent[] = []
  private batchTimer: ReturnType<typeof setTimeout> | null = null

  // Adaptive heartbeat state
  private adaptiveHeartbeat: boolean
  private minHeartbeatMs: number
  private maxHeartbeatMs: number
  private currentHeartbeatMs: number
  private lastEventTime = 0
  private eventRateWindow: number[] = []
  private heartbeatTimer: ReturnType<typeof setInterval> | null = null

  // Buffer management for backpressure
  private maxBufferSize: number
  private currentBufferSize = 0

  // Compression support
  private enableCompression: boolean

  // Performance metrics
  private metrics = {
    eventsReceived: 0,
    eventsBatched: 0,
    batchesFlushed: 0,
    reconnections: 0,
    bytesReceived: 0,
  }

  constructor(config: CDCSseTransportConfig | CDCSseTransportOptimizedConfig) {
    const optimizedConfig = config as CDCSseTransportOptimizedConfig
    this.config = {
      url: config.url,
      retryMs: config.retryMs ?? DEFAULTS.retryMs,
    }
    if (config.apiKey !== undefined) {
      this.config.apiKey = config.apiKey
    }
    if (config.resumeFrom !== undefined) {
      this.config.resumeFrom = config.resumeFrom
    }
    if (config.fetch !== undefined) {
      this.config.fetch = config.fetch
    }
    if (config.headers !== undefined) {
      this.config.headers = config.headers
    }
    this.fetchImpl = config.fetch || globalThis.fetch
    this.lastEventId = config.resumeFrom ?? null

    // Initialize optimization settings
    this.enableBatching = optimizedConfig.enableBatching ?? false
    this.batchSize = optimizedConfig.batchSize ?? DEFAULTS.batchSize
    this.batchTimeoutMs = optimizedConfig.batchTimeoutMs ?? DEFAULTS.batchTimeoutMs
    this.enableCompression = optimizedConfig.enableCompression ?? true
    this.adaptiveHeartbeat = optimizedConfig.adaptiveHeartbeat ?? true
    this.minHeartbeatMs = optimizedConfig.minHeartbeatMs ?? DEFAULTS.minHeartbeatMs
    this.maxHeartbeatMs = optimizedConfig.maxHeartbeatMs ?? DEFAULTS.maxHeartbeatMs
    this.currentHeartbeatMs = this.minHeartbeatMs
    this.maxBufferSize = optimizedConfig.maxBufferSize ?? DEFAULTS.maxBufferSize
  }

  /**
   * Connect to the CDC stream
   */
  async connect(): Promise<void> {
    if (this.connected) {
      return
    }

    this.shouldReconnect = true
    this.abortController = new AbortController()

    await this._connect()
  }

  /**
   * Internal connect implementation with optimized headers
   */
  private async _connect(): Promise<void> {
    const headers: Record<string, string> = {
      Accept: 'text/event-stream',
      'Cache-Control': 'no-cache',
      // Keep-alive connection for persistent streaming
      Connection: 'keep-alive',
      ...this.config.headers,
    }

    // Request compression if enabled
    if (this.enableCompression) {
      headers['Accept-Encoding'] = 'gzip, deflate'
    }

    if (this.config.apiKey) {
      headers['Authorization'] = `Bearer ${this.config.apiKey}`
    }

    if (this.lastEventId) {
      headers['Last-Event-ID'] = this.lastEventId
    }

    // Build URL with subscription filters
    const url = this.buildSubscriptionUrl()

    try {
      const requestInit: RequestInit = {
        method: 'GET',
        headers,
        keepalive: true,
      }
      if (this.abortController?.signal) {
        requestInit.signal = this.abortController.signal
      }
      const response = await this.fetchImpl(url, requestInit)

      if (!response.ok) {
        const errorBody = await response.text().catch(() => '')
        throw new ConnectionError(`SSE connection failed: ${response.status} ${response.statusText}${errorBody ? ` - ${errorBody}` : ''}`)
      }

      const contentType = response.headers.get('Content-Type')
      if (!contentType || !contentType.includes('text/event-stream')) {
        throw new ConnectionError(`Invalid Content-Type: expected text/event-stream, got ${contentType}`)
      }

      this.connected = true
      this.reconnectAttempt = 0

      // Resolve any pending subscriptions since SSE subscriptions are URL-based
      for (const [subscriptionId, { resolve }] of this.pendingSubscriptions) {
        resolve()
        this.pendingSubscriptions.delete(subscriptionId)
      }

      // Process the stream
      await this.processStream(response)
    } catch (error) {
      if (error instanceof Error && error.name === 'AbortError') {
        // Intentional abort - don't reconnect
        return
      }

      // Reject pending subscriptions on error
      for (const [subscriptionId, { reject }] of this.pendingSubscriptions) {
        reject(error instanceof Error ? error : new Error(String(error)))
        this.pendingSubscriptions.delete(subscriptionId)
      }

      throw error
    }
  }

  /**
   * Build URL with subscription parameters
   */
  private buildSubscriptionUrl(): string {
    const url = new URL(this.config.url)

    // Add table filter if there are active subscriptions
    const tables = Array.from(this.activeSubscriptions.values()).map(s => s.table)
    if (tables.length > 0) {
      url.searchParams.set('tables', tables.join(','))
    }

    // Add schema filter
    const schemas = new Set(Array.from(this.activeSubscriptions.values()).map(s => s.schema))
    if (schemas.size > 0) {
      url.searchParams.set('schemas', Array.from(schemas).join(','))
    }

    // Add subscription IDs
    const subscriptionIds = Array.from(this.activeSubscriptions.keys())
    if (subscriptionIds.length > 0) {
      url.searchParams.set('subscriptionIds', subscriptionIds.join(','))
    }

    // Add common options from first subscription
    const firstSub = Array.from(this.activeSubscriptions.values())[0]
    if (firstSub) {
      if (firstSub.options.events) {
        url.searchParams.set('events', firstSub.options.events.join(','))
      }
      if (firstSub.options.filter) {
        url.searchParams.set('filter', firstSub.options.filter)
      }
      if (firstSub.options.includeOldRow) {
        url.searchParams.set('includeOldRow', 'true')
      }
      if (firstSub.options.trackChangedColumns) {
        url.searchParams.set('trackChangedColumns', 'true')
      }
      if (firstSub.options.batchSize) {
        url.searchParams.set('batchSize', String(firstSub.options.batchSize))
      }
    }

    return url.toString()
  }

  /**
   * Process the SSE stream with optimized buffering and memory management
   */
  private async processStream(response: Response): Promise<void> {
    if (!response.body) {
      throw new ConnectionError('Response body is null')
    }

    const reader = response.body.getReader()
    let buffer = ''

    // Start adaptive heartbeat monitoring if enabled
    if (this.adaptiveHeartbeat) {
      this.startHeartbeatMonitor()
    }

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

        if (done) {
          // Stream ended - flush any remaining batched events
          this.flushEventBatch()
          this.stopHeartbeatMonitor()

          // Stream ended - schedule reconnect
          this.connected = false
          if (this.shouldReconnect) {
            this.scheduleReconnect()
          }
          return
        }

        // Track bytes received for metrics
        if (value) {
          this.metrics.bytesReceived += value.length
          this.currentBufferSize += value.length
        }

        // Use reusable decoder for memory efficiency
        buffer += this.decoder.decode(value, { stream: true })

        // Apply backpressure if buffer is too large
        if (this.currentBufferSize > this.maxBufferSize) {
          // Process events immediately to reduce memory pressure
          await this.processBufferWithBackpressure(buffer)
          buffer = ''
          this.currentBufferSize = 0
          continue
        }

        // Process complete events (separated by double newline)
        const events = buffer.split('\n\n')
        buffer = events.pop() || '' // Keep incomplete event in buffer

        // Update current buffer size estimate
        this.currentBufferSize = buffer.length

        for (const eventText of events) {
          if (eventText.trim()) {
            this.parseAndHandleEvent(eventText)
          }
        }
      }
    } catch (error) {
      this.connected = false
      this.flushEventBatch()
      this.stopHeartbeatMonitor()

      if (error instanceof Error && error.name === 'AbortError') {
        // Intentional abort
        return
      }

      // Emit error and schedule reconnect
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, error instanceof Error ? error : new Error(String(error)))
      }

      if (this.shouldReconnect) {
        this.scheduleReconnect()
      }
    }
  }

  /**
   * Process buffer when under backpressure - yields to event loop
   */
  private async processBufferWithBackpressure(buffer: string): Promise<void> {
    const events = buffer.split('\n\n')

    for (let i = 0; i < events.length; i++) {
      const eventText = events[i]
      if (eventText && eventText.trim()) {
        this.parseAndHandleEvent(eventText)
      }

      // Yield to event loop every 50 events to prevent blocking
      if (i > 0 && i % 50 === 0) {
        await new Promise(resolve => setTimeout(resolve, 0))
      }
    }
  }

  /**
   * Parse and handle an SSE event
   */
  private parseAndHandleEvent(eventText: string): void {
    let id: string | undefined
    let eventType: string | undefined
    let data: string | undefined

    const lines = eventText.split('\n')
    for (const line of lines) {
      if (line.startsWith(':')) {
        // Comment line - ignore
        continue
      }

      const colonIndex = line.indexOf(':')
      if (colonIndex === -1) {
        continue
      }

      const field = line.substring(0, colonIndex)
      // Skip the space after colon if present
      const value = line.substring(colonIndex + 1).trimStart()

      switch (field) {
        case 'id':
          id = value
          break
        case 'event':
          eventType = value
          break
        case 'data':
          data = data ? data + '\n' + value : value
          break
        case 'retry': {
          const retryMs = parseInt(value, 10)
          if (!isNaN(retryMs)) {
            this.config.retryMs = retryMs
          }
          break
        }
      }
    }

    // Update last event ID
    if (id) {
      this.lastEventId = id
    }

    if (!data) {
      return
    }

    // Parse data JSON
    let eventData: SseEventData
    try {
      eventData = JSON.parse(data)
    } catch (error) {
      // Emit parse error
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, new Error(`Failed to parse SSE event data: ${error instanceof Error ? error.message : String(error)}`))
      }
      return
    }

    // Handle different event types
    this.handleParsedEvent(id, eventType || eventData.type, eventData)
  }

  /**
   * Handle a parsed event with optional batching
   */
  private handleParsedEvent(id: string | undefined, eventType: string, eventData: SseEventData): void {
    const subscriptionId = eventData.subscriptionId || Array.from(this.activeSubscriptions.keys())[0]

    // Update event timing for adaptive heartbeat
    this.lastEventTime = Date.now()
    this.metrics.eventsReceived++

    switch (eventType) {
      case 'insert':
      case 'update':
      case 'delete':
      case 'truncate': {
        if (!subscriptionId) return

        const changeEvent: CDCChangeEvent = {
          id: id || `${Date.now()}-${Math.random()}`,
          lsn: id || '',
          operation: eventType.toUpperCase() as CDCOperation,
          table: eventData.table || '',
          schema: eventData.schema || 'public',
          timestamp: new Date(),
          xid: eventData.xid || 0,
          isLastInTransaction: eventData.isLastInTransaction || false,
        }

        if (eventType === 'update' && eventData.data) {
          const updateData = eventData.data as UpdateEventData
          changeEvent.oldRow = updateData.old ?? eventData.old
          changeEvent.newRow = updateData.new ?? eventData.new
        } else if (eventType === 'insert') {
          changeEvent.newRow = eventData.data
        } else if (eventType === 'delete') {
          changeEvent.oldRow = eventData.data
        }

        if (eventData.changedColumns) {
          changeEvent.changedColumns = eventData.changedColumns
        }

        // Use batching if enabled, otherwise emit immediately
        if (this.enableBatching) {
          this.addToBatch(subscriptionId, changeEvent)
        } else {
          this.eventHandler?.(subscriptionId, changeEvent)
        }
        break
      }

      case 'keepalive':
        // Update heartbeat timing - used for adaptive intervals
        this.updateHeartbeatFromServer()
        break

      case 'begin':
      case 'commit':
        // Transaction boundaries - flush batch on commit for consistency
        if (eventType === 'commit' && this.enableBatching) {
          this.flushEventBatch()
        }
        break

      case 'error':
        if (subscriptionId) {
          this.errorHandler?.(subscriptionId, new Error(eventData.message || eventData.error || 'Unknown error'))
        }
        break

      case 'subscribe_result':
        if (eventData.subscriptionId) {
          const pending = this.pendingSubscriptions.get(eventData.subscriptionId)
          if (pending) {
            if (eventData.success) {
              pending.resolve()
            } else {
              pending.reject(new Error(eventData.error || 'Subscription failed'))
            }
            this.pendingSubscriptions.delete(eventData.subscriptionId)
          }
        }
        break

      case 'unsubscribe_result':
        if (eventData.subscriptionId) {
          const pending = this.pendingUnsubscriptions.get(eventData.subscriptionId)
          if (pending) {
            pending.resolve()
            this.pendingUnsubscriptions.delete(eventData.subscriptionId)
          }
        }
        break
    }
  }

  // ========== Event Batching ==========

  /**
   * Add event to batch for coalesced delivery
   */
  private addToBatch(subscriptionId: string, event: CDCChangeEvent): void {
    this.eventBatch.push({
      subscriptionId,
      event,
      timestamp: Date.now(),
    })
    this.metrics.eventsBatched++

    // Flush if batch is full
    if (this.eventBatch.length >= this.batchSize) {
      this.flushEventBatch()
      return
    }

    // Start batch timer if not already running
    if (!this.batchTimer) {
      this.batchTimer = setTimeout(() => {
        this.flushEventBatch()
      }, this.batchTimeoutMs)
    }
  }

  /**
   * Flush all batched events to handlers
   */
  private flushEventBatch(): void {
    if (this.batchTimer) {
      clearTimeout(this.batchTimer)
      this.batchTimer = null
    }

    if (this.eventBatch.length === 0) {
      return
    }

    // Group events by subscription for efficient delivery
    const eventsBySubscription = new Map<string, CDCChangeEvent[]>()

    for (const { subscriptionId, event } of this.eventBatch) {
      const events = eventsBySubscription.get(subscriptionId) || []
      events.push(event)
      eventsBySubscription.set(subscriptionId, events)
    }

    // Emit events grouped by subscription
    for (const [subscriptionId, events] of eventsBySubscription) {
      for (const event of events) {
        this.eventHandler?.(subscriptionId, event)
      }
    }

    this.metrics.batchesFlushed++
    this.eventBatch = []
  }

  // ========== Adaptive Heartbeat ==========

  /**
   * Start monitoring connection health with adaptive intervals
   */
  private startHeartbeatMonitor(): void {
    if (!this.adaptiveHeartbeat) return

    this.stopHeartbeatMonitor()
    this.lastEventTime = Date.now()
    this.eventRateWindow = []

    this.heartbeatTimer = setInterval(() => {
      this.checkConnectionHealth()
    }, this.currentHeartbeatMs)
  }

  /**
   * Stop the heartbeat monitor
   */
  private stopHeartbeatMonitor(): void {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer)
      this.heartbeatTimer = null
    }
  }

  /**
   * Check connection health and adjust heartbeat interval
   */
  private checkConnectionHealth(): void {
    const now = Date.now()
    const timeSinceLastEvent = now - this.lastEventTime

    // If no events for 3x the current heartbeat interval, connection may be stale
    if (timeSinceLastEvent > this.currentHeartbeatMs * 3) {
      // Emit warning - connection might be stale
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, new Error('Connection appears stale - no events received'))
      }
    }

    // Adapt heartbeat interval based on event rate
    this.adaptHeartbeatInterval()
  }

  /**
   * Adapt heartbeat interval based on observed event rate
   */
  private adaptHeartbeatInterval(): void {
    // Track event rate in sliding window (last 10 samples)
    this.eventRateWindow.push(this.metrics.eventsReceived)
    if (this.eventRateWindow.length > 10) {
      this.eventRateWindow.shift()
    }

    // Calculate events per second
    const recentEvents = this.eventRateWindow.length > 1
      ? this.eventRateWindow[this.eventRateWindow.length - 1]! - this.eventRateWindow[0]!
      : 0
    const eventsPerSecond = recentEvents / (this.eventRateWindow.length * (this.currentHeartbeatMs / 1000))

    // High event rate = shorter heartbeat (faster detection of issues)
    // Low event rate = longer heartbeat (reduce overhead)
    let newInterval: number

    if (eventsPerSecond > 100) {
      // High throughput - use minimum interval
      newInterval = this.minHeartbeatMs
    } else if (eventsPerSecond < 1) {
      // Low throughput - increase interval
      newInterval = Math.min(
        this.currentHeartbeatMs * DEFAULTS.heartbeatScaleFactor,
        this.maxHeartbeatMs
      )
    } else {
      // Moderate throughput - decrease interval slightly
      newInterval = Math.max(
        this.currentHeartbeatMs / DEFAULTS.heartbeatScaleFactor,
        this.minHeartbeatMs
      )
    }

    // Update interval if changed significantly (>10% difference)
    if (Math.abs(newInterval - this.currentHeartbeatMs) / this.currentHeartbeatMs > 0.1) {
      this.currentHeartbeatMs = Math.round(newInterval)
      this.stopHeartbeatMonitor()
      this.startHeartbeatMonitor()
    }
  }

  /**
   * Update heartbeat timing when server sends keepalive
   */
  private updateHeartbeatFromServer(): void {
    this.lastEventTime = Date.now()
  }

  /**
   * Schedule reconnection with exponential backoff and jitter
   *
   * Uses a decorrelated jitter algorithm for optimal reconnection behavior:
   * - Exponential backoff: delay = retryMs * 1.5^attempt
   * - Jitter: adds random variation to prevent thundering herd
   * - Capped at maxReconnectDelay to prevent excessive waits
   */
  private scheduleReconnect(): void {
    if (!this.shouldReconnect) {
      return
    }

    this.reconnectAttempt++
    this.metrics.reconnections++

    if (this.reconnectAttempt > DEFAULTS.maxReconnectAttempts) {
      const error = new ConnectionError('Max reconnection attempts exceeded')
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, error)
      }
      this.closeHandler?.()
      return
    }

    // Calculate base delay with exponential backoff
    const baseDelay = this.config.retryMs * Math.pow(1.5, this.reconnectAttempt - 1)

    // Add jitter: random value between 0 and 50% of base delay
    // This helps prevent thundering herd when many clients reconnect
    const jitter = Math.random() * baseDelay * 0.5

    // Final delay with cap at 60 seconds
    const maxDelay = 60000
    const delay = Math.min(baseDelay + jitter, maxDelay)

    this.reconnectTimer = setTimeout(async () => {
      this.reconnectHandler?.(this.reconnectAttempt, this.lastEventId ?? undefined)

      try {
        this.abortController = new AbortController()
        await this._connect()
      } catch (error) {
        // Will schedule another reconnect
        if (this.shouldReconnect) {
          this.scheduleReconnect()
        }
      }
    }, delay)
  }

  /**
   * Disconnect from the CDC stream
   */
  async disconnect(): Promise<void> {
    this.shouldReconnect = false

    // Clean up all timers
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer)
      this.reconnectTimer = null
    }

    if (this.batchTimer) {
      clearTimeout(this.batchTimer)
      this.batchTimer = null
    }

    this.stopHeartbeatMonitor()

    // Flush any remaining batched events before disconnecting
    this.flushEventBatch()

    if (this.abortController) {
      this.abortController.abort()
      this.abortController = null
    }

    this.connected = false
    this.activeSubscriptions.clear()

    // Reset buffer state
    this.currentBufferSize = 0

    // Reject all pending requests
    for (const [, { reject }] of this.pendingSubscriptions) {
      reject(new ConnectionError('Connection closed'))
    }
    this.pendingSubscriptions.clear()

    for (const [, { resolve }] of this.pendingUnsubscriptions) {
      resolve() // Unsubscriptions resolve on disconnect
    }
    this.pendingUnsubscriptions.clear()

    this.closeHandler?.()
  }

  /**
   * Get transport performance metrics
   */
  getMetrics(): Readonly<typeof this.metrics> {
    return { ...this.metrics }
  }

  /**
   * Reset performance metrics
   */
  resetMetrics(): void {
    this.metrics = {
      eventsReceived: 0,
      eventsBatched: 0,
      batchesFlushed: 0,
      reconnections: 0,
      bytesReceived: 0,
    }
  }

  /**
   * Get current heartbeat interval (useful for monitoring)
   */
  getCurrentHeartbeatMs(): number {
    return this.currentHeartbeatMs
  }

  /**
   * Check if connected
   */
  isConnected(): boolean {
    return this.connected
  }

  /**
   * Get the last event ID received
   */
  getLastEventId(): string | null {
    return this.lastEventId
  }

  /**
   * Subscribe to a table
   * For SSE, this adds to the subscription set and reconnects
   */
  async subscribe(
    subscriptionId: string,
    table: string,
    schema: string,
    options: CDCSubscriptionOptions
  ): Promise<void> {
    // Store the subscription
    this.activeSubscriptions.set(subscriptionId, { table, schema, options })

    // If already connected, need to reconnect with new subscription
    if (this.connected) {
      // Create a pending subscription promise that resolves after reconnect
      const promise = new Promise<void>((resolve, reject) => {
        this.pendingSubscriptions.set(subscriptionId, { resolve, reject })
      })

      // Reconnect with updated subscriptions
      if (this.abortController) {
        this.abortController.abort()
      }
      this.abortController = new AbortController()
      this._connect().catch(() => {
        // Error handled in _connect
      })

      return promise
    } else {
      // Not connected yet - subscription will be included when connecting
      return new Promise<void>((resolve, reject) => {
        this.pendingSubscriptions.set(subscriptionId, { resolve, reject })
        resolve() // Resolve immediately for pre-connect subscriptions
      })
    }
  }

  /**
   * Unsubscribe from a subscription
   * For SSE, this removes from the subscription set and reconnects
   */
  async unsubscribe(subscriptionId: string): Promise<void> {
    this.activeSubscriptions.delete(subscriptionId)

    // If no more subscriptions, just disconnect
    if (this.activeSubscriptions.size === 0) {
      await this.disconnect()
      return
    }

    // Otherwise reconnect with remaining subscriptions
    if (this.connected) {
      return new Promise<void>((resolve, reject) => {
        this.pendingUnsubscriptions.set(subscriptionId, { resolve, reject })

        // Reconnect with updated subscriptions
        if (this.abortController) {
          this.abortController.abort()
        }
        this.abortController = new AbortController()
        this._connect().catch(() => {
          // Error handled in _connect
        })

        // Resolve immediately for SSE since unsubscription is URL-based
        setTimeout(() => {
          const pending = this.pendingUnsubscriptions.get(subscriptionId)
          if (pending) {
            pending.resolve()
            this.pendingUnsubscriptions.delete(subscriptionId)
          }
        }, 100)
      })
    }
  }

  /**
   * Set the event handler
   */
  onEvent(handler: (subscriptionId: string, event: CDCChangeEvent) => void): void {
    this.eventHandler = handler
  }

  /**
   * Set the error handler
   */
  onError(handler: (subscriptionId: string, error: Error) => void): void {
    this.errorHandler = handler
  }

  /**
   * Set the reconnect handler
   */
  onReconnect(handler: (attempt: number, lastLsn: string | undefined) => void): void {
    this.reconnectHandler = handler
  }

  /**
   * Set the close handler
   */
  onClose(handler: () => void): void {
    this.closeHandler = handler
  }
}

/**
 * Create a CDC SSE transport
 */
export function createCdcSseTransport(config: CDCSseTransportConfig): CDCSseTransport {
  return new CDCSseTransport(config)
}