/**
 * PostgreSQL client types for postgres.do
 * Compatible with Drizzle ORM and direct usage
 */

import {
  PostgresError as SharedPostgresError,
  ConnectionError as SharedConnectionError,
  ConnectionTimeoutError as SharedConnectionTimeoutError,
  type ErrorContext,
} from '@dotdo/postgres-shared/errors'

/** A single row returned from a query */
export type Row = Record<string, unknown>

/** Result of a query execution */
export interface QueryResult<T extends Row = Row> {
  /** Rows returned by the query */
  rows: T[]
  /** Number of rows affected (for INSERT/UPDATE/DELETE) */
  rowCount: number
  /** Column metadata */
  fields: FieldInfo[]
  /** Command type (SELECT, INSERT, etc.) */
  command: string
}

/** Column/field metadata */
export interface FieldInfo {
  /** Column name */
  name: string
  /** PostgreSQL OID for the column type */
  dataTypeID: number
  /** Table OID (0 if not from a table) */
  tableID: number
  /** Column position in table */
  columnID: number
  /** Data type size in bytes (-1 for variable) */
  dataTypeSize: number
  /** Type modifier */
  dataTypeModifier: number
  /** Format code (0 = text, 1 = binary) */
  format: number
}

/** Type parser function */
export type TypeParser = (value: string) => unknown

/** Configuration options for the postgres client */
export interface PostgresConfig {
  /** Database connection URL (postgres://...) */
  url?: string

  /** API key for postgres.do authentication */
  apiKey?: string

  /** Transport type: 'http' (default) or 'ws' */
  transport?: 'http' | 'ws'

  /** Custom type parsers by OID */
  parsers?: Record<number, TypeParser>

  /** Connection timeout in milliseconds */
  connectTimeout?: number

  /** Query timeout in milliseconds */
  queryTimeout?: number

  /** Maximum number of retry attempts */
  maxRetries?: number

  /** Whether to use SSL/TLS */
  ssl?: boolean | 'require' | 'prefer' | 'allow' | 'disable'

  /** Custom fetch implementation (for Cloudflare Workers) */
  fetch?: typeof fetch

  /** Custom WebSocket implementation */
  WebSocket?: typeof WebSocket

  /** Enable debug logging */
  debug?: boolean

  /** Transform column names (e.g., snake_case to camelCase) */
  transform?: {
    column?: (name: string) => string
    value?: (value: unknown, column: string) => unknown
  }

  /**
   * Default transaction retry configuration.
   * These settings are used as defaults for all transactions initiated with begin().
   * Transaction-level options can override these defaults.
   */
  transactionRetry?: {
    /** Automatically retry on serialization failure (error code 40001) */
    retryOnSerializationFailure?: boolean
    /** Automatically retry on deadlock (error code 40P01) */
    retryOnDeadlock?: boolean
    /** Maximum retry attempts (default: 3) */
    maxRetries?: number
    /** Base delay in milliseconds for exponential backoff (default: 10) */
    backoffMs?: number
  }
}

/** Pending query that can be executed */
export interface PendingQuery<T extends Row = Row> extends Promise<T[]> {
  /** Execute the query and return rows */
  execute(): Promise<T[]>

  /** Get the parameterized SQL string */
  toString(): string

  /** Get query values/parameters */
  values: unknown[]

  /** Get the SQL template string */
  strings: TemplateStringsArray
}

/** Transaction isolation levels */
export type IsolationLevel =
  | 'read uncommitted'
  | 'read committed'
  | 'repeatable read'
  | 'serializable'

/** Lock mode types */
export type LockMode =
  | 'access share'
  | 'row share'
  | 'row exclusive'
  | 'share update exclusive'
  | 'share'
  | 'share row exclusive'
  | 'exclusive'
  | 'access exclusive'

