/**
 * PostgreSQL client for postgres.do
 * SQL tagged template function with Drizzle ORM compatibility
 */

import type {
  Sql,
  TransactionSql,
  ReservedSql,
  PendingQuery,
  Row,
  PostgresConfig,
  Transport,
  TransactionOptions,
  TypeParser,
  QueryResult,
  StreamResult,
  CDCTransport,
  CDCClientSubscription,
  CDCClientSubscribeOptions,
  CDCClientSubscriptionInfo,
} from './types'
import { PostgresError, ConnectionError, TimeoutError } from './types'
import { createHttpTransport } from './transport/http'
import { createRpcTransport } from './transport/rpc'
import { createCdcWsTransport } from './transport/cdc-ws'
import { createCdcSseTransport } from './transport/cdc-sse'
import type { QueryMiddleware, QueryRequest } from './middleware'
import { generateQueryId, executeWithMiddleware } from './middleware'
import { createStreamResult, buildParameterizedSql as buildStreamSql } from './streaming'
import {
  CDCSubscription,
  validateTableName,
  generateSubscriptionId,
  type SubscribeOptions,
} from './subscription'

/** Default type parsers for common PostgreSQL types */
const DEFAULT_PARSERS: Record<number, TypeParser> = {
  // Boolean
  16: (value: string) => value === 't' || value === 'true',
  // Integer types
  20: (value: string) => BigInt(value), // int8/bigint
  21: (value: string) => parseInt(value, 10), // int2
  23: (value: string) => parseInt(value, 10), // int4
  26: (value: string) => parseInt(value, 10), // oid
  // Float types
  700: (value: string) => parseFloat(value), // float4
  701: (value: string) => parseFloat(value), // float8
  1700: (value: string) => parseFloat(value), // numeric
  // Date/time types
  1082: (value: string) => value, // date (keep as string for flexibility)
  1083: (value: string) => value, // time
  1114: (value: string) => new Date(value), // timestamp
  1184: (value: string) => new Date(value), // timestamptz
  // JSON types
  114: (value: string) => JSON.parse(value), // json
  3802: (value: string) => JSON.parse(value), // jsonb
  // Array types are handled by the server
}

/**
 * Parse a postgres:// URL into config
 */
function parseConnectionUrl(url: string): Partial<PostgresConfig> {
  try {
    const parsed = new URL(url)
    const config: Partial<PostgresConfig> = {}

    // Extract protocol (postgres:// or postgresql://)
    if (!parsed.protocol.startsWith('postgres')) {
      throw new Error('Invalid protocol: must be postgres:// or postgresql://')
    }

    // Use the host directly for postgres.do
    // URL format: postgres://db.postgres.do/database_name
    const baseUrl = `https://${parsed.host}`

    // Extract API key from password or query param
    if (parsed.password) {
      config.apiKey = decodeURIComponent(parsed.password)
    }
    if (parsed.searchParams.has('apiKey')) {
      config.apiKey = parsed.searchParams.get('apiKey')!
    }

    // SSL setting - use conditional to avoid exactOptionalPropertyTypes issue
    const sslMode = parsed.searchParams.get('sslmode')
    if (sslMode && (sslMode === 'require' || sslMode === 'prefer' || sslMode === 'allow' || sslMode === 'disable')) {
      config.ssl = sslMode
    }

    // Store the URL for transport
    config.url = baseUrl + parsed.pathname

    return config
  } catch (error) {
    throw new ConnectionError(`Invalid connection URL: ${error instanceof Error ? error.message : String(error)}`)
  }
}

/**
 * Build parameterized SQL from template strings
 */
function buildParameterizedSql(
  strings: TemplateStringsArray,
  values: unknown[]
): string {
  const parts: string[] = []
  for (let i = 0; i < strings.length; i++) {
    parts.push(strings[i]!)
    if (i < values.length) {
      parts.push(`$${i + 1}`)
    }
  }
  return parts.join('')
}

/**
 * Create a pending query that can be executed
 */
function createPendingQuery<T extends Row>(
  transport: Transport,
  strings: TemplateStringsArray,
  values: unknown[],
  parsers: Record<number, TypeParser>,
  middlewares: QueryMiddleware[] = []
): PendingQuery<T> {
  const sql = buildParameterizedSql(strings, values)

  // Create the promise that executes the query
  const execute = async (): Promise<T[]> => {
    if (middlewares.length === 0) {
      // No middlewares - execute directly
      const result = await transport.query<T>(sql, values)
      return parseRows(result.rows, result.fields, parsers)
    }

    // Execute through middleware chain
    const request: QueryRequest = {
      sql,
      params: values,
      timestamp: Date.now(),
      queryId: generateQueryId(),
    }

    const response = await executeWithMiddleware<T>(
      middlewares as QueryMiddleware<T>[],
      request,
      async () => transport.query<T>(sql, values)
    )

    if (!response.success && response.error) {
      throw response.error
    }

    return parseRows(response.result.rows, response.result.fields, parsers)
  }

  // Create the pending query object
  const promise = execute() as PendingQuery<T>

  // Add metadata
  Object.defineProperty(promise, 'strings', { value: strings })
  Object.defineProperty(promise, 'values', { value: values })
  Object.defineProperty(promise, 'execute', { value: execute })
  Object.defineProperty(promise, 'toString', {
    value: () => sql,
  })

  return promise
}

/**
 * Parse rows using type parsers
 */
