// ❌ One connection per consumer — TCP overhead compounds with consumer count
const conn1 = await amqplib.connect('amqp://rabbitmq:5672')
const conn2 = await amqplib.connect('amqp://rabbitmq:5672')
const conn3 = await amqplib.connect('amqp://rabbitmq:5672')
// Each connection = one TCP socket + TLS handshake + auth round-trip

// ✅ One connection per process; one channel per consumer coroutine
const connection = await amqplib.connect('amqp://rabbitmq:5672')
const channel1   = await connection.createChannel()   // consumer 1
const channel2   = await connection.createChannel()   // consumer 2
const channel3   = await connection.createChannel()   // consumer 3
// AMQP multiplexes all three over the same TCP connection via channel IDs

// ❌ No prefetch — RabbitMQ floods the consumer with all available messages
// 50,000 messages outstanding, consumer OOMs or processes them serially anyway
const channel = await connection.createChannel()
await channel.consume('order.fulfillment', handler)

// ✅ Prefetch = 1: broker holds back until consumer acks the in-flight message
// Ensures fair dispatch when multiple consumers share a queue
await channel.prefetch(1)
await channel.consume('order.fulfillment', async (msg) => {
  if (!msg) return
  try {
    await processOrder(JSON.parse(msg.content.toString()))
    channel.ack(msg)                    // next message released only after ack
  } catch (err) {
    channel.nack(msg, false, false)     // reject without requeue → dead-letter exchange
  }
})

// ✅ Queue configured with a dead-letter exchange — rejected messages route to DLX
await channel.assertExchange('orders.dlx', 'direct', { durable: true })
await channel.assertQueue('orders.dlq', { durable: true })
await channel.bindQueue('orders.dlq', 'orders.dlx', 'order.fulfillment')

await channel.assertQueue('order.fulfillment', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange':    'orders.dlx',
    'x-dead-letter-routing-key': 'order.fulfillment',  // routing key in DLX
    'x-message-ttl':             86_400_000,            // 24h TTL — expired → DLX
    'x-queue-type':              'quorum',               // mandatory for production
  }
})

// ❌ nack + requeue: true — creates a busy-wait message storm
// Message returns to front of queue immediately, consumer re-processes it instantly
// CPU at 100%, queue depth stays constant, real work stops
channel.nack(msg, false, true)   // DO NOT DO THIS in a tight loop

// ❌ nack + requeue: true with no backoff — same problem
await sleep(100)
channel.nack(msg, false, true)   // still a loop, just slower

// ✅ nack + requeue: false → message routes to DLX → dead-letter queue
// Human or retry consumer processes DLQ separately, with controlled rate
channel.nack(msg, false, false)

// ✅ For transient errors: publish to a retry exchange with per-message TTL
// Message waits in the retry queue until TTL expires, then routes back to main queue
async function requeueWithDelay(
  channel: amqplib.Channel,
  msg:     amqplib.ConsumeMessage,
  delayMs: number
): Promise<void> {
  const retryCount = (msg.properties.headers?.['x-retry-count'] ?? 0) as number
  if (retryCount >= 3) {
    channel.nack(msg, false, false)   // max retries exceeded → DLQ
    return
  }
  await channel.assertQueue('orders.retry', {
    durable: true,
    arguments: {
      'x-dead-letter-exchange':    'orders',
      'x-dead-letter-routing-key': msg.fields.routingKey,
      'x-message-ttl':             delayMs,
      'x-queue-type':              'quorum',
    }
  })
  channel.publish('', 'orders.retry', msg.content, {
    headers: { 'x-retry-count': retryCount + 1 },
    persistent: true,
  })
  channel.ack(msg)   // ack the original — the retry copy is now in flight
}

# ❌ Classic mirrored queue policy — DEPRECATED in 3.12, REMOVED in 4.0
rabbitmqctl set_policy mirror-all "^" \
  '{"ha-mode":"all","ha-sync-mode":"automatic"}' \
  --apply-to queues
# ha-sync-mode=automatic: synchronisation blocks the entire queue during promotion
# Network partition + promotion = data loss

# ✅ Quorum Queue — Raft-based, no split-brain
rabbitmqctl set_policy quorum-critical "^order\." \
  '{"queue-mode":"lazy","x-queue-type":"quorum"}' \
  --apply-to queues

// ✅ Declaring a Quorum Queue — x-queue-type must be set at declaration time
// Cannot be changed after creation without deleting and recreating the queue
await channel.assertQueue('order.fulfillment', {
  durable: true,
  arguments: {
    'x-queue-type':                   'quorum',
    'x-dead-letter-exchange':         'orders.dlx',
    'x-dead-letter-routing-key':      'order.fulfillment',
    'x-quorum-initial-group-size':    3,    // Raft group size — must be odd
    'x-delivery-limit':               5,    // nack'd more than 5 times → DLX automatically
  }
})

// ✅ x-delivery-limit replaces manual retry counting in the consumer
await channel.assertQueue('order.fulfillment', {
  durable: true,
  arguments: {
    'x-queue-type':       'quorum',
    'x-delivery-limit':   5,            // after 5 nacks, broker dead-letters automatically
    'x-dead-letter-exchange': 'orders.dlx',
  }
})

// Consumer can simply nack on any failure — retry tracking is broker-side
await channel.consume('order.fulfillment', async (msg) => {
  if (!msg) return
  try {
    await processOrder(JSON.parse(msg.content.toString()))
    channel.ack(msg)
  } catch {
    channel.nack(msg, false, false)  // broker counts; auto-DLX at limit
  }
})