/** Transaction options */
export interface TransactionOptions {
  /** Isolation level */
  isolationLevel?: IsolationLevel
  /** Read only transaction */
  readOnly?: boolean
  /** Deferrable (only with serializable + read only) */
  deferrable?: boolean
  /** Named transaction (for debugging/logging) */
  name?: string
  /**
   * Transaction timeout in milliseconds.
   * This is the total timeout across all retry attempts.
   * When retries are enabled, the timeout counts from the start of the first attempt.
   */
  timeout?: number
  /**
   * Per-retry timeout in milliseconds.
   * When set, each retry attempt gets a fresh timeout of this duration.
   * This is useful when you want to allow each attempt to take up to N ms,
   * but the total time can exceed that due to retries.
   * If both `timeout` and `perRetryTimeout` are set, `timeout` takes precedence
   * (the total timeout will be enforced).
   */
  perRetryTimeout?: number
  /** Lock mode for explicit table locking */
  lockMode?: LockMode
  /** Automatically retry on serialization failure (error code 40001) */
  retryOnSerializationFailure?: boolean
  /** Automatically retry on deadlock (error code 40P01) */
  retryOnDeadlock?: boolean
  /** Maximum retry attempts (default: 3) */
  maxRetries?: number
  /** Base delay in milliseconds for exponential backoff (default: 10) */
  backoffMs?: number
  /** Callback called before each retry attempt */
  onRetry?: (attempt: number, error: Error) => void | Promise<void>
}

/**
 * Options for streaming queries
 */
export interface StreamOptions {
  /** Batch size for fetching rows (default: 100) */
  batchSize?: number
  /** Custom cursor name (auto-generated if not provided) */
  cursorName?: string
  /** Transaction isolation level for the cursor */
  isolationLevel?: IsolationLevel
  /** Whether the cursor should be read-only (default: true) */
  readOnly?: boolean
}

/**
 * Result of a cursor fetch operation
 */
export interface CursorFetchResult<T extends Row = Row> {
  /** Rows fetched in this batch */
  rows: T[]
  /** Whether there are more rows to fetch */
  hasMore: boolean
  /** Total number of rows fetched so far */
  totalFetched: number
}

/**
 * Cursor interface for manual control over streaming
 */
export interface Cursor<T extends Row = Row> extends AsyncIterable<T> {
  /** Whether the cursor is still open */
  readonly isOpen: boolean
  /** Total number of rows fetched so far */
  readonly totalFetched: number
  /** Column metadata from the query */
  readonly fields: FieldInfo[]

  /**
   * Fetch the next batch of rows
   * @param count - Number of rows to fetch (default: batch size)
   * @returns Array of rows (empty array if no more rows)
   */
  fetch(count?: number): Promise<T[]>

  /**
   * Fetch the next batch of rows with additional metadata
   * @param count - Number of rows to fetch (default: batch size)
   * @returns Result object with rows, hasMore flag, and total fetched count
   */
  fetchWithInfo(count?: number): Promise<CursorFetchResult<T>>

  /**
   * Close the cursor and release resources
   * Always call this when done, or use for-await-of which closes automatically
   */
  close(): Promise<void>
}

/**
 * Stream result interface for fluent streaming API
 */
export interface StreamResult<T extends Row = Row> extends AsyncIterable<T> {
  /** The SQL query being streamed */
  readonly sql: string
  /** The query parameters */
  readonly params: unknown[]

  /**
   * Stream rows in batches
   * @param size - Number of rows per batch
   * @returns AsyncIterable of row arrays
   * @example
   * for await (const batch of sql.stream`SELECT * FROM users`.batch(1000)) {
   *   processBatch(batch)
   * }
   */
  batch(size: number): AsyncIterable<T[]>

  /**
   * Get a cursor for manual control
   * @returns A cursor that can be used to fetch rows manually
   * @example
   * const cursor = await sql.stream`SELECT * FROM users`.cursor()
   * while (const rows = await cursor.fetch(100)) {
   *   process(rows)
   * }
   * await cursor.close()
   */
  cursor(): Promise<Cursor<T>>
}

/**
 * Query middleware function signature (forward declaration for types)
 * Full implementation is in middleware.ts
 */
