/**
 * @module @dotdo/postgres-shared/sync-primitives
 *
 * Common sync primitives for CDC/Electric integrations.
 * Provides:
 * - Delta compression for repeated column patterns
 * - Transaction batching for multi-row operations
 * - Bloom filter conflict detection
 * - Merge strategies (server-wins, client-wins, field-merge, CRDT-like)
 * - Sync progress tracking
 */

import {
  FNV_OFFSET_BASIS,
  FNV_PRIME,
  MURMUR_HASH_MULTIPLIER,
  THROUGHPUT_WINDOW_MS,
  DEFAULT_EXPECTED_CHANGES,
} from './constants.js'

// =============================================================================
// Type Definitions
// =============================================================================

/**
 * Represents a single row change
 */
export interface RowChange<T = Record<string, unknown>> {
  /** Unique change ID (usually LSN) */
  id: string
  /** Table name */
  table: string
  /** Schema name */
  schema: string
  /** Operation type */
  operation: 'INSERT' | 'UPDATE' | 'DELETE' | 'TRUNCATE'
  /** New row data for INSERT/UPDATE */
  newRow?: T | undefined
  /** Old row data for UPDATE/DELETE */
  oldRow?: T | undefined
  /** Changed columns for UPDATE */
  changedColumns?: string[] | undefined
  /** Transaction ID */
  xid?: number | undefined
  /** Timestamp */
  timestamp: Date
}

/**
 * Delta-compressed change for efficient transmission
 */
export interface DeltaCompressedChange<T = Record<string, unknown>> {
  /** Change ID */
  id: string
  /** Reference to schema definition (index into column schema) */
  schemaRef?: number
  /** Operation code: 0=INSERT, 1=UPDATE, 2=DELETE, 3=TRUNCATE */
  op: 0 | 1 | 2 | 3
  /** New values (only changed fields for UPDATE) */
  n?: Partial<T>
  /** Old values (only for UPDATE/DELETE, only included fields) */
  o?: Partial<T>
  /** Changed column indices (instead of names) */
  c?: number[]
  /** Transaction ID */
  x?: number
  /** Timestamp delta from batch base (ms) */
  td?: number
}

/**
 * Batch of delta-compressed changes
 */
export interface DeltaCompressedBatch<T = Record<string, unknown>> {
  /** Schema definitions: array of column names per table */
  schemas: Record<string, string[]>
  /** Base timestamp (ISO string) */
  baseTs: string
  /** Table name (if all changes are for same table) */
  table?: string
  /** Schema name (if all changes are for same schema) */
  schema?: string
  /** Compressed changes */
  changes: DeltaCompressedChange<T>[]
}

/**
 * Merge strategy types
 */
export type MergeStrategy =
  | 'server-wins'
  | 'client-wins'
  | 'field-merge'
  | 'timestamp-wins'
  | 'custom'

/**
 * Merge conflict information
 */
// T is kept for type consistency across the merge system
// eslint-disable-next-line @typescript-eslint/no-unused-vars
export interface MergeConflict<_T = Record<string, unknown>> {
  /** Conflicting field name */
  field: string
  /** Server value */
  serverValue: unknown
  /** Client value */
  clientValue: unknown
  /** Base value (before both changes) */
  baseValue?: unknown
  /** Server timestamp */
  serverTimestamp?: Date
  /** Client timestamp */
  clientTimestamp?: Date
}

/**
 * Merge result
 */
export interface MergeResult<T = Record<string, unknown>> {
  /** Resolved row */
  resolved: T
  /** Which strategy was applied */
  strategy: MergeStrategy
  /** Conflicts that were detected */
  conflicts: MergeConflict<T>[]
  /** Fields that required resolution */
  resolvedFields: string[]
  /** Whether any conflicts occurred */
  hadConflicts: boolean
}

/**
 * Custom merge resolver function
 */
export type MergeResolver<T = Record<string, unknown>> = (
  conflict: MergeConflict<T>,
  context: { serverRow: T; clientRow: T; baseRow?: T }
) => unknown

/**
 * Merge configuration
 */
