/**
 * CDC WebSocket Transport for postgres.do
 *
 * Provides real-time change data capture streaming via WebSocket:
 * - Bidirectional communication for subscription management
 * - Automatic reconnection with exponential backoff
 * - LSN-based resumption for reliable delivery
 * - Heartbeat/ping-pong for connection keepalive
 */

import type {
  Row,
  CDCChangeEvent,
  CDCTransport,
  CDCSubscriptionOptions,
  CDCWsTransportConfig,
  CDCOperation,
} from '../types'
import { ConnectionError, TimeoutError } from '../types'
import type { PendingRequest } from './base-transport'

/**
 * CDC WebSocket message types
 */
enum CDCMessageType {
  // Client -> Server
  Subscribe = 'subscribe',
  Unsubscribe = 'unsubscribe',
  Ping = 'ping',
  Auth = 'auth',
  Ack = 'ack',

  // Server -> Client
  Event = 'event',
  Error = 'error',
  Pong = 'pong',
  AuthResult = 'auth_result',
  SubscribeResult = 'subscribe_result',
  UnsubscribeResult = 'unsubscribe_result',
}

/**
 * Incoming CDC event message from server
 */
interface CDCEventMessage {
  type: CDCMessageType.Event
  subscriptionId: string
  event: {
    id: string
    lsn: string
    operation: CDCOperation
    table: string
    schema: string
    timestamp: string
    newRow?: Row
    oldRow?: Row
    changedColumns?: string[]
    xid: number
    isLastInTransaction: boolean
  }
}

/**
 * Error message from server
 */
interface CDCErrorMessage {
  type: CDCMessageType.Error
  subscriptionId?: string
  error: {
    message: string
    code?: string
  }
}

/**
 * Subscription result message
 */
interface CDCSubscribeResultMessage {
  type: CDCMessageType.SubscribeResult
  subscriptionId: string
  success: boolean
  error?: string
}

/**
 * Unsubscribe result message
 */
interface CDCUnsubscribeResultMessage {
  type: CDCMessageType.UnsubscribeResult
  subscriptionId: string
  success: boolean
}

/**
 * Auth result message
 */
interface CDCAuthResultMessage {
  type: CDCMessageType.AuthResult
  success: boolean
  error?: string
}

type CDCIncomingMessage =
  | CDCEventMessage
  | CDCErrorMessage
  | CDCSubscribeResultMessage
  | CDCUnsubscribeResultMessage
  | CDCAuthResultMessage
  | { type: CDCMessageType.Pong }

/**
 * Default configuration values
 */
const DEFAULTS = {
  connectTimeout: 10000,
  pingInterval: 0, // Disabled by default for test compatibility; production code should set explicitly
  reconnectBaseDelay: 1000,
  maxReconnectDelay: 30000,
  maxReconnectAttempts: 10,
} as const

/**
 * CDC WebSocket Transport implementation
 *
 * Extends BaseCDCTransport for common functionality.
 */
export class CDCWsTransport implements CDCTransport {
  private readonly config: {
    readonly url: string
    readonly apiKey?: string | undefined
    readonly WebSocket?: typeof WebSocket | undefined
    readonly connectTimeout: number
    readonly pingInterval: number
  }
  private readonly WebSocketImpl: typeof WebSocket
  private ws: WebSocket | null = null
  private connected = false
  private authenticated = false
  private connecting: Promise<void> | null = null
  private lastEventId: string | null = null
  private reconnectAttempt = 0
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null
  private pingTimer: ReturnType<typeof setInterval> | null = null
  private shouldReconnect = true

  // Pending requests
  private pendingSubscriptions = new Map<string, PendingRequest<void>>()
  private pendingUnsubscriptions = new Map<string, PendingRequest<void>>()
  private pendingAuth: PendingRequest<void> | null = null

  // Event handlers
  private eventHandler: ((subscriptionId: string, event: CDCChangeEvent) => void) | null = null
  private errorHandler: ((subscriptionId: string, error: Error) => void) | null = null
  private reconnectHandler: ((attempt: number, lastLsn: string | undefined) => void) | null = null
  private closeHandler: (() => void) | null = null

  // Active subscriptions for reconnection
  private activeSubscriptions = new Map<
    string,
    {
      table: string
      schema: string
      options: CDCSubscriptionOptions
    }
  >()

