/**
 * Streaming API for postgres.do
 *
 * Provides cursor-based streaming for large datasets to avoid OOM errors
 * in Cloudflare Workers' 128MB memory limit.
 *
 * Uses PostgreSQL cursors under the hood:
 * - DECLARE cursor FOR SELECT ...
 * - FETCH n FROM cursor
 * - CLOSE cursor
 *
 * @example
 * ```typescript
 * // 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 = sql.cursor`SELECT * FROM huge_table`
 * while (const rows = await cursor.fetch(100)) {
 *   process(rows)
 * }
 * await cursor.close()
 * ```
 *
 * ## Security Considerations
 *
 * ### Cursor Name Injection Prevention
 *
 * Cursor names are used in SQL statements (DECLARE, FETCH, CLOSE) and must be
 * carefully validated to prevent SQL injection attacks. This module implements
 * multiple layers of protection:
 *
 * 1. **PostgreSQL Identifier Validation**: Cursor names must conform to PostgreSQL
 *    identifier rules - start with letter or underscore, contain only alphanumeric
 *    characters and underscores, max 63 characters.
 *
 * 2. **Cryptographic Randomness**: Auto-generated cursor names use crypto.getRandomValues()
 *    to prevent name prediction attacks. The format is:
 *    `pg_cursor_{timestamp}_{counter}_{8-char-hex-random}`
 *
 * 3. **Double-Quote Escaping**: Cursor names are always wrapped in double quotes
 *    in SQL statements, preventing injection through special characters.
 *
 * 4. **No User Input Echoing**: Error messages do not echo potentially malicious
 *    cursor names back to the user, preventing information leakage.
 *
 * ### Isolation Level Validation
 *
 * Transaction isolation levels are validated against a whitelist to prevent
 * SQL injection through the isolationLevel option.
 *
 * ### Defense in Depth
 *
 * Even if validation were bypassed, the cursor name quoting provides additional
 * protection. However, users should never rely solely on quoting - always use
 * validated cursor names or let the system generate them automatically.
 *
 * @see validateCursorName for the validation implementation
 * @see generateCursorName for the secure name generation
 */

import type { Row, Transport, QueryResult, FieldInfo, TypeParser, StreamOptions, Cursor, CursorFetchResult, StreamResult } from './types'
import { isValidIdentifier, MAX_IDENTIFIER_LENGTH } from '@dotdo/postgres-shared/validation'

/**
 * Valid PostgreSQL transaction isolation levels (case-insensitive).
 * Used to validate the isolationLevel option and prevent SQL injection.
 */
const VALID_ISOLATION_LEVELS = Object.freeze([
  'read uncommitted',
  'read committed',
  'repeatable read',
  'serializable',
] as const)

/**
 * Type for valid isolation levels
 */
type IsolationLevel = typeof VALID_ISOLATION_LEVELS[number]

/**
 * Default batch size for streaming operations
 */
const DEFAULT_BATCH_SIZE = 100

/**
 * Parse rows using type parsers
 */
function parseRows<T extends Row>(
  rows: T[],
  fields: FieldInfo[],
  parsers: Record<number, TypeParser>
): T[] {
  if (!fields || fields.length === 0 || Object.keys(parsers).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
  })
}

/**
 * Generate a cryptographically random hex string
 */
function randomHex(length: number): string {
  // Use crypto.getRandomValues for secure randomness
  // Works in both browser and Node.js environments
  if (typeof crypto !== 'undefined' && crypto.getRandomValues) {
    const bytes = new Uint8Array(Math.ceil(length / 2))
    crypto.getRandomValues(bytes)
    return Array.from(bytes, (b) => b.toString(16).padStart(2, '0'))
      .join('')
      .slice(0, length)
  }
  // Fallback for environments without crypto (should not happen in Workers)
  let result = ''
  const chars = '0123456789abcdef'
  for (let i = 0; i < length; i++) {
    result += chars[Math.floor(Math.random() * chars.length)]
  }
  return result
}

/**
 * Counter for generating unique cursor names within a session.
 * Combined with timestamp and cryptographic random bytes for uniqueness.
 *
 * ## Serverless Safety
 *
 * In serverless environments, this counter persists across requests within the
 * same isolate. This is acceptable because:
 *
 * 1. The primary uniqueness comes from the 8-char cryptographic random component
 * 2. The timestamp provides time-based uniqueness
 * 3. The counter only needs to be unique within the same millisecond
 * 4. Counter is periodically reset with jitter to reduce predictability
 *
 * The counter adds defense-in-depth but is not the sole uniqueness mechanism.
 */