export interface MergeConfig<T = Record<string, unknown>> {
  /** Default merge strategy */
  strategy: MergeStrategy
  /** Field-specific strategies */
  fieldStrategies?: Record<string, MergeStrategy>
  /** Custom resolver for 'custom' strategy */
  customResolver?: MergeResolver<T>
  /** Timestamp field name for timestamp-wins strategy */
  timestampField?: string
  /** Counter fields for CRDT-like increment merge */
  counterFields?: string[]
  /** Set fields for CRDT-like set union merge */
  setFields?: string[]
}

/**
 * Sync progress state
 */
export interface SyncProgress {
  /** Current LSN position */
  currentLsn: string
  /** Target LSN (if known, e.g., from snapshot) */
  targetLsn?: string
  /** Total changes processed */
  changesProcessed: number
  /** Changes in current batch */
  batchSize: number
  /** Start time */
  startedAt: Date
  /** Last update time */
  updatedAt: Date
  /** Estimated completion percentage (0-100) */
  progress: number
  /** Current phase */
  phase: 'initial-sync' | 'streaming' | 'catching-up' | 'idle'
  /** Error if any */
  error?: string
  /** Bytes transferred */
  bytesTransferred: number
  /** Throughput (changes/second) */
  throughput: number
}

/**
 * Transaction batch for multi-row operations
 */
export interface TransactionBatch<T = Record<string, unknown>> {
  /** Transaction ID */
  xid: number
  /** Changes in this transaction */
  changes: RowChange<T>[]
  /** Batch start LSN */
  startLsn: string
  /** Batch end LSN */
  endLsn: string
  /** Transaction timestamp */
  timestamp: Date
  /** Whether transaction is committed */
  committed: boolean
}

// =============================================================================
// Delta Compression
// =============================================================================

const OP_MAP: Record<string, 0 | 1 | 2 | 3> = {
  INSERT: 0,
  UPDATE: 1,
  DELETE: 2,
  TRUNCATE: 3,
}

const REVERSE_OP_MAP: Record<number, 'INSERT' | 'UPDATE' | 'DELETE' | 'TRUNCATE'> = {
  0: 'INSERT',
  1: 'UPDATE',
  2: 'DELETE',
  3: 'TRUNCATE',
}

/**
 * Delta compression encoder for CDC changes.
 * Reduces payload size by:
 * - Using numeric operation codes
 * - Storing column schemas separately
 * - Using column indices instead of names
 * - Storing timestamps as deltas from batch base
 * - Only including changed fields for updates
 */
export class DeltaCompressor<T = Record<string, unknown>> {
  private schemaCache = new Map<string, string[]>()
  private schemaIndex = new Map<string, number>()
  private nextSchemaIndex = 0

  /**
   * Get or create schema for a table
   */
  private getOrCreateSchema(table: string, row: Record<string, unknown>): { index: number; columns: string[] } {
    const key = table
    let columns = this.schemaCache.get(key)

    if (!columns) {
      columns = Object.keys(row).sort()
      this.schemaCache.set(key, columns)
      this.schemaIndex.set(key, this.nextSchemaIndex)
      this.nextSchemaIndex++
    }

    return { index: this.schemaIndex.get(key)!, columns }
  }

  /**
   * Compress a single change
   */
  compressChange(change: RowChange<T>, baseTimestamp: Date): DeltaCompressedChange<T> {
    const opCode = OP_MAP[change.operation]
    if (opCode === undefined) {
      throw new Error(`Unknown operation: ${change.operation}`)
    }
    const result: DeltaCompressedChange<T> = {
      id: change.id,
      op: opCode,
    }

    // Get schema for this table
    const row = change.newRow || change.oldRow
    if (row) {
      const { index, columns } = this.getOrCreateSchema(change.table, row as Record<string, unknown>)
      result.schemaRef = index

      // For INSERT, include all new values
      if (change.operation === 'INSERT' && change.newRow) {
        result.n = change.newRow
      }

      // For UPDATE, only include changed fields
      if (change.operation === 'UPDATE') {
        if (change.changedColumns && change.changedColumns.length > 0) {
          result.c = change.changedColumns.map(col => columns.indexOf(col)).filter(i => i >= 0)
          result.n = {} as Partial<T>
          for (const col of change.changedColumns) {
            if (change.newRow && col in (change.newRow as Record<string, unknown>)) {
              (result.n as Record<string, unknown>)[col] = (change.newRow as Record<string, unknown>)[col]
            }
          }
        } else if (change.newRow) {
          result.n = change.newRow
        }
        if (change.oldRow) {
          result.o = change.oldRow
        }
      }

      // For DELETE, include old row
      if (change.operation === 'DELETE' && change.oldRow) {
        result.o = change.oldRow
      }
    }

    // Transaction ID
    if (change.xid) {
      result.x = change.xid
    }

    // Timestamp delta
    const timeDelta = change.timestamp.getTime() - baseTimestamp.getTime()
    if (timeDelta !== 0) {
      result.td = timeDelta
    }

    return result
  }

