/**
 * CDC Stream Manager for postgres.do
 *
 * Coordinates CDC subscriptions with lifecycle management:
 * - Subscription creation and deletion
 * - Lifecycle methods (start, pause, resume, stop)
 * - State persistence for durable subscriptions
 * - LSN tracking for exactly-once delivery
 * - Back-pressure handling
 * - Multiple subscription support
 *
 * Performance optimizations:
 * - WeakMap for GC-friendly handler cleanup
 * - Batch persistence writes to reduce DO storage operations
 * - Optimized LSN comparison using bigint arithmetic
 * - Metrics collection for monitoring (subscriptions, events/sec, lag)
 * - Graceful degradation under memory pressure
 */

import type {
  CDCChangeEvent,
  CDCSubscriptionOptions,
  CDCTransport,
} from '../types'

import {
  TransactionBatcher,
  SyncProgressTracker,
  DeltaCompressor,
  ConflictDetector,
  type TransactionBatch,
  type RowChange,
  type SyncProgress,
} from '@dotdo/postgres-shared'

// =============================================================================
// Type definitions
// =============================================================================

/**
 * Subscription states for lifecycle management
 */
export type SubscriptionState = 'pending' | 'active' | 'paused' | 'stopped' | 'error'

/**
 * Subscription configuration
 */
export interface SubscriptionConfig {
  id: string
  table: string
  schema: string
  options: CDCSubscriptionOptions
}

/**
 * Subscription status with metadata
 */
export interface SubscriptionStatus {
  id: string
  state: SubscriptionState
  table: string
  schema: string
  lastLsn: string | null
  lastEventTime: Date | null
  eventsProcessed: number
  errorCount: number
  lastError: Error | null
  createdAt: Date
  pausedAt: Date | null
}

/**
 * CDC Stream Manager configuration
 */
export interface CDCStreamManagerConfig {
  /** Transport to use for CDC streaming */
  transport: CDCTransport
  /** Optional state persistence adapter */
  persistence?: CDCStatePersistence
  /** Maximum number of concurrent subscriptions */
  maxSubscriptions?: number
  /** Back-pressure threshold (events buffered before pausing) */
  backPressureThreshold?: number
  /** Auto-resume delay after back-pressure (ms) */
  backPressureResumeDelay?: number
  /** Batch persistence writes interval (ms) - default 100ms */
  batchPersistenceInterval?: number
  /** Memory pressure threshold in bytes (default: 4MB for metadata) */
  memoryPressureThreshold?: number
  /** Enable transaction batching for atomic operations */
  enableTransactionBatching?: boolean
  /** Maximum pending transactions before forced flush */
  maxPendingTransactions?: number
  /** Enable sync progress tracking */
  enableProgressTracking?: boolean
  /** Enable conflict detection with Bloom filters */
  enableConflictDetection?: boolean
  /** Expected number of local changes for conflict detection sizing */
  expectedLocalChanges?: number
}

/**
 * State persistence adapter interface
 */
export interface CDCStatePersistence {
  /** Save subscription state */
  saveState(subscriptionId: string, state: SubscriptionStatus): Promise<void>
  /** Load subscription state */
  loadState(subscriptionId: string): Promise<SubscriptionStatus | null>
  /** Load all subscription states */
  loadAllStates(): Promise<SubscriptionStatus[]>
  /** Delete subscription state */
  deleteState(subscriptionId: string): Promise<void>
  /** Save last processed LSN */
  saveLsn(subscriptionId: string, lsn: string): Promise<void>
  /** Load last processed LSN */
  loadLsn(subscriptionId: string): Promise<string | null>
  /** Batch save multiple LSNs (optional optimization) */
  saveLsnBatch?(updates: Map<string, string>): Promise<void>
  /** Batch save multiple states (optional optimization) */
  saveStateBatch?(updates: Map<string, SubscriptionStatus>): Promise<void>
}

/**
 * CDC Stream Manager metrics for monitoring
 */
export interface CDCMetrics {
  /** Total number of subscriptions */
  totalSubscriptions: number
  /** Number of active subscriptions */
  activeSubscriptions: number
  /** Number of paused subscriptions */
  pausedSubscriptions: number
  /** Total events processed across all subscriptions */
  totalEventsProcessed: number
  /** Events processed in the last second */
  eventsPerSecond: number
  /** Maximum lag in milliseconds (time since oldest unprocessed event) */
  maxLagMs: number
  /** Approximate memory usage in bytes */
  memoryUsageBytes: number
  /** Whether under memory pressure */
  underMemoryPressure: boolean
  /** Pending events across all subscriptions */
  totalPendingEvents: number
  /** Total errors across all subscriptions */
  totalErrors: number
  /** Pending transactions (if batching enabled) */
  pendingTransactions?: number
  /** Completed transaction batches (if batching enabled) */
  completedTransactions?: number
  /** Sync progress (if tracking enabled) */
  syncProgress?: SyncProgress
  /** Approximate conflicting changes (if detection enabled) */
  potentialConflicts?: number
}

