/**
 * Transactions Module for postgres.do
 *
 * This module handles:
 * - Transaction begin/commit/rollback
 * - Transaction context (txid, name, hooks)
 * - Savepoints
 * - Two-phase commit (2PC)
 * - Transaction options (isolation level, read-only, deferrable)
 * - Serialization failure retry
 */

import type {
  Row,
  PendingQuery,
  TypeParser,
  Transport,
  TransactionSql,
  TransactionOptions,
  StreamResult,
} from './types'
import { PostgresError, TimeoutError } from './types'
import type { QueryMiddleware } from './middleware'
import { createPendingQuery, parseRows } from './query-execution'
import {
  validateLockKey,
  validateTwoKeyLock,
  buildAdvisoryLockSql,
} from './advisory-locks'
import { createStreamResult, buildParameterizedSql as buildStreamSql } from './streaming'

// ============================================================================
// 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
 * ```
 */
export 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')
 */
export 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)
 */
export 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
}

/**
 * Create a fresh transaction context.
 *
 * @param options - Transaction options containing the optional name
 * @returns A new transaction context
 */
export function createTransactionContext(options?: TransactionOptions): TransactionContext {
  return {
    txid: undefined,
    txName: options?.name,
    onCommitCallbacks: [],
    onRollbackCallbacks: [],
    isPrepared: false,
    preparedTxId: undefined,
  }
}

// ============================================================================
// Transaction SQL Creation
// ============================================================================

/**
 * Create a transaction SQL client.
 *
 * This creates a SQL tagged template function that executes queries within
 * a transaction context. It includes:
 * - Tagged template SQL execution
 * - Savepoint support
 * - Advisory lock methods
 * - Two-phase commit (prepare)
 * - Transaction hooks (onCommit, onRollback)
 *
 * @param transport - The transport to execute queries on
 * @param parsers - Type parsers for result rows
 * @param savepointCounter - Mutable counter for generating unique savepoint names
 * @param txContext - The transaction context
 * @returns A TransactionSql interface
 */
export 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)
    },

    // CDC methods - not available in transactions (throws or no-op)
    subscribe: async () => {
      throw new PostgresError('subscribe() cannot be called within a transaction', {
        code: '25001',
        severity: 'ERROR',
      })
    },

    unsubscribe: async () => {
      throw new PostgresError('unsubscribe() cannot be called within a transaction', {
        code: '25001',
        severity: 'ERROR',
      })
    },

    subscriptions: () => [],
  }) 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,
  })

  return txSql
}

// ============================================================================
// BEGIN Statement Building
// ============================================================================

/**
 * Valid PostgreSQL isolation levels.
 */
export const VALID_ISOLATION_LEVELS = [
  'read uncommitted',
  'read committed',
  'repeatable read',
  'serializable',
] as const

/**
 * Build the BEGIN statement with transaction options.
 *
 * @param options - Transaction options
 * @returns The complete BEGIN statement
 * @throws {Error} If isolation level is invalid
 */
export function buildBeginStatement(options?: TransactionOptions): string {
  const beginParts = ['BEGIN']

  if (options?.isolationLevel) {
    // Validate isolation level to prevent SQL injection
    const normalized = options.isolationLevel.toLowerCase()
    if (!VALID_ISOLATION_LEVELS.includes(normalized as typeof VALID_ISOLATION_LEVELS[number])) {
      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')
  }

  return beginParts.join(' ')
}

// ============================================================================
// Transaction Execution
// ============================================================================

/**
 * Execute a transaction with the given callback.
 *
 * This handles:
 * - BEGIN with options (isolation level, read-only, deferrable)
 * - Fetching the transaction ID
 * - Timeout handling
 * - COMMIT on success
 * - ROLLBACK on error
 * - Transaction hooks (onCommit, onRollback)
 * - Two-phase commit (PREPARE TRANSACTION)
 * - Serialization failure retry
 *
 * @param transport - The transport to execute queries on
 * @param parsers - Type parsers for result rows
 * @param fn - The transaction callback
 * @param options - Transaction options
 * @returns The result of the callback
 */
export async function executeTransaction<T>(
  transport: Transport,
  parsers: Record<number, TypeParser>,
  fn: (sql: TransactionSql) => Promise<T>,
  options?: TransactionOptions
): Promise<T> {
  const maxRetries = options?.maxRetries ?? 3
  const shouldRetryOnSerializationFailure = options?.retryOnSerializationFailure ?? false

  const executeTransactionAttempt = async (attempt: number): Promise<T> => {
    // Start transaction
    const beginStatement = buildBeginStatement(options)
    await transport.query(beginStatement)

    // Create transaction context
    const txContext = createTransactionContext(options)

    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
    let timeoutHandle: ReturnType<typeof setTimeout> | undefined

    const runWithTimeout = async (): Promise<T> => {
      if (options?.timeout) {
        return new Promise<T>((resolve, reject) => {
          timeoutHandle = setTimeout(() => {
            reject(new TimeoutError(`Transaction timed out after ${options.timeout}ms`))
          }, options.timeout)

          fn(txSql)
            .then(resolve)
            .catch(reject)
            .finally(() => {
              if (timeoutHandle) {
                clearTimeout(timeoutHandle)
              }
            })
        })
      }
      return fn(txSql)
    }

    try {
      const result = await runWithTimeout()

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

      await transport.query('COMMIT')

      // 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)
      }

      // 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')

      // Check for serialization failure and retry if enabled
      if (
        shouldRetryOnSerializationFailure &&
        error instanceof PostgresError &&
        error.pgCode === '40001' &&
        attempt < maxRetries
      ) {
        // Exponential backoff: 10ms, 20ms, 40ms, etc.
        const delay = 10 * Math.pow(2, attempt - 1)
        await new Promise((resolve) => setTimeout(resolve, delay))
        return executeTransactionAttempt(attempt + 1)
      }

      throw error
    }
  }

  return executeTransactionAttempt(1)
}