  /**
   * Compress a batch of changes
   */
  compressBatch(changes: RowChange<T>[], table?: string, schema?: string): DeltaCompressedBatch<T> {
    const batch: DeltaCompressedBatch<T> = {
      schemas: {},
      baseTs: new Date().toISOString(),
      changes: [],
    }

    if (table !== undefined) {
      batch.table = table
    }
    if (schema !== undefined) {
      batch.schema = schema
    }

    if (changes.length === 0) {
      return batch
    }

    const baseTs = changes[0]!.timestamp
    batch.baseTs = baseTs.toISOString()
    batch.changes = changes.map(c => this.compressChange(c, baseTs))

    // Build schema map from cache
    const schemas: Record<string, string[]> = {}
    for (const [key, columns] of this.schemaCache) {
      schemas[key] = columns
    }
    batch.schemas = schemas

    return batch
  }

  /**
   * Clear the schema cache (call when switching tables/schemas)
   */
  clearCache(): void {
    this.schemaCache.clear()
    this.schemaIndex.clear()
    this.nextSchemaIndex = 0
  }

  /**
   * Get compression statistics
   */
  getStats(): { schemasCount: number; columnCount: number } {
    let columnCount = 0
    for (const columns of this.schemaCache.values()) {
      columnCount += columns.length
    }
    return {
      schemasCount: this.schemaCache.size,
      columnCount,
    }
  }
}

/**
 * Decompress a batch of delta-compressed changes
 */
export function decompressBatch<T = Record<string, unknown>>(
  batch: DeltaCompressedBatch<T>
): RowChange<T>[] {
  const baseTs = new Date(batch.baseTs)
  const tableSchemas = batch.schemas

  return batch.changes.map(change => {
    // Get schema columns if available
    const schemaKey = batch.table ?? Object.keys(tableSchemas)[change.schemaRef ?? 0]
    const columns = schemaKey ? tableSchemas[schemaKey] : undefined

    const operation = REVERSE_OP_MAP[change.op]
    if (!operation) {
      throw new Error(`Unknown operation code: ${change.op}`)
    }

    const result: RowChange<T> = {
      id: change.id,
      table: batch.table ?? schemaKey ?? '',
      schema: batch.schema ?? 'public',
      operation,
      timestamp: change.td !== undefined ? new Date(baseTs.getTime() + change.td) : baseTs,
    }

    if (change.n !== undefined) {
      result.newRow = change.n as T
    }

    if (change.o !== undefined) {
      result.oldRow = change.o as T
    }

    if (change.c !== undefined && columns) {
      result.changedColumns = change.c.map(i => columns[i]).filter((c): c is string => c !== undefined)
    }

    if (change.x !== undefined) {
      result.xid = change.x
    }

    return result
  })
}

/**
 * Calculate compression ratio for a batch
 */
export function calculateCompressionRatio<T>(
  original: RowChange<T>[],
  compressed: DeltaCompressedBatch<T>
): number {
  const originalSize = JSON.stringify(original).length
  const compressedSize = JSON.stringify(compressed).length
  return compressedSize > 0 ? originalSize / compressedSize : 1
}

// =============================================================================
// Transaction Batching
// =============================================================================

/**
 * Batches changes by transaction for atomic application.
 * Ensures all changes in a transaction are applied together.
 */
export class TransactionBatcher<T = Record<string, unknown>> {
  private pendingBatches = new Map<number, TransactionBatch<T>>()
  private completedBatches: TransactionBatch<T>[] = []
  private maxPendingBatches: number
  private flushCallback?: (batch: TransactionBatch<T>) => void | Promise<void>