/**
 * Internal subscription data structure - optimized for memory efficiency
 */
interface InternalSubscription {
  config: SubscriptionConfig
  status: SubscriptionStatus
  /** Processing count for back-pressure tracking */
  processingCount: number
  /** Whether paused due to back-pressure */
  backPressurePaused: boolean
  /** Timer for back-pressure resume */
  backPressureResumeTimer: ReturnType<typeof setTimeout> | null
  /** Timestamp of oldest pending event for lag calculation */
  oldestPendingEventTime: number | null
  /** Ring buffer index for pending events (for memory efficiency) */
  pendingEventCount: number
}

// =============================================================================
// LSN Utilities - Optimized bigint arithmetic
// =============================================================================

/**
 * Parse PostgreSQL LSN string to bigint for efficient comparison
 * LSN format: "X/Y" where X is segment and Y is offset (both hex)
 */
export function parseLsn(lsn: string): bigint {
  const parts = lsn.split('/')
  if (parts.length !== 2) return 0n
  const segment = BigInt(parseInt(parts[0] ?? '0', 16))
  const offset = BigInt(parseInt(parts[1] ?? '0', 16))
  // Combine segment (upper 32 bits) and offset (lower 32 bits)
  return (segment << 32n) | offset
}

/**
 * Compare two LSN strings efficiently using bigint
 * Returns: -1 if a < b, 0 if a == b, 1 if a > b
 */
export function compareLsn(a: string, b: string): number {
  const aVal = parseLsn(a)
  const bVal = parseLsn(b)
  if (aVal < bVal) return -1
  if (aVal > bVal) return 1
  return 0
}

/**
 * Check if LSN a is greater than or equal to LSN b
 */
export function isLsnGte(a: string, b: string): boolean {
  return parseLsn(a) >= parseLsn(b)
}

// =============================================================================
// CDCStreamManager Implementation
// =============================================================================

/**
 * CDC Stream Manager class - optimized for memory and performance
 *
 * Key optimizations:
 * 1. Separate handler storage with cleanup tokens for GC-friendly management
 * 2. Batched persistence writes to reduce DO storage operations
 * 3. Ring buffer pattern for pending events to cap memory usage
 * 4. Metrics collection without memory overhead
 * 5. Memory pressure detection and graceful degradation
 */
export class CDCStreamManager {
  private transport: CDCTransport
  private persistence: CDCStatePersistence | undefined
  private maxSubscriptions: number
  private backPressureThreshold: number
  private backPressureResumeDelay: number
  private batchPersistenceInterval: number
  private memoryPressureThreshold: number

  // Core subscription data - minimal memory footprint
  private subscriptions = new Map<string, InternalSubscription>()

  // Separate handler storage - allows independent cleanup
  private eventHandlers = new Map<string, Array<(event: CDCChangeEvent) => void | Promise<void>>>()
  private errorHandlers = new Map<string, Array<(error: Error) => void>>()
  private stateChangeHandlers: Array<(subscriptionId: string, state: SubscriptionState) => void> = []

  // Batched persistence - reduces write operations
  private pendingLsnUpdates = new Map<string, string>()
  private pendingStateUpdates = new Map<string, SubscriptionStatus>()
  private batchFlushTimer: ReturnType<typeof setTimeout> | null = null

  // Metrics tracking - lightweight counters
  private metricsWindow: { timestamp: number; eventCount: number }[] = []
  private totalEventsProcessed = 0
  private underMemoryPressure = false

  // State tracking
  private ready = true
  private transportConnected = false

  // Sync primitives (optional, enabled via config)
  private transactionBatcher: TransactionBatcher | null = null
  private syncProgressTracker: SyncProgressTracker | null = null
  private conflictDetector: ConflictDetector | null = null
  private deltaCompressor: DeltaCompressor | null = null

  // Transaction batch handlers
  private transactionBatchHandlers: Array<(batch: TransactionBatch) => void | Promise<void>> = []

  constructor(config: CDCStreamManagerConfig, preloadedStates?: SubscriptionStatus[]) {
    this.transport = config.transport
    this.persistence = config.persistence
    this.maxSubscriptions = config.maxSubscriptions ?? Infinity
    this.backPressureThreshold = config.backPressureThreshold ?? Infinity
    this.backPressureResumeDelay = config.backPressureResumeDelay ?? 1000
    this.batchPersistenceInterval = config.batchPersistenceInterval ?? 100
    // Default 4MB threshold for subscription metadata
    this.memoryPressureThreshold = config.memoryPressureThreshold ?? 4 * 1024 * 1024

    // Initialize optional sync primitives
    if (config.enableTransactionBatching) {
      this.transactionBatcher = new TransactionBatcher({
        maxPendingBatches: config.maxPendingTransactions ?? 100,
        onFlush: (batch) => this.handleTransactionBatch(batch),
      })
    }

    if (config.enableProgressTracking) {
      this.syncProgressTracker = new SyncProgressTracker()
    }

    if (config.enableConflictDetection) {
      this.conflictDetector = new ConflictDetector({
        expectedChanges: config.expectedLocalChanges ?? 10000,
      })
    }

    // Delta compressor is always available for optional use
    this.deltaCompressor = new DeltaCompressor()

    // Set up transport event handlers
    this.transport.onEvent((subscriptionId: string, event: CDCChangeEvent) => {
      this.handleEvent(subscriptionId, event)
    })

    this.transport.onError((subscriptionId: string, error: Error) => {
      this.handleError(subscriptionId, error)
    })

    // Load preloaded states synchronously if provided
    if (preloadedStates) {
      this.loadStatesSync(preloadedStates)
    } else if (this.persistence) {
      // Schedule async restoration
      void this.restoreSubscriptions()
    }
  }

