/**
 * CDC Subscription for postgres.do
 *
 * Provides a user-friendly API for subscribing to database changes:
 * - Callback API (onChange, onInsert, onUpdate, onDelete)
 * - Async iterator API for use with for-await-of loops
 * - TypeScript generics for typed change payloads
 * - Automatic connection management and reconnection
 */

import type {
  Row,
  CDCChangeEvent,
  CDCSubscriptionOptions,
} from './types'
import type { CDCTransport } from './types'

// =============================================================================
// Types
// =============================================================================

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

/**
 * Typed subscribe options with generics
 */
export interface TypedSubscribeOptions<T extends Row> extends Omit<SubscribeOptions, 'onChange' | 'onInsert' | 'onUpdate' | 'onDelete'> {
  /** 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
}

/**
 * Subscription interface for CDC streaming
 */
export interface Subscription<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
}

// =============================================================================
// Implementation
// =============================================================================

/**
 * Internal state for managing the subscription
 */
interface SubscriptionState {
  active: boolean
  lastLsn: string | null
  changeHandlers: Array<(event: CDCChangeEvent) => void | Promise<void>>
  insertHandlers: Array<(row: Row) => void | Promise<void>>
  updateHandlers: Array<(newRow: Row, oldRow?: Row) => void | Promise<void>>
  deleteHandlers: Array<(oldRow: Row) => void | Promise<void>>
  errorHandler: ((error: Error) => void) | null
  eventQueue: CDCChangeEvent[]
  eventResolvers: Array<(result: IteratorResult<CDCChangeEvent>) => void>
  iteratorDone: boolean
}

/**
 * CDC Subscription implementation
 */
export class CDCSubscription<T extends Row = Row> implements Subscription<T> {
  readonly id: string
  readonly table: string
  readonly schema: string

  private state: SubscriptionState
  private unsubscribeCallback: () => Promise<void>

  constructor(
    id: string,
    table: string,
    schema: string,
    transport: CDCTransport,
    unsubscribeCallback: () => Promise<void>,
    options: SubscribeOptions = {}
  ) {
    this.id = id
    this.table = table
    this.schema = schema
    void transport // Transport reference kept for future reconnection support
    this.unsubscribeCallback = unsubscribeCallback

    this.state = {
      active: true,
      lastLsn: options.resumeFrom ?? null,
      changeHandlers: [],
      insertHandlers: [],
      updateHandlers: [],
      deleteHandlers: [],
      errorHandler: options.onError ?? null,
      eventQueue: [],
      eventResolvers: [],
      iteratorDone: false,
    }

    // Register initial handlers from options
    if (options.onChange) {
      this.state.changeHandlers.push(options.onChange)
    }
    if (options.onInsert) {
      this.state.insertHandlers.push(options.onInsert)
    }
    if (options.onUpdate) {
      this.state.updateHandlers.push(options.onUpdate)
    }
    if (options.onDelete) {
      this.state.deleteHandlers.push(options.onDelete)
    }
  }

  get isActive(): boolean {
    return this.state.active
  }

  get lastLsn(): string | null {
    return this.state.lastLsn
  }

  /**
   * Handle an incoming event from the transport
   */
  handleEvent(event: CDCChangeEvent): void {
    if (!this.state.active) return

    // Update last LSN
    this.state.lastLsn = event.lsn

    // Call change handlers
    for (const handler of this.state.changeHandlers) {
      try {
        void handler(event)
      } catch (error) {
        this.handleError(error instanceof Error ? error : new Error(String(error)))
      }
    }

    // Call operation-specific handlers
    switch (event.operation) {
      case 'INSERT':
        if (event.newRow) {
          for (const handler of this.state.insertHandlers) {
            try {
              void handler(event.newRow)
            } catch (error) {
              this.handleError(error instanceof Error ? error : new Error(String(error)))
            }
          }
        }
        break

      case 'UPDATE':
        if (event.newRow) {
          for (const handler of this.state.updateHandlers) {
            try {
              void handler(event.newRow, event.oldRow)
            } catch (error) {
              this.handleError(error instanceof Error ? error : new Error(String(error)))
            }
          }
        }
        break

      case 'DELETE':
        if (event.oldRow) {
          for (const handler of this.state.deleteHandlers) {
            try {
              void handler(event.oldRow)
            } catch (error) {
              this.handleError(error instanceof Error ? error : new Error(String(error)))
            }
          }
        }
        break
    }

    // If there are pending iterator resolvers, resolve the next one
    if (this.state.eventResolvers.length > 0) {
      const resolver = this.state.eventResolvers.shift()!
      resolver({ value: event, done: false })
    } else {
      // Otherwise queue the event
      this.state.eventQueue.push(event)
    }
  }