  constructor(options: {
    /** Maximum pending batches before forced flush */
    maxPendingBatches?: number
    /** Callback when a batch is ready to flush */
    onFlush?: (batch: TransactionBatch<T>) => void | Promise<void>
  } = {}) {
    this.maxPendingBatches = options.maxPendingBatches ?? 100
    if (options.onFlush !== undefined) {
      this.flushCallback = options.onFlush
    }
  }

  /**
   * Add a change to the appropriate transaction batch
   */
  addChange(change: RowChange<T>, isLastInTransaction: boolean): TransactionBatch<T> | null {
    const xid = change.xid ?? 0

    // Get or create batch for this transaction
    let batch = this.pendingBatches.get(xid)
    if (!batch) {
      batch = {
        xid,
        changes: [],
        startLsn: change.id,
        endLsn: change.id,
        timestamp: change.timestamp,
        committed: false,
      }
      this.pendingBatches.set(xid, batch)
    }

    // Add change to batch
    batch.changes.push(change)
    batch.endLsn = change.id

    // If this is the last change in transaction, complete the batch
    if (isLastInTransaction) {
      batch.committed = true
      this.pendingBatches.delete(xid)
      this.completedBatches.push(batch)

      // Trigger callback if configured
      if (this.flushCallback) {
        void this.flushCallback(batch)
      }

      return batch
    }

    // Check if we need to force flush incomplete batches
    if (this.pendingBatches.size > this.maxPendingBatches) {
      this.forceFlushOldest()
    }

    return null
  }

  /**
   * Force flush the oldest incomplete batch
   */
  private forceFlushOldest(): void {
    if (this.pendingBatches.size === 0) return

    // Get oldest batch
    const oldestXid = this.pendingBatches.keys().next().value as number
    const batch = this.pendingBatches.get(oldestXid)

    if (batch) {
      batch.committed = false // Mark as incomplete
      this.pendingBatches.delete(oldestXid)
      this.completedBatches.push(batch)

      if (this.flushCallback) {
        void this.flushCallback(batch)
      }
    }
  }

  /**
   * Get all completed batches and clear them
   */
  flushCompleted(): TransactionBatch<T>[] {
    const batches = [...this.completedBatches]
    this.completedBatches = []
    return batches
  }

  /**
   * Get pending batch count
   */
  getPendingCount(): number {
    return this.pendingBatches.size
  }

  /**
   * Get completed batch count
   */
  getCompletedCount(): number {
    return this.completedBatches.length
  }

  /**
   * Force complete all pending batches
   */
  forceCompleteAll(): TransactionBatch<T>[] {
    for (const batch of this.pendingBatches.values()) {
      batch.committed = false // Mark as incomplete
      this.completedBatches.push(batch)
    }
    this.pendingBatches.clear()
    return this.flushCompleted()
  }

  /**
   * Clear all batches
   */
  clear(): void {
    this.pendingBatches.clear()
    this.completedBatches = []
  }
}

// =============================================================================
// Bloom Filter for Conflict Detection
// =============================================================================

/**
 * Simple Bloom filter implementation for efficient conflict detection.
 * Used to quickly check if a row ID has been modified locally before
 * applying remote changes.
 */
export class BloomFilter {
  private bitArray: Uint8Array
  private size: number
  private hashCount: number

  /**
   * Create a new Bloom filter
   * @param expectedItems Expected number of items to store
   * @param falsePositiveRate Desired false positive rate (default 0.01 = 1%)
   */
  constructor(expectedItems: number, falsePositiveRate = 0.01) {
    // Calculate optimal size and hash count
    // m = -n * ln(p) / (ln(2)^2)
    // k = m/n * ln(2)
    const n = Math.max(expectedItems, 1)
    const p = Math.max(falsePositiveRate, 0.0001)

    this.size = Math.ceil(-n * Math.log(p) / (Math.LN2 * Math.LN2))
    this.hashCount = Math.ceil((this.size / n) * Math.LN2)
    this.bitArray = new Uint8Array(Math.ceil(this.size / 8))
  }

  /**
   * Generate hash values for an item
   */
  private hash(item: string): number[] {
    const hashes: number[] = []

    // Use two base hashes and derive k hashes (enhanced double hashing)
    // h_i(x) = (h1(x) + i * h2(x) + i^2) mod m
    const h1 = this.fnv1a(item)
    const h2 = this.murmur3(item)

    for (let i = 0; i < this.hashCount; i++) {
      const hash = Math.abs((h1 + i * h2 + i * i) % this.size)
      hashes.push(hash)
    }

    return hashes
  }

