// ✅ Typed saga event interface — every event carries causal chain headers
interface SagaEvent {
  eventId:       string   // unique event ID (idempotency key for the consumer)
  correlationId: string   // original orderId — stable for the entire saga lifetime
  causationId:   string   // eventId of the event that caused THIS event
  orderId:       string
  timestamp:     number
}

interface OrderCreated extends SagaEvent {
  type:        'order.created'
  customerId:  string
  amountCents: number
  items:       Array<{ sku: string; quantity: number }>
}

interface InventoryReserved extends SagaEvent {
  type:              'inventory.reserved'
  reservationId:     string
  items:             Array<{ sku: string; quantity: number }>
}

interface InventoryReservationFailed extends SagaEvent {
  type:   'inventory.reservation.failed'
  reason: 'out_of_stock' | 'sku_not_found'
  skus:   string[]
}

// Compensation events — must be typed, not generic
interface PaymentRefunded extends SagaEvent {
  type:        'payment.refunded'
  refundId:    string
  amountCents: number
  reason:      string
}

Before pivot — compensatable:
  PaymentAuthorized    → compensate: PaymentVoided (card not yet captured)
  InventoryReserved    → compensate: InventoryReleased (un-reserve SKUs)

Pivot:
  PaymentCaptured      → card is charged; compensation = refund (different operation)

After pivot — must retry until success:
  OrderShipped         → package is in transit; cannot un-ship
  TrackingNotified     → email sent; cannot un-send

// ✅ Saga compensation: Inventory service listens for InventoryReservationFailed
// to release reservations it holds when the saga is compensating

const consumer = kafka.consumer({ groupId: 'inventory-service-compensation' })
await consumer.subscribe({ topics: ['payment.voided', 'order.cancelled'] })

await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString()) as SagaEvent

    // Idempotency gate: compensation events processed twice must be safe
    const result = await db.query(`
      INSERT INTO compensation_log (idempotency_key, order_id, action, processed_at)
      VALUES ($1, $2, 'inventory.released', NOW())
      ON CONFLICT (idempotency_key) DO NOTHING
      RETURNING id
    `, [`${event.correlationId}:inventory.released`, event.orderId])

    if (result.rows.length === 0) return   // already compensated — idempotent

    // Perform compensation — append a release record, never delete the reservation
    await db.query(`
      INSERT INTO inventory_releases (reservation_id, order_id, released_at, reason)
      SELECT id, $1, NOW(), $2
      FROM inventory_reservations
      WHERE order_id = $1 AND status = 'reserved'
    `, [event.orderId, 'saga_compensation'])

    await db.query(`
      UPDATE inventory_reservations SET status = 'released' WHERE order_id = $1
    `, [event.orderId])

    await producer.send({
      topic:    'inventory.released',
      messages: [{
        key:   event.orderId,
        value: JSON.stringify({
          eventId:       crypto.randomUUID(),
          correlationId: event.correlationId,
          causationId:   event.eventId,
          orderId:       event.orderId,
          type:          'inventory.released',
          timestamp:     Date.now(),
        } satisfies InventoryReleased)
      }]
    })
  }
})

// ✅ Semantic lock: mark the reservation as 'pending' during the saga
// Prevents concurrent orders from reserving the same SKU while this saga is in-flight
await db.query(`
  INSERT INTO inventory_reservations (id, order_id, sku, quantity, status)
  VALUES (uuid_generate_v4(), $1, $2, $3, 'pending')
  -- status = 'pending' = semantic lock: this inventory is in-flight
  -- Other saga steps that check availability must exclude 'pending' rows
`, [orderId, sku, quantity])

// Only after InventoryReserved event is published: update to 'reserved'
// On compensation (InventoryReleased): update to 'released'
// Stock availability query: WHERE status NOT IN ('pending', 'reserved')

// When choreography becomes too complex: Temporal.io workflow
// Provides orchestration with durable execution, explicit compensation, and full audit trail
import { proxyActivities, sleep } from '@temporalio/workflow'

export async function orderWorkflow(orderId: string): Promise<void> {
  const { chargePayment, reserveInventory, createShipment, sendNotification,
          refundPayment, releaseInventory } = proxyActivities({ startToCloseTimeout: '30s' })

  let paymentCharged = false
  let inventoryReserved = false

  try {
    await chargePayment(orderId)
    paymentCharged = true

    await reserveInventory(orderId)
    inventoryReserved = true

    await createShipment(orderId)  // pivot — past this point, retry not compensate
    await sendNotification(orderId)
  } catch (err) {
    // Compensate in reverse order — only undo what was done
    if (inventoryReserved) await releaseInventory(orderId)
    if (paymentCharged)    await refundPayment(orderId)
    throw err
  }
}
// Temporal: workflow state is durable — process crash = resume from last checkpoint
// Compensation is explicit and sequential — no event routing needed