function parseRows<T extends Row>(
  rows: T[],
  fields: QueryResult['fields'],
  parsers: Record<number, TypeParser>
): T[] {
  if (!fields || fields.length === 0) {
    return rows
  }

  // Create a map of column name to parser
  const columnParsers = new Map<string, TypeParser>()
  for (const field of fields) {
    const parser = parsers[field.dataTypeID]
    if (parser) {
      columnParsers.set(field.name, parser)
    }
  }

  // If no parsers apply, return rows as-is
  if (columnParsers.size === 0) {
    return rows
  }

  // Parse each row
  return rows.map((row) => {
    const parsed: Record<string, unknown> = {}
    for (const [key, value] of Object.entries(row)) {
      const parser = columnParsers.get(key)
      if (parser && typeof value === 'string') {
        try {
          parsed[key] = parser(value)
        } catch {
          parsed[key] = value // Keep original if parsing fails
        }
      } else {
        parsed[key] = value
      }
    }
    return parsed as T
  })
}

// ============================================================================
// Advisory Lock Helpers
// ============================================================================

/** PostgreSQL bigint max value: 2^63 - 1 */
const BIGINT_MAX = BigInt('9223372036854775807')

/** PostgreSQL int4 max value: 2^31 - 1 */
const INT4_MAX = 2147483647

/**
 * Validate a single advisory lock key.
 *
 * Lock keys must be:
 * - Non-negative integers
 * - Within PostgreSQL bigint range (0 to 2^63 - 1)
 *
 * @param key - The lock key to validate
 * @throws {PostgresError} If the key is invalid (PostgreSQL error code '22023')
 */
function validateLockKey(key: number | bigint): void {
  // Check for non-integer FIRST (before BigInt conversion throws its own error)
  if (typeof key === 'number' && !Number.isInteger(key)) {
    throw new PostgresError('Advisory lock key must be an integer', {
      code: '22023',
      severity: 'ERROR',
    })
  }

  const keyBigInt = BigInt(key)

  // Check for negative values
  if (keyBigInt < BigInt(0)) {
    throw new PostgresError('Advisory lock key must be non-negative', {
      code: '22023',
      severity: 'ERROR',
    })
  }

  // Check for overflow (PostgreSQL bigint max)
  if (keyBigInt > BIGINT_MAX) {
    throw new PostgresError('Advisory lock key exceeds PostgreSQL bigint maximum', {
      code: '22023',
      severity: 'ERROR',
    })
  }
}

/**
 * Validate a two-key advisory lock pair.
 *
 * Two-key locks use PostgreSQL's int4 range for both classid and objid:
 * - Non-negative integers
 * - Within int4 range (0 to 2^31 - 1)
 *
 * @param classId - The lock class/namespace identifier
 * @param objId - The object identifier within the namespace
 * @throws {PostgresError} If either key is invalid (PostgreSQL error code '22023')
 */
function validateTwoKeyLock(classId: number, objId: number): void {
  // Check for non-integers
  if (!Number.isInteger(classId) || !Number.isInteger(objId)) {
    throw new PostgresError('Advisory lock keys must be integers', {
      code: '22023',
      severity: 'ERROR',
    })
  }

  // Check for negative values
  if (classId < 0 || objId < 0) {
    throw new PostgresError('Advisory lock keys must be non-negative', {
      code: '22023',
      severity: 'ERROR',
    })
  }

  // Check for int4 overflow
  if (classId > INT4_MAX || objId > INT4_MAX) {
    throw new PostgresError('Advisory lock key exceeds PostgreSQL int4 maximum', {
      code: '22023',
      severity: 'ERROR',
    })
  }
}

/**
 * Build the SQL for an advisory lock operation.
 *
 * @param lockFn - The PostgreSQL lock function name
 * @param args - The lock key arguments (single bigint or two int4s)
 * @returns The SQL string for the lock operation
 */
function buildAdvisoryLockSql(lockFn: string, ...args: Array<number | bigint>): string {
  if (args.length === 1) {
    return `SELECT ${lockFn}(${BigInt(args[0]!)})`
  }
  return `SELECT ${lockFn}(${args[0]}, ${args[1]})`
}

// ============================================================================
// Two-Phase Commit (2PC) Helpers
// ============================================================================

/**
 * Regular expression for validating two-phase commit transaction IDs.
 *
 * Transaction IDs must:
 * - Start with a letter or underscore
 * - Contain only alphanumeric characters and underscores
 * - Be non-empty
 *
 * This prevents SQL injection in PREPARE TRANSACTION, COMMIT PREPARED,
 * and ROLLBACK PREPARED statements.
 */
const TRANSACTION_ID_REGEX = /^[a-zA-Z_][a-zA-Z0-9_]*$/

/**
 * Validate a transaction ID for two-phase commit operations.
 *
 * Two-phase commit (2PC) allows a transaction to be prepared on one
 * connection and committed/rolled back on another, enabling distributed
 * transactions across multiple databases or services.
 *
 * @param transactionId - The transaction ID to validate
 * @throws {PostgresError} If the transaction ID is invalid (PostgreSQL error code '22023')
 *
 * @example
 * ```typescript
 * validateTransactionId('my_tx_123')  // OK
 * validateTransactionId('_private')   // OK
 * validateTransactionId('123_bad')    // throws - cannot start with number
 * validateTransactionId('has space')  // throws - no spaces allowed
 * validateTransactionId("'; DROP")    // throws - SQL injection attempt blocked
 * ```
 */