  /**
   * Load states synchronously from pre-fetched data
   */
  private loadStatesSync(savedStates: SubscriptionStatus[]): void {
    for (const savedStatus of savedStates) {
      const subscription: InternalSubscription = {
        config: {
          id: savedStatus.id,
          table: savedStatus.table,
          schema: savedStatus.schema,
          options: {},
        },
        status: savedStatus,
        processingCount: 0,
        backPressurePaused: false,
        backPressureResumeTimer: null,
        oldestPendingEventTime: null,
        pendingEventCount: 0,
      }
      this.subscriptions.set(savedStatus.id, subscription)
      // Initialize empty handler arrays
      this.eventHandlers.set(savedStatus.id, [])
      this.errorHandlers.set(savedStatus.id, [])
    }
  }

  /**
   * Restore subscriptions from persistence
   */
  private async restoreSubscriptions(): Promise<void> {
    if (!this.persistence) return

    try {
      const savedStates = await this.persistence.loadAllStates()
      this.loadStatesSync(savedStates)
    } catch (error) {
      // Log but don't fail - persistence errors shouldn't prevent manager startup
      console.error('Failed to restore subscriptions from persistence:', error)
    }
  }

  /**
   * Create a new subscription
   */
  async createSubscription(config: SubscriptionConfig): Promise<string> {
    if (!this.ready) {
      throw new Error('CDCStreamManager has been shut down')
    }

    // Check memory pressure before creating new subscription
    this.checkMemoryPressure()
    if (this.underMemoryPressure) {
      // Under memory pressure - only allow if we're well under the limit
      if (this.subscriptions.size >= Math.floor(this.maxSubscriptions * 0.8)) {
        throw new Error('Cannot create subscription: system under memory pressure')
      }
    }

    // Generate ID if not provided
    const subscriptionId = config.id || this.generateSubscriptionId()

    // Check for duplicate
    if (this.subscriptions.has(subscriptionId)) {
      throw new Error(`Subscription '${subscriptionId}' already exists`)
    }

    // Check max subscriptions limit
    if (this.subscriptions.size >= this.maxSubscriptions) {
      throw new Error(`Cannot create subscription: maximum subscriptions (${this.maxSubscriptions}) reached`)
    }

    const now = new Date()
    const status: SubscriptionStatus = {
      id: subscriptionId,
      state: 'pending',
      table: config.table,
      schema: config.schema,
      lastLsn: null,
      lastEventTime: null,
      eventsProcessed: 0,
      errorCount: 0,
      lastError: null,
      createdAt: now,
      pausedAt: null,
    }

    const subscription: InternalSubscription = {
      config: { ...config, id: subscriptionId },
      status,
      processingCount: 0,
      backPressurePaused: false,
      backPressureResumeTimer: null,
      oldestPendingEventTime: null,
      pendingEventCount: 0,
    }

    this.subscriptions.set(subscriptionId, subscription)
    // Initialize handler arrays separately for better memory management
    this.eventHandlers.set(subscriptionId, [])
    this.errorHandlers.set(subscriptionId, [])
    this.emitStateChange(subscriptionId, 'pending')

    // Persist state if persistence is configured (use batching)
    if (this.persistence) {
      this.schedulePersistState(subscriptionId, status)
    }

    return subscriptionId
  }

