// Producer
interface OrderCreatedEvent {
  eventId: string;
  eventType: 'order.created';
  occurredAt: string; // ISO-8601
  payload: {
    orderId: string;
    customerId: string;
    totalAmount: number;
    currency: string;
  };
}

const event: OrderCreatedEvent = {
  eventId: crypto.randomUUID(),
  eventType: 'order.created',
  occurredAt: new Date().toISOString(),
  payload: {
    orderId: order.id,
    customerId: order.customerId,
    totalAmount: order.total,
    currency: 'USD',
  },
};

await producer.send({
  topic: 'orders',
  messages: [{ 
    key: order.id,  // Ensures ordering per order
    value: JSON.stringify(event),
  }],
});

// Bad: Consumer must call back to get order details
interface OrderCreatedEvent {
  orderId: string; // Consumer needs to GET /orders/{id}
}

// Good: Include all needed state
interface OrderCreatedEvent {
  orderId: string;
  customerId: string;
  customerEmail: string;
  items: Array<{ sku: string; quantity: number; price: number }>;
  totalAmount: number;
  shippingAddress: Address;
}

// WRONG — not atomic
await db.orders.create(order);
await kafka.produce('orders', orderCreatedEvent); // What if this fails?

// CORRECT — atomic via outbox
await db.transaction(async (tx) => {
  await tx.orders.create(order);
  
  // Write to outbox table in same transaction
  await tx.outbox.create({
    id: crypto.randomUUID(),
    topic: 'orders',
    key: order.id,
    payload: JSON.stringify(orderCreatedEvent),
    createdAt: new Date(),
    publishedAt: null,
  });
});

// Separate process reads outbox and publishes
// (Debezium CDC is the gold standard for this)

const consumer = kafka.consumer({ 
  groupId: 'payment-service',  // All instances share work
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
});

await consumer.subscribe({ 
  topics: ['orders'],
  fromBeginning: false,
});

await consumer.run({
  partitionsConsumedConcurrently: 3,
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value!.toString());
    
    try {
      await processOrderPayment(event);
      // Kafka auto-commits offset after successful processing
    } catch (error) {
      if (isRetryable(error)) {
        // Don't commit — message will be redelivered
        throw error;
      }
      // Non-retryable: send to DLQ
      await sendToDeadLetterQueue(topic, message, error);
    }
  },
});

async function sendToDeadLetterQueue(
  originalTopic: string,
  message: KafkaMessage,
  error: Error,
): Promise<void> {
  await producer.send({
    topic: `${originalTopic}.dlq`,
    messages: [{
      key: message.key,
      value: message.value,
      headers: {
        'dlq-original-topic': originalTopic,
        'dlq-error-message': error.message,
        'dlq-error-stack': error.stack ?? '',
        'dlq-failed-at': new Date().toISOString(),
        'dlq-consumer-group': 'payment-service',
      },
    }],
  });
}

// Producer-side: Enable idempotent producer
const producer = kafka.producer({
  idempotent: true,         // Prevents duplicate sends
  transactionalId: 'payment-service-txn',
  maxInFlightRequests: 5,
});

// Transactional produce + consume
await producer.transaction(async (tx) => {
  // Read from input topic
  const messages = await consumeMessages();
  
  // Process and produce to output topic
  await tx.send({
    topic: 'payment-results',
    messages: messages.map(processMessage),
  });
  
  // Commit offsets atomically
  await tx.sendOffsets({
    consumerGroupId: 'payment-processor',
    topics: [{ topic: 'orders', partitions: [...] }],
  });
});

// Integration test with embedded Kafka (Testcontainers)
import { KafkaContainer } from '@testcontainers/kafka';

describe('Order processing', () => {
  let kafka: StartedKafkaContainer;
  
  beforeAll(async () => {
    kafka = await new KafkaContainer('confluentinc/cp-kafka:7.5.0')
      .withExposedPorts(9093)
      .start();
  });
  
  it('publishes OrderCreated event when order is saved', async () => {
    const brokers = [`${kafka.getHost()}:${kafka.getMappedPort(9093)}`];
    const consumer = new Kafka({ brokers }).consumer({ groupId: 'test' });
    
    await consumer.subscribe({ topics: ['orders'] });
    
    const receivedMessages: KafkaMessage[] = [];
    await consumer.run({
      eachMessage: async ({ message }) => {
        receivedMessages.push(message);
      },
    });
    
    // Trigger the action
    await orderService.createOrder({ customerId: 'cust_123', items: [...] });
    
    // Wait for message propagation
    await waitFor(() => receivedMessages.length > 0, { timeout: 5000 });
    
    const event = JSON.parse(receivedMessages[0].value!.toString());
    expect(event.eventType).toBe('order.created');
    expect(event.payload.customerId).toBe('cust_123');
  });
});