/**
 * RPC Transport for postgres.do
 *
 * Connects to PostgresDO via capnweb RPC for:
 * - Direct Worker-to-DO communication without HTTP overhead
 * - Promise pipelining for efficient batched operations
 * - Magic map support for N+1 query elimination
 *
 * @see packages/postgres/src/worker/rpc.ts for the server-side RPC API
 * @see packages/postgres/src/worker/proxy.ts for the proxy session pool
 */

import type {
  Transport,
  QueryResult,
  Row,
  TransactionOptions,
  FieldInfo,
} from '../types'
import { PostgresError, ConnectionError, TimeoutError } from '../types'
import { RpcPromise, createRpcPromise, type RpcExecutionContext, type BatchResult } from '../rpc/rpc-promise'

/**
 * RPC Transport configuration
 */
export interface RpcTransportConfig {
  /** WebSocket URL for the capnweb RPC endpoint */
  url: string
  /** API key for authentication */
  apiKey?: string | undefined
  /** Custom WebSocket implementation */
  WebSocket?: typeof WebSocket | undefined
  /** Connection timeout in milliseconds */
  connectTimeout?: number | undefined
  /** Request timeout in milliseconds */
  requestTimeout?: number | undefined
  /** Enable automatic batching */
  autoBatch?: boolean | undefined
  /** Batch delay in milliseconds (how long to wait for more queries) */
  batchDelayMs?: number | undefined
}

/** Default configuration values */
const DEFAULTS = {
  url: 'wss://db.postgres.do/rpc',
  connectTimeout: 10000,
  requestTimeout: 30000,
  autoBatch: true,
  batchDelayMs: 0, // No delay by default - use microtask batching
}

/** RPC message types */
enum RpcMessageType {
  // Requests
  Query = 'query',
  Batch = 'batch',
  BatchTransaction = 'batch_tx',
  Transaction = 'transaction',
  TransactionQuery = 'tx_query',
  TransactionCommit = 'tx_commit',
  TransactionRollback = 'tx_rollback',
  Ping = 'ping',
  Auth = 'auth',

  // Responses
  QueryResult = 'query_result',
  BatchResult = 'batch_result',
  TransactionResult = 'tx_result',
  Error = 'error',
  Pong = 'pong',
  AuthResult = 'auth_result',
}

/** Pending request waiting for a response */
interface PendingRequest<T = unknown> {
  resolve: (value: T) => void
  reject: (error: Error) => void
  timeout: ReturnType<typeof setTimeout>
}

/** Pending batch of queries */
interface PendingBatch {
  queries: Array<{ sql: string; params?: unknown[] }>
  resolvers: Array<{
    resolve: (value: BatchResult) => void
    reject: (error: Error) => void
  }>
  timer: ReturnType<typeof setTimeout> | null
}

/** Resolved config with optional properties handled */
interface ResolvedConfig {
  url: string
  apiKey: string | undefined
  WebSocket: typeof WebSocket | undefined
  connectTimeout: number
  requestTimeout: number
  autoBatch: boolean
  batchDelayMs: number
}

/**
 * RPC Transport implementation
 *
 * Provides capnweb-style RPC communication with PostgresDO:
 * - Automatic query batching for N+1 elimination
 * - Promise pipelining support
 * - WebSocket-based persistent connection
 */
export class RpcTransport implements Transport, RpcExecutionContext {
  private readonly config: ResolvedConfig
  private readonly WebSocketImpl: typeof WebSocket
  private ws: WebSocket | null = null
  private connected = false
  private authenticated = false
  private connecting: Promise<void> | null = null
  private requestId = 0
  private pendingRequests = new Map<number, PendingRequest>()
  private pendingBatch: PendingBatch | null = null
  private collecting = false

  constructor(config: RpcTransportConfig) {
    this.config = {
      url: config.url ?? DEFAULTS.url,
      apiKey: config.apiKey,
      WebSocket: config.WebSocket,
      connectTimeout: config.connectTimeout ?? DEFAULTS.connectTimeout,
      requestTimeout: config.requestTimeout ?? DEFAULTS.requestTimeout,
      autoBatch: config.autoBatch ?? DEFAULTS.autoBatch,
      batchDelayMs: config.batchDelayMs ?? DEFAULTS.batchDelayMs,
    }
    this.WebSocketImpl = config.WebSocket || globalThis.WebSocket
  }