export type QueryMiddlewareFn<T extends Row = Row> = (
  request: { sql: string; params: unknown[]; timestamp: number; queryId: string; context?: Record<string, unknown> },
  next: () => Promise<{ result: QueryResult<T>; durationMs: number; success: boolean; error?: Error; metadata?: Record<string, unknown> }>
) => Promise<{ result: QueryResult<T>; durationMs: number; success: boolean; error?: Error; metadata?: Record<string, unknown> }>

/**
 * Main SQL client interface
 * Compatible with Drizzle ORM integration
 */
export interface Sql {
  /**
   * Execute a SQL query using tagged template literal
   * @example
   * const users = await sql`SELECT * FROM users WHERE id = ${userId}`
   */
  <T extends Row = Row>(
    strings: TemplateStringsArray,
    ...values: unknown[]
  ): PendingQuery<T>

  /**
   * Execute raw SQL (use with caution - no parameterization)
   * This is used by Drizzle ORM for generated queries
   * @param query - Raw SQL query string
   * @param params - Optional query parameters
   */
  unsafe<T extends Row = Row>(
    query: string,
    params?: unknown[]
  ): Promise<T[]>

  /**
   * Begin a transaction
   * Compatible with Drizzle ORM transaction interface
   * @param fn - Transaction callback function
   * @param options - Transaction options
   */
  begin<T>(
    fn: (sql: TransactionSql) => Promise<T>,
    options?: TransactionOptions
  ): Promise<T>

  /**
   * Create a savepoint within a transaction
   * For nested transaction support
   * @param fn - Savepoint callback function
   */
  savepoint<T>(
    fn: (sql: TransactionSql) => Promise<T>
  ): Promise<T>

  /**
   * End the connection/session
   */
  end(): Promise<void>

  /**
   * Stream large result sets using PostgreSQL cursors
   * Keeps memory usage constant regardless of result size
   * @example
   * // Stream rows one at a time
   * for await (const row of sql.stream`SELECT * FROM huge_table`) {
   *   process(row)
   * }
   *
   * // Stream in batches
   * for await (const batch of sql.stream`SELECT * FROM huge_table`.batch(1000)) {
   *   processBatch(batch)
   * }
   *
   * // Manual cursor control
   * const cursor = await sql.stream`SELECT * FROM huge_table`.cursor()
   * while (const rows = await cursor.fetch(100)) {
   *   process(rows)
   * }
   * await cursor.close()
   */
  stream<T extends Row = Row>(
    strings: TemplateStringsArray,
    ...values: unknown[]
  ): StreamResult<T>

  /**
   * Add a query middleware to the chain
   * Middlewares are executed in the order they are added
   * @param middleware - The middleware function to add
   * @returns The same Sql instance for chaining
   * @example
   * const sql = postgres()
   *   .use(loggingMiddleware({ level: 'info' }))
   *   .use(metricsMiddleware({ onMetric: console.log }))
   */
  use(middleware: QueryMiddlewareFn): this

  /**
   * Client configuration options
   * Includes type parsers and serializers for Drizzle compatibility
   */
  options: {
    parsers: Record<number, TypeParser>
    serializers: Record<number, (value: unknown) => string>
    [key: string]: unknown
  }

  /**
   * Reserve a connection from the pool
   * Returns a client with exclusive connection
   */
  reserve(): Promise<ReservedSql>

  /**
   * Commit a prepared transaction (two-phase commit)
   * @param transactionId - The ID returned from prepare()
   */
  commitPrepared(transactionId: string): Promise<void>

  /**
   * Rollback a prepared transaction (two-phase commit)
   * @param transactionId - The ID returned from prepare()
   */
  rollbackPrepared(transactionId: string): Promise<void>

  /**
   * Subscribe to changes on a table
   * @param table - Table name (optionally with schema: "schema.table")
   * @param options - Subscription options
   * @returns A subscription that can be iterated over or used with callbacks
   * @example
   * // Using async iteration
   * const sub = await sql.subscribe('users')
   * for await (const event of sub) {
   *   console.log(event.operation, event.newRow)
   * }
   *
   * // Using callbacks
   * const sub = await sql.subscribe('users', {
   *   onInsert: (row) => console.log('New user:', row),
   *   onUpdate: (newRow, oldRow) => console.log('Updated:', newRow),
   *   onDelete: (oldRow) => console.log('Deleted:', oldRow)
   * })
   */
  subscribe<T extends Row = Row>(
    table: string,
    options?: CDCClientSubscribeOptions<T>
  ): Promise<CDCClientSubscription<T>>