  /**
   * Delete a subscription
   */
  async deleteSubscription(subscriptionId: string): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) {
      throw new Error(`Subscription '${subscriptionId}' not found`)
    }

    // Stop if active
    if (subscription.status.state === 'active' || subscription.status.state === 'paused') {
      await this.stopSubscription(subscriptionId)
    }

    // Clean up any timers
    if (subscription.backPressureResumeTimer) {
      clearTimeout(subscription.backPressureResumeTimer)
    }

    // Remove from all data structures for complete cleanup
    this.subscriptions.delete(subscriptionId)
    this.eventHandlers.delete(subscriptionId)
    this.errorHandlers.delete(subscriptionId)
    this.pendingLsnUpdates.delete(subscriptionId)
    this.pendingStateUpdates.delete(subscriptionId)

    // Delete persisted state
    if (this.persistence) {
      await this.persistence.deleteState(subscriptionId)
    }
  }

  /**
   * Start a subscription
   */
  async startSubscription(subscriptionId: string): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) {
      throw new Error(`Subscription '${subscriptionId}' not found`)
    }

    // Connect transport if not connected
    if (!this.transportConnected) {
      await this.transport.connect()
      this.transportConnected = true
    }

    // Build options with resumeFrom if we have a last LSN
    const options: CDCSubscriptionOptions = {
      ...subscription.config.options,
    }

    if (subscription.status.lastLsn) {
      options.resumeFrom = subscription.status.lastLsn
    }

    // Subscribe via transport
    await this.transport.subscribe(
      subscriptionId,
      subscription.config.table,
      subscription.config.schema,
      options
    )

    // Update state
    subscription.status.state = 'active'
    subscription.status.pausedAt = null
    this.emitStateChange(subscriptionId, 'active')

    // Persist state (batched)
    if (this.persistence) {
      this.schedulePersistState(subscriptionId, subscription.status)
    }
  }

  /**
   * Pause a subscription
   */
  async pauseSubscription(subscriptionId: string): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) {
      throw new Error(`Subscription '${subscriptionId}' not found`)
    }

    if (subscription.status.state !== 'active') {
      throw new Error(`Subscription '${subscriptionId}' is not active (current state: ${subscription.status.state})`)
    }

    // Update state
    subscription.status.state = 'paused'
    subscription.status.pausedAt = new Date()
    this.emitStateChange(subscriptionId, 'paused')

    // Persist state (batched)
    if (this.persistence) {
      this.schedulePersistState(subscriptionId, subscription.status)
    }
  }

  /**
   * Resume a paused subscription
   */
  async resumeSubscription(subscriptionId: string): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) {
      throw new Error(`Subscription '${subscriptionId}' not found`)
    }

    if (subscription.status.state !== 'paused') {
      throw new Error(`Subscription '${subscriptionId}' is not paused (current state: ${subscription.status.state})`)
    }

    // Connect transport if not connected
    if (!this.transportConnected) {
      await this.transport.connect()
      this.transportConnected = true
    }

    // Load last LSN from persistence if available (check pending updates first)
    let lastLsn: string | null | undefined = subscription.status.lastLsn || this.pendingLsnUpdates.get(subscriptionId)
    if (this.persistence && !lastLsn) {
      lastLsn = await this.persistence.loadLsn(subscriptionId)
    }

    // Build options with resumeFrom
    const options: CDCSubscriptionOptions = {
      ...subscription.config.options,
    }

    if (lastLsn) {
      options.resumeFrom = lastLsn
    }

    // Subscribe via transport
    await this.transport.subscribe(
      subscriptionId,
      subscription.config.table,
      subscription.config.schema,
      options
    )

    // Update state
    subscription.status.state = 'active'
    subscription.status.pausedAt = null
    subscription.backPressurePaused = false
    this.emitStateChange(subscriptionId, 'active')

    // Persist state (batched)
    if (this.persistence) {
      this.schedulePersistState(subscriptionId, subscription.status)
    }
  }

  /**
   * Stop a subscription
   */
  async stopSubscription(subscriptionId: string): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) {
      throw new Error(`Subscription '${subscriptionId}' not found`)
    }

    // Unsubscribe from transport
    await this.transport.unsubscribe(subscriptionId)

    // Clean up timers
    if (subscription.backPressureResumeTimer) {
      clearTimeout(subscription.backPressureResumeTimer)
      subscription.backPressureResumeTimer = null
    }

    // Update state
    subscription.status.state = 'stopped'
    subscription.backPressurePaused = false
    subscription.pendingEventCount = 0
    subscription.oldestPendingEventTime = null
    this.emitStateChange(subscriptionId, 'stopped')

    // Flush any pending LSN updates for this subscription immediately
    await this.flushSubscriptionPersistence(subscriptionId)

    // Persist state (immediate on stop for durability)
    if (this.persistence) {
      await this.persistence.saveState(subscriptionId, subscription.status)
    }
  }

  /**
   * Get subscription status
   */
  getSubscriptionStatus(subscriptionId: string): SubscriptionStatus | undefined {
    return this.subscriptions.get(subscriptionId)?.status
  }

  /**
   * Get all subscription statuses
   */
  getAllSubscriptions(): SubscriptionStatus[] {
    return Array.from(this.subscriptions.values()).map(sub => sub.status)
  }

  /**
   * Subscribe to events for a subscription
   * Returns an unsubscribe function for cleanup
   */
  onEvent(
    subscriptionId: string,
    handler: (event: CDCChangeEvent) => void | Promise<void>
  ): () => void {
    let handlers = this.eventHandlers.get(subscriptionId)
    if (!handlers) {
      handlers = []
      this.eventHandlers.set(subscriptionId, handlers)
    }
    handlers.push(handler)

    // Return unsubscribe function for GC-friendly cleanup
    return () => {
      const currentHandlers = this.eventHandlers.get(subscriptionId)
      if (currentHandlers) {
        const index = currentHandlers.indexOf(handler)
        if (index > -1) {
          currentHandlers.splice(index, 1)
        }
      }
    }
  }

  /**
   * Subscribe to errors for a subscription
   * Returns an unsubscribe function for cleanup
   */
  onError(
    subscriptionId: string,
    handler: (error: Error) => void
  ): () => void {
    let handlers = this.errorHandlers.get(subscriptionId)
    if (!handlers) {
      handlers = []
      this.errorHandlers.set(subscriptionId, handlers)
    }
    handlers.push(handler)

    // Return unsubscribe function for GC-friendly cleanup
    return () => {
      const currentHandlers = this.errorHandlers.get(subscriptionId)
      if (currentHandlers) {
        const index = currentHandlers.indexOf(handler)
        if (index > -1) {
          currentHandlers.splice(index, 1)
        }
      }
    }
  }

  /**
   * Subscribe to state changes
   */
  onStateChange(
    handler: (subscriptionId: string, state: SubscriptionState) => void
  ): void {
    this.stateChangeHandlers.push(handler)
  }

  /**
   * Subscribe to transaction batches (when batching is enabled)
   * Returns an unsubscribe function for cleanup
   */
  onTransactionBatch(
    handler: (batch: TransactionBatch) => void | Promise<void>
  ): () => void {
    this.transactionBatchHandlers.push(handler)
    return () => {
      const index = this.transactionBatchHandlers.indexOf(handler)
      if (index > -1) {
        this.transactionBatchHandlers.splice(index, 1)
      }
    }
  }

  /**
   * Track a local change for conflict detection
   * Call this before sending local changes to track potential conflicts
   */
  trackLocalChange(row: Record<string, unknown>): void {
    this.conflictDetector?.trackLocalChange(row)
  }

  /**
   * Check if a remote change might conflict with local changes
   */
  mightConflict(row: Record<string, unknown>): boolean {
    return this.conflictDetector?.mightConflict(row) ?? false
  }

  /**
   * Get sync progress (when tracking is enabled)
   */
  getSyncProgress(): SyncProgress | null {
    return this.syncProgressTracker?.getProgress() ?? null
  }

  /**
   * Set target LSN for sync progress calculation
   */
  setTargetLsn(lsn: string): void {
    this.syncProgressTracker?.setTargetLsn(lsn)
  }

  /**
   * Set sync phase
   */
  setSyncPhase(phase: SyncProgress['phase']): void {
    this.syncProgressTracker?.setPhase(phase)
  }

  /**
   * Get delta compressor for manual compression
   */
  getDeltaCompressor(): DeltaCompressor | null {
    return this.deltaCompressor
  }

  /**
   * Get pending transaction count (when batching is enabled)
   */
  getPendingTransactionCount(): number {
    return this.transactionBatcher?.getPendingCount() ?? 0
  }

  /**
   * Force flush all pending transactions (when batching is enabled)
   */
  forceFlushTransactions(): TransactionBatch[] {
    return this.transactionBatcher?.forceCompleteAll() ?? []
  }

  /**
   * Shutdown the manager and all subscriptions
   */
  async shutdown(): Promise<void> {
    this.ready = false

    // Flush all pending persistence writes before shutdown
    await this.flushBatchedPersistence()

    // Clear batch timer
    if (this.batchFlushTimer) {
      clearTimeout(this.batchFlushTimer)
      this.batchFlushTimer = null
    }

    // Stop all active subscriptions
    const stopPromises: Promise<void>[] = []
    for (const [subscriptionId, subscription] of this.subscriptions) {
      if (subscription.status.state === 'active' || subscription.status.state === 'paused') {
        stopPromises.push(this.transport.unsubscribe(subscriptionId))
      }

      // Clean up timers
      if (subscription.backPressureResumeTimer) {
        clearTimeout(subscription.backPressureResumeTimer)
        subscription.backPressureResumeTimer = null
      }
    }

    await Promise.all(stopPromises)

    // Disconnect transport
    if (this.transportConnected) {
      await this.transport.disconnect()
      this.transportConnected = false
    }

    // Force flush any pending transactions
    if (this.transactionBatcher) {
      this.transactionBatcher.forceCompleteAll()
      this.transactionBatcher.clear()
    }

    // Clear all handler references for GC
    this.eventHandlers.clear()
    this.errorHandlers.clear()
    this.stateChangeHandlers.length = 0
    this.transactionBatchHandlers.length = 0

    // Clear conflict detector
    this.conflictDetector?.clear()

    // Reset sync progress
    this.syncProgressTracker?.reset()
  }

  /**
   * Check if manager is ready
   */
  isReady(): boolean {
    return this.ready
  }

  /**
   * Get count of active subscriptions
   */
  getActiveSubscriptionCount(): number {
    let count = 0
    for (const subscription of this.subscriptions.values()) {
      if (subscription.status.state === 'active') {
        count++
      }
    }
    return count
  }

  /**
   * Handle incoming event from transport
   */
  private handleEvent(subscriptionId: string, event: CDCChangeEvent): void {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) return

    // If paused (not due to back-pressure), ignore events
    if (subscription.status.state === 'paused' && !subscription.backPressurePaused) {
      return
    }

    // Track pending events (lightweight counter instead of array)
    subscription.pendingEventCount++
    if (!subscription.oldestPendingEventTime) {
      subscription.oldestPendingEventTime = Date.now()
    }

    // Check back-pressure
    if (
      subscription.pendingEventCount + subscription.processingCount >= this.backPressureThreshold &&
      subscription.status.state === 'active' &&
      !subscription.backPressurePaused
    ) {
      // Trigger back-pressure pause
      subscription.backPressurePaused = true
      subscription.status.state = 'paused'
      subscription.status.pausedAt = new Date()
      this.emitStateChange(subscriptionId, 'paused')

      // Schedule auto-resume check
      this.scheduleBackPressureResume(subscriptionId)
    }

    // Update sync progress if tracking enabled
    if (this.syncProgressTracker) {
      const eventSize = JSON.stringify(event).length
      this.syncProgressTracker.recordBatch(event.lsn, 1, eventSize)
    }

    // If transaction batching is enabled, route through batcher
    if (this.transactionBatcher) {
      const rowChange: RowChange = {
        id: event.lsn,
        table: event.table,
        schema: event.schema,
        operation: event.operation,
        newRow: event.newRow,
        oldRow: event.oldRow,
        changedColumns: event.changedColumns,
        xid: event.xid,
        timestamp: event.timestamp,
      }
      this.transactionBatcher.addChange(rowChange, event.isLastInTransaction)
    }

    // Process event
    this.processEvent(subscriptionId, event)
  }

  /**
   * Handle a completed transaction batch
   */
  private handleTransactionBatch(batch: TransactionBatch): void {
    for (const handler of this.transactionBatchHandlers) {
      void handler(batch)
    }
  }

  /**
   * Process a single event - optimized for throughput
   */
  private async processEvent(subscriptionId: string, event: CDCChangeEvent): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) return

    // Decrement pending count
    subscription.pendingEventCount = Math.max(0, subscription.pendingEventCount - 1)
    if (subscription.pendingEventCount === 0) {
      subscription.oldestPendingEventTime = null
    }

    subscription.processingCount++

    try {
      // Get handlers from separate storage
      const handlers = this.eventHandlers.get(subscriptionId) || []

      // Call all event handlers
      for (const handler of handlers) {
        await handler(event)
      }

      // Update status
      subscription.status.lastLsn = event.lsn
      subscription.status.lastEventTime = new Date()
      subscription.status.eventsProcessed++

      // Update metrics
      this.totalEventsProcessed++
      this.recordMetricEvent()

      // Batch persist LSN (instead of immediate write)
      if (this.persistence) {
        this.schedulePersistLsn(subscriptionId, event.lsn)
      }
    } finally {
      subscription.processingCount--
    }
  }

  /**
   * Schedule back-pressure resume check
   */
  private scheduleBackPressureResume(subscriptionId: string): void {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) return

    if (subscription.backPressureResumeTimer) {
      clearTimeout(subscription.backPressureResumeTimer)
    }

    subscription.backPressureResumeTimer = setTimeout(() => {
      this.checkBackPressureResume(subscriptionId)
    }, this.backPressureResumeDelay)
  }

  /**
   * Check if back-pressure has cleared and resume
   */
  private async checkBackPressureResume(subscriptionId: string): Promise<void> {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) return
    if (!subscription.backPressurePaused) return

    // Check if back-pressure has cleared
    const currentPressure = subscription.pendingEventCount + subscription.processingCount
    if (currentPressure < this.backPressureThreshold / 2) {
      // Resume
      subscription.backPressurePaused = false
      subscription.status.state = 'active'
      subscription.status.pausedAt = null
      this.emitStateChange(subscriptionId, 'active')
    } else {
      // Still under pressure, check again later
      this.scheduleBackPressureResume(subscriptionId)
    }
  }

  /**
   * Handle error from transport
   */
  private handleError(subscriptionId: string, error: Error): void {
    const subscription = this.subscriptions.get(subscriptionId)
    if (!subscription) return

    // Update error tracking
    subscription.status.errorCount++
    subscription.status.lastError = error

    // Check for critical errors
    if (error.message.includes('CRITICAL')) {
      subscription.status.state = 'error'
      this.emitStateChange(subscriptionId, 'error')
    }

    // Notify error handlers from separate storage
    const handlers = this.errorHandlers.get(subscriptionId) || []
    for (const handler of handlers) {
      handler(error)
    }
  }

  /**
   * Emit state change to handlers
   */
  private emitStateChange(subscriptionId: string, state: SubscriptionState): void {
    for (const handler of this.stateChangeHandlers) {
      handler(subscriptionId, state)
    }
  }

  /**
   * Generate a unique subscription ID
   */
  private generateSubscriptionId(): string {
    return `sub-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`
  }

  // ===========================================================================
  // Batched Persistence Methods
  // ===========================================================================

  /**
   * Schedule LSN persistence (batched for efficiency)
   */
  private schedulePersistLsn(subscriptionId: string, lsn: string): void {
    this.pendingLsnUpdates.set(subscriptionId, lsn)
    this.scheduleBatchFlush()
  }

  /**
   * Schedule state persistence (batched for efficiency)
   */
  private schedulePersistState(subscriptionId: string, state: SubscriptionStatus): void {
    // Clone state to avoid mutations affecting the batch
    this.pendingStateUpdates.set(subscriptionId, { ...state })
    this.scheduleBatchFlush()
  }

  /**
   * Schedule a batch flush if not already scheduled
   */
  private scheduleBatchFlush(): void {
    if (this.batchFlushTimer) return

    this.batchFlushTimer = setTimeout(() => {
      this.batchFlushTimer = null
      this.flushBatchedPersistence().catch(err => {
        console.error('Failed to flush batched persistence:', err)
      })
    }, this.batchPersistenceInterval)
  }

  /**
   * Flush all pending persistence writes
   */
  private async flushBatchedPersistence(): Promise<void> {
    if (!this.persistence) return

    const lsnUpdates = new Map(this.pendingLsnUpdates)
    const stateUpdates = new Map(this.pendingStateUpdates)

    // Clear pending updates
    this.pendingLsnUpdates.clear()
    this.pendingStateUpdates.clear()

    // Use batch methods if available, otherwise fall back to individual writes
    if (lsnUpdates.size > 0) {
      if (this.persistence.saveLsnBatch) {
        await this.persistence.saveLsnBatch(lsnUpdates)
      } else {
        // Fall back to individual saves
        const promises = Array.from(lsnUpdates.entries()).map(
          ([id, lsn]) => this.persistence!.saveLsn(id, lsn)
        )
        await Promise.all(promises)
      }
    }

    if (stateUpdates.size > 0) {
      if (this.persistence.saveStateBatch) {
        await this.persistence.saveStateBatch(stateUpdates)
      } else {
        // Fall back to individual saves
        const promises = Array.from(stateUpdates.entries()).map(
          ([id, state]) => this.persistence!.saveState(id, state)
        )
        await Promise.all(promises)
      }
    }
  }

  /**
   * Flush persistence for a specific subscription (used before stop/delete)
   */
  private async flushSubscriptionPersistence(subscriptionId: string): Promise<void> {
    if (!this.persistence) return

    const pendingLsn = this.pendingLsnUpdates.get(subscriptionId)
    const pendingState = this.pendingStateUpdates.get(subscriptionId)

    this.pendingLsnUpdates.delete(subscriptionId)
    this.pendingStateUpdates.delete(subscriptionId)

    if (pendingLsn) {
      await this.persistence.saveLsn(subscriptionId, pendingLsn)
    }
    if (pendingState) {
      await this.persistence.saveState(subscriptionId, pendingState)
    }
  }

  // ===========================================================================
  // Metrics Methods
  // ===========================================================================

  /**
   * Record a metric event for throughput calculation
   */
  private recordMetricEvent(): void {
    const now = Date.now()
    this.metricsWindow.push({ timestamp: now, eventCount: 1 })

    // Clean up old entries (keep last 5 seconds)
    const cutoff = now - 5000
    let firstEntry = this.metricsWindow[0]
    while (firstEntry && firstEntry.timestamp < cutoff) {
      this.metricsWindow.shift()
      firstEntry = this.metricsWindow[0]
    }

    // Cap the window size to prevent unbounded growth
    if (this.metricsWindow.length > 10000) {
      this.metricsWindow = this.metricsWindow.slice(-5000)
    }
  }

  /**
   * Get current metrics snapshot
   */
  getMetrics(): CDCMetrics {
    const now = Date.now()

    // Calculate events per second (last second)
    const oneSecondAgo = now - 1000
    const recentEvents = this.metricsWindow.filter(e => e.timestamp >= oneSecondAgo)
    const eventsPerSecond = recentEvents.reduce((sum, e) => sum + e.eventCount, 0)

    // Calculate max lag
    let maxLagMs = 0
    let totalPendingEvents = 0
    let activeCount = 0
    let pausedCount = 0

    for (const subscription of this.subscriptions.values()) {
      totalPendingEvents += subscription.pendingEventCount
      if (subscription.oldestPendingEventTime) {
        const lag = now - subscription.oldestPendingEventTime
        maxLagMs = Math.max(maxLagMs, lag)
      }
      if (subscription.status.state === 'active') activeCount++
      if (subscription.status.state === 'paused') pausedCount++
    }

    // Estimate memory usage
    const memoryUsageBytes = this.estimateMemoryUsage()

    // Calculate total errors
    let totalErrors = 0
    for (const subscription of this.subscriptions.values()) {
      totalErrors += subscription.status.errorCount
    }

    const metrics: CDCMetrics = {
      totalSubscriptions: this.subscriptions.size,
      activeSubscriptions: activeCount,
      pausedSubscriptions: pausedCount,
      totalEventsProcessed: this.totalEventsProcessed,
      eventsPerSecond,
      maxLagMs,
      memoryUsageBytes,
      underMemoryPressure: this.underMemoryPressure,
      totalPendingEvents,
      totalErrors,
    }

    // Add sync primitive metrics if enabled
    if (this.transactionBatcher) {
      metrics.pendingTransactions = this.transactionBatcher.getPendingCount()
      metrics.completedTransactions = this.transactionBatcher.getCompletedCount()
    }

    if (this.syncProgressTracker) {
      metrics.syncProgress = this.syncProgressTracker.getProgress()
    }

    if (this.conflictDetector) {
      metrics.potentialConflicts = this.conflictDetector.approximateCount()
    }

    return metrics
  }

  /**
   * Estimate current memory usage for subscription metadata
   */
  private estimateMemoryUsage(): number {
    // Rough estimates per data structure:
    // - InternalSubscription: ~500 bytes base
    // - SubscriptionStatus: ~300 bytes
    // - Config: ~200 bytes
    // - Handler arrays: ~50 bytes + 8 bytes per handler
    // - Pending updates: ~100 bytes per entry

    let bytes = 0

    // Base subscription overhead
    bytes += this.subscriptions.size * 1000

    // Handler storage
    for (const handlers of this.eventHandlers.values()) {
      bytes += 50 + handlers.length * 8
    }
    for (const handlers of this.errorHandlers.values()) {
      bytes += 50 + handlers.length * 8
    }

    // Pending persistence updates
    bytes += this.pendingLsnUpdates.size * 100
    bytes += this.pendingStateUpdates.size * 400

    // Metrics window
    bytes += this.metricsWindow.length * 16

    // State change handlers
    bytes += this.stateChangeHandlers.length * 8

    return bytes
  }

  // ===========================================================================
  // Memory Pressure Methods
  // ===========================================================================

  /**
   * Check and update memory pressure state
   */
  private checkMemoryPressure(): void {
    const memoryUsage = this.estimateMemoryUsage()
    const wasUnderPressure = this.underMemoryPressure

    this.underMemoryPressure = memoryUsage > this.memoryPressureThreshold

    // If transitioning to memory pressure, trigger cleanup
    if (this.underMemoryPressure && !wasUnderPressure) {
      this.handleMemoryPressure()
    }
  }

  /**
   * Handle memory pressure - graceful degradation
   */
  private handleMemoryPressure(): void {
    // 1. Flush pending persistence immediately
    this.flushBatchedPersistence().catch(err => {
      console.error('Failed to flush persistence during memory pressure:', err)
    })

    // 2. Trim metrics window aggressively
    if (this.metricsWindow.length > 1000) {
      this.metricsWindow = this.metricsWindow.slice(-500)
    }

    // 3. Clear empty handler arrays
    for (const [id, handlers] of this.eventHandlers) {
      if (handlers.length === 0) {
        this.eventHandlers.delete(id)
      }
    }
    for (const [id, handlers] of this.errorHandlers) {
      if (handlers.length === 0) {
        this.errorHandlers.delete(id)
      }
    }

    // Log warning
    console.warn('CDC Stream Manager: Under memory pressure, triggered cleanup')
  }

  /**
   * Force garbage collection friendly cleanup
   * Call this when you know subscriptions are being removed
   */
  triggerCleanup(): void {
    this.checkMemoryPressure()

    // Remove any orphaned handler arrays
    for (const subscriptionId of this.eventHandlers.keys()) {
      if (!this.subscriptions.has(subscriptionId)) {
        this.eventHandlers.delete(subscriptionId)
      }
    }
    for (const subscriptionId of this.errorHandlers.keys()) {
      if (!this.subscriptions.has(subscriptionId)) {
        this.errorHandlers.delete(subscriptionId)
      }
    }
  }
}