  // ========== RpcExecutionContext implementation ==========

  /**
   * Add a query to the current batch
   */
  addQuery(sql: string, params?: unknown[]): number {
    if (!this.pendingBatch) {
      this.pendingBatch = {
        queries: [],
        resolvers: [],
        timer: null,
      }

      // Schedule batch execution on next microtask
      if (this.config.batchDelayMs === 0) {
        queueMicrotask(() => this.flushBatch())
      } else {
        this.pendingBatch.timer = setTimeout(
          () => this.flushBatch(),
          this.config.batchDelayMs
        )
      }
    }

    const queryId = this.pendingBatch.queries.length
    this.pendingBatch.queries.push({ sql, params: params ?? [] })

    return queryId
  }

  /**
   * Execute all batched queries
   */
  async executeBatch(): Promise<BatchResult[]> {
    if (!this.pendingBatch || this.pendingBatch.queries.length === 0) {
      return []
    }

    await this.ensureConnected()

    const batch = this.pendingBatch
    this.pendingBatch = null

    if (batch.timer) {
      clearTimeout(batch.timer)
    }

    const id = this.nextRequestId()
    const response = await this.sendRequest<RpcBatchResponse>(id, {
      type: RpcMessageType.Batch,
      id,
      queries: batch.queries,
    })

    return response.results.map((r, i) => ({
      queryId: i,
      rows: r.rows,
      fields: r.fields,
      rowCount: r.rowCount,
      durationMs: r.durationMs,
    }))
  }

  /**
   * Get a pending result by query ID
   */
  async getResult(_queryId: number): Promise<unknown> {
    // Results are returned in order from executeBatch
    // This method is for future optimization where results
    // can be streamed as they complete
    throw new Error('getResult is not yet implemented for streaming results')
  }

  /**
   * Check if the context is currently collecting queries
   */
  isCollecting(): boolean {
    return this.collecting
  }

  /**
   * Start collecting queries for batching
   */
  startCollecting(): void {
    this.collecting = true
  }

  /**
   * Stop collecting and execute batch
   */
  async stopCollecting(): Promise<void> {
    this.collecting = false
    await this.flushBatch()
  }

  /**
   * Flush the pending batch immediately
   */
  private async flushBatch(): Promise<void> {
    if (!this.pendingBatch || this.pendingBatch.queries.length === 0) {
      return
    }

    const batch = this.pendingBatch
    this.pendingBatch = null

    if (batch.timer) {
      clearTimeout(batch.timer)
    }

    try {
      await this.ensureConnected()

      const id = this.nextRequestId()
      const response = await this.sendRequest<RpcBatchResponse>(id, {
        type: RpcMessageType.Batch,
        id,
        queries: batch.queries,
      })

      // Resolve all pending queries with their results
      for (let i = 0; i < batch.resolvers.length; i++) {
        const resolver = batch.resolvers[i]
        const result = response.results[i]
        if (resolver && result) {
          resolver.resolve({
            queryId: i,
            rows: result.rows,
            fields: result.fields,
            rowCount: result.rowCount,
            durationMs: result.durationMs,
          })
        } else if (resolver) {
          resolver.reject(new Error(`No result for query ${i}`))
        }
      }
    } catch (error) {
      // Reject all pending queries with the error
      for (const resolver of batch.resolvers) {
        if (resolver) {
          resolver.reject(error instanceof Error ? error : new Error(String(error)))
        }
      }
    }
  }

  // ========== Transport interface implementation ==========

  /**
   * Execute a SQL query with RpcPromise support
   */
  queryRpc<T extends Row = Row>(
    sql: string,
    params?: unknown[]
  ): RpcPromise<T[]> {
    return createRpcPromise(async () => {
      const result = await this.query<T>(sql, params)
      return result.rows
    }, this)
  }