let cursorCounter = 0

/**
 * Track number of cursor names generated to periodically reset counter.
 */
let cursorsSinceReset = 0

/**
 * Number of cursor names before resetting counter with jitter.
 * Prevents long-lived predictable counter sequences.
 */
const CURSOR_COUNTER_RESET_INTERVAL = 1000

/**
 * Prefix for all auto-generated cursor names.
 * This prefix is part of the security model - it ensures generated names
 * follow a predictable, safe format.
 */
const CURSOR_NAME_PREFIX = 'pg_cursor_' as const

/**
 * Generate a unique cursor name
 *
 * Uses cryptographic randomness to prevent name prediction attacks.
 * The generated format is: `pg_cursor_{timestamp}_{counter}_{random}`
 *
 * Security properties:
 * - Timestamp provides uniqueness across time
 * - Counter provides uniqueness within millisecond (with periodic jitter reset)
 * - 8 hex chars (32 bits) of cryptographic randomness prevents prediction
 * - Prefix ensures name always starts with letter (pg)
 * - Format guarantees valid PostgreSQL identifier
 *
 * Serverless safety:
 * - Counter is reset with random jitter every 1000 generations
 * - 8 random hex chars provide primary unpredictability
 * - Timestamp ensures no collision across time
 *
 * @returns A unique, cryptographically unpredictable cursor name
 */
function generateCursorName(): string {
  // Periodically reset counter with random jitter for serverless safety
  if (cursorsSinceReset >= CURSOR_COUNTER_RESET_INTERVAL) {
    cursorCounter = Math.floor(Math.random() * 1000000)
    cursorsSinceReset = 0
  }

  const timestamp = Date.now()
  const counter = ++cursorCounter
  cursorsSinceReset++
  const random = randomHex(8)
  // Format: pg_cursor_1234567890123_1_a1b2c3d4
  // Always valid: starts with 'p', contains only alphanumerics and underscores
  return `${CURSOR_NAME_PREFIX}${timestamp}_${counter}_${random}`
}

/**
 * Reset cursor counter state. Intended for testing only.
 *
 * @security In production, this should never be exposed to external callers.
 *           Only exported for use in test cleanup between test cases.
 */
export function resetCursorCounterState(): void {
  cursorCounter = 0
  cursorsSinceReset = 0
}

/**
 * Error class for cursor validation failures.
 * Provides structured error information without leaking potentially malicious input.
 */
export class CursorValidationError extends Error {
  /** Error code for programmatic handling */
  readonly code: string

  constructor(message: string, code: string = 'INVALID_CURSOR_NAME') {
    super(message)
    this.name = 'CursorValidationError'
    this.code = code
  }
}

/**
 * Validate a cursor name against PostgreSQL identifier rules
 * and security requirements.
 *
 * This function implements multiple layers of validation:
 *
 * 1. **Non-empty check**: Cursor names cannot be empty
 * 2. **Length check**: Cannot exceed PostgreSQL's 63-character limit
 * 3. **PostgreSQL identifier rules**: Must start with letter or underscore,
 *    contain only alphanumeric characters and underscores
 *
 * Security notes:
 * - Error messages do not echo the input back to prevent information leakage
 * - The regex pattern in isValidIdentifier explicitly rejects:
 *   - Quotes (single, double, backtick)
 *   - Semicolons and other SQL delimiters
 *   - Whitespace (spaces, tabs, newlines)
 *   - Dollar signs (used for dollar-quoting in PostgreSQL)
 *   - All other special characters
 *
 * @param cursorName - The cursor name to validate
 * @throws CursorValidationError if the cursor name is invalid
 *
 * @example
 * ```typescript
 * validateCursorName('my_cursor')      // OK
 * validateCursorName('_private')       // OK
 * validateCursorName('Cursor123')      // OK
 * validateCursorName('123cursor')      // Throws - starts with number
 * validateCursorName('my cursor')      // Throws - contains space
 * validateCursorName('cursor"; DROP')  // Throws - contains special chars
 * ```
 */
