// ✅ One connection per process; one channel per consumer thread
const connection = await amqplib.connect('amqp://rabbitmq:5672')
const channel    = await connection.createChannel()

await channel.assertExchange('orders', 'topic', { durable: true })
await channel.assertQueue('order.fulfillment', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': 'orders.dlx',
    'x-message-ttl': 86_400_000,
    'x-queue-type': 'quorum',
  }
})
await channel.bindQueue('order.fulfillment', 'orders', 'order.placed')

// ✅ Topic exchange routing patterns
channel.publish('orders', 'order.placed.uk',    Buffer.from(...)) // UK placement
channel.publish('orders', 'order.placed.us',    Buffer.from(...)) // US placement
channel.publish('orders', 'order.cancelled.uk', Buffer.from(...)) // UK cancellation

// Bindings:
await channel.bindQueue('fulfillment-all',   'orders', 'order.#')        // all order events
await channel.bindQueue('fulfillment-placed','orders', 'order.placed.*') // placed only
await channel.bindQueue('fulfillment-uk',    'orders', '*.*.uk')          // UK only

// ✅ Fanout exchange — every bound service receives the message simultaneously
await channel.assertExchange('cache.invalidations', 'fanout', { durable: true })

// Each service creates its own exclusive queue and binds to the fanout
const { queue } = await channel.assertQueue('', { exclusive: true, autoDelete: true })
await channel.bindQueue(queue, 'cache.invalidations', '') // routing key ignored for fanout

channel.consume(queue, (msg) => {
  if (!msg) return
  const { entityType, entityId } = JSON.parse(msg.content.toString())
  localCache.invalidate(entityType, entityId)
  channel.ack(msg)
})

// ❌ Common mistake — queue is durable but message is not
await channel.assertQueue('orders', { durable: true })
channel.sendToQueue('orders', Buffer.from(JSON.stringify(order)), { persistent: false })
// → message LOST on broker restart even though queue survives

// ✅ All three durable
await channel.assertExchange('orders', 'topic', { durable: true })   // (1) durable exchange
await channel.assertQueue('order.fulfillment', { durable: true })    // (2) durable queue
channel.publish('orders', 'order.placed',
  Buffer.from(JSON.stringify(order)),
  { persistent: true }  // (3) persistent message (deliveryMode: 2)
)

// ✅ Quorum queue declaration
await channel.assertQueue('order.fulfillment', {
  durable: true,
  arguments: {
    'x-queue-type': 'quorum',
    'x-delivery-limit': 3,
    'x-dead-letter-exchange': 'orders.dlx',
  }
})

// ❌ Auto-ACK — message deleted from queue on delivery, BEFORE your code processes it
channel.consume('order.fulfillment', async (msg) => {
  if (!msg) return
  await fulfillmentService.ship(JSON.parse(msg.content.toString()))
  // If ship() throws: message PERMANENTLY LOST — already ACK'd on delivery
}, { noAck: true })

// ✅ Manual ACK — message stays in queue until explicitly acknowledged
channel.consume('order.fulfillment', async (msg) => {
  if (!msg) return
  try {
    await fulfillmentService.ship(JSON.parse(msg.content.toString()))
    channel.ack(msg)
  } catch (err) {
    const requeue = isTransientError(err)
    channel.nack(msg, false, requeue)
    // requeue=true  → returns to queue head for redelivery
    // requeue=false → routes to DLX if configured, or discards
  }
}, { noAck: false })

// ✅ Limit unacknowledged messages per consumer — prevents one slow consumer hoarding all work
await channel.prefetch(10)
// Consumer holds at most 10 unacked messages; messages 11+ remain in queue for other consumers

// ✅ Step 1: Declare the DLX and its dead-letter queue
await channel.assertExchange('orders.dlx', 'topic', { durable: true })
await channel.assertQueue('orders.dead-letter', {
  durable: true,
  arguments: { 'x-queue-type': 'quorum' }
})
await channel.bindQueue('orders.dead-letter', 'orders.dlx', '#')

// ✅ Step 2: Main queue points at DLX for failures
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.failed',
    'x-message-ttl': 300_000,   // expire unprocessed after 5min
    'x-delivery-limit': 3,      // DLX after 3 failed attempts
  }
})

// ✅ Publisher confirms — broker guarantees message written to disk before ack
await channel.confirmSelect()

channel.publish('orders', 'order.placed',
  Buffer.from(JSON.stringify(order)),
  { persistent: true }
)
await channel.waitForConfirms()
// After this line: message guaranteed to survive broker restart and leader failover