Timeline:
T0: Order service updates status to "shipped"
T1: OrderShipped event published to broker (50ms later)
T2: Notification service processes event (200ms later)
T3: Analytics service processes event (500ms later)
T4: Customer dashboard updated (800ms later)

Between T0 and T4:
- Order service says: "shipped"
- Analytics service says: "processing" (hasn't gotten the event yet)
- Customer sees: "processing" on dashboard (hasn't been updated yet)

// What you expect:
// 1. OrderPlaced (orderId: 123)
// 2. OrderPaid (orderId: 123)
// 3. OrderShipped (orderId: 123)

// What might actually arrive:
// 1. OrderPlaced (orderId: 123)
// 2. OrderShipped (orderId: 123)   <-- arrived before OrderPaid!
// 3. OrderPaid (orderId: 123)

interface VersionedEvent {
  type: string;
  entityId: string;
  version: number;    // Monotonically increasing per entity
  data: unknown;
}

async function handleEvent(event: VersionedEvent) {
  const current = await db.entities.findOne({ id: event.entityId });

  // Only process if this is the next expected version
  if (current && event.version <= current.lastProcessedVersion) {
    console.log(`Skipping event v${event.version}, already at v${current.lastProcessedVersion}`);
    return; // Already processed or out of order
  }

  if (current && event.version > current.lastProcessedVersion + 1) {
    // Gap detected: we missed an event. Buffer this one and wait.
    await db.bufferedEvents.insert({
      entityId: event.entityId,
      version: event.version,
      payload: event
    });
    console.log(`Buffering event v${event.version}, waiting for v${current.lastProcessedVersion + 1}`);
    return;
  }

  // Process the event
  await processEvent(event);
  await db.entities.update(event.entityId, { lastProcessedVersion: event.version });

  // Check if any buffered events can now be processed
  await processBufferedEvents(event.entityId, event.version + 1);
}

async function handleProfileUpdate(event: { userId: string; updatedAt: string; name: string }) {
  const current = await db.profiles.findOne({ userId: event.userId });

  // Only update if this event is newer than what we have
  if (current && new Date(event.updatedAt) <= new Date(current.updatedAt)) {
    console.log('Ignoring older event');
    return;
  }

  await db.profiles.upsert({
    userId: event.userId,
    name: event.name,
    updatedAt: event.updatedAt
  });
}

// Kafka: use orderId as partition key
await producer.send({
  topic: 'order-events',
  messages: [{
    key: order.id,  // All events for this order go to the same partition
    value: JSON.stringify(event)
  }]
});
// Partition guarantees: OrderPlaced -> OrderPaid -> OrderShipped (in order for this orderId)

// NOT idempotent: processes payment every time the event is received
async function handleOrderPlaced(event: OrderPlacedEvent) {
  await paymentService.charge(event.data.customerId, event.data.total);
  // If this event is delivered twice, the customer is charged twice!
}

// Idempotent: checks before processing
async function handleOrderPlaced(event: OrderPlacedEvent) {
  // Check if we already processed this event
  const alreadyProcessed = await db.processedEvents.findOne({ eventId: event.id });
  if (alreadyProcessed) {
    console.log(`Event ${event.id} already processed, skipping`);
    return;
  }

  // Process within a transaction
  await db.transaction(async (trx) => {
    // Record that we processed this event
    await trx('processed_events').insert({
      event_id: event.id,
      processed_at: new Date()
    });

    // Idempotent payment: use orderId as idempotency key
    await paymentService.charge(event.data.customerId, event.data.total, {
      idempotencyKey: event.data.orderId
    });
  });
}

// Forward action: charge payment
async function chargePayment(orderId: string, amount: number) {
  const payment = await stripe.charges.create({ amount, currency: 'usd' });
  return payment.id;
}

// Compensating action: refund payment
async function refundPayment(paymentId: string) {
  await stripe.refunds.create({ charge: paymentId });
}

// Forward action: reserve inventory
async function reserveInventory(orderId: string, items: Item[]) {
  for (const item of items) {
    await db('inventory')
      .where('product_id', item.productId)
      .decrement('available', item.quantity);
  }
}

// Compensating action: release inventory
async function releaseInventory(orderId: string, items: Item[]) {
  for (const item of items) {
    await db('inventory')
      .where('product_id', item.productId)
      .increment('available', item.quantity);
  }
}