/**
 * Batched Change Processing for High Throughput
 *
 * Processes CDC events in batches for improved performance:
 * - Configurable batch size and timeout
 * - Memory-efficient event coalescing
 * - Transaction-aware batching
 * - Backpressure handling
 */

import type { CDCChangeEvent, Row } from '../types'
import { CDCResourceError, CDCErrorCode } from './errors'
import { CDCLogger, defaultLogger } from './logger'

/**
 * Batch of events ready for processing
 */
export interface EventBatch<T extends Row = Row> {
  /** Events in this batch */
  events: Array<CDCChangeEvent & { newRow?: T; oldRow?: T }>
  /** Subscription ID */
  subscriptionId: string
  /** Batch creation timestamp */
  createdAt: Date
  /** Batch ready timestamp (when it was flushed) */
  readyAt: Date
  /** Number of events */
  size: number
  /** First LSN in batch */
  firstLsn: string
  /** Last LSN in batch */
  lastLsn: string
  /** Whether batch contains transaction boundary */
  hasTransactionBoundary: boolean
}

/**
 * Batch processor configuration
 */
export interface BatchProcessorConfig {
  /** Maximum events per batch (default: 100) */
  maxBatchSize?: number
  /** Maximum time to wait before flushing batch (ms) (default: 50) */
  maxBatchTimeMs?: number
  /** Maximum memory for pending batches (bytes) (default: 10MB) */
  maxMemoryBytes?: number
  /** Whether to flush batch on transaction commit (default: true) */
  flushOnTransactionEnd?: boolean
  /** Callback when batch is ready */
  onBatch?: <T extends Row>(batch: EventBatch<T>) => void | Promise<void>
  /** Callback for backpressure events */
  onBackpressure?: (memoryUsed: number, maxMemory: number) => void
  /** Logger instance */
  logger?: CDCLogger
}

/**
 * Estimated event memory size
 */
function estimateEventSize(event: CDCChangeEvent): number {
  // Base event overhead
  let size = 200

  // Estimate row sizes
  if (event.newRow) {
    size += JSON.stringify(event.newRow).length * 2 // UTF-16 estimate
  }
  if (event.oldRow) {
    size += JSON.stringify(event.oldRow).length * 2
  }

  return size
}

/**
 * Pending batch for a subscription
 */
interface PendingBatch {
  events: CDCChangeEvent[]
  createdAt: Date
  memoryEstimate: number
  firstLsn: string
  lastLsn: string
}

/**
 * Batch Processor for CDC Events
 */
export class BatchProcessor<T extends Row = Row> {
  private batches = new Map<string, PendingBatch>()
  private batchTimers = new Map<string, ReturnType<typeof setTimeout>>()
  private totalMemory = 0
  private paused = false
  private stats = {
    batchesProcessed: 0,
    eventsProcessed: 0,
    backpressureEvents: 0,
    flushesOnTimeout: 0,
    flushesOnSize: 0,
    flushesOnTransaction: 0,
  }

  private readonly config: Required<Omit<BatchProcessorConfig, 'onBatch' | 'onBackpressure' | 'logger'>> & {
    onBatch?: BatchProcessorConfig['onBatch']
    onBackpressure?: BatchProcessorConfig['onBackpressure']
    logger: CDCLogger
  }

  constructor(config: BatchProcessorConfig = {}) {
    this.config = {
      maxBatchSize: config.maxBatchSize ?? 100,
      maxBatchTimeMs: config.maxBatchTimeMs ?? 50,
      maxMemoryBytes: config.maxMemoryBytes ?? 10 * 1024 * 1024, // 10MB
      flushOnTransactionEnd: config.flushOnTransactionEnd ?? true,
      onBatch: config.onBatch,
      onBackpressure: config.onBackpressure,
      logger: config.logger ?? defaultLogger,
    }
  }