  /**
   * Execute a SQL query and return full result
   */
  async query<T extends Row = Row>(
    sql: string,
    params?: unknown[]
  ): Promise<QueryResult<T>> {
    // If autoBatch is enabled and we're in a batching context
    if (this.config.autoBatch && this.collecting) {
      return new Promise((resolve, reject) => {
        const queryId = this.addQuery(sql, params)

        if (!this.pendingBatch) {
          reject(new Error('Batch was flushed before query could be added'))
          return
        }

        this.pendingBatch.resolvers[queryId] = {
          resolve: (result) => {
            resolve({
              rows: result.rows as T[],
              fields: result.fields as FieldInfo[],
              rowCount: result.rowCount,
              command: 'SELECT',
            })
          },
          reject,
        }
      })
    }

    await this.ensureConnected()

    const id = this.nextRequestId()
    const response = await this.sendRequest<RpcQueryResponse<T>>(id, {
      type: RpcMessageType.Query,
      id,
      sql,
      params: params || [],
    })

    return {
      rows: response.rows || [],
      fields: (response.fields || []) as FieldInfo[],
      rowCount: response.rowCount ?? response.rows?.length ?? 0,
      command: response.command || 'SELECT',
    }
  }

  /**
   * Execute a batch of queries
   */
  async batch<T extends Row = Row>(
    queries: Array<{ sql: string; params?: unknown[] }>
  ): Promise<Array<QueryResult<T>>> {
    await this.ensureConnected()

    const id = this.nextRequestId()
    const response = await this.sendRequest<RpcBatchResponse>(id, {
      type: RpcMessageType.Batch,
      id,
      queries,
    })

    return response.results.map(r => ({
      rows: r.rows as T[],
      fields: r.fields as FieldInfo[],
      rowCount: r.rowCount,
      command: 'SELECT',
    }))
  }

  /**
   * Execute a batch of queries within a transaction
   */
  async batchTransaction<T extends Row = Row>(
    queries: Array<{ sql: string; params?: unknown[] }>,
    options?: TransactionOptions
  ): Promise<Array<QueryResult<T>>> {
    await this.ensureConnected()

    const id = this.nextRequestId()
    const response = await this.sendRequest<RpcBatchResponse>(id, {
      type: RpcMessageType.BatchTransaction,
      id,
      queries,
      options: options ? this.serializeTransactionOptions(options) : undefined,
    })

    return response.results.map(r => ({
      rows: r.rows as T[],
      fields: r.fields as FieldInfo[],
      rowCount: r.rowCount,
      command: 'SELECT',
    }))
  }

  /**
   * Execute multiple queries in a transaction (Transport interface)
   */
  async transaction<T>(
    queries: Array<{ sql: string; params?: unknown[] }>,
    options?: TransactionOptions
  ): Promise<T> {
    const results = await this.batchTransaction(queries, options)
    return results as T
  }

  /**
   * Close the WebSocket connection
   */
  async close(): Promise<void> {
    // Flush any pending batch
    if (this.pendingBatch) {
      await this.flushBatch()
    }

    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.connecting = null
  }

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

  // ========== Private methods ==========

  /**
   * Ensure the WebSocket is connected and authenticated
   */
  private async ensureConnected(): Promise<void> {
    if (this.isConnected()) {
      return
    }

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

    this.connecting = this.connect()
    try {
      await this.connecting
    } finally {
      this.connecting = null
    }
  }

  /**
   * Establish WebSocket connection
   */
  private async 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 = async () => {
        this.connected = true

        // Authenticate if API key is provided
        if (this.config.apiKey) {
          try {
            await this.authenticate()
            clearTimeout(timeoutId)
            resolve()
          } catch (error) {
            clearTimeout(timeoutId)
            reject(error)
          }
        } else {
          this.authenticated = true
          clearTimeout(timeoutId)
          resolve()
        }
      }

      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

