/**
 * CDC SSE (Server-Sent Events) Transport for postgres.do
 *
 * Provides HTTP-based change data capture streaming via SSE:
 * - Works in environments where WebSocket is blocked
 * - Uses standard EventSource/fetch with streaming body
 * - Automatic reconnection with Last-Event-ID header
 * - LSN-based resumption for reliable delivery
 */

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

/**
 * SSE event types for CDC
 */
type SseEventType = 'insert' | 'update' | 'delete' | 'truncate' | 'begin' | 'commit' | 'keepalive' | 'error' | 'subscribe_result' | 'unsubscribe_result'

/**
 * SSE event data structure
 */
interface SseEventData {
  type: SseEventType
  subscriptionId?: string
  table?: string
  schema?: string
  data?: Row
  old?: Row
  new?: Row
  changedColumns?: string[]
  xid?: number
  isLastInTransaction?: boolean
  success?: boolean
  error?: string
  message?: string
  code?: string
}

/**
 * Default configuration values
 */
const DEFAULTS = {
  retryMs: 3000,
  connectTimeout: 10000,
  maxReconnectAttempts: 10,
}

/**
 * CDC SSE Transport implementation
 */
export class CDCSseTransport implements CDCTransport {
  private config: {
    url: string
    apiKey?: string
    resumeFrom?: string
    retryMs: number
    fetch?: typeof fetch
    headers?: Record<string, string>
  }
  private fetchImpl: typeof fetch
  private connected = false
  private lastEventId: string | null = null
  private shouldReconnect = true
  private reconnectAttempt = 0
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null
  private abortController: AbortController | 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 URL construction
  private activeSubscriptions = new Map<string, {
    table: string
    schema: string
    options: CDCSubscriptionOptions
  }>()

  // Pending subscription/unsubscription requests
  private pendingSubscriptions = new Map<string, {
    resolve: () => void
    reject: (error: Error) => void
  }>()
  private pendingUnsubscriptions = new Map<string, {
    resolve: () => void
    reject: (error: Error) => void
  }>()

  constructor(config: CDCSseTransportConfig) {
    this.config = {
      url: config.url,
      retryMs: config.retryMs ?? DEFAULTS.retryMs,
    }
    // Only set optional properties if defined
    if (config.apiKey !== undefined) {
      this.config.apiKey = config.apiKey
    }
    if (config.resumeFrom !== undefined) {
      this.config.resumeFrom = config.resumeFrom
    }
    if (config.fetch !== undefined) {
      this.config.fetch = config.fetch
    }
    if (config.headers !== undefined) {
      this.config.headers = config.headers
    }
    this.fetchImpl = config.fetch ?? globalThis.fetch
    this.lastEventId = config.resumeFrom ?? null
  }

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

    this.shouldReconnect = true
    this.abortController = new AbortController()