function validateTransactionId(transactionId: string): void {
  if (!TRANSACTION_ID_REGEX.test(transactionId)) {
    throw new PostgresError(
      'Invalid transaction ID: must be alphanumeric with underscores, starting with letter or underscore',
      {
        code: '22023',
        severity: 'ERROR',
      }
    )
  }
}

// ============================================================================
// Transaction Hook Helpers
// ============================================================================

/**
 * Execute an array of transaction lifecycle callbacks sequentially.
 *
 * This is used for transaction hooks (onCommit/onRollback) where:
 * - All callbacks should be attempted even if earlier ones fail
 * - Callback errors are logged but not propagated to the caller
 * - Callbacks execute in registration order (FIFO)
 *
 * @param callbacks - Array of callback functions to execute
 * @param phase - The phase name for error logging ('commit' or 'rollback')
 */
async function executeTransactionCallbacks(
  callbacks: Array<() => void | Promise<void>>,
  phase: 'commit' | 'rollback'
): Promise<void> {
  for (const callback of callbacks) {
    try {
      await callback()
    } catch (error) {
      // Log but don't propagate - ensures all callbacks are attempted
      console.error(`Transaction ${phase} callback error:`, error)
    }
  }
}

// ============================================================================
// Transaction Context and State
// ============================================================================

/**
 * Transaction context for tracking mutable state during a transaction.
 *
 * This context is created fresh for each transaction attempt and holds:
 * - Transaction metadata (txid, name)
 * - Lifecycle hooks (onCommit, onRollback callbacks)
 * - Two-phase commit state (prepared flag and transaction ID)
 * - Retry state (current attempt number)
 */
interface TransactionContext {
  /** PostgreSQL transaction ID fetched via txid_current() */
  txid: string | undefined
  /** User-provided transaction name for debugging/logging */
  txName: string | undefined
  /** Callbacks to execute after successful COMMIT */
  onCommitCallbacks: Array<() => void | Promise<void>>
  /** Callbacks to execute after ROLLBACK (errors ignored) */
  onRollbackCallbacks: Array<() => void | Promise<void>>
  /** Whether PREPARE TRANSACTION has been called (2PC mode) */
  isPrepared: boolean
  /** The transaction ID used in PREPARE TRANSACTION */
  preparedTxId: string | undefined
  /** Current retry attempt number (0 for first attempt, 1 for first retry, etc.) */
  retryCount: number
}

/**
 * Create a transaction SQL client
 */