        if (!event.wasClean) {
          const error = new ConnectionError(`WebSocket closed unexpectedly: ${event.code} ${event.reason}`)
          this.rejectAllPending(error)
          reject(error)
        }
      }
    })
  }

  /**
   * Authenticate with the API key
   */
  private async authenticate(): Promise<void> {
    const id = this.nextRequestId()
    const response = await this.sendRequest<RpcAuthResponse>(id, {
      type: RpcMessageType.Auth,
      id,
      apiKey: this.config.apiKey,
    })

    if (!response.success) {
      throw new ConnectionError('Authentication failed')
    }

    this.authenticated = true
  }

  /**
   * Send a request and wait for response
   */
  private sendRequest<T>(id: number, message: unknown): Promise<T> {
    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        this.pendingRequests.delete(id)
        reject(new TimeoutError('Request timed out'))
      }, this.config.requestTimeout)

      this.pendingRequests.set(id, {
        resolve: resolve as (value: unknown) => void,
        reject,
        timeout,
      })

      if (this.ws && this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(JSON.stringify(message))
      } else {
        clearTimeout(timeout)
        this.pendingRequests.delete(id)
        reject(new ConnectionError('WebSocket is not connected'))
      }
    })
  }

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

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

    // Handle pong messages
    if (message.type === RpcMessageType.Pong) {
      return
    }

    // Handle responses to pending requests
    if ('id' in message && typeof message.id === 'number') {
      const pending = this.pendingRequests.get(message.id)
      if (pending) {
        clearTimeout(pending.timeout)
        this.pendingRequests.delete(message.id)

        if (message.type === RpcMessageType.Error) {
          const errorMessage = message as RpcErrorMessage
          pending.reject(new PostgresError(errorMessage.error.message || 'Database error', {
            code: errorMessage.error.code,
            severity: errorMessage.error.severity || 'ERROR',
            detail: errorMessage.error.detail,
            hint: errorMessage.error.hint,
            position: errorMessage.error.position,
            schema: errorMessage.error.schema,
            table: errorMessage.error.table,
            column: errorMessage.error.column,
            constraint: errorMessage.error.constraint,
          }))
        } else {
          pending.resolve(message)
        }
      }
    }
  }

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

    // Also reject pending batch
    if (this.pendingBatch) {
      for (const resolver of this.pendingBatch.resolvers) {
        resolver.reject(error)
      }
      if (this.pendingBatch.timer) {
        clearTimeout(this.pendingBatch.timer)
      }
      this.pendingBatch = null
    }
  }

  /**
   * Get next request ID
   */
  private nextRequestId(): number {
    return ++this.requestId
  }

  /**
   * Serialize transaction options
   */
  private serializeTransactionOptions(options: TransactionOptions): Record<string, unknown> {
    return {
      isolationLevel: options.isolationLevel,
      readOnly: options.readOnly,
      deferrable: options.deferrable,
    }
  }
}

/** RPC message type definitions */
interface RpcQueryResponse<T extends Row = Row> {
  type: RpcMessageType.QueryResult
  id: number
  rows?: T[]
  rowCount?: number
  fields?: Array<{ name: string; dataTypeID: number }>
  command?: string
}

interface RpcBatchResponse {
  type: RpcMessageType.BatchResult
  id: number
  results: Array<{
    rows: Row[]
    fields: Array<{ name: string; dataTypeID: number }>
    rowCount: number
    durationMs: number
  }>
  durationMs: number
}

interface RpcAuthResponse {
  type: RpcMessageType.AuthResult
  id: number
  success: boolean
}

interface RpcErrorMessage {
  type: RpcMessageType.Error
  id: number
  error: {
    message?: string
    code?: string
    severity?: string
    detail?: string
    hint?: string
    position?: number
    schema?: string
    table?: string
    column?: string
    constraint?: string
  }
}

type RpcIncomingMessage =
  | RpcQueryResponse
  | RpcBatchResponse
  | RpcAuthResponse
  | RpcErrorMessage
  | { type: RpcMessageType.Pong }

/**
 * Create an RPC transport instance
 */
export function createRpcTransport(config: RpcTransportConfig): RpcTransport {
  return new RpcTransport(config)
}