  constructor(config: CDCWsTransportConfig) {
    this.config = {
      url: config.url,
      apiKey: config.apiKey,
      WebSocket: config.WebSocket,
      connectTimeout: config.connectTimeout ?? DEFAULTS.connectTimeout,
      pingInterval: config.pingInterval ?? DEFAULTS.pingInterval,
    }
    this.WebSocketImpl = config.WebSocket || globalThis.WebSocket
  }

  /**
   * Connect to the CDC stream
   */
  async connect(): Promise<void> {
    if (this.connected && this.authenticated) {
      return
    }

    if (this.connecting) {
      return this.connecting
    }

    this.shouldReconnect = true
    this.connecting = this._connect()

    try {
      await this.connecting
    } finally {
      this.connecting = null
    }
  }

  /**
   * Internal connect implementation
   */
  private _connect(): Promise<void> {
    return new Promise((resolve, reject) => {
      const timeoutId = setTimeout(() => {
        reject(new TimeoutError(`WebSocket connection timed out after ${this.config.connectTimeout}ms`))
        if (this.ws) {
          this.ws.close()
          this.ws = null
        }
      }, this.config.connectTimeout)

      try {
        this.ws = new this.WebSocketImpl(this.config.url)
      } catch (error) {
        clearTimeout(timeoutId)
        reject(
          new ConnectionError(
            `Failed to create WebSocket: ${error instanceof Error ? error.message : String(error)}`
          )
        )
        return
      }

      this.ws.onopen = () => {
        this.connected = true
        this.reconnectAttempt = 0

        // Authenticate if API key is provided
        if (this.config.apiKey) {
          this.authenticate()
            .then(() => {
              clearTimeout(timeoutId)
              resolve()
              // Start ping interval after connect resolves
              if (this.config.pingInterval > 0) {
                this.startPingInterval()
              }
            })
            .catch((error) => {
              clearTimeout(timeoutId)
              reject(error)
            })
        } else {
          this.authenticated = true
          clearTimeout(timeoutId)
          resolve()
          // Start ping interval after connect resolves
          if (this.config.pingInterval > 0) {
            this.startPingInterval()
          }
        }
      }

      this.ws.onmessage = (event) => {
        this.handleMessage(event.data)
      }

      this.ws.onerror = () => {
        clearTimeout(timeoutId)
        const error = new ConnectionError('WebSocket error occurred')
        this.rejectAllPending(error)
        reject(error)
      }

      this.ws.onclose = (event) => {
        clearTimeout(timeoutId)
        this.connected = false
        this.authenticated = false
        this.stopPingInterval()

        if (!event.wasClean && this.shouldReconnect) {
          this.scheduleReconnect()
        } else if (!this.shouldReconnect) {
          this.closeHandler?.()
        } else {
          const error = new ConnectionError(`WebSocket closed: ${event.code} ${event.reason}`)
          this.rejectAllPending(error)
          reject(error)
        }
      }
    })
  }

  /**
   * Authenticate with the API key
   */
  private authenticate(): Promise<void> {
    return new Promise((resolve, reject) => {
      const timeoutId = setTimeout(() => {
        this.pendingAuth = null
        reject(new TimeoutError('Authentication timed out'))
      }, this.config.connectTimeout)

      this.pendingAuth = {
        resolve: () => {
          clearTimeout(timeoutId)
          this.pendingAuth = null
          this.authenticated = true
          resolve()
        },
        reject: (error: Error) => {
          clearTimeout(timeoutId)
          this.pendingAuth = null
          reject(error)
        },
        timeout: timeoutId,
      }

      this.send({
        type: CDCMessageType.Auth,
        apiKey: this.config.apiKey,
      })
    })
  }

