// ❌ At-most-once on Kafka — auto-commit before processing
const consumer = kafka.consumer({ groupId: 'analytics' })
await consumer.run({
  // enable.auto.commit = true (kafkajs default): offset committed on interval
  // regardless of whether eachMessage handler has finished
  eachMessage: async ({ message }) => {
    // If process crashes here — after auto-commit, before insert — message is lost
    await analyticsDb.insert(JSON.parse(message.value!.toString()))
  }
})

// ❌ At-most-once on RabbitMQ — autoAck: true
await channel.consume('analytics', (msg) => {
  if (!msg) return
  // Message acked by broker the moment it is delivered — before this line runs
  analyticsDb.insert(JSON.parse(msg.content.toString()))
  // If process crashes mid-insert: message already acked, permanently gone
}, { noAck: true })

// ✅ At-least-once on Kafka — manual offset commit AFTER processing
const consumer = kafka.consumer({ groupId: 'fulfillment' })
await consumer.run({
  autoCommit: false,   // disable auto-commit
  eachMessage: async ({ topic, partition, message }) => {
    // 1. Process first
    await fulfillmentDb.insert(JSON.parse(message.value!.toString()))
    // 2. Commit offset AFTER successful processing
    await consumer.commitOffsets([{
      topic,
      partition,
      offset: (BigInt(message.offset) + 1n).toString(),
    }])
    // Crash window: if process dies after insert but before commitOffsets,
    // consumer restarts at previous offset → re-processes the same message
    // → duplicate insert → idempotency required
  }
})

// ✅ At-least-once on RabbitMQ — manual ack AFTER processing
await channel.prefetch(10)
await channel.consume('fulfillment', async (msg) => {
  if (!msg) return
  await fulfillmentDb.insert(JSON.parse(msg.content.toString()))
  channel.ack(msg)   // ack AFTER insert — crash before ack = redelivery
})

// ✅ Idempotency gate — PostgreSQL ON CONFLICT DO NOTHING
async function processPaymentEvent(event: PaymentEvent): Promise<void> {
  await db.query(`
    INSERT INTO payment_processed (idempotency_key, payment_id, amount, processed_at)
    VALUES ($1, $2, $3, NOW())
    ON CONFLICT (idempotency_key) DO NOTHING
  `, [event.idempotencyKey, event.paymentId, event.amount])
  // If this row already exists: no error, no duplicate charge, no side effect
  // Second delivery of the same event is silently discarded
}

// ✅ Redis-based idempotency gate — for consumers without DB write
async function processAnalyticsEvent(event: AnalyticsEvent): Promise<void> {
  const key = `idem:analytics:${event.idempotencyKey}`
  const isNew = await redis.set(key, '1', 'NX', 'EX', 86400) // 24h TTL
  if (!isNew) return   // duplicate — skip processing
  await analyticsQueue.add(event)
}

// ✅ Kafka transactional producer — exactly-once write + offset commit
const producer = kafka.producer({
  transactionalId: 'payment-processor-1',   // stable across restarts — must be unique per instance
  idempotent: true,                          // required for transactions
  maxInFlightRequests: 1,                    // transactions require sequential inflight
})
await producer.connect()
await producer.transaction(async (tx) => {
  // 1. Produce the output event
  await tx.send({
    topic: 'fulfillment.requests',
    messages: [{ key: event.paymentId, value: JSON.stringify(event) }]
  })
  // 2. Commit the source consumer's offset atomically with the produce
  await tx.sendOffsets({
    consumerGroupId: 'payment-processor',
    topics: [{ topic: 'payments.charged', partitions: [{ partition, offset }] }]
  })
  // Both succeed or both roll back — never a state where offset is committed
  // without the output event being written, or vice versa
})

// ❌ Consumer generates idempotency key — not idempotent across retries
// A new UUID is generated on every message delivery (including redeliveries)
async function processOrder(msg: Message): Promise<void> {
  const idempotencyKey = uuid()   // different value on every invocation
  await db.insert({ idempotencyKey, orderId: msg.orderId })
  // Re-delivery → new UUID → INSERT succeeds → duplicate row
}

// ✅ Idempotency key from the event itself — stable across redeliveries
interface OrderEvent {
  eventId:         string   // UUID generated at the point of order creation, never changes
  orderId:         string
  customerId:      string
  totalCents:      number
}

async function processOrder(event: OrderEvent): Promise<void> {
  await db.query(`
    INSERT INTO orders_processed (idempotency_key, order_id, customer_id, total_cents)
    VALUES ($1, $2, $3, $4)
    ON CONFLICT (idempotency_key) DO NOTHING
  `, [event.eventId, event.orderId, event.customerId, event.totalCents])
}

// ✅ Database idempotency table — permanent, no TTL needed for financial events
CREATE TABLE payment_idempotency (
  idempotency_key  VARCHAR(128) PRIMARY KEY,
  payment_id       UUID NOT NULL,
  processed_at     TIMESTAMPTZ DEFAULT NOW()
);
-- Keep indefinitely — financial audit trail requires it

// ✅ Redis idempotency — appropriate for events with a natural time window
const TTL_SECONDS = 7 * 24 * 3600   // 7 days — longer than max broker retention
await redis.set(
  `idem:order:${event.eventId}`,
  event.orderId,
  'NX',    // set only if not exists
  'EX',    // with expiry
  TTL_SECONDS
)
// TTL must exceed max possible redelivery window (broker retention + consumer lag)