  /**
   * Unsubscribe from a subscription by ID
   * @param subscriptionId - The subscription ID to unsubscribe from
   */
  unsubscribe(subscriptionId: string): Promise<void>

  /**
   * Get all active subscriptions
   * @returns Array of active subscriptions
   */
  subscriptions(): CDCClientSubscriptionInfo[]
}

/** Transaction-specific SQL interface */
export interface TransactionSql extends Omit<Sql, 'begin' | 'end' | 'reserve' | 'commitPrepared' | 'rollbackPrepared'> {
  /**
   * Create a savepoint within the transaction
   */
  savepoint<T>(
    fn: (sql: TransactionSql) => Promise<T>
  ): Promise<T>

  /**
   * Acquire an exclusive advisory lock that is released when the transaction ends (blocking).
   * @param lockId - Lock identifier (number or bigint for single-key)
   * @example await tx.advisoryLock(12345)
   */
  advisoryLock(lockId: number | bigint): Promise<void>
  /**
   * Acquire an exclusive advisory lock with namespace (blocking).
   * @param classId - Lock class/namespace identifier (int4)
   * @param objId - Object identifier within the namespace (int4)
   * @example await tx.advisoryLock(1, 42) // namespace 1, object 42
   */
  advisoryLock(classId: number, objId: number): Promise<void>

  /**
   * Try to acquire an exclusive advisory lock without blocking.
   * @param lockId - Lock identifier (number or bigint for single-key)
   * @returns true if lock was acquired, false otherwise
   * @example const acquired = await tx.tryAdvisoryLock(12345)
   */
  tryAdvisoryLock(lockId: number | bigint): Promise<boolean>
  /**
   * Try to acquire an exclusive advisory lock with namespace without blocking.
   * @param classId - Lock class/namespace identifier (int4)
   * @param objId - Object identifier within the namespace (int4)
   * @returns true if lock was acquired, false otherwise
   * @example const acquired = await tx.tryAdvisoryLock(1, 42)
   */
  tryAdvisoryLock(classId: number, objId: number): Promise<boolean>

  /**
   * Acquire a shared advisory lock that is released when the transaction ends (blocking).
   * Multiple sessions can hold a shared lock simultaneously; blocks only if exclusive lock is held.
   * @param lockId - Lock identifier (number or bigint for single-key)
   * @example await tx.advisoryLockShared(12345)
   */
  advisoryLockShared(lockId: number | bigint): Promise<void>
  /**
   * Acquire a shared advisory lock with namespace (blocking).
   * @param classId - Lock class/namespace identifier (int4)
   * @param objId - Object identifier within the namespace (int4)
   * @example await tx.advisoryLockShared(1, 42)
   */
  advisoryLockShared(classId: number, objId: number): Promise<void>

  /**
   * Try to acquire a shared advisory lock without blocking.
   * @param lockId - Lock identifier (number or bigint for single-key)
   * @returns true if lock was acquired, false if exclusive lock is held
   * @example const acquired = await tx.tryAdvisoryLockShared(12345)
   */
  tryAdvisoryLockShared(lockId: number | bigint): Promise<boolean>
  /**
   * Try to acquire a shared advisory lock with namespace without blocking.
   * @param classId - Lock class/namespace identifier (int4)
   * @param objId - Object identifier within the namespace (int4)
   * @returns true if lock was acquired, false if exclusive lock is held
   * @example const acquired = await tx.tryAdvisoryLockShared(1, 42)
   */
  tryAdvisoryLockShared(classId: number, objId: number): Promise<boolean>