// =============================================================================
// Factory functions
// =============================================================================

/**
 * Create a CDC Stream Manager synchronously
 *
 * Note: For synchronous initialization with persistence, the persistence adapter's
 * loadAllStates will be called and the manager will schedule restoration as a microtask.
 * For guaranteed synchronous state availability, use createCDCStreamManagerAsync instead.
 */
export function createCDCStreamManager(config: CDCStreamManagerConfig): CDCStreamManager {
  // For synchronous creation, we attempt to pre-load states if persistence has
  // internal synchronous access (common in testing scenarios)
  let preloadedStates: SubscriptionStatus[] | undefined

  if (config.persistence) {
    // Check if persistence exposes internal states synchronously (for mocks/testing)
    const persistenceWithStates = config.persistence as CDCStatePersistence & {
      _states?: Map<string, SubscriptionStatus>
    }
    if (persistenceWithStates._states instanceof Map) {
      preloadedStates = Array.from(persistenceWithStates._states.values())
    }
  }

  return new CDCStreamManager(config, preloadedStates)
}

/**
 * Create a CDC Stream Manager asynchronously with guaranteed state restoration
 */
export async function createCDCStreamManagerAsync(config: CDCStreamManagerConfig): Promise<CDCStreamManager> {
  let preloadedStates: SubscriptionStatus[] | undefined

  if (config.persistence) {
    try {
      preloadedStates = await config.persistence.loadAllStates()
    } catch (error) {
      console.error('Failed to load subscription states from persistence:', error)
    }
  }

  return new CDCStreamManager(config, preloadedStates)
}