  /**
   * FNV-1a hash function
   * @see https://en.wikipedia.org/wiki/Fowler%E2%80%93Noll%E2%80%93Vo_hash_function
   */
  private fnv1a(str: string): number {
    let hash = FNV_OFFSET_BASIS
    for (let i = 0; i < str.length; i++) {
      hash ^= str.charCodeAt(i)
      hash = Math.imul(hash, FNV_PRIME)
    }
    return hash >>> 0 // Ensure positive
  }

  /**
   * Simple Murmur3-like hash function
   * @see https://en.wikipedia.org/wiki/MurmurHash
   */
  private murmur3(str: string): number {
    let h = 0xdeadbeef
    for (let i = 0; i < str.length; i++) {
      h = Math.imul(h ^ str.charCodeAt(i), MURMUR_HASH_MULTIPLIER)
    }
    return ((h ^ (h >>> 16)) >>> 0)
  }

  /**
   * Add an item to the filter
   */
  add(item: string): void {
    const hashes = this.hash(item)
    for (const h of hashes) {
      const byteIndex = Math.floor(h / 8)
      const bitIndex = h % 8
      this.bitArray[byteIndex]! |= (1 << bitIndex)
    }
  }

  /**
   * Check if an item might be in the filter
   * Returns true if item might exist (may be false positive)
   * Returns false if item definitely doesn't exist
   */
  mightContain(item: string): boolean {
    const hashes = this.hash(item)
    for (const h of hashes) {
      const byteIndex = Math.floor(h / 8)
      const bitIndex = h % 8
      if ((this.bitArray[byteIndex]! & (1 << bitIndex)) === 0) {
        return false
      }
    }
    return true
  }

  /**
   * Merge another filter into this one (union operation)
   */
  merge(other: BloomFilter): void {
    if (this.size !== other.size) {
      throw new Error('Cannot merge filters of different sizes')
    }
    for (let i = 0; i < this.bitArray.length; i++) {
      this.bitArray[i]! |= other.bitArray[i]!
    }
  }

  /**
   * Clear the filter
   */
  clear(): void {
    this.bitArray.fill(0)
  }

  /**
   * Get approximate number of items in filter
   * Uses formula: -(m/k) * ln(1 - X/m) where X is set bits
   */
  approximateCount(): number {
    let setBits = 0
    for (let i = 0; i < this.bitArray.length; i++) {
      // Count set bits using Brian Kernighan's algorithm
      let byte = this.bitArray[i]!
      while (byte) {
        setBits++
        byte &= byte - 1
      }
    }

    if (setBits === 0) return 0
    if (setBits >= this.size) return this.size // Saturated

    const ratio = setBits / this.size
    return Math.round(-(this.size / this.hashCount) * Math.log(1 - ratio))
  }

  /**
   * Export filter state for serialization
   */
  export(): { size: number; hashCount: number; data: string } {
    return {
      size: this.size,
      hashCount: this.hashCount,
      // Base64 encode the bit array
      data: btoa(String.fromCharCode(...this.bitArray)),
    }
  }

  /**
   * Import filter state from serialization
   */
  static import(state: { size: number; hashCount: number; data: string }): BloomFilter {
    const filter = Object.create(BloomFilter.prototype) as BloomFilter
    filter.size = state.size
    filter.hashCount = state.hashCount
    // Base64 decode
    const binaryStr = atob(state.data)
    filter.bitArray = new Uint8Array(binaryStr.length)
    for (let i = 0; i < binaryStr.length; i++) {
      filter.bitArray[i] = binaryStr.charCodeAt(i)
    }
    return filter
  }
}

/**
 * Conflict detector using Bloom filter for efficient local change tracking
 */
export class ConflictDetector<T = Record<string, unknown>> {
  private localChanges: BloomFilter
  private changeTimestamps = new Map<string, number>()
  private primaryKeyField: string

  constructor(options: {
    /** Expected number of local changes to track */
    expectedChanges?: number
    /** False positive rate for Bloom filter */
    falsePositiveRate?: number
    /** Primary key field name */
    primaryKeyField?: string
  } = {}) {
    this.localChanges = new BloomFilter(
      options.expectedChanges ?? DEFAULT_EXPECTED_CHANGES,
      options.falsePositiveRate ?? 0.01
    )
    this.primaryKeyField = options.primaryKeyField ?? 'id'
  }