  /**
   * Prepare transaction for two-phase commit
   * After calling this, the transaction must be committed with commitPrepared()
   * @param transactionId - Unique identifier for the prepared transaction
   * @returns The transaction ID
   */
  prepare(transactionId: string): Promise<string>

  /**
   * Register a callback to run after transaction commits
   * @param fn - Callback function
   */
  onCommit(fn: () => void | Promise<void>): void

  /**
   * Register a callback to run after transaction rolls back
   * @param fn - Callback function
   */
  onRollback(fn: () => void | Promise<void>): void

  /**
   * Current transaction ID (xid)
   * Lazily fetched from the database
   */
  readonly txid: string | undefined

  /**
   * Transaction name (if provided in options)
   */
  readonly name: string | undefined

  /**
   * Current retry attempt number (0 for first attempt, 1 for first retry, etc.)
   * Only set when retryOnSerializationFailure or retryOnDeadlock is enabled.
   */
  readonly retryCount: number
}

/** Reserved connection SQL interface */
export interface ReservedSql extends Sql {
  /**
   * Release the reserved connection back to the pool
   */
  release(): Promise<void>
}

/** Transport interface for sending queries */
export interface Transport {
  /** Send a query and receive results */
  query<T extends Row = Row>(
    sql: string,
    params?: unknown[]
  ): Promise<QueryResult<T>>

  /** Execute multiple queries in a transaction */
  transaction<T>(
    queries: Array<{ sql: string; params?: unknown[] }>,
    options?: TransactionOptions
  ): Promise<T>

  /** Close the transport connection */
  close(): Promise<void>

  /** Check if transport is connected */
  isConnected(): boolean
}

/** HTTP transport configuration */
export interface HttpTransportConfig {
  /** Base URL for the postgres.do API */
  baseUrl: string
  /** API key for authentication */
  apiKey?: string | undefined
  /** Custom fetch implementation */
  fetch?: typeof fetch | undefined
  /** Request timeout in milliseconds */
  timeout?: number | undefined
  /** Custom headers */
  headers?: Record<string, string> | undefined
}

/** WebSocket transport configuration */
export interface WsTransportConfig {
  /** WebSocket URL for postgres.do */
  url: string
  /** API key for authentication */
  apiKey?: string | undefined
  /** Custom WebSocket implementation */
  WebSocket?: typeof WebSocket | undefined
  /** Connection timeout in milliseconds */
  connectTimeout?: number | undefined
  /** Ping interval in milliseconds */
  pingInterval?: number | undefined
}

// =============================================================================
// CDC (Change Data Capture) Types
// =============================================================================

/**
 * CDC operation types for PostgreSQL change notifications
 */
export type CDCOperation = 'INSERT' | 'UPDATE' | 'DELETE' | 'TRUNCATE'

/**
 * CDC change event representing a single database change
 */
export interface CDCChangeEvent {
  /** Unique event ID (typically LSN-based) */
  id: string
  /** PostgreSQL Log Sequence Number for this change */
  lsn: string
  /** Type of operation that caused this change */
  operation: CDCOperation
  /** Table name where the change occurred */
  table: string
  /** Schema name */
  schema: string
  /** Timestamp when the change occurred */
  timestamp: Date
  /** New row data (for INSERT and UPDATE) */
  newRow?: Row | undefined
  /** Old row data (for UPDATE and DELETE) */
  oldRow?: Row | undefined
  /** Columns that were changed (for UPDATE) */
  changedColumns?: string[] | undefined
  /** Transaction ID */
  xid: number
  /** Whether this is the last change in the transaction */
  isLastInTransaction: boolean
}

/**
 * Options for CDC subscriptions
 */
export interface CDCSubscriptionOptions {
  /** Event types to subscribe to (default: all) */
  events?: CDCOperation[] | undefined
  /** SQL filter expression for row-level filtering */
  filter?: string | undefined
  /** Resume from this LSN (for exactly-once delivery) */
  resumeFrom?: string | undefined
  /** Include old row data for UPDATE/DELETE events */
  includeOldRow?: boolean | undefined
  /** Track which columns changed for UPDATE events */
  trackChangedColumns?: boolean | undefined
  /** Batch size for event delivery */
  batchSize?: number | undefined
  /** Heartbeat interval in milliseconds */
  heartbeatInterval?: number | undefined
}