  /**
   * Add an event to the batch
   */
  add(subscriptionId: string, event: CDCChangeEvent & { newRow?: T; oldRow?: T }): void {
    // Check memory limit
    const eventSize = estimateEventSize(event)
    if (this.totalMemory + eventSize > this.config.maxMemoryBytes) {
      this.handleBackpressure()
      throw new CDCResourceError('memory', this.totalMemory + eventSize, this.config.maxMemoryBytes, {
        code: CDCErrorCode.BUFFER_OVERFLOW,
        subscriptionId,
      })
    }

    // Get or create batch
    let batch = this.batches.get(subscriptionId)
    if (!batch) {
      batch = {
        events: [],
        createdAt: new Date(),
        memoryEstimate: 0,
        firstLsn: event.lsn,
        lastLsn: event.lsn,
      }
      this.batches.set(subscriptionId, batch)

      // Start batch timer
      this.startBatchTimer(subscriptionId)
    }

    // Add event
    batch.events.push(event)
    batch.memoryEstimate += eventSize
    batch.lastLsn = event.lsn
    this.totalMemory += eventSize

    // Check if batch should be flushed
    if (batch.events.length >= this.config.maxBatchSize) {
      this.stats.flushesOnSize++
      this.flush(subscriptionId)
    } else if (this.config.flushOnTransactionEnd && event.isLastInTransaction) {
      this.stats.flushesOnTransaction++
      this.flush(subscriptionId)
    }
  }

  /**
   * Flush a specific subscription's batch
   */
  flush(subscriptionId: string): void {
    const batch = this.batches.get(subscriptionId)
    if (!batch || batch.events.length === 0) {
      return
    }

    // Cancel timer
    const timer = this.batchTimers.get(subscriptionId)
    if (timer) {
      clearTimeout(timer)
      this.batchTimers.delete(subscriptionId)
    }

    // Create batch object
    const eventBatch: EventBatch<T> = {
      events: batch.events as Array<CDCChangeEvent & { newRow?: T; oldRow?: T }>,
      subscriptionId,
      createdAt: batch.createdAt,
      readyAt: new Date(),
      size: batch.events.length,
      firstLsn: batch.firstLsn,
      lastLsn: batch.lastLsn,
      hasTransactionBoundary: batch.events.some((e) => e.isLastInTransaction),
    }

    // Clear batch
    this.totalMemory -= batch.memoryEstimate
    this.batches.delete(subscriptionId)

    // Update stats
    this.stats.batchesProcessed++
    this.stats.eventsProcessed += eventBatch.size

    // Call handler
    this.config.logger.debug('Flushing batch', {
      subscriptionId,
      data: { size: eventBatch.size, memory: batch.memoryEstimate },
    })

    try {
      const result = this.config.onBatch?.(eventBatch)
      if (result instanceof Promise) {
        result.catch((error) => {
          this.config.logger.error('Batch handler error', {
            subscriptionId,
            error: error instanceof Error ? error : new Error(String(error)),
          })
        })
      }
    } catch (error) {
      this.config.logger.error('Batch handler error', {
        subscriptionId,
        error: error instanceof Error ? error : new Error(String(error)),
      })
    }
  }

  /**
   * Flush all pending batches
   */
  flushAll(): void {
    for (const subscriptionId of this.batches.keys()) {
      this.flush(subscriptionId)
    }
  }

  /**
   * Pause batch processing
   */
  pause(): void {
    this.paused = true
    // Cancel all timers
    for (const timer of this.batchTimers.values()) {
      clearTimeout(timer)
    }
    this.batchTimers.clear()
  }

  /**
   * Resume batch processing
   */
  resume(): void {
    this.paused = false
    // Restart timers for existing batches
    for (const subscriptionId of this.batches.keys()) {
      this.startBatchTimer(subscriptionId)
    }
  }

  /**
   * Get pending batch size for a subscription
   */
  getPendingSize(subscriptionId: string): number {
    return this.batches.get(subscriptionId)?.events.length ?? 0
  }

  /**
   * Get total pending events across all subscriptions
   */
  getTotalPending(): number {
    let total = 0
    for (const batch of this.batches.values()) {
      total += batch.events.length
    }
    return total
  }

  /**
   * Get memory usage
   */
  getMemoryUsage(): { used: number; max: number; percent: number } {
    return {
      used: this.totalMemory,
      max: this.config.maxMemoryBytes,
      percent: Math.round((this.totalMemory / this.config.maxMemoryBytes) * 100),
    }
  }

  /**
   * Get processing statistics
   */
  getStats(): Readonly<typeof this.stats> {
    return { ...this.stats }
  }