  /**
   * Track a local change
   */
  trackLocalChange(row: T): void {
    const key = this.getRowKey(row)
    this.localChanges.add(key)
    this.changeTimestamps.set(key, Date.now())
  }

  /**
   * Check if a remote change might conflict with local changes
   * Returns true if there MIGHT be a conflict (Bloom filter positive)
   */
  mightConflict(row: T): boolean {
    return this.localChanges.mightContain(this.getRowKey(row))
  }

  /**
   * Get the timestamp of a tracked local change (if tracked exactly)
   */
  getLocalChangeTime(row: T): number | undefined {
    return this.changeTimestamps.get(this.getRowKey(row))
  }

  /**
   * Clear all tracked changes
   */
  clear(): void {
    this.localChanges.clear()
    this.changeTimestamps.clear()
  }

  /**
   * Get approximate number of tracked changes
   */
  approximateCount(): number {
    return this.localChanges.approximateCount()
  }

  /**
   * Get row key for conflict detection
   */
  private getRowKey(row: T): string {
    const r = row as Record<string, unknown>
    const pk = r[this.primaryKeyField]
    return pk !== undefined ? String(pk) : JSON.stringify(row)
  }
}

// =============================================================================
// Merge Strategies
// =============================================================================

/**
 * Merge engine for resolving conflicts between server and client changes.
 * Supports multiple strategies and CRDT-like merging for specific fields.
 */
export class MergeEngine<T = Record<string, unknown>> {
  private strategy: MergeStrategy
  private fieldStrategies: Record<string, MergeStrategy>
  private customResolver?: MergeResolver<T>
  private timestampField: string
  private counterFields: string[]
  private setFields: string[]

  constructor(config: MergeConfig<T>) {
    this.strategy = config.strategy
    this.fieldStrategies = config.fieldStrategies ?? {}
    if (config.customResolver !== undefined) {
      this.customResolver = config.customResolver
    }
    this.timestampField = config.timestampField ?? 'updated_at'
    this.counterFields = config.counterFields ?? []
    this.setFields = config.setFields ?? []
  }

  /**
   * Merge server and client changes
   */
  merge(
    serverRow: T,
    clientRow: T,
    baseRow?: T
  ): MergeResult<T> {
    const conflicts: MergeConflict<T>[] = []
    const resolvedFields: string[] = []
    const resolved: Record<string, unknown> = {}

    // Get all field names
    const allFields = new Set([
      ...Object.keys(serverRow as Record<string, unknown>),
      ...Object.keys(clientRow as Record<string, unknown>),
    ])

    for (const field of allFields) {
      const serverValue = (serverRow as Record<string, unknown>)[field]
      const clientValue = (clientRow as Record<string, unknown>)[field]
      const baseValue = baseRow ? (baseRow as Record<string, unknown>)[field] : undefined

      // No conflict if values are the same
      if (this.valuesEqual(serverValue, clientValue)) {
        resolved[field] = serverValue
        continue
      }

      // Check if only one side changed from base
      if (baseRow !== undefined) {
        const serverChanged = !this.valuesEqual(serverValue, baseValue)
        const clientChanged = !this.valuesEqual(clientValue, baseValue)

        if (serverChanged && !clientChanged) {
          resolved[field] = serverValue
          continue
        }

        if (clientChanged && !serverChanged) {
          resolved[field] = clientValue
          continue
        }
      }

      // Conflict detected
      const conflict: MergeConflict<T> = {
        field,
        serverValue,
        clientValue,
        baseValue,
      }
      conflicts.push(conflict)
      resolvedFields.push(field)

      // Determine strategy for this field
      const strategy = this.fieldStrategies?.[field] ?? this.strategy

      // Resolve based on strategy
      const resolveContext: { serverRow: T; clientRow: T; baseRow?: T } = { serverRow, clientRow }
      if (baseRow !== undefined) {
        resolveContext.baseRow = baseRow
      }
      resolved[field] = this.resolveField(field, conflict, resolveContext, strategy)
    }

    return {
      resolved: resolved as T,
      strategy: this.strategy,
      conflicts,
      resolvedFields,
      hadConflicts: conflicts.length > 0,
    }
  }