/**
 * CDC Transport interface for streaming database changes
 */
export interface CDCTransport {
  /** Connect to the CDC stream */
  connect(): Promise<void>
  /** Disconnect from the CDC stream */
  disconnect(): Promise<void>
  /** Check if connected */
  isConnected(): boolean
  /** Get the last event ID received */
  getLastEventId(): string | null
  /** Subscribe to a table */
  subscribe(
    subscriptionId: string,
    table: string,
    schema: string,
    options: CDCSubscriptionOptions
  ): Promise<void>
  /** Unsubscribe from a subscription */
  unsubscribe(subscriptionId: string): Promise<void>
  /** Set the event handler */
  onEvent(handler: (subscriptionId: string, event: CDCChangeEvent) => void): void
  /** Set the error handler */
  onError(handler: (subscriptionId: string, error: Error) => void): void
  /** Set the reconnect handler */
  onReconnect(handler: (attempt: number, lastLsn: string | undefined) => void): void
  /** Set the close handler */
  onClose(handler: () => void): void
}

/**
 * CDC WebSocket transport configuration
 */
export interface CDCWsTransportConfig {
  /** WebSocket URL for the CDC endpoint */
  url: string
  /** API key for authentication */
  apiKey?: string | undefined
  /** Custom WebSocket implementation */
  WebSocket?: typeof WebSocket | undefined
  /** Connection timeout in milliseconds */
  connectTimeout?: number | undefined
  /** Ping interval for keepalive in milliseconds */
  pingInterval?: number | undefined
}

/**
 * CDC SSE (Server-Sent Events) transport configuration
 */
export interface CDCSseTransportConfig {
  /** URL for the CDC SSE endpoint */
  url: string
  /** API key for authentication */
  apiKey?: string
  /** Resume from this event ID (Last-Event-ID) */
  resumeFrom?: string
  /** Reconnection delay in milliseconds */
  retryMs?: number
  /** Custom fetch implementation */
  fetch?: typeof fetch
  /** Custom headers */
  headers?: Record<string, string>
}

/** PostgreSQL error fields specific to postgres.do */
export interface PostgresErrorFields {
  /** PostgreSQL error code (e.g., '23505' for unique violation) */
  code?: string | undefined
  /** Error severity */
  severity?: string | undefined
  /** Detailed error message */
  detail?: string | undefined
  /** Hint for fixing the error */
  hint?: string | undefined
  /** Position in query where error occurred */
  position?: number | undefined
  /** Schema name if relevant */
  schema?: string | undefined
  /** Table name if relevant */
  table?: string | undefined
  /** Column name if relevant */
  column?: string | undefined
  /** Constraint name if relevant */
  constraint?: string | undefined
}

/**
 * Error thrown by postgres.do client.
 * Extends the shared PostgresError with postgres.do-specific properties.
 */
export class PostgresError extends SharedPostgresError {
  /** Error severity */
  readonly severity: string
  /** Detailed error message */
  readonly detail: string | undefined
  /** Hint for fixing the error */
  readonly hint: string | undefined
  /** Position in query where error occurred */
  readonly position: number | undefined
  /** PostgreSQL error code (e.g., '23505' for unique violation) */
  readonly pgCode: string

  constructor(message: string, fields: PostgresErrorFields = {}) {
    // Build context object only with defined values to satisfy exactOptionalPropertyTypes
    const context: ErrorContext = { pgCode: fields.code ?? 'UNKNOWN' }
    if (fields.schema !== undefined) context.schema = fields.schema
    if (fields.table !== undefined) context.table = fields.table
    if (fields.column !== undefined) context.column = fields.column
    if (fields.constraint !== undefined) context.constraint = fields.constraint

    super(message, {
      code: 'QUERY_ERROR',
      context,
    })
    // Override name property
    Object.defineProperty(this, 'name', { value: 'PostgresError', writable: true, configurable: true })
    // Override code property to return the PostgreSQL error code instead of internal code
    Object.defineProperty(this, 'code', { value: fields.code ?? 'UNKNOWN', writable: false, configurable: true })

    this.severity = fields.severity ?? 'ERROR'
    this.detail = fields.detail
    this.hint = fields.hint
    this.position = fields.position
    this.pgCode = fields.code ?? 'UNKNOWN'
  }

