Producer & Consumer: Properti Pesan dan Siklus Hidupnya

┌─ Properties (baku AMQP) ────────────────────────┐
│ content_type   : application/json               │
│ delivery_mode  : 2  (persistent)                │
│ message_id     : 8f2c-4a1b-…                    │
│ correlation_id : untuk RPC / pelacakan          │
│ reply_to       : nama queue balasan             │
│ timestamp      : 1785312000                     │
│ expiration     : "60000"  (TTL per pesan, ms)   │
│ priority       : 0–9                            │
│ app_id         : layanan-pesanan                │
├─ Headers (bebas) ───────────────────────────────┤
│ x-trace-id     : 4bf92f3577b34da6               │
│ x-versi-skema  : 2                              │
│ x-percobaan    : 0                              │
├─ Body ──────────────────────────────────────────┤
│ {"pesanan_id":"P-1042","total":150000}          │
└─────────────────────────────────────────────────┘

const amqp = require('amqplib');
const { randomUUID } = require('crypto');

let koneksi = null;
let channel = null;

async function siapkan() {
  koneksi = await amqp.connect({
    protocol: 'amqp',
    hostname: process.env.RMQ_HOST || 'localhost',
    port: 5672,
    username: process.env.RMQ_USER,
    password: process.env.RMQ_PASS,
    vhost: '/',
    heartbeat: 60,                       // deteksi koneksi mati
    clientProperties: {
      connection_name: 'layanan-pesanan-producer'   // terlihat di UI
    }
  });

  koneksi.on('error', (e) => console.error('koneksi error:', e.message));
  koneksi.on('close', () => { channel = null; jadwalkanSambungUlang(); });

  // ConfirmChannel: broker mengonfirmasi tiap publish
  channel = await koneksi.createConfirmChannel();

  await channel.assertExchange('pesanan', 'topic', {
    durable: true,
    arguments: { 'alternate-exchange': 'pesanan.tak-terkirim' }
  });

  // pesan yang tak bisa dirutekan dikembalikan ke sini
  channel.on('return', (msg) => {
    console.error('TIDAK TERKIRIM:', msg.fields.routingKey,
                  msg.content.toString());
  });
}

async function kirimPeristiwa(peristiwa, wilayah, data) {
  const routingKey = 'pesanan.' + wilayah + '.' + peristiwa;
  const body = Buffer.from(JSON.stringify(data));

  return new Promise((resolve, reject) => {
    channel.publish('pesanan', routingKey, body, {
      persistent: true,                  // = delivery_mode 2
      mandatory: true,                   // kembalikan bila tak terkirim
      contentType: 'application/json',
      contentEncoding: 'utf-8',
      messageId: randomUUID(),           // dasar idempotensi konsumer
      timestamp: Math.floor(Date.now() / 1000),
      appId: 'layanan-pesanan',
      headers: {
        'x-versi-skema': 2,
        'x-trace-id': ambilTraceId()
      }
    }, (err) => err ? reject(err) : resolve());   // publisher confirm
  });
}

async function mulaiKonsumsi() {
  const channel = await koneksi.createChannel();

  await channel.assertQueue('pesanan.email', {
    durable: true,
    arguments: {
      'x-queue-type': 'quorum',
      'x-dead-letter-exchange': 'pesanan.dlx',
      'x-delivery-limit': 5
    }
  });
  await channel.bindQueue('pesanan.email', 'pesanan', 'pesanan.*.dibuat');

  // WAJIB: batasi berapa pesan boleh "in-flight" tanpa ack
  await channel.prefetch(20);

  await channel.consume('pesanan.email', async (msg) => {
    if (msg === null) return;            // konsumer dibatalkan broker

    try {
      const data = JSON.parse(msg.content.toString());
      await prosesKirimEmail(data, msg.properties.messageId);
      channel.ack(msg);                  // baru ack SETELAH berhasil
    } catch (e) {
      const bisaDiulang = e.sementara === true;
      // requeue=false → jatuh ke DLX, bukan berputar tanpa henti
      channel.nack(msg, false, bisaDiulang);
      console.error('gagal memproses:', e.message);
    }
  }, {
    noAck: false,                        // ack manual — WAJIB
    consumerTag: 'email-worker-' + process.pid
  });
}