  /**
   * Resolve a single field conflict
   */
  private resolveField(
    field: string,
    conflict: MergeConflict<T>,
    context: { serverRow: T; clientRow: T; baseRow?: T },
    strategy: MergeStrategy
  ): unknown {
    // Check for CRDT-like fields first
    if (this.counterFields?.includes(field)) {
      return this.mergeCounter(conflict)
    }

    if (this.setFields?.includes(field)) {
      return this.mergeSet(conflict)
    }

    switch (strategy) {
      case 'server-wins':
        return conflict.serverValue

      case 'client-wins':
        return conflict.clientValue

      case 'timestamp-wins':
        return this.mergeByTimestamp(conflict, context)

      case 'field-merge':
        // For objects, try to merge fields
        if (
          typeof conflict.serverValue === 'object' &&
          typeof conflict.clientValue === 'object' &&
          conflict.serverValue !== null &&
          conflict.clientValue !== null
        ) {
          return { ...conflict.serverValue as object, ...conflict.clientValue as object }
        }
        // Fall back to server-wins for non-objects
        return conflict.serverValue

      case 'custom':
        if (this.customResolver) {
          return this.customResolver(conflict, context)
        }
        return conflict.serverValue

      default:
        return conflict.serverValue
    }
  }

  /**
   * Merge by timestamp (latest wins)
   */
  private mergeByTimestamp(
    conflict: MergeConflict<T>,
    context: { serverRow: T; clientRow: T }
  ): unknown {
    const timestampField = this.timestampField!
    const serverTs = (context.serverRow as Record<string, unknown>)[timestampField]
    const clientTs = (context.clientRow as Record<string, unknown>)[timestampField]

    const serverTime = serverTs instanceof Date ? serverTs.getTime() : new Date(serverTs as string).getTime()
    const clientTime = clientTs instanceof Date ? clientTs.getTime() : new Date(clientTs as string).getTime()

    if (isNaN(serverTime) && isNaN(clientTime)) {
      return conflict.serverValue // Default to server if no valid timestamps
    }
    if (isNaN(serverTime)) return conflict.clientValue
    if (isNaN(clientTime)) return conflict.serverValue

    return clientTime > serverTime ? conflict.clientValue : conflict.serverValue
  }

  /**
   * CRDT-like counter merge (add deltas from base)
   */
  private mergeCounter(conflict: MergeConflict<T>): number {
    const base = typeof conflict.baseValue === 'number' ? conflict.baseValue : 0
    const server = typeof conflict.serverValue === 'number' ? conflict.serverValue : 0
    const client = typeof conflict.clientValue === 'number' ? conflict.clientValue : 0

    // Calculate deltas from base and sum them
    const serverDelta = server - base
    const clientDelta = client - base

    return base + serverDelta + clientDelta
  }

  /**
   * CRDT-like set merge (union of changes)
   */
  private mergeSet(conflict: MergeConflict<T>): unknown[] {
    const serverSet = new Set(Array.isArray(conflict.serverValue) ? conflict.serverValue : [])
    const clientSet = new Set(Array.isArray(conflict.clientValue) ? conflict.clientValue : [])

    // Union of both sets
    return [...new Set([...serverSet, ...clientSet])]
  }

  /**
   * Check if two values are equal
   */
  private valuesEqual(a: unknown, b: unknown): boolean {
    if (a === b) return true
    if (a === null || b === null) return a === b
    if (a === undefined || b === undefined) return a === b

    // Date comparison
    if (a instanceof Date && b instanceof Date) {
      return a.getTime() === b.getTime()
    }

    // Array comparison
    if (Array.isArray(a) && Array.isArray(b)) {
      if (a.length !== b.length) return false
      return a.every((v, i) => this.valuesEqual(v, b[i]))
    }

    // Object comparison
    if (typeof a === 'object' && typeof b === 'object') {
      const keysA = Object.keys(a as object)
      const keysB = Object.keys(b as object)
      if (keysA.length !== keysB.length) return false
      return keysA.every(key =>
        this.valuesEqual((a as Record<string, unknown>)[key], (b as Record<string, unknown>)[key])
      )
    }

    return false
  }
}

// =============================================================================
// Sync Progress Tracker
// =============================================================================

