// ❌ Mocked Kafka — tests producer call signatures, not broker behaviour
const mockProducer = {
  send: jest.fn().mockResolvedValue({ topicName: 'orders', partition: 0 })
}
// This test passes even if:
// - The topic does not exist
// - The message format is wrong
// - The consumer cannot deserialize the message
// - The consumer group offset logic is incorrect
// Mocked tests give false confidence in async correctness

// ✅ Testcontainers — real Kafka, real broker semantics, runs in Docker
// False positives are impossible: if the broker can't handle it, the test fails

// test/setup/kafka.ts — shared Testcontainers setup
import { KafkaContainer, StartedKafkaContainer } from '@testcontainers/kafka'
import { Kafka } from 'kafkajs'

let container: StartedKafkaContainer
let kafka:     Kafka

export async function startKafka(): Promise<Kafka> {
  container = await new KafkaContainer('confluentinc/cp-kafka:7.6.0')
    .withExposedPorts(9093)
    .start()

  kafka = new Kafka({
    clientId: 'test-client',
    brokers:  [`${container.getHost()}:${container.getMappedPort(9093)}`],
  })
  return kafka
}

export async function stopKafka(): Promise<void> {
  await container.stop()
}

// vitest.config.ts — single Kafka container shared across all integration tests
// Start once → run all tests → teardown

// orders.integration.test.ts
import { describe, it, expect, beforeAll, afterAll } from 'vitest'
import { startKafka, stopKafka } from './setup/kafka'
import { Kafka, Admin } from 'kafkajs'
import { OrderService }   from '../src/services/order.service'
import { db }             from '../src/db'

let kafka: Kafka
let admin: Admin

beforeAll(async () => {
  kafka = await startKafka()
  admin = kafka.admin()
  await admin.connect()

  // Create topics needed by this test file
  await admin.createTopics({
    topics: [
      { topic: 'orders.created',  numPartitions: 3 },
      { topic: 'orders.dlq',      numPartitions: 1 },
    ]
  })
}, 60_000)  // Kafka startup: allow 60s

afterAll(async () => {
  await admin.disconnect()
  await stopKafka()
})

// test/setup/rabbitmq.ts
import { RabbitMQContainer, StartedRabbitMQContainer } from '@testcontainers/rabbitmq'
import amqplib from 'amqplib'

let container: StartedRabbitMQContainer

export async function startRabbitMQ(): Promise<amqplib.Connection> {
  container = await new RabbitMQContainer('rabbitmq:3.13-management')
    .withExposedPorts(5672)
    .start()

  return amqplib.connect(
    `amqp://${container.getHost()}:${container.getMappedPort(5672)}`
  )
}

export async function stopRabbitMQ(): Promise<void> {
  await container.stop()
}

// ❌ Shared consumer group — concurrent tests step on each other
describe('order processing', () => {
  it('processes a single order', async () => {
    const consumer = kafka.consumer({ groupId: 'test-fulfillment' }) // shared!
    // ...
  })

  it('handles duplicate orders idempotently', async () => {
    const consumer = kafka.consumer({ groupId: 'test-fulfillment' }) // same group!
    // When both tests run concurrently: consumer 2 joins group, triggers rebalance
    // Partitions redistributed mid-test → first test may lose messages to second consumer
    // → flaky: first test receives 2 messages, not 3
  })
})

// ✅ Unique groupId per test run — zero shared state, no rebalance interference
import { randomUUID } from 'crypto'

function uniqueGroupId(prefix: string): string {
  return `${prefix}-${randomUUID()}`  // 'test-fulfillment-a3f8...'
}

describe('order processing', () => {
  it('processes a single order', async () => {
    const consumer = kafka.consumer({ groupId: uniqueGroupId('test-fulfillment') })
    // ...
  })

  it('handles duplicate orders idempotently', async () => {
    const consumer = kafka.consumer({ groupId: uniqueGroupId('test-fulfillment') })
    // Different group ID — no interaction with the concurrent test above
  })
})

// ❌ Anti-pattern 1: fixed sleep — flaky on slow CI, wastes time on fast machines
await new Promise(resolve => setTimeout(resolve, 500))
expect(receivedMessages).toHaveLength(3)

// ❌ Anti-pattern 2: polling with no timeout — hangs CI forever on bugs
while (receivedMessages.length < 3) {
  await new Promise(resolve => setTimeout(resolve, 10))
}

// ❌ Anti-pattern 3: asserting exact timing — environment-dependent
const start = Date.now()
await waitForMessages(3)
expect(Date.now() - start).toBeLessThan(200)  // fails on GH Actions vs local