  /**
   * Reset statistics
   */
  resetStats(): void {
    this.stats = {
      batchesProcessed: 0,
      eventsProcessed: 0,
      backpressureEvents: 0,
      flushesOnTimeout: 0,
      flushesOnSize: 0,
      flushesOnTransaction: 0,
    }
  }

  /**
   * Clear all pending batches without processing
   */
  clear(): void {
    for (const timer of this.batchTimers.values()) {
      clearTimeout(timer)
    }
    this.batchTimers.clear()
    this.batches.clear()
    this.totalMemory = 0
  }

  /**
   * Destroy the processor
   */
  destroy(): void {
    this.clear()
  }

  /**
   * Start batch timer
   */
  private startBatchTimer(subscriptionId: string): void {
    if (this.paused) return

    // Cancel existing timer
    const existing = this.batchTimers.get(subscriptionId)
    if (existing) {
      clearTimeout(existing)
    }

    // Start new timer
    const timer = setTimeout(() => {
      this.batchTimers.delete(subscriptionId)
      this.stats.flushesOnTimeout++
      this.flush(subscriptionId)
    }, this.config.maxBatchTimeMs)

    this.batchTimers.set(subscriptionId, timer)
  }

  /**
   * Handle backpressure condition
   */
  private handleBackpressure(): void {
    this.stats.backpressureEvents++
    this.config.logger.warn('Backpressure triggered', {
      data: { memory: this.totalMemory, max: this.config.maxMemoryBytes },
    })

    // Notify handler
    this.config.onBackpressure?.(this.totalMemory, this.config.maxMemoryBytes)

    // Force flush all batches to reduce memory
    this.flushAll()
  }
}

/**
 * Create a batch processor instance
 */
export function createBatchProcessor<T extends Row = Row>(
  config?: BatchProcessorConfig
): BatchProcessor<T> {
  return new BatchProcessor<T>(config)
}

/**
 * Utility to coalesce multiple updates to the same row
 */
export function coalesceEvents<T extends Row>(
  events: Array<CDCChangeEvent & { newRow?: T; oldRow?: T }>
): Array<CDCChangeEvent & { newRow?: T; oldRow?: T }> {
  const byKey = new Map<string, CDCChangeEvent & { newRow?: T; oldRow?: T }>()

  for (const event of events) {
    // Create a key from table + primary key (if available)
    const pk =
      (event.newRow as Record<string, unknown>)?.['id'] ??
      (event.oldRow as Record<string, unknown>)?.['id'] ??
      ''
    const key = `${event.schema}.${event.table}:${pk}`

    const existing = byKey.get(key)
    if (!existing) {
      byKey.set(key, event)
    } else {
      // Coalesce: keep latest, merge old/new appropriately
      const coalesced: CDCChangeEvent & { newRow?: T; oldRow?: T } = {
        ...event,
      }
      // Keep the original oldRow from the first event
      if (existing.oldRow !== undefined) {
        coalesced.oldRow = existing.oldRow
      } else if (event.oldRow !== undefined) {
        coalesced.oldRow = event.oldRow
      }
      // Use the latest newRow
      if (event.newRow !== undefined) {
        coalesced.newRow = event.newRow
      } else if (existing.newRow !== undefined) {
        coalesced.newRow = existing.newRow
      }

      // Handle special cases
      if (existing.operation === 'INSERT' && event.operation === 'DELETE') {
        // INSERT followed by DELETE = no-op, remove from map
        byKey.delete(key)
      } else if (existing.operation === 'INSERT' && event.operation === 'UPDATE') {
        // INSERT followed by UPDATE = INSERT with final value
        coalesced.operation = 'INSERT'
        byKey.set(key, coalesced)
      } else if (existing.operation === 'DELETE' && event.operation === 'INSERT') {
        // DELETE followed by INSERT = UPDATE (row was recreated)
        coalesced.operation = 'UPDATE'
        if (existing.oldRow !== undefined) {
          coalesced.oldRow = existing.oldRow
        }
        byKey.set(key, coalesced)
      } else {
        byKey.set(key, coalesced)
      }
    }
  }

  return Array.from(byKey.values())
}