/**
 * Phoenix Event Bus Service
 * Durable events, replay, versioning, consumer idempotency
 */

import { getDb } from '../../db/index.js'
import { logger } from '../../lib/logger.js'

export interface Event {
  eventId: string
  eventType: string
  aggregateId: string
  version: number
  payload: Record<string, any>
  metadata: Record<string, any>
  timestamp: Date
  correlationId: string
}

export interface ConsumerOffset {
  consumerId: string
  eventId: string
  processedAt: Date
}

class EventBusService {
  /**
   * Publish an event (via outbox pattern)
   */
  async publishEvent(
    eventType: string,
    aggregateId: string,
    payload: Record<string, any>,
    correlationId: string,
    metadata: Record<string, any> = {}
  ): Promise<Event> {
    const db = getDb()
    
    // Get next version for this aggregate
    const versionResult = await db.query(
      `SELECT COALESCE(MAX(version), 0) + 1 as next_version
       FROM events
       WHERE aggregate_id = $1 AND event_type = $2`,
      [aggregateId, eventType]
    )
    const version = parseInt(versionResult.rows[0].next_version)

    // Insert into outbox (atomic with business logic)
    const result = await db.query(
      `INSERT INTO event_outbox (
        event_type, aggregate_id, version, payload, metadata, correlation_id, status
      ) VALUES ($1, $2, $3, $4, $5, $6, 'PENDING')
      RETURNING *`,
      [
        eventType,
        aggregateId,
        version,
        JSON.stringify(payload),
        JSON.stringify(metadata),
        correlationId
      ]
    )

    logger.info('Event published to outbox', { 
      eventId: result.rows[0].id, 
      eventType, 
      correlationId 
    })

    // Process outbox (would be done by background worker)
    await this.processOutbox()

    return this.mapEvent(result.rows[0])
  }

  /**
   * Process outbox (typically run by background worker)
   */
  async processOutbox(): Promise<void> {
    const db = getDb()
    
    // Get pending events
    const pending = await db.query(
      `SELECT * FROM event_outbox WHERE status = 'PENDING' ORDER BY created_at LIMIT 100`
    )

    for (const event of pending.rows) {
      try {
        // Publish to actual event bus (Kafka/Redpanda/NATS)
        await this.publishToBus(event)
        
        // Mark as published
        await db.query(
          `UPDATE event_outbox SET status = 'PUBLISHED', published_at = NOW() WHERE id = $1`,
          [event.id]
        )

        // Insert into events table
        await db.query(
          `INSERT INTO events (
            event_id, event_type, aggregate_id, version, payload, metadata, correlation_id, timestamp
          ) VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
          ON CONFLICT (event_id) DO NOTHING`,
          [
            event.id,
            event.event_type,
            event.aggregate_id,
            event.version,
            event.payload,
            event.metadata,
            event.correlation_id
          ]
        )

        logger.info('Event processed from outbox', { eventId: event.id })
      } catch (error) {
        logger.error('Failed to process event from outbox', { eventId: event.id, error })
        // Would implement retry logic here
      }
    }
  }

  /**
   * Consume events with idempotency
   */
  async consumeEvents(
    consumerId: string,
    eventType: string,
    limit: number = 100
  ): Promise<Event[]> {
    const db = getDb()
    
    // Get last processed event
    const lastOffset = await db.query(
      `SELECT event_id FROM consumer_offsets 
       WHERE consumer_id = $1 AND event_type = $2 
       ORDER BY processed_at DESC LIMIT 1`,
      [consumerId, eventType]
    )

    const lastEventId = lastOffset.rows[0]?.event_id || null

    // Get events after last processed
    const query = lastEventId
      ? `SELECT * FROM events 
         WHERE event_type = $1 AND id > $2 
         ORDER BY timestamp ASC LIMIT $3`
      : `SELECT * FROM events 
         WHERE event_type = $1 
         ORDER BY timestamp ASC LIMIT $2`

    const params = lastEventId ? [eventType, lastEventId, limit] : [eventType, limit]
    const result = await db.query(query, params)

    // Record offsets
    for (const event of result.rows) {
      await db.query(
        `INSERT INTO consumer_offsets (consumer_id, event_id, event_type, processed_at)
         VALUES ($1, $2, $3, NOW())
         ON CONFLICT (consumer_id, event_id) DO NOTHING`,
        [consumerId, event.id, eventType]
      )
    }

    return result.rows.map(this.mapEvent)
  }

  private async publishToBus(event: any): Promise<void> {
    // This would publish to Kafka/Redpanda/NATS
    logger.info('Publishing to event bus', { eventId: event.id })
    // Placeholder - would implement actual bus publishing
  }

  private mapEvent(row: any): Event {
    return {
      eventId: row.id || row.event_id,
      eventType: row.event_type,
      aggregateId: row.aggregate_id,
      version: row.version,
      payload: typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload,
      metadata: typeof row.metadata === 'string' ? JSON.parse(row.metadata) : row.metadata,
      timestamp: row.timestamp || row.created_at,
      correlationId: row.correlation_id
    }
  }
}

export const eventBusService = new EventBusService()