/**
 * Tracks sync progress and provides estimates for UI feedback.
 */
export class SyncProgressTracker {
  private progress: SyncProgress
  private recentChanges: { timestamp: number; count: number }[] = []
  private throughputWindowMs = THROUGHPUT_WINDOW_MS

  constructor(initialLsn?: string) {
    this.progress = {
      currentLsn: initialLsn ?? '0/0',
      changesProcessed: 0,
      batchSize: 0,
      startedAt: new Date(),
      updatedAt: new Date(),
      progress: 0,
      phase: 'initial-sync',
      bytesTransferred: 0,
      throughput: 0,
    }
  }

  /**
   * Update progress with a new batch of changes
   */
  recordBatch(lsn: string, changeCount: number, bytesTransferred: number): void {
    const now = Date.now()

    this.progress.currentLsn = lsn
    this.progress.changesProcessed += changeCount
    this.progress.batchSize = changeCount
    this.progress.bytesTransferred += bytesTransferred
    this.progress.updatedAt = new Date()

    // Record for throughput calculation
    this.recentChanges.push({ timestamp: now, count: changeCount })

    // Clean up old entries
    const cutoff = now - this.throughputWindowMs
    this.recentChanges = this.recentChanges.filter(r => r.timestamp >= cutoff)

    // Calculate throughput
    if (this.recentChanges.length > 0) {
      const totalChanges = this.recentChanges.reduce((sum, r) => sum + r.count, 0)
      const windowMs = now - this.recentChanges[0]!.timestamp
      this.progress.throughput = windowMs > 0 ? (totalChanges / windowMs) * 1000 : 0
    }

    // Update progress estimate if we have target
    if (this.progress.targetLsn) {
      this.updateProgressEstimate()
    }
  }

  /**
   * Set target LSN for progress calculation
   */
  setTargetLsn(lsn: string): void {
    this.progress.targetLsn = lsn
    this.updateProgressEstimate()
  }

  /**
   * Set the sync phase
   */
  setPhase(phase: SyncProgress['phase']): void {
    this.progress.phase = phase
  }

  /**
   * Set an error
   */
  setError(error: string): void {
    this.progress.error = error
  }

  /**
   * Clear error
   */
  clearError(): void {
    delete this.progress.error
  }

  /**
   * Get current progress snapshot
   */
  getProgress(): SyncProgress {
    return { ...this.progress }
  }

  /**
   * Update progress estimate based on LSN comparison
   */
  private updateProgressEstimate(): void {
    if (!this.progress.targetLsn) {
      this.progress.progress = 0
      return
    }

    // Parse LSNs for comparison
    const current = this.parseLsn(this.progress.currentLsn)
    const target = this.parseLsn(this.progress.targetLsn)

    if (target === 0n) {
      this.progress.progress = 100
      return
    }

    // Simple linear progress estimate
    const progress = Number((current * 100n) / target)
    this.progress.progress = Math.min(Math.max(progress, 0), 100)

    // Update phase based on progress
    if (this.progress.progress >= 95) {
      this.progress.phase = 'streaming'
    } else if (this.progress.progress > 0) {
      this.progress.phase = 'catching-up'
    }
  }

  /**
   * Parse PostgreSQL LSN string to bigint
   */
  private 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))
    return (segment << 32n) | offset
  }

  /**
   * Get estimated time remaining in milliseconds
   */
  getEstimatedTimeRemaining(): number | null {
    if (!this.progress.targetLsn || this.progress.throughput === 0) {
      return null
    }

    const current = this.parseLsn(this.progress.currentLsn)
    const target = this.parseLsn(this.progress.targetLsn)
    const remaining = Number(target - current)

    if (remaining <= 0) return 0

    // Rough estimate: assume 1 LSN unit ≈ 1 change
    return (remaining / this.progress.throughput) * 1000
  }

  /**
   * Reset progress
   */
  reset(): void {
    this.progress = {
      currentLsn: '0/0',
      changesProcessed: 0,
      batchSize: 0,
      startedAt: new Date(),
      updatedAt: new Date(),
      progress: 0,
      phase: 'initial-sync',
      bytesTransferred: 0,
      throughput: 0,
    }
    this.recentChanges = []
  }
}

// =============================================================================
// Exports
// =============================================================================

export {
  OP_MAP,
  REVERSE_OP_MAP,
}