// ✅ poll-until: retries with timeout, fails deterministically
async function pollUntil<T>(
  fn:        () => T | Promise<T>,
  predicate: (value: T) => boolean,
  opts:      { timeoutMs?: number; intervalMs?: number } = {}
): Promise<T> {
  const { timeoutMs = 10_000, intervalMs = 50 } = opts
  const deadline = Date.now() + timeoutMs

  while (Date.now() < deadline) {
    const value = await fn()
    if (predicate(value)) return value
    await new Promise(resolve => setTimeout(resolve, intervalMs))
  }
  throw new Error(`pollUntil: condition not met after ${timeoutMs}ms`)
}

// ✅ Promise-based barrier: resolve exactly when N messages arrive
function createMessageBarrier(expectedCount: number): {
  barrier: Promise<KafkaMessage[]>
  onMessage: (msg: KafkaMessage) => void
} {
  const received: KafkaMessage[] = []
  let resolve!: (msgs: KafkaMessage[]) => void
  let reject!:  (err: Error) => void

  const barrier = new Promise<KafkaMessage[]>((res, rej) => {
    resolve = res
    reject  = rej
  })

  // Auto-timeout: fail test if messages don't arrive within 10s
  const timer = setTimeout(() => {
    reject(new Error(`Expected ${expectedCount} messages, received ${received.length}`))
  }, 10_000)

  const onMessage = (msg: KafkaMessage) => {
    received.push(msg)
    if (received.length >= expectedCount) {
      clearTimeout(timer)
      resolve(received)
    }
  }

  return { barrier, onMessage }
}

// orders.integration.test.ts
it('consumer processes order and writes to DB', async () => {
  const groupId = uniqueGroupId('test-fulfillment')
  const consumer = kafka.consumer({ groupId })
  const producer = kafka.producer()

  await consumer.connect()
  await producer.connect()
  await consumer.subscribe({ topic: 'orders.created', fromBeginning: false })

  // Set up message barrier: wait for exactly 1 message
  const { barrier, onMessage } = createMessageBarrier(1)

  await consumer.run({
    eachMessage: async ({ message }) => {
      // Run the real handler — writes to DB
      await orderService.process(JSON.parse(message.value!.toString()))
      onMessage(message)
    }
  })

  const order = { orderId: 'ORD-42', customerId: 'C-1', amountCents: 9900 }
  await producer.send({
    topic:    'orders.created',
    messages: [{ key: order.orderId, value: JSON.stringify(order) }]
  })

  // Wait deterministically — fails after 10s, not after 500ms sleep
  const [received] = await barrier

  // Assert the DB write happened
  const row = await db.query('SELECT * FROM orders WHERE order_id = $1', [order.orderId])
  expect(row.rows).toHaveLength(1)
  expect(row.rows[0].amount_cents).toBe(9900)

  await consumer.disconnect()
  await producer.disconnect()
}, 30_000)

// ❌ Checks that messages arrived — does NOT verify order
const messages = await collectMessages(consumer, 3)
expect(messages).toHaveLength(3)    // passes even if order is wrong

// ❌ Partial order check — passes for ABCDE when order ABCDE is correct
// but also passes for BACDE which violates per-entity ordering
expect(messages.map(m => m.key?.toString())).toContain('ORD-1')

// ✅ Explicit sequence assertion — fails if any message arrives out of order
it('processes order events in sequence within a partition', async () => {
  const events = [
    { orderId: 'ORD-1', event: 'created',   sequence: 1 },
    { orderId: 'ORD-1', event: 'confirmed', sequence: 2 },
    { orderId: 'ORD-1', event: 'shipped',   sequence: 3 },
  ]

  const { barrier, onMessage } = createMessageBarrier(3)

  // Publish all events for the SAME key → same partition → guaranteed order
  await producer.send({
    topic:    'orders.lifecycle',
    messages: events.map(e => ({
      key:   e.orderId,    // same key → same partition → strict ordering
      value: JSON.stringify(e),
    }))
  })

  const received = await barrier

  // Assert exact sequence — not just presence
  const sequences = received.map(m => JSON.parse(m.value!.toString()).sequence)
  expect(sequences).toEqual([1, 2, 3])  // strict ordering assertion
})

// ✅ Assert that duplicate delivery produces exactly one DB row
it('consumer is idempotent on duplicate delivery', async () => {
  const orderId = `ORD-${randomUUID()}`
  const message = { orderId, amountCents: 9900 }
  const encoded = Buffer.from(JSON.stringify(message))

  const { barrier, onMessage } = createMessageBarrier(2)

  // Publish the same logical message twice (simulate duplicate delivery)
  await producer.send({
    topic:    'orders.created',
    messages: [
      { key: orderId, value: encoded },
      { key: orderId, value: encoded },  // exact duplicate
    ]
  })

  await barrier  // wait for both to be consumed

  // Assert: two messages delivered, but only ONE DB row created
  const rows = await db.query('SELECT * FROM orders WHERE order_id = $1', [orderId])
  expect(rows.rows).toHaveLength(1)   // idempotency gate: ON CONFLICT DO NOTHING
})