  /**
   * Handle an error from the transport
   */
  handleError(error: Error): void {
    if (this.state.errorHandler) {
      this.state.errorHandler(error)
    }
  }

  /**
   * Mark the subscription as inactive
   */
  setInactive(): void {
    this.state.active = false
    this.state.iteratorDone = true

    // Resolve any pending iterator requests with done
    while (this.state.eventResolvers.length > 0) {
      const resolver = this.state.eventResolvers.shift()!
      resolver({ value: undefined as unknown as CDCChangeEvent, done: true })
    }
  }

  async unsubscribe(): Promise<void> {
    if (!this.state.active) return

    await this.unsubscribeCallback()
    this.setInactive()
  }

  onChange(handler: (event: CDCChangeEvent & { newRow?: T; oldRow?: T }) => void | Promise<void>): this {
    this.state.changeHandlers.push(handler as (event: CDCChangeEvent) => void | Promise<void>)
    return this
  }

  onInsert(handler: (row: T) => void | Promise<void>): this {
    this.state.insertHandlers.push(handler as (row: Row) => void | Promise<void>)
    return this
  }

  onUpdate(handler: (newRow: T, oldRow?: T) => void | Promise<void>): this {
    this.state.updateHandlers.push(handler as (newRow: Row, oldRow?: Row) => void | Promise<void>)
    return this
  }

  onDelete(handler: (oldRow: T) => void | Promise<void>): this {
    this.state.deleteHandlers.push(handler as (oldRow: Row) => void | Promise<void>)
    return this
  }

  /**
   * Async iterator implementation for for-await-of support
   */
  [Symbol.asyncIterator](): AsyncIterator<CDCChangeEvent & { newRow?: T; oldRow?: T }> {
    return {
      next: async (): Promise<IteratorResult<CDCChangeEvent & { newRow?: T; oldRow?: T }>> => {
        // If subscription is inactive and queue is empty, we're done
        if (this.state.iteratorDone && this.state.eventQueue.length === 0) {
          return { value: undefined as unknown as CDCChangeEvent, done: true }
        }

        // If there's an event in the queue, return it
        if (this.state.eventQueue.length > 0) {
          const event = this.state.eventQueue.shift()!
          return { value: event as CDCChangeEvent & { newRow?: T; oldRow?: T }, done: false }
        }

        // Wait for the next event
        return new Promise((resolve) => {
          this.state.eventResolvers.push(resolve as (result: IteratorResult<CDCChangeEvent>) => void)
        })
      },

      return: async (): Promise<IteratorResult<CDCChangeEvent & { newRow?: T; oldRow?: T }>> => {
        // Called when breaking out of for-await-of
        // Don't unsubscribe - allow the user to continue later
        return { value: undefined as unknown as CDCChangeEvent, done: true }
      },
    }
  }
}

// =============================================================================
// Validation Helpers
// =============================================================================

/**
 * Regular expression for valid table/schema names
 * Must start with letter or underscore, contain only alphanumeric and underscores
 */
const VALID_IDENTIFIER_REGEX = /^[a-zA-Z_][a-zA-Z0-9_]*$/

/**
 * Validate a table name to prevent SQL injection
 */
export function validateTableName(name: string): { table: string; schema: string } {
  if (!name || name.trim() === '') {
    throw new Error('Table name cannot be empty')
  }

  // Check for SQL injection patterns
  if (name.includes(';') || name.includes('--') || name.includes('/*') || name.includes('*/')) {
    throw new Error('Invalid table name: contains forbidden characters')
  }

  // Split into schema.table if present
  const parts = name.split('.')
  if (parts.length > 2) {
    throw new Error('Invalid table name: too many parts')
  }

  let schema = 'public'
  let table: string

  if (parts.length === 2) {
    schema = parts[0]!
    table = parts[1]!
  } else {
    table = parts[0]!
  }

  // Validate schema name
  if (!VALID_IDENTIFIER_REGEX.test(schema)) {
    throw new Error(`Invalid schema name: "${schema}"`)
  }

  // Validate table name
  if (!VALID_IDENTIFIER_REGEX.test(table)) {
    throw new Error(`Invalid table name: "${table}"`)
  }

  return { table, schema }
}

/**
 * Generate a unique subscription ID
 */
export function generateSubscriptionId(): string {
  return `sub-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`
}