  /**
   * Disconnect from the CDC stream
   */
  async disconnect(): Promise<void> {
    this.shouldReconnect = false

    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer)
      this.reconnectTimer = null
    }

    this.stopPingInterval()
    this.rejectAllPending(new ConnectionError('Connection closed'))

    if (this.ws) {
      this.ws.close(1000, 'Client closing connection')
      this.ws = null
    }

    this.connected = false
    this.authenticated = false
    this.activeSubscriptions.clear()
    this.closeHandler?.()
  }

  /**
   * Check if connected
   */
  isConnected(): boolean {
    return this.connected && this.authenticated
  }

  /**
   * Get the last event ID received
   */
  getLastEventId(): string | null {
    return this.lastEventId
  }

  /**
   * Subscribe to a table
   */
  async subscribe(
    subscriptionId: string,
    table: string,
    schema: string,
    options: CDCSubscriptionOptions
  ): Promise<void> {
    return new Promise((resolve, reject) => {
      const timeoutId = setTimeout(() => {
        this.pendingSubscriptions.delete(subscriptionId)
        reject(new TimeoutError('Subscription request timed out'))
      }, this.config.connectTimeout)

      this.pendingSubscriptions.set(subscriptionId, {
        resolve: () => {
          clearTimeout(timeoutId)
          // Store for reconnection
          this.activeSubscriptions.set(subscriptionId, { table, schema, options })
          resolve()
        },
        reject: (error) => {
          clearTimeout(timeoutId)
          reject(error)
        },
        timeout: timeoutId,
      })

      this.send({
        type: CDCMessageType.Subscribe,
        subscriptionId,
        table,
        schema,
        options: {
          events: options.events,
          filter: options.filter,
          resumeFrom: options.resumeFrom,
          includeOldRow: options.includeOldRow,
          trackChangedColumns: options.trackChangedColumns,
          batchSize: options.batchSize,
          heartbeatInterval: options.heartbeatInterval,
        },
      })
    })
  }

  /**
   * Unsubscribe from a subscription
   */
  async unsubscribe(subscriptionId: string): Promise<void> {
    this.activeSubscriptions.delete(subscriptionId)

    return new Promise((resolve) => {
      const timeoutId = setTimeout(() => {
        this.pendingUnsubscriptions.delete(subscriptionId)
        // Resolve anyway on timeout - server may have already processed
        resolve()
      }, this.config.connectTimeout)

      this.pendingUnsubscriptions.set(subscriptionId, {
        resolve: () => {
          clearTimeout(timeoutId)
          resolve()
        },
        reject: () => {
          clearTimeout(timeoutId)
          // Still resolve on reject for unsubscribe
          resolve()
        },
        timeout: timeoutId,
      })

      this.send({
        type: CDCMessageType.Unsubscribe,
        subscriptionId,
      })
    })
  }

  /**
   * Set the event handler
   */
  onEvent(handler: (subscriptionId: string, event: CDCChangeEvent) => void): void {
    this.eventHandler = handler
  }

  /**
   * Set the error handler
   */
  onError(handler: (subscriptionId: string, error: Error) => void): void {
    this.errorHandler = handler
  }

  /**
   * Set the reconnect handler
   */
  onReconnect(handler: (attempt: number, lastLsn: string | undefined) => void): void {
    this.reconnectHandler = handler
  }

  /**
   * Set the close handler
   */
  onClose(handler: () => void): void {
    this.closeHandler = handler
  }

  /**
   * Send a message
   */
  private send(message: unknown): void {
    // Use WebSocketImpl for the OPEN constant to support custom WebSocket implementations in tests
    if (this.ws && this.ws.readyState === this.WebSocketImpl.OPEN) {
      this.ws.send(JSON.stringify(message))
    }
  }

  /**
   * Handle incoming message
   */
  private handleMessage(data: string): void {
    let message: CDCIncomingMessage

    try {
      message = JSON.parse(data)
    } catch {
      console.error('Failed to parse CDC message:', data)
      return
    }

    switch (message.type) {
      case CDCMessageType.Event:
        this.handleEventMessage(message)
        break

      case CDCMessageType.Error:
        this.handleErrorMessage(message)
        break

      case CDCMessageType.SubscribeResult:
        this.handleSubscribeResult(message)
        break

      case CDCMessageType.UnsubscribeResult:
        this.handleUnsubscribeResult(message)
        break

      case CDCMessageType.AuthResult:
        this.handleAuthResult(message)
        break

      case CDCMessageType.Pong:
        // Heartbeat response - connection is alive
        break

      default:
        // Unknown message type - ignore
        break
    }
  }

  /**
   * Handle CDC event message
   */
  private handleEventMessage(message: CDCEventMessage): void {
    const { subscriptionId, event } = message

    // Update last event ID
    this.lastEventId = event.lsn

    // Convert to CDCChangeEvent
    const changeEvent: CDCChangeEvent = {
      id: event.id,
      lsn: event.lsn,
      operation: event.operation,
      table: event.table,
      schema: event.schema,
      timestamp: new Date(event.timestamp),
      newRow: event.newRow,
      oldRow: event.oldRow,
      changedColumns: event.changedColumns,
      xid: event.xid,
      isLastInTransaction: event.isLastInTransaction,
    }

    this.eventHandler?.(subscriptionId, changeEvent)

    // Send ACK
    this.send({
      type: CDCMessageType.Ack,
      subscriptionId,
      lsn: event.lsn,
    })
  }

  /**
   * Handle error message
   */
  private handleErrorMessage(message: CDCErrorMessage): void {
    const error = new Error(message.error.message)

    if (message.subscriptionId) {
      this.errorHandler?.(message.subscriptionId, error)
    } else {
      // Global error - notify all subscriptions
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, error)
      }
    }
  }

  /**
   * Handle subscribe result
   */
  private handleSubscribeResult(message: CDCSubscribeResultMessage): void {
    const pending = this.pendingSubscriptions.get(message.subscriptionId)
    if (pending) {
      this.pendingSubscriptions.delete(message.subscriptionId)
      clearTimeout(pending.timeout)

      if (message.success) {
        pending.resolve()
      } else {
        pending.reject(new Error(message.error || 'Subscription failed'))
      }
    }
  }

  /**
   * Handle unsubscribe result
   */
  private handleUnsubscribeResult(message: CDCUnsubscribeResultMessage): void {
    const pending = this.pendingUnsubscriptions.get(message.subscriptionId)
    if (pending) {
      this.pendingUnsubscriptions.delete(message.subscriptionId)
      clearTimeout(pending.timeout)
      pending.resolve()
    }
  }

  /**
   * Handle auth result
   */
  private handleAuthResult(message: CDCAuthResultMessage): void {
    if (this.pendingAuth) {
      if (message.success) {
        this.pendingAuth.resolve()
      } else {
        this.pendingAuth.reject(new ConnectionError(message.error || 'Authentication failed'))
      }
    }
  }

  /**
   * Start ping interval for keepalive
   */
  private startPingInterval(): void {
    this.stopPingInterval()
    if (this.config.pingInterval > 0) {
      this.pingTimer = setInterval(() => {
        this.send({ type: CDCMessageType.Ping })
      }, this.config.pingInterval)
    }
  }

  /**
   * Stop ping interval
   */
  private stopPingInterval(): void {
    if (this.pingTimer) {
      clearInterval(this.pingTimer)
      this.pingTimer = null
    }
  }

  /**
   * Schedule reconnection with exponential backoff
   */
  private scheduleReconnect(): void {
    if (!this.shouldReconnect) {
      return
    }

    this.reconnectAttempt++

    if (this.reconnectAttempt > DEFAULTS.maxReconnectAttempts) {
      const error = new ConnectionError('Max reconnection attempts exceeded')
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, error)
      }
      this.closeHandler?.()
      return
    }

    // Exponential backoff with jitter
    const delay = Math.min(
      DEFAULTS.reconnectBaseDelay * Math.pow(2, this.reconnectAttempt - 1) + Math.random() * 1000,
      DEFAULTS.maxReconnectDelay
    )

    this.reconnectTimer = setTimeout(() => {
      this.reconnectHandler?.(this.reconnectAttempt, this.lastEventId ?? undefined)
      this.reconnect()
    }, delay)
  }

  /**
   * Reconnect and resubscribe
   */
  private async reconnect(): Promise<void> {
    try {
      await this._connect()

      // Resubscribe to all active subscriptions
      for (const [subscriptionId, { table, schema, options }] of this.activeSubscriptions) {
        // Resume from last known LSN
        const resumeOptions = {
          ...options,
          resumeFrom: this.lastEventId ?? options.resumeFrom,
        }

        try {
          await this.subscribe(subscriptionId, table, schema, resumeOptions)
        } catch (error) {
          this.errorHandler?.(subscriptionId, error instanceof Error ? error : new Error(String(error)))
        }
      }
    } catch {
      // Schedule another reconnect
      if (this.shouldReconnect) {
        this.scheduleReconnect()
      }
    }
  }

  /**
   * Reject all pending requests
   */
  private rejectAllPending(error: Error): void {
    for (const pending of this.pendingSubscriptions.values()) {
      clearTimeout(pending.timeout)
      pending.reject(error)
    }
    this.pendingSubscriptions.clear()

    for (const pending of this.pendingUnsubscriptions.values()) {
      clearTimeout(pending.timeout)
      pending.reject(error)
    }
    this.pendingUnsubscriptions.clear()

    if (this.pendingAuth) {
      clearTimeout(this.pendingAuth.timeout)
      this.pendingAuth.reject(error)
      this.pendingAuth = null
    }
  }
}

/**
 * Create a CDC WebSocket transport
 */
export function createCdcWsTransport(config: CDCWsTransportConfig): CDCWsTransport {
  return new CDCWsTransport(config)
}