export function validateCursorName(cursorName: string): void {
  // Empty check with clear error
  if (!cursorName || cursorName.length === 0) {
    throw new CursorValidationError(
      'Invalid cursor name: cannot be empty',
      'CURSOR_NAME_EMPTY'
    )
  }

  // Length check - PostgreSQL truncates identifiers silently at 63 chars,
  // but we reject to prevent unexpected behavior
  if (cursorName.length > MAX_IDENTIFIER_LENGTH) {
    throw new CursorValidationError(
      `Invalid cursor name: exceeds maximum length of ${MAX_IDENTIFIER_LENGTH} characters`,
      'CURSOR_NAME_TOO_LONG'
    )
  }

  // Use the shared validation function for PostgreSQL identifier rules
  // This rejects names with:
  // - Special characters (quotes, semicolons, spaces, etc.)
  // - Names not starting with letter or underscore
  // - SQL injection payloads
  // - Dollar signs (valid in PG but excluded for security)
  if (!isValidIdentifier(cursorName)) {
    throw new CursorValidationError(
      'Invalid cursor name: must start with a letter or underscore, and contain only letters, numbers, and underscores',
      'CURSOR_NAME_INVALID_FORMAT'
    )
  }
}

/**
 * Validates an isolation level against the allowed whitelist.
 *
 * @param level - The isolation level to validate
 * @returns The normalized (lowercase) isolation level
 * @throws CursorValidationError if the isolation level is invalid
 */
function validateIsolationLevel(level: string): IsolationLevel {
  const normalized = level.toLowerCase() as IsolationLevel
  if (!VALID_ISOLATION_LEVELS.includes(normalized)) {
    throw new CursorValidationError(
      'Invalid isolation level: must be one of: read uncommitted, read committed, repeatable read, serializable',
      'INVALID_ISOLATION_LEVEL'
    )
  }
  return normalized
}

/**
 * Internal cursor implementation
 */
class CursorImpl<T extends Row = Row> implements Cursor<T> {
  private _isOpen = true
  private _totalFetched = 0
  private _fields: FieldInfo[] = []
  private readonly cursorName: string
  private readonly batchSize: number
  private readonly transport: Transport
  private readonly parsers: Record<number, TypeParser>
  private inTransaction = false

  constructor(
    transport: Transport,
    cursorName: string,
    batchSize: number,
    parsers: Record<number, TypeParser>
  ) {
    this.transport = transport
    this.cursorName = cursorName
    this.batchSize = batchSize
    this.parsers = parsers
  }

  /**
   * Initialize the cursor (called after DECLARE)
   */
  _setFields(fields: FieldInfo[]): void {
    this._fields = fields
  }

  _setInTransaction(inTransaction: boolean): void {
    this.inTransaction = inTransaction
  }

  get isOpen(): boolean {
    return this._isOpen
  }

  get totalFetched(): number {
    return this._totalFetched
  }

  get fields(): FieldInfo[] {
    return this._fields
  }

  async fetch(count?: number): Promise<T[]> {
    const result = await this.fetchWithInfo(count)
    return result.rows
  }

  async fetchWithInfo(count?: number): Promise<CursorFetchResult<T>> {
    if (!this._isOpen) {
      return { rows: [], hasMore: false, totalFetched: this._totalFetched }
    }

    const fetchCount = count ?? this.batchSize
    const result = await this.transport.query<T>(
      `FETCH ${fetchCount} FROM "${this.cursorName}"`
    )

    // Store fields from first fetch if not already set
    if (this._fields.length === 0 && result.fields.length > 0) {
      this._fields = result.fields
    }

    const parsedRows = parseRows(result.rows, this._fields, this.parsers)
    this._totalFetched += parsedRows.length

    const hasMore = parsedRows.length === fetchCount

    return {
      rows: parsedRows,
      hasMore,
      totalFetched: this._totalFetched,
    }
  }

  async close(): Promise<void> {
    if (!this._isOpen) {
      return
    }

    this._isOpen = false

    try {
      // Close the cursor
      await this.transport.query(`CLOSE "${this.cursorName}"`)

      // Commit the transaction if we started one
      if (this.inTransaction) {
        await this.transport.query('COMMIT')
      }
    } catch (error) {
      // Try to rollback on error
      if (this.inTransaction) {
        try {
          await this.transport.query('ROLLBACK')
        } catch {
          // Ignore rollback errors
        }
      }
      throw error
    }
  }