    await this._connect()
  }

  /**
   * Internal connect implementation
   */
  private async _connect(): Promise<void> {
    const headers: Record<string, string> = {
      Accept: 'text/event-stream',
      'Cache-Control': 'no-cache',
      ...this.config.headers,
    }

    if (this.config.apiKey) {
      headers['Authorization'] = `Bearer ${this.config.apiKey}`
    }

    if (this.lastEventId) {
      headers['Last-Event-ID'] = this.lastEventId
    }

    // Build URL with subscription filters
    const url = this.buildSubscriptionUrl()

    try {
      const response = await this.fetchImpl(url, {
        method: 'GET',
        headers,
        signal: this.abortController?.signal ?? null,
      })

      if (!response.ok) {
        const errorBody = await response.text().catch(() => '')
        throw new ConnectionError(`SSE connection failed: ${response.status} ${response.statusText}${errorBody ? ` - ${errorBody}` : ''}`)
      }

      const contentType = response.headers.get('Content-Type')
      if (!contentType || !contentType.includes('text/event-stream')) {
        throw new ConnectionError(`Invalid Content-Type: expected text/event-stream, got ${contentType}`)
      }

      this.connected = true
      this.reconnectAttempt = 0

      // Resolve any pending subscriptions since SSE subscriptions are URL-based
      for (const [subscriptionId, { resolve }] of this.pendingSubscriptions) {
        resolve()
        this.pendingSubscriptions.delete(subscriptionId)
      }

      // Process the stream
      await this.processStream(response)
    } catch (error) {
      if (error instanceof Error && error.name === 'AbortError') {
        // Intentional abort - don't reconnect
        return
      }

      // Reject pending subscriptions on error
      for (const [subscriptionId, { reject }] of this.pendingSubscriptions) {
        reject(error instanceof Error ? error : new Error(String(error)))
        this.pendingSubscriptions.delete(subscriptionId)
      }

      throw error
    }
  }

  /**
   * Build URL with subscription parameters
   */
  private buildSubscriptionUrl(): string {
    const url = new URL(this.config.url)

    // Add table filter if there are active subscriptions
    const tables = Array.from(this.activeSubscriptions.values()).map(s => s.table)
    if (tables.length > 0) {
      url.searchParams.set('tables', tables.join(','))
    }

    // Add schema filter
    const schemas = new Set(Array.from(this.activeSubscriptions.values()).map(s => s.schema))
    if (schemas.size > 0) {
      url.searchParams.set('schemas', Array.from(schemas).join(','))
    }

    // Add subscription IDs
    const subscriptionIds = Array.from(this.activeSubscriptions.keys())
    if (subscriptionIds.length > 0) {
      url.searchParams.set('subscriptionIds', subscriptionIds.join(','))
    }

    // Add common options from first subscription
    const firstSub = Array.from(this.activeSubscriptions.values())[0]
    if (firstSub) {
      if (firstSub.options.events) {
        url.searchParams.set('events', firstSub.options.events.join(','))
      }
      if (firstSub.options.filter) {
        url.searchParams.set('filter', firstSub.options.filter)
      }
      if (firstSub.options.includeOldRow) {
        url.searchParams.set('includeOldRow', 'true')
      }
      if (firstSub.options.trackChangedColumns) {
        url.searchParams.set('trackChangedColumns', 'true')
      }
      if (firstSub.options.batchSize) {
        url.searchParams.set('batchSize', String(firstSub.options.batchSize))
      }
    }

    return url.toString()
  }

  /**
   * Process the SSE stream
   */
  private async processStream(response: Response): Promise<void> {
    if (!response.body) {
      throw new ConnectionError('Response body is null')
    }

    const reader = response.body.getReader()
    const decoder = new TextDecoder()
    let buffer = ''

    try {
      while (true) {
        const { done, value } = await reader.read()

        if (done) {
          // Stream ended - schedule reconnect
          this.connected = false
          if (this.shouldReconnect) {
            this.scheduleReconnect()
          }
          return
        }

        buffer += decoder.decode(value, { stream: true })

        // Process complete events (separated by double newline)
        const events = buffer.split('\n\n')
        buffer = events.pop() || '' // Keep incomplete event in buffer

        for (const eventText of events) {
          if (eventText.trim()) {
            this.parseAndHandleEvent(eventText)
          }
        }
      }
    } catch (error) {
      this.connected = false

      if (error instanceof Error && error.name === 'AbortError') {
        // Intentional abort
        return
      }

      // Emit error and schedule reconnect
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, error instanceof Error ? error : new Error(String(error)))
      }

      if (this.shouldReconnect) {
        this.scheduleReconnect()
      }
    }
  }

  /**
   * Parse and handle an SSE event
   */
  private parseAndHandleEvent(eventText: string): void {
    let id: string | undefined
    let eventType: string | undefined
    let data: string | undefined
    let retry: number | undefined

    const lines = eventText.split('\n')
    for (const line of lines) {
      if (line.startsWith(':')) {
        // Comment line - ignore
        continue
      }

      const colonIndex = line.indexOf(':')
      if (colonIndex === -1) {
        continue
      }

      const field = line.substring(0, colonIndex)
      // Skip the space after colon if present
      const value = line.substring(colonIndex + 1).trimStart()

      switch (field) {
        case 'id':
          id = value
          break
        case 'event':
          eventType = value
          break
        case 'data':
          data = data ? data + '\n' + value : value
          break
        case 'retry':
          retry = parseInt(value, 10)
          if (!isNaN(retry)) {
            this.config.retryMs = retry
          }
          break
      }
    }

    // Update last event ID
    if (id) {
      this.lastEventId = id
    }

    if (!data) {
      return
    }

    // Parse data JSON
    let eventData: SseEventData
    try {
      eventData = JSON.parse(data)
    } catch (error) {
      // Emit parse error
      for (const subscriptionId of this.activeSubscriptions.keys()) {
        this.errorHandler?.(subscriptionId, new Error(`Failed to parse SSE event data: ${error instanceof Error ? error.message : String(error)}`))
      }
      return
    }

    // Handle different event types
    this.handleParsedEvent(id, eventType || eventData.type, eventData)
  }

  /**
   * Handle a parsed event
   */
  private handleParsedEvent(id: string | undefined, eventType: string, eventData: SseEventData): void {
    const subscriptionId = eventData.subscriptionId || Array.from(this.activeSubscriptions.keys())[0]

    switch (eventType) {
      case 'insert':
      case 'update':
      case 'delete':
      case 'truncate': {
        if (!subscriptionId) return

        const changeEvent: CDCChangeEvent = {
          id: id || `${Date.now()}-${Math.random()}`,
          lsn: id || '',
          operation: eventType.toUpperCase() as CDCOperation,
          table: eventData.table || '',
          schema: eventData.schema || 'public',
          timestamp: new Date(),
          xid: eventData.xid || 0,
          isLastInTransaction: eventData.isLastInTransaction || false,
        }

        if (eventType === 'update' && eventData.data) {
          const updateData = eventData.data as { old?: Row; new?: Row }
          changeEvent.oldRow = updateData.old || eventData.old
          changeEvent.newRow = updateData.new || eventData.new
        } else if (eventType === 'insert') {
          changeEvent.newRow = eventData.data
        } else if (eventType === 'delete') {
          changeEvent.oldRow = eventData.data
        }

        if (eventData.changedColumns) {
          changeEvent.changedColumns = eventData.changedColumns
        }

        this.eventHandler?.(subscriptionId, changeEvent)
        break
      }

      case 'keepalive':
      case 'begin':
      case 'commit':
        // Just update the last event ID, no action needed
        break

      case 'error':
        if (subscriptionId) {
          this.errorHandler?.(subscriptionId, new Error(eventData.message || eventData.error || 'Unknown error'))
        }
        break

      case 'subscribe_result':
        if (eventData.subscriptionId) {
          const pending = this.pendingSubscriptions.get(eventData.subscriptionId)
          if (pending) {
            if (eventData.success) {
              pending.resolve()
            } else {
              pending.reject(new Error(eventData.error || 'Subscription failed'))
            }
            this.pendingSubscriptions.delete(eventData.subscriptionId)
          }
        }
        break

      case 'unsubscribe_result':
        if (eventData.subscriptionId) {
          const pending = this.pendingUnsubscriptions.get(eventData.subscriptionId)
          if (pending) {
            pending.resolve()
            this.pendingUnsubscriptions.delete(eventData.subscriptionId)
          }
        }
        break
    }
  }

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

    // Use server-specified retry or exponential backoff
    const delay = this.config.retryMs * Math.pow(1.5, this.reconnectAttempt - 1)

    this.reconnectTimer = setTimeout(async () => {
      this.reconnectHandler?.(this.reconnectAttempt, this.lastEventId ?? undefined)

      try {
        this.abortController = new AbortController()
        await this._connect()
      } catch (error) {
        // Will schedule another reconnect
        if (this.shouldReconnect) {
          this.scheduleReconnect()
        }
      }
    }, delay)
  }

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

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

    if (this.abortController) {
      this.abortController.abort()
      this.abortController = null
    }

    this.connected = false
    this.activeSubscriptions.clear()

    // Reject all pending requests
    for (const [, { reject }] of this.pendingSubscriptions) {
      reject(new ConnectionError('Connection closed'))
    }
    this.pendingSubscriptions.clear()

    for (const [, { resolve }] of this.pendingUnsubscriptions) {
      resolve() // Unsubscriptions resolve on disconnect
    }
    this.pendingUnsubscriptions.clear()

    this.closeHandler?.()
  }

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

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

  /**
   * Subscribe to a table
   * For SSE, this adds to the subscription set and reconnects
   */
  async subscribe(
    subscriptionId: string,
    table: string,
    schema: string,
    options: CDCSubscriptionOptions
  ): Promise<void> {
    // Store the subscription
    this.activeSubscriptions.set(subscriptionId, { table, schema, options })

    // If already connected, need to reconnect with new subscription
    if (this.connected) {
      // Create a pending subscription promise
      const promise = new Promise<void>((resolve, reject) => {
        this.pendingSubscriptions.set(subscriptionId, { resolve, reject })
      })

      // Reconnect with updated subscriptions
      if (this.abortController) {
        this.abortController.abort()
      }
      this.abortController = new AbortController()
      this._connect().catch(() => {
        // Error handled in _connect
      })

      return promise
    } else {
      // Not connected yet - subscription will be included when connecting
      return new Promise<void>((resolve, reject) => {
        this.pendingSubscriptions.set(subscriptionId, { resolve, reject })
      })
    }
  }

  /**
   * Unsubscribe from a subscription
   * For SSE, this removes from the subscription set and reconnects
   */
  async unsubscribe(subscriptionId: string): Promise<void> {
    this.activeSubscriptions.delete(subscriptionId)

    // If no more subscriptions, just disconnect
    if (this.activeSubscriptions.size === 0) {
      await this.disconnect()
      return
    }

    // Otherwise reconnect with remaining subscriptions
    if (this.connected) {
      return new Promise<void>((resolve, reject) => {
        this.pendingUnsubscriptions.set(subscriptionId, { resolve, reject })

        // Reconnect with updated subscriptions
        if (this.abortController) {
          this.abortController.abort()
        }
        this.abortController = new AbortController()
        this._connect().catch(() => {
          // Error handled in _connect
        })

        // Resolve immediately for SSE since unsubscription is URL-based
        setTimeout(() => {
          const pending = this.pendingUnsubscriptions.get(subscriptionId)
          if (pending) {
            pending.resolve()
            this.pendingUnsubscriptions.delete(subscriptionId)
          }
        }, 100)
      })
    }
  }

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

/**
 * Create a CDC SSE transport
 */
export function createCdcSseTransport(config: CDCSseTransportConfig): CDCSseTransport {
  return new CDCSseTransport(config)
}