function createTransactionSql(
  transport: Transport,
  parsers: Record<number, TypeParser>,
  savepointCounter: { count: number },
  txContext: TransactionContext
): TransactionSql {
  // Create the base tagged template function
  const taggedTemplate = <T extends Row = Row>(
    strings: TemplateStringsArray,
    ...values: unknown[]
  ): PendingQuery<T> => {
    return createPendingQuery<T>(transport, strings, values, parsers)
  }

  // Middleware array for TransactionSql (usually empty within transaction)
  const txMiddlewares: QueryMiddleware[] = []

  // Create the transaction sql object with all required methods
  const txSql = Object.assign(taggedTemplate, {
    unsafe: async <T extends Row = Row>(
      query: string,
      params?: unknown[]
    ): Promise<T[]> => {
      const result = await transport.query<T>(query, params)
      return parseRows(result.rows, result.fields, parsers)
    },

    savepoint: async <T>(
      fn: (sql: TransactionSql) => Promise<T>
    ): Promise<T> => {
      const savepointName = `sp_${++savepointCounter.count}`

      // Create savepoint
      await transport.query(`SAVEPOINT ${savepointName}`)

      try {
        const result = await fn(txSql as unknown as TransactionSql)
        // Release savepoint on success
        await transport.query(`RELEASE SAVEPOINT ${savepointName}`)
        return result
      } catch (error) {
        // Rollback to savepoint on error
        await transport.query(`ROLLBACK TO SAVEPOINT ${savepointName}`)
        throw error
      }
    },

    stream: <T extends Row = Row>(
      strings: TemplateStringsArray,
      ...values: unknown[]
    ): StreamResult<T> => {
      const { sql: query, params } = buildStreamSql(strings, values)
      return createStreamResult<T>(transport, query, params, parsers)
    },

    use: function (middleware: QueryMiddleware): TransactionSql {
      txMiddlewares.push(middleware)
      return txSql as unknown as TransactionSql
    },

    options: {
      parsers,
      serializers: {} as Record<number, (value: unknown) => string>,
    },

    // =========================================================================
    // Advisory Lock Methods
    // =========================================================================
    // PostgreSQL advisory locks are application-level cooperative locks.
    // Transaction-scoped locks (*_xact_*) are automatically released when
    // the transaction ends (commit or rollback).

    /**
     * Acquire an exclusive advisory lock (blocking).
     * Supports single-key (bigint) and two-key (int4, int4) variants.
     */
    advisoryLock: async (keyOrClassId: number | bigint, objId?: number): Promise<void> => {
      if (objId !== undefined) {
        // Two-key variant: pg_advisory_xact_lock(classid, objid)
        validateTwoKeyLock(keyOrClassId as number, objId)
        await transport.query(buildAdvisoryLockSql('pg_advisory_xact_lock', keyOrClassId as number, objId))
      } else {
        // Single-key variant: pg_advisory_xact_lock(key)
        validateLockKey(keyOrClassId)
        await transport.query(buildAdvisoryLockSql('pg_advisory_xact_lock', keyOrClassId))
      }
    },

    /**
     * Try to acquire an exclusive advisory lock (non-blocking).
     * Returns true if lock acquired, false if lock is held by another session.
     */
    tryAdvisoryLock: async (keyOrClassId: number | bigint, objId?: number): Promise<boolean> => {
      let sql: string
      if (objId !== undefined) {
        validateTwoKeyLock(keyOrClassId as number, objId)
        sql = buildAdvisoryLockSql('pg_try_advisory_xact_lock', keyOrClassId as number, objId)
      } else {
        validateLockKey(keyOrClassId)
        sql = buildAdvisoryLockSql('pg_try_advisory_xact_lock', keyOrClassId)
      }
      const result = await transport.query<{ pg_try_advisory_xact_lock: boolean }>(sql)
      return result.rows[0]?.pg_try_advisory_xact_lock ?? false
    },

    /**
     * Acquire a shared advisory lock (blocking).
     * Multiple sessions can hold a shared lock simultaneously.
     * Blocks if an exclusive lock is held.
     */
    advisoryLockShared: async (keyOrClassId: number | bigint, objId?: number): Promise<void> => {
      if (objId !== undefined) {
        validateTwoKeyLock(keyOrClassId as number, objId)
        await transport.query(buildAdvisoryLockSql('pg_advisory_xact_lock_shared', keyOrClassId as number, objId))
      } else {
        validateLockKey(keyOrClassId)
        await transport.query(buildAdvisoryLockSql('pg_advisory_xact_lock_shared', keyOrClassId))
      }
    },

    /**
     * Try to acquire a shared advisory lock (non-blocking).
     * Returns true if lock acquired, false if an exclusive lock is held.
     */
    tryAdvisoryLockShared: async (keyOrClassId: number | bigint, objId?: number): Promise<boolean> => {
      let sql: string
      if (objId !== undefined) {
        validateTwoKeyLock(keyOrClassId as number, objId)
        sql = buildAdvisoryLockSql('pg_try_advisory_xact_lock_shared', keyOrClassId as number, objId)
      } else {
        validateLockKey(keyOrClassId)
        sql = buildAdvisoryLockSql('pg_try_advisory_xact_lock_shared', keyOrClassId)
      }
      const result = await transport.query<{ pg_try_advisory_xact_lock_shared: boolean }>(sql)
      return result.rows[0]?.pg_try_advisory_xact_lock_shared ?? false
    },

    // Two-phase commit support - allows distributed transaction coordination
    prepare: async (transactionId: string): Promise<string> => {
      validateTransactionId(transactionId)
      txContext.isPrepared = true
      txContext.preparedTxId = transactionId
      await transport.query(`PREPARE TRANSACTION '${transactionId}'`)
      return transactionId
    },

    // Transaction hooks
    onCommit: (fn: () => void | Promise<void>): void => {
      txContext.onCommitCallbacks.push(fn)
    },

    onRollback: (fn: () => void | Promise<void>): void => {
      txContext.onRollbackCallbacks.push(fn)
    },
  }) as unknown as TransactionSql

  // Define 'txid' as a getter that reads from context (Object.assign doesn't preserve getters).
  // The txid is eagerly fetched after BEGIN via txid_current() and stored in txContext.
  Object.defineProperty(txSql, 'txid', {
    get: () => txContext.txid,
    configurable: true,
    enumerable: true,
  })

  // Define 'name' as a getter for the user-provided transaction name (function.name is read-only)
  Object.defineProperty(txSql, 'name', {
    get: () => txContext.txName,
    configurable: true,
    enumerable: true,
  })

  // Define 'retryCount' as a getter for the current retry attempt number
  Object.defineProperty(txSql, 'retryCount', {
    get: () => txContext.retryCount,
    configurable: true,
    enumerable: true,
  })

  return txSql
}

/**
 * Create the main SQL client
 */