  async *[Symbol.asyncIterator](): AsyncIterableIterator<T> {
    try {
      while (this._isOpen) {
        const { rows, hasMore } = await this.fetchWithInfo()

        for (const row of rows) {
          yield row
        }

        if (!hasMore) {
          break
        }
      }
    } finally {
      await this.close()
    }
  }
}

/**
 * Create a cursor for a query
 */
export async function createCursor<T extends Row = Row>(
  transport: Transport,
  sql: string,
  params: unknown[],
  parsers: Record<number, TypeParser>,
  options: StreamOptions = {}
): Promise<Cursor<T>> {
  // Generate or validate cursor name
  const cursorName = options.cursorName ?? generateCursorName()

  // Validate the cursor name (whether user-provided or generated)
  // This prevents SQL injection attacks through cursor names
  validateCursorName(cursorName)

  const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE
  const readOnly = options.readOnly ?? true

  // Build transaction options
  const txParts: string[] = ['BEGIN']
  if (options.isolationLevel) {
    // Validate isolation level to prevent SQL injection
    // This uses a whitelist approach - only known-safe values are accepted
    const normalized = validateIsolationLevel(options.isolationLevel)
    txParts.push(`ISOLATION LEVEL ${normalized.toUpperCase()}`)
  }
  if (readOnly) {
    txParts.push('READ ONLY')
  }

  // Start transaction and declare cursor
  await transport.query(txParts.join(' '))

  try {
    // Declare the cursor
    // For parameterized queries, we need to use a prepared statement approach
    // PostgreSQL cursors don't directly support parameters in DECLARE
    // So we build the query with proper escaping for the cursor
    const declareResult = await declareCursor<T>(transport, cursorName, sql, params)

    const cursor = new CursorImpl<T>(transport, cursorName, batchSize, parsers)
    cursor._setFields(declareResult.fields)
    cursor._setInTransaction(true)

    return cursor
  } catch (error) {
    // Rollback on error
    try {
      await transport.query('ROLLBACK')
    } catch {
      // Ignore rollback errors
    }
    throw error
  }
}

/**
 * Declare a cursor for a query with parameters
 */
async function declareCursor<T extends Row>(
  transport: Transport,
  cursorName: string,
  sql: string,
  params: unknown[]
): Promise<QueryResult<T>> {
  // For queries without parameters, declare directly
  if (params.length === 0) {
    return transport.query<T>(`DECLARE "${cursorName}" CURSOR FOR ${sql}`)
  }

  // For parameterized queries, we use a slightly different approach
  // PostgreSQL's DECLARE doesn't support $n parameters directly,
  // but the transport will handle parameter substitution
  // We can use EXECUTE with a prepared statement
  //
  // However, for HTTP transport, parameters are handled server-side
  // So we send the DECLARE with the query and parameters
  return transport.query<T>(
    `DECLARE "${cursorName}" CURSOR FOR ${sql}`,
    params
  )
}

/**
 * Create a stream result for a query
 */
export function createStreamResult<T extends Row = Row>(
  transport: Transport,
  sql: string,
  params: unknown[],
  parsers: Record<number, TypeParser>,
  options: StreamOptions = {}
): StreamResult<T> {
  const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE

  return {
    sql,
    params,

    async *[Symbol.asyncIterator](): AsyncIterableIterator<T> {
      const cursor = await createCursor<T>(transport, sql, params, parsers, {
        ...options,
        batchSize,
      })

      try {
        for await (const row of cursor) {
          yield row
        }
      } finally {
        if (cursor.isOpen) {
          await cursor.close()
        }
      }
    },

    batch(size: number): AsyncIterable<T[]> {
      return {
        async *[Symbol.asyncIterator](): AsyncIterableIterator<T[]> {
          const cursor = await createCursor<T>(transport, sql, params, parsers, {
            ...options,
            batchSize: size,
          })

          try {
            while (cursor.isOpen) {
              const { rows, hasMore } = await cursor.fetchWithInfo(size)

              if (rows.length > 0) {
                yield rows
              }

              if (!hasMore) {
                break
              }
            }
          } finally {
            if (cursor.isOpen) {
              await cursor.close()
            }
          }
        },
      }
    },

    async cursor(): Promise<Cursor<T>> {
      return createCursor<T>(transport, sql, params, parsers, options)
    },
  }
}

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