  /** Schema name if relevant */
  get schema(): string | undefined {
    return this.context.schema
  }

  /** Table name if relevant */
  get table(): string | undefined {
    return this.context.table
  }

  /** Column name if relevant */
  get column(): string | undefined {
    return this.context.column
  }

  /** Constraint name if relevant */
  get constraint(): string | undefined {
    return this.context.constraint
  }
}

/**
 * Connection error.
 * Extends the shared ConnectionError.
 */
export class ConnectionError extends SharedConnectionError {
  constructor(message: string) {
    super(message)
    // Override name property
    Object.defineProperty(this, 'name', { value: 'ConnectionError', writable: true, configurable: true })
  }
}

/**
 * Query timeout error.
 * Extends the shared ConnectionTimeoutError.
 */
export class TimeoutError extends SharedConnectionTimeoutError {
  constructor(message: string = 'Query timed out') {
    // Extract timeout from message if possible, default to 0
    const timeoutMatch = message.match(/(\d+)\s*ms/)
    const matchedMs = timeoutMatch?.[1]
    const timeoutMs = matchedMs ? parseInt(matchedMs, 10) : 0
    super(timeoutMs)
    // Override name property
    Object.defineProperty(this, 'name', { value: 'TimeoutError', writable: true, configurable: true })
    // Override the message if a custom one was provided
    if (message !== 'Query timed out' && message !== `Connection timed out after ${timeoutMs}ms`) {
      Object.defineProperty(this, 'message', { value: message, writable: true, configurable: true })
    }
  }
}

// =============================================================================
// CDC Client Types (High-Level API)
// =============================================================================

/**
 * Options for CDC client subscription
 */
export interface CDCClientSubscribeOptions<T extends Row = Row> extends CDCSubscriptionOptions {
  /** Called for every change event */
  onChange?: (event: CDCChangeEvent & { newRow?: T; oldRow?: T }) => void | Promise<void>
  /** Called for INSERT events */
  onInsert?: (row: T) => void | Promise<void>
  /** Called for UPDATE events */
  onUpdate?: (newRow: T, oldRow?: T) => void | Promise<void>
  /** Called for DELETE events */
  onDelete?: (oldRow: T) => void | Promise<void>
  /** Called when an error occurs */
  onError?: (error: Error) => void
}

/**
 * Client subscription interface for CDC streaming
 */
export interface CDCClientSubscription<T extends Row = Row> extends AsyncIterable<CDCChangeEvent & { newRow?: T; oldRow?: T }> {
  /** Unique subscription ID */
  readonly id: string
  /** Table name */
  readonly table: string
  /** Schema name */
  readonly schema: string
  /** Whether the subscription is currently active */
  readonly isActive: boolean
  /** Last processed LSN for resumption */
  readonly lastLsn: string | null

  /** Unsubscribe and stop receiving events */
  unsubscribe(): Promise<void>

  /** Register a handler for all change events */
  onChange(handler: (event: CDCChangeEvent & { newRow?: T; oldRow?: T }) => void | Promise<void>): this

  /** Register a handler for INSERT events */
  onInsert(handler: (row: T) => void | Promise<void>): this

  /** Register a handler for UPDATE events */
  onUpdate(handler: (newRow: T, oldRow?: T) => void | Promise<void>): this

  /** Register a handler for DELETE events */
  onDelete(handler: (oldRow: T) => void | Promise<void>): this
}

/**
 * Information about an active subscription
 */
export interface CDCClientSubscriptionInfo {
  /** Unique subscription ID */
  id: string
  /** Table name */
  table: string
  /** Schema name */
  schema: string
  /** Whether the subscription is currently active */
  isActive: boolean
}