export function createClient(config: PostgresConfig = {}): Sql {
  // Merge default parsers with custom parsers
  const parsers: Record<number, TypeParser> = {
    ...DEFAULT_PARSERS,
    ...config.parsers,
  }

  // Determine base URL
  let baseUrl = config.url || 'https://db.postgres.do'
  if (!baseUrl.startsWith('http')) {
    // Parse as postgres:// URL
    const parsed = parseConnectionUrl(baseUrl)
    baseUrl = parsed.url || 'https://db.postgres.do'
    config = { ...config, ...parsed }
  }

  // Create transport
  const transport: Transport = config.transport === 'ws'
    ? createRpcTransport({
        url: baseUrl.replace('https://', 'wss://').replace('http://', 'ws://') + '/rpc',
        apiKey: config.apiKey,
        WebSocket: config.WebSocket,
        connectTimeout: config.connectTimeout,
      })
    : createHttpTransport({
        baseUrl,
        apiKey: config.apiKey,
        fetch: config.fetch,
        timeout: config.queryTimeout,
      })

  // Middleware array - mutable to support .use() chaining
  const middlewares: QueryMiddleware[] = []

  /**
   * Main SQL tagged template function
   */
  const sql = function <T extends Row = Row>(
    strings: TemplateStringsArray,
    ...values: unknown[]
  ): PendingQuery<T> {
    return createPendingQuery<T>(transport, strings, values, parsers, middlewares)
  } as Sql

  /**
   * Execute raw SQL query (used by Drizzle ORM)
   */
  sql.unsafe = async <T extends Row = Row>(
    query: string,
    params?: unknown[]
  ): Promise<T[]> => {
    if (middlewares.length === 0) {
      const result = await transport.query<T>(query, params)
      return parseRows(result.rows, result.fields, parsers)
    }

    // Execute through middleware chain
    const request: QueryRequest = {
      sql: query,
      params: params || [],
      timestamp: Date.now(),
      queryId: generateQueryId(),
    }

    const response = await executeWithMiddleware<T>(
      middlewares as QueryMiddleware<T>[],
      request,
      async () => transport.query<T>(query, params)
    )

    if (!response.success && response.error) {
      throw response.error
    }

    return parseRows(response.result.rows, response.result.fields, parsers)
  }

  /**
   * Begin a transaction
   */
  sql.begin = async <T>(
    fn: (sql: TransactionSql) => Promise<T>,
    options?: TransactionOptions
  ): Promise<T> => {
    // Merge client-level transactionRetry defaults with transaction-level options
    // Transaction-level options override client-level defaults
    const clientRetryConfig = config.transactionRetry ?? {}
    const maxRetries = options?.maxRetries ?? clientRetryConfig.maxRetries ?? 3
    const shouldRetryOnSerializationFailure = options?.retryOnSerializationFailure ?? clientRetryConfig.retryOnSerializationFailure ?? false
    const shouldRetryOnDeadlock = options?.retryOnDeadlock ?? clientRetryConfig.retryOnDeadlock ?? false
    const backoffMs = options?.backoffMs ?? clientRetryConfig.backoffMs ?? 10
    const onRetry = options?.onRetry

    /**
     * Check if an error is retryable based on the configured options.
     * Retryable errors are:
     * - 40001: Serialization failure (when retryOnSerializationFailure is enabled)
     * - 40P01: Deadlock detected (when retryOnDeadlock is enabled)
     */
    const isRetryableError = (error: unknown): boolean => {
      if (!(error instanceof PostgresError)) {
        return false
      }
      if (shouldRetryOnSerializationFailure && error.pgCode === '40001') {
        return true
      }
      if (shouldRetryOnDeadlock && error.pgCode === '40P01') {
        return true
      }
      return false
    }

    // Track start time for total timeout across retries
    // Note: totalTimeout can be 0 (immediate timeout), undefined (no timeout), or negative (treated as no timeout)
    // Fall back to client-level queryTimeout if transaction-level timeout is not specified
    const effectiveTimeout = options?.timeout ?? config.queryTimeout
    // Negative timeouts are treated as "no timeout" for backward compatibility
    const hasTotalTimeout = effectiveTimeout !== undefined && effectiveTimeout >= 0
    const totalTimeout = hasTotalTimeout ? effectiveTimeout : undefined
    const startTime = hasTotalTimeout ? Date.now() : 0
    const perRetryTimeout = options?.perRetryTimeout

    /**
     * Calculate the timeout for the current attempt.
     *
     * Timeout precedence:
     * 1. If `timeout` is set (including 0), use the remaining time from the total timeout
     * 2. If only `perRetryTimeout` is set, use it directly (fresh timeout per attempt)
     * 3. If neither is set, return undefined (no timeout)
     *
     * Returns undefined if no timeout is set.
     * Throws TimeoutError if total timeout has been exceeded.
     */
    const getAttemptTimeout = (): number | undefined => {
      // Total timeout takes precedence (including 0)
      if (hasTotalTimeout && totalTimeout !== undefined) {
        const elapsed = Date.now() - startTime
        const remaining = totalTimeout - elapsed
        if (remaining <= 0) {
          throw new TimeoutError(`Transaction timed out after ${totalTimeout}ms`)
        }
        return remaining
      }

      // Per-retry timeout: each attempt gets a fresh timeout
      if (perRetryTimeout) {
        return perRetryTimeout
      }

      return undefined
    }

    const executeTransaction = async (attempt: number): Promise<T> => {
      // Check if we've exceeded total timeout before starting a new attempt
      // This check is for retry attempts where time has already elapsed
      if (hasTotalTimeout && attempt > 1) {
        getAttemptTimeout() // Throws if timeout exceeded during retries
      }

      // Build BEGIN statement with options
      const beginParts = ['BEGIN']

      if (options?.isolationLevel) {
        // Validate isolation level to prevent SQL injection
        const validLevels = ['read uncommitted', 'read committed', 'repeatable read', 'serializable']
        const normalized = options.isolationLevel.toLowerCase()
        if (!validLevels.includes(normalized)) {
          throw new Error(`Invalid isolation level: "${options.isolationLevel}"`)
        }
        beginParts.push(`ISOLATION LEVEL ${normalized.toUpperCase()}`)
      }
      if (options?.readOnly) {
        beginParts.push('READ ONLY')
      }
      if (options?.deferrable) {
        beginParts.push('DEFERRABLE')
      }

      // Start transaction - may fail with serialization failure in rare cases
      try {
        await transport.query(beginParts.join(' '))
      } catch (beginError) {
        // Check if BEGIN failed with a retryable error
        if (isRetryableError(beginError) && attempt < maxRetries) {
          // Call onRetry callback if provided
          if (onRetry) {
            await onRetry(attempt, beginError as Error)
          }
          // Exponential backoff
          const delay = backoffMs * Math.pow(2, attempt - 1)
          await new Promise((resolve) => setTimeout(resolve, delay))
          return executeTransaction(attempt + 1)
        }
        throw beginError
      }

      // Create transaction context
      const txContext: TransactionContext = {
        txid: undefined,
        txName: options?.name,
        onCommitCallbacks: [],
        onRollbackCallbacks: [],
        isPrepared: false,
        preparedTxId: undefined,
        retryCount: attempt - 1, // 0 for first attempt, 1 for first retry, etc.
      }

      const savepointCounter = { count: 0 }
      const txSql = createTransactionSql(transport, parsers, savepointCounter, txContext)

      // Eagerly fetch PostgreSQL transaction ID (xid) so it's available throughout the callback.
      // This queries txid_current() which returns the internal transaction identifier,
      // useful for debugging, logging, and correlating with pg_stat_activity or pg_locks.
      // Errors are silently ignored since txid is optional debugging metadata.
      try {
        const txidResult = await transport.query<{ txid_current: string }>('SELECT txid_current()::text')
        txContext.txid = txidResult.rows[0]?.txid_current
      } catch {
        // Non-fatal: txid is optional metadata for debugging/logging
      }

      // Handle timeout if specified - uses remaining time from total timeout or per-retry timeout
      // The timeout wraps the entire transaction including user callback AND COMMIT
      let timeoutHandle: ReturnType<typeof setTimeout> | undefined
      let timedOut = false

      const attemptTimeout = getAttemptTimeout()
      if (attemptTimeout !== undefined) {
        timeoutHandle = setTimeout(() => {
          timedOut = true
        }, attemptTimeout)
      }

      /**
       * Check if we've timed out and throw if so.
       * This is called at key points during the transaction.
       */
      const checkTimeout = (): void => {
        if (timedOut) {
          const timeoutForMessage = totalTimeout ?? perRetryTimeout ?? attemptTimeout
          throw new TimeoutError(`Transaction timed out after ${timeoutForMessage}ms`)
        }
      }

      try {
        // Execute user callback
        const result = await fn(txSql)

        // Check for timeout after user callback
        checkTimeout()

        // If transaction was prepared for 2PC, don't COMMIT
        if (txContext.isPrepared) {
          // onCommit callbacks will be called when commitPrepared is called
          if (timeoutHandle) clearTimeout(timeoutHandle)
          return result
        }

        // COMMIT is also covered by the timeout
        await transport.query('COMMIT')

        // Check for timeout after COMMIT
        checkTimeout()

        // Clear timeout after successful commit
        if (timeoutHandle) clearTimeout(timeoutHandle)

        // Execute onCommit callbacks (errors logged but not propagated)
        await executeTransactionCallbacks(txContext.onCommitCallbacks, 'commit')

        return result
      } catch (error) {
        // Clear timeout if still pending
        if (timeoutHandle) {
          clearTimeout(timeoutHandle)
        }

        // If we timed out during execution, throw TimeoutError
        if (timedOut && !(error instanceof TimeoutError)) {
          const timeoutForMessage = totalTimeout ?? perRetryTimeout ?? attemptTimeout
          throw new TimeoutError(`Transaction timed out after ${timeoutForMessage}ms`)
        }

        // Always attempt ROLLBACK on error (including timeout) unless prepared for 2PC.
        // Even if the transaction timed out, PostgreSQL still has an open transaction
        // that needs cleanup to avoid leaving connections in bad state.
        if (!txContext.isPrepared) {
          try {
            await transport.query('ROLLBACK')
          } catch {
            // Ignore rollback errors - connection might be in invalid state
          }
        }

        // Execute onRollback callbacks (errors logged but not propagated)
        await executeTransactionCallbacks(txContext.onRollbackCallbacks, 'rollback')

        // Don't retry if we've hit a timeout
        if (error instanceof TimeoutError) {
          throw error
        }

        // Check for retryable errors (serialization failure or deadlock)
        if (isRetryableError(error) && attempt < maxRetries) {
          // Call onRetry callback if provided
          if (onRetry) {
            await onRetry(attempt, error as Error)
          }
          // Exponential backoff
          const delay = backoffMs * Math.pow(2, attempt - 1)
          await new Promise((resolve) => setTimeout(resolve, delay))
          return executeTransaction(attempt + 1)
        }

        throw error
      }
    }

    return executeTransaction(1)
  }

  /**
   * Create a savepoint (only valid within a transaction)
   */
  sql.savepoint = async <T>(
    _fn: (sql: TransactionSql) => Promise<T>
  ): Promise<T> => {
    throw new PostgresError('savepoint() can only be called within a transaction', {
      code: '25001',
      severity: 'ERROR',
    })
  }

  /**
   * Stream large result sets using PostgreSQL cursors
   * Memory stays constant regardless of result size
   */
  sql.stream = <T extends Row = Row>(
    strings: TemplateStringsArray,
    ...values: unknown[]
  ): StreamResult<T> => {
    const { sql: query, params } = buildStreamSql(strings, values)
    return createStreamResult<T>(transport, query, params, parsers)
  }

  /**
   * End the connection
   */
  sql.end = async (): Promise<void> => {
    await transport.close()
  }

  /**
   * Reserve a connection (for connection pooling scenarios)
   */
  sql.reserve = async (): Promise<ReservedSql> => {
    // For HTTP transport, we create a new instance
    // For WS transport, we could potentially implement true connection reservation
    const reservedSql = createClient(config) as ReservedSql
    reservedSql.release = async () => {
      await reservedSql.end()
    }
    return reservedSql
  }

  /**
   * Commit a prepared transaction (two-phase commit)
   */
  sql.commitPrepared = async (transactionId: string): Promise<void> => {
    validateTransactionId(transactionId)
    await transport.query(`COMMIT PREPARED '${transactionId}'`)
  }

  /**
   * Rollback a prepared transaction (two-phase commit)
   */
  sql.rollbackPrepared = async (transactionId: string): Promise<void> => {
    validateTransactionId(transactionId)
    await transport.query(`ROLLBACK PREPARED '${transactionId}'`)
  }

  /**
   * Add a middleware to the chain
   * Returns the same Sql instance for chaining
   */
  sql.use = function (middleware: QueryMiddleware): Sql {
    middlewares.push(middleware)
    return sql
  }

  /**
   * Options object (for Drizzle compatibility)
   * Includes both parsers and serializers that Drizzle expects
   */
  sql.options = {
    parsers,
    serializers: {} as Record<number, (value: unknown) => string>,
  }

  // ===========================================================================
  // CDC Subscription Implementation
  // ===========================================================================
  //
  // CDC (Change Data Capture) provides real-time streaming of database changes.
  // This implementation uses a single shared transport (WebSocket or SSE) for
  // all subscriptions to minimize resource usage and connection overhead.

  // ---------------------------------------------------------------------------
  // CDC State Management
  // ---------------------------------------------------------------------------
  //
  // Transport lifecycle:
  //   null -> pending (creating) -> connected -> null (disconnected)
  //
  // cdcTransport: Holds the active connection once established
  // cdcTransportPending: Acts as a mutex/lock to prevent duplicate transports
  //                      when multiple subscribe() calls race during initialization
  let cdcTransport: CDCTransport | null = null
  let cdcTransportPending: Promise<CDCTransport> | null = null
  const activeSubscriptions = new Map<string, CDCSubscription>()

  /**
   * Create a CDC transport instance based on configuration.
   *
   * Encapsulates transport creation logic, keeping the main concurrency
   * control function focused on its primary responsibility.
   *
   * @param cdcUrl - The CDC endpoint URL (wss:// for WebSocket, https:// for SSE)
   * @returns The created transport instance (not yet connected)
   */
  function createCdcTransportInstance(cdcUrl: string): CDCTransport {
    if (config.transport === 'ws') {
      // WebSocket transport: bidirectional, lower latency, requires WS support
      const wsConfig: Parameters<typeof createCdcWsTransport>[0] = { url: cdcUrl }
      if (config.apiKey !== undefined) wsConfig.apiKey = config.apiKey
      if (config.WebSocket !== undefined) wsConfig.WebSocket = config.WebSocket
      if (config.connectTimeout !== undefined) wsConfig.connectTimeout = config.connectTimeout
      return createCdcWsTransport(wsConfig)
    } else {
      // SSE transport: HTTP-based, works through proxies, wider compatibility
      const sseConfig: Parameters<typeof createCdcSseTransport>[0] = { url: cdcUrl }
      if (config.apiKey !== undefined) sseConfig.apiKey = config.apiKey
      if (config.fetch !== undefined) sseConfig.fetch = config.fetch
      return createCdcSseTransport(sseConfig)
    }
  }

  /**
   * Configure event handlers for a CDC transport.
   *
   * Routes transport-level events to the appropriate subscription instances
   * and handles transport lifecycle events (close/disconnect).
   *
   * @param transport - The CDC transport to configure
   */
  function configureCdcTransportHandlers(transport: CDCTransport): void {
    // Route CDC change events to the subscription that owns them
    transport.onEvent((subscriptionId: string, event) => {
      const subscription = activeSubscriptions.get(subscriptionId)
      if (subscription) {
        subscription.handleEvent(event)
      }
    })

    // Route errors to the subscription that should handle them
    transport.onError((subscriptionId: string, error) => {
      const subscription = activeSubscriptions.get(subscriptionId)
      if (subscription) {
        subscription.handleError(error)
      }
    })

    // Handle transport close: mark all subscriptions inactive and reset state
    transport.onClose(() => {
      activeSubscriptions.forEach(subscription => {
        subscription.setInactive()
      })
      activeSubscriptions.clear()
      // Reset transport state to allow fresh reconnection on next subscribe()
      cdcTransport = null
      cdcTransportPending = null
    })
  }

  /**
   * Get or create the shared CDC transport.
   *
   * CONCURRENCY CONTROL - PENDING PROMISE PATTERN:
   * -----------------------------------------------
   * This function uses a "pending promise" pattern to serialize transport
   * creation and prevent race conditions when multiple subscribe() calls
   * happen concurrently. Without this pattern, the following race occurs:
   *
   *   Call A: checks cdcTransport (null) -> starts creating transport
   *   Call B: checks cdcTransport (null) -> starts creating ANOTHER transport
   *   Call A: finishes, sets cdcTransport
   *   Call B: finishes, OVERWRITES cdcTransport (resource leak!)
   *
   * The pattern works in four steps:
   *
   * STEP 1 - FAST PATH: If cdcTransport exists, return it immediately.
   *          This is the common case after initial connection.
   *
   * STEP 2 - AWAIT PENDING: If cdcTransportPending is set, another call is
   *          already creating the transport. Await that same promise to share
   *          the result. This prevents duplicate transports.
   *
   * STEP 3 - CREATE NEW: We're the first caller. Set cdcTransportPending
   *          SYNCHRONOUSLY (before any await) so concurrent calls will see it
   *          and wait for our promise. Then perform async creation.
   *
   * STEP 4 - CLEANUP: In the finally block, clear cdcTransportPending.
   *          On success: cdcTransport is set, subsequent calls use fast path
   *          On failure: cdcTransport is null, next call can retry fresh
   *
   * @returns The shared CDC transport instance
   * @throws {ConnectionError} If transport creation or connection fails
   */
  const getOrCreateCdcTransport = async (): Promise<CDCTransport> => {
    // -------------------------------------------------------------------------
    // STEP 1: FAST PATH - Return existing transport
    // -------------------------------------------------------------------------
    if (cdcTransport) {
      return cdcTransport
    }

    // -------------------------------------------------------------------------
    // STEP 2: AWAIT PENDING - Join existing creation in progress
    // -------------------------------------------------------------------------
    // CRITICAL: This check must happen BEFORE we set cdcTransportPending below.
    // cdcTransportPending acts as a mutex - if set, await it instead of racing.
    if (cdcTransportPending) {
      return cdcTransportPending
    }

    // -------------------------------------------------------------------------
    // STEP 3: CREATE NEW - We're first, set pending SYNCHRONOUSLY then await
    // -------------------------------------------------------------------------
    // Set cdcTransportPending IMMEDIATELY (no await before this line).
    // Any concurrent calls arriving during our async work will see this
    // pending promise and await it instead of starting their own creation.
    cdcTransportPending = (async (): Promise<CDCTransport> => {
      // Build CDC endpoint URL (wss:// for WebSocket, https:// for SSE)
      const cdcUrl = baseUrl.replace('https://', config.transport === 'ws' ? 'wss://' : 'https://') + '/cdc'

      // Create transport instance (helper function for clarity)
      const transport = createCdcTransportInstance(cdcUrl)

      // Wire up event handlers to route events to subscriptions
      configureCdcTransportHandlers(transport)

      // Connect to server - THIS is the async operation that creates the race
      // window that cdcTransportPending protects against
      await transport.connect()

      // Store for fast path on subsequent calls
      cdcTransport = transport
      return transport
    })()

    try {
      return await cdcTransportPending
    } finally {
      // -----------------------------------------------------------------------
      // STEP 4: CLEANUP - Clear pending regardless of success/failure
      // -----------------------------------------------------------------------
      // SUCCESS: cdcTransport is set -> subsequent calls use fast path
      // FAILURE: cdcTransport is null -> next call retries fresh
      //
      // The finally block ensures we never leave cdcTransportPending set
      // indefinitely, which would block all future subscribe() calls.
      cdcTransportPending = null
    }
  }

  /**
   * Subscribe to changes on a table
   */
  sql.subscribe = async <T extends Row = Row>(
    tableName: string,
    options?: CDCClientSubscribeOptions<T>
  ): Promise<CDCClientSubscription<T>> => {
    // Validate table name to prevent SQL injection
    const { table, schema } = validateTableName(tableName)

    // Get or create CDC transport
    const transport = await getOrCreateCdcTransport()

    // Generate subscription ID
    const subscriptionId = generateSubscriptionId()

    // Create unsubscribe callback
    const unsubscribeCallback = async (): Promise<void> => {
      activeSubscriptions.delete(subscriptionId)
      await transport.unsubscribe(subscriptionId)

      // If no more subscriptions, disconnect the transport
      if (activeSubscriptions.size === 0 && cdcTransport) {
        await cdcTransport.disconnect()
        cdcTransport = null
      }
    }

    // Create subscription instance
    const subscription = new CDCSubscription<T>(
      subscriptionId,
      table,
      schema,
      transport,
      unsubscribeCallback,
      options as SubscribeOptions
    )

    // Register subscription
    activeSubscriptions.set(subscriptionId, subscription as CDCSubscription)

    // Subscribe via transport - build options object only with defined values
    const subscriptionOptions: Parameters<typeof transport.subscribe>[3] = {}
    if (options?.events !== undefined) subscriptionOptions.events = options.events
    if (options?.filter !== undefined) subscriptionOptions.filter = options.filter
    if (options?.resumeFrom !== undefined) subscriptionOptions.resumeFrom = options.resumeFrom
    if (options?.includeOldRow !== undefined) subscriptionOptions.includeOldRow = options.includeOldRow
    if (options?.trackChangedColumns !== undefined) subscriptionOptions.trackChangedColumns = options.trackChangedColumns
    if (options?.batchSize !== undefined) subscriptionOptions.batchSize = options.batchSize
    if (options?.heartbeatInterval !== undefined) subscriptionOptions.heartbeatInterval = options.heartbeatInterval
    await transport.subscribe(subscriptionId, table, schema, subscriptionOptions)

    return subscription
  }

  /**
   * Unsubscribe from a subscription by ID
   */
  sql.unsubscribe = async (subscriptionId: string): Promise<void> => {
    const subscription = activeSubscriptions.get(subscriptionId)
    if (subscription) {
      await subscription.unsubscribe()
    }
    // If subscription not found, silently succeed (idempotent behavior)
  }

  /**
   * Get all active subscriptions
   */
  sql.subscriptions = (): CDCClientSubscriptionInfo[] => {
    return Array.from(activeSubscriptions.values()).map(sub => ({
      id: sub.id,
      table: sub.table,
      schema: sub.schema,
      isActive: sub.isActive,
    }))
  }

  return sql
}

/**
 * Main entry point - create a postgres client
 * @param urlOrConfig - Connection URL or config object
 * @param options - Additional options
 */
export function postgres(
  urlOrConfig?: string | PostgresConfig,
  options?: PostgresConfig
): Sql {
  if (typeof urlOrConfig === 'string') {
    const parsed = parseConnectionUrl(urlOrConfig)
    return createClient({ ...parsed, ...options })
  }

  return createClient({ ...urlOrConfig, ...options })
}

// Default export
export default postgres