// ✅ Default: direct HTTP call — lowest latency, simplest failure model
const score = await fraudClient.score({ transactionId, amount, userId })
// Round-trip: ~5–20ms over HTTP/2 gRPC
// Failure: immediate 4xx/5xx with structured error
// No broker dependency, no correlation ID, no timeout management

// ❌ Request-reply over messaging adds at least 2× broker latency
// publish → broker → consumer → publish response → broker → caller
// Round-trip: ~20–100ms minimum (2 broker round-trips)
// Adds correlation ID management, timeout handling, orphaned response cleanup

// ❌ Shared response topic — multiple callers consume each other's responses
const correlationId = uuid()

// Caller A publishes request
await producer.send({
  topic: 'fraud.requests',
  messages: [{ value: JSON.stringify({ correlationId, transactionId: 'TXN-1' }) }]
})

// Caller A and Caller B both subscribe to the same fraud.responses topic
// They are in the SAME consumer group — Kafka assigns each partition to ONE consumer
// Caller A's response may be delivered to Caller B

const consumer = kafka.consumer({ groupId: 'fraud-response-consumers' }) // shared!
await consumer.subscribe({ topic: 'fraud.responses' })
await consumer.run({
  eachMessage: async ({ message }) => {
    const { correlationId: id, score } = JSON.parse(message.value!.toString())
    // Is this OUR response? Maybe. Maybe it belongs to another caller instance.
    const pending = pendingRequests.get(id)
    if (pending) pending.resolve(score)
    // If not found: silently dropped — other caller is stuck waiting
  }
})

// ✅ RabbitMQ Direct Reply-To — broker manages per-caller queue lifecycle
import amqplib from 'amqplib'

const connection = await amqplib.connect('amqp://rabbitmq:5672')
const channel    = await connection.createChannel()

class FraudRpcClient {
  private pending = new Map<string, {
    resolve: (score: number) => void
    reject:  (err: Error)   => void
    timer:   NodeJS.Timeout
  }>()

  async init(): Promise<void> {
    // Subscribe to the pseudo-queue — only THIS connection receives its own replies
    await channel.consume(
      'amq.rabbitmq.reply-to',
      (msg) => {
        if (!msg) return
        const { correlationId, score } = JSON.parse(msg.content.toString())
        const pending = this.pending.get(correlationId)
        if (!pending) return   // timed out or already resolved
        clearTimeout(pending.timer)
        this.pending.delete(correlationId)
        pending.resolve(score)
      },
      { noAck: true }   // reply-to pseudo-queue requires noAck
    )
  }

  async score(transactionId: string, timeoutMs = 5000): Promise<number> {
    const correlationId = crypto.randomUUID()

    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.pending.delete(correlationId)
        reject(new Error(`Fraud score timeout after ${timeoutMs}ms [corrId=${correlationId}]`))
      }, timeoutMs)

      this.pending.set(correlationId, { resolve, reject, timer })

      // correlationId set by caller — echoed verbatim by responder
      channel.sendToQueue('fraud.requests', Buffer.from(JSON.stringify({ transactionId })), {
        correlationId,
        replyTo:    'amq.rabbitmq.reply-to',   // broker routes response back to this queue
        persistent: true,
      })
    })
  }
}

// ✅ Fraud service responder — echoes correlationId and replyTo
await channel.prefetch(20)
await channel.consume('fraud.requests', async (msg) => {
  if (!msg) return
  const { transactionId } = JSON.parse(msg.content.toString())

  const score = await fraudModel.evaluate(transactionId)

  // CRITICAL: echo correlationId from the request, never generate a new one
  channel.sendToQueue(
    msg.properties.replyTo,   // routes directly to caller's pseudo-queue
    Buffer.from(JSON.stringify({
      correlationId: msg.properties.correlationId,  // verbatim echo
      score,
    })),
    { correlationId: msg.properties.correlationId }
  )
  channel.ack(msg)
})

// ✅ Kafka per-caller inbox — caller has its own topic, no shared consumer group
const CALLER_REPLY_TOPIC = `fraud.responses.checkout-service`   // unique per service

// Caller: subscribes to its own reply topic in its own group
const replyConsumer = kafka.consumer({ groupId: 'checkout-fraud-reply' })
await replyConsumer.subscribe({ topic: CALLER_REPLY_TOPIC })

const pending = new Map<string, {
  resolve: (score: number) => void
  reject:  (err: Error)   => void
  timer:   NodeJS.Timeout
}>()

await replyConsumer.run({
  eachMessage: async ({ message }) => {
    const { correlationId, score } = JSON.parse(message.value!.toString())
    const req = pending.get(correlationId)
    if (!req) return
    clearTimeout(req.timer)
    pending.delete(correlationId)
    req.resolve(score)
  }
})

// Caller: publishes request with replyTopic header
async function requestFraudScore(transactionId: string, timeoutMs = 5000): Promise<number> {
  const correlationId = crypto.randomUUID()

  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      pending.delete(correlationId)
      reject(new Error(`Timeout [corrId=${correlationId}]`))
    }, timeoutMs)

    pending.set(correlationId, { resolve, reject, timer })

    producer.send({
      topic: 'fraud.requests',
      messages: [{
        key:   correlationId,
        value: JSON.stringify({ correlationId, transactionId }),
        headers: {
          'reply-topic': CALLER_REPLY_TOPIC,   // responder reads this header
          'correlation-id': correlationId,
        },
      }]
    })
  })
}

// ✅ Fraud responder: routes to reply-topic from request headers
await consumer.run({
  eachMessage: async ({ message }) => {
    const { correlationId, transactionId } = JSON.parse(message.value!.toString())
    const replyTopic    = message.headers?.['reply-topic']?.toString()
    const correlationId = message.headers?.['correlation-id']?.toString()

    if (!replyTopic || !correlationId) {
      // Fire-and-forget request — no reply expected
      await processEvent(transactionId)
      return
    }

    const score = await fraudModel.evaluate(transactionId)

    await producer.send({
      topic: replyTopic,
      messages: [{
        key:     correlationId,   // same key → same partition → ordered replies per caller
        value:   JSON.stringify({ correlationId, score }),
        headers: { 'correlation-id': correlationId },
      }]
    })
  }
})

// ✅ Timeout contract: caller cleans up pending state on expiry
const timer = setTimeout(() => {
  pending.delete(correlationId)
  reject(new Error(`Timeout after ${timeoutMs}ms`))
  // The response may still arrive after the timeout:
  // - It will be consumed from the reply topic/queue
  // - The pending.get(correlationId) lookup will return undefined
  // - The response is silently discarded — no memory leak, no crash
}, timeoutMs)

// ✅ Cleanup on graceful shutdown
process.on('SIGTERM', () => {
  for (const [id, req] of pending.entries()) {
    clearTimeout(req.timer)
    req.reject(new Error('Service shutting down'))
    pending.delete(id)
  }
})