// services/order/src/events/publisher.ts
import { Kafka, Partitioners } from 'kafkajs'
 
const kafka = new Kafka({
  clientId: 'order-service',
  brokers: [process.env.KAFKA_BROKER!],
  ssl: true,
  sasl: {
    mechanism: 'scram-sha-256',
    username: process.env.KAFKA_USERNAME!,
    password: process.env.KAFKA_PASSWORD!,
  },
})
 
const producer = kafka.producer({
  createPartitioner: Partitioners.LegacyPartitioner,
  idempotent: true,                     // Exactly-once delivery
  maxInFlightRequests: 5,
  retry: { retries: 5 },
})
 
export async function publishOrderCreated(order: Order) {
  await producer.send({
    topic: 'orders.created',
    messages: [{
      key: order.id,                     // Partition by order ID
      value: JSON.stringify({
        eventId: crypto.randomUUID(),
        eventType: 'ORDER_CREATED',
        timestamp: new Date().toISOString(),
        data: {
          orderId: order.id,
          userId: order.userId,
          items: order.items,
          totalAmount: order.totalAmount,
        },
        metadata: {
          source: 'order-service',
          version: '1.0',
          correlationId: order.correlationId,
        },
      }),
      headers: {
        'event-type': 'ORDER_CREATED',
        'content-type': 'application/json',
      },
    }],
  })
}

// services/payment/src/events/consumer.ts
const consumer = kafka.consumer({
  groupId: 'payment-service',
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
})
 
await consumer.subscribe({
  topics: ['orders.created'],
  fromBeginning: false,
})
 
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value!.toString())
 
    try {
      // Idempotency check
      const processed = await redis.get(`event:${event.eventId}`)
      if (processed) {
        logger.info(`Event ${event.eventId} already processed, skipping`)
        return
      }
 
      // Process payment
      const payment = await processPayment({
        orderId: event.data.orderId,
        userId: event.data.userId,
        amount: event.data.totalAmount,
      })
 
      // Mark as processed
      await redis.set(`event:${event.eventId}`, 'processed', 'EX', 86400)
 
      // Publish payment result
      await publishPaymentProcessed(payment)
 
      logger.info(`Payment processed for order ${event.data.orderId}`)
 
    } catch (error) {
      logger.error(`Failed to process payment`, {
        eventId: event.eventId,
        error: error.message,
      })
 
      // Dead letter queue for failed messages
      await publishToDeadLetterQueue(topic, message, error)
    }
  },
})