// ❌ Traditional: store current state — history is discarded
await db.query(`
  UPDATE orders SET status = 'shipped', shipped_at = NOW() WHERE id = $1
`, [orderId])
// The previous 'confirmed' status is gone. What changed it? When? Who?

// ✅ Event Sourcing: append domain events — current state is derived
await eventStore.append('order', orderId, [
  {
    type: 'OrderShipped',
    orderId,
    trackingNumber,
    carrier,
    shippedAt: new Date().toISOString(),
    version: currentVersion + 1,
  }
])
// Current state = replay of all events for this order
// History: OrderPlaced(v1) → PaymentCharged(v2) → OrderConfirmed(v3) → OrderShipped(v4)
// Any version reconstructable at any time

// ✅ Event store backed by Kafka — one topic per aggregate type, key = aggregateId
const ORDER_EVENTS_TOPIC = 'order.event-store'

class KafkaEventStore {
  async append(aggregateId: string, events: DomainEvent[]): Promise<void> {
    await this.producer.send({
      topic: ORDER_EVENTS_TOPIC,
      messages: events.map(event => ({
        key: aggregateId,           // same aggregateId always routes to same partition
        value: JSON.stringify(event),
        headers: {
          'event-type': event.type,
          'aggregate-version': event.version.toString(),
          'occurred-at': event.occurredAt,
        }
      }))
    })
  }

  async loadEvents(aggregateId: string, fromVersion = 0): Promise<DomainEvent[]> {
    // Subscribe to the partition owning this aggregateId
    // Replay all events for this key from the beginning of the partition
    const consumer = this.kafka.consumer({ groupId: `load-${aggregateId}-${Date.now()}` })
    await consumer.subscribe({ topic: ORDER_EVENTS_TOPIC, fromBeginning: true })

    const events: DomainEvent[] = []
    await consumer.run({
      eachMessage: async ({ message }) => {
        if (message.key?.toString() === aggregateId) {
          const event = JSON.parse(message.value!.toString()) as DomainEvent
          if (event.version >= fromVersion) events.push(event)
        }
      }
    })
    return events.sort((a, b) => a.version - b.version)
  }
}

// ✅ Reconstruct current state by replaying all events for an aggregate
class OrderAggregate {
  static reconstruct(events: DomainEvent[]): OrderAggregate {
    const order = new OrderAggregate()
    for (const event of events) {
      order.apply(event) // each event mutates in-memory state
    }
    return order
  }

  private apply(event: DomainEvent): void {
    switch (event.type) {
      case 'OrderPlaced':
        this.orderId    = event.orderId
        this.userId     = event.userId
        this.lineItems  = event.lineItems
        this.totalCents = event.totalCents
        this.status     = 'pending'
        this.version    = event.version
        break
      case 'PaymentCharged':
        this.chargeId = event.chargeId
        this.status   = 'payment_confirmed'
        this.version  = event.version
        break
      case 'OrderShipped':
        this.trackingNumber = event.trackingNumber
        this.status         = 'shipped'
        this.version        = event.version
        break
      case 'OrderCancelled':
        this.status = 'cancelled'
        this.version = event.version
        break
    }
  }
}

// ✅ Snapshot-based reconstruction — O(snapshot_gap) instead of O(all events)
class SnapshotStore {
  async saveSnapshot(aggregateId: string, state: OrderAggregate): Promise<void> {
    await db.query(`
      INSERT INTO order_snapshots (aggregate_id, version, state, created_at)
      VALUES ($1, $2, $3, NOW())
      ON CONFLICT (aggregate_id)
      DO UPDATE SET version = $2, state = $3, created_at = NOW()
    `, [aggregateId, state.version, JSON.stringify(state.toJSON())])
  }

  async loadSnapshot(aggregateId: string): Promise<{ state: OrderAggregate; version: number } | null> {
    const row = await db.query(
      `SELECT state, version FROM order_snapshots WHERE aggregate_id = $1`,
      [aggregateId]
    )
    if (!row.rows[0]) return null
    return {
      state: OrderAggregate.fromJSON(row.rows[0].state),
      version: row.rows[0].version
    }
  }
}

// ✅ Reconstruction with snapshot shortcut
async function loadOrder(aggregateId: string): Promise<OrderAggregate> {
  const snapshot = await snapshotStore.loadSnapshot(aggregateId)
  const fromVersion = snapshot?.version ?? 0

  const recentEvents = await eventStore.loadEvents(aggregateId, fromVersion + 1)

  if (snapshot && recentEvents.length === 0) {
    return snapshot.state // fully up-to-date snapshot — no replay needed
  }

  const base = snapshot?.state ?? new OrderAggregate()
  for (const event of recentEvents) {
    base.apply(event)
  }

  // Refresh snapshot every 50 events to keep replay gap bounded
  if (recentEvents.length >= 50) {
    await snapshotStore.saveSnapshot(aggregateId, base)
  }

  return base
}

{
  "type": "record",
  "name": "OrderPlaced",
  "namespace": "com.company.orders.v1",
  "fields": [
    { "name": "orderId",     "type": "string" },
    { "name": "userId",      "type": "string" },
    { "name": "totalCents",  "type": "long" },
    { "name": "placedAt",    "type": "string" },
    { "name": "currency",    "type": "string", "default": "USD" }
  ]
}

// ✅ Backward-compatible evolution — add optional fields with defaults
// Old consumers (compiled against v1) ignore new fields → no breaking change
const OrderPlacedV2 = {
  type: 'record',
  name: 'OrderPlaced',
  namespace: 'com.company.orders.v1',
  fields: [
    { name: 'orderId',    type: 'string' },
    { name: 'userId',     type: 'string' },
    { name: 'totalCents', type: 'long' },
    { name: 'placedAt',   type: 'string' },
    { name: 'currency',   type: 'string', default: 'USD' },
    // NEW: optional field with default — backward compatible ✅
    { name: 'promoCode',  type: ['null', 'string'], default: null },
  ]
}

// ❌ Breaking changes — never do these in the same schema version
// - Removing an existing field (old events no longer deserialize)
// - Renaming a field (field number/name change)
// - Changing a field type (int → string)

// ✅ Schema Registry enforces compatibility at publish time
const schemaRegistry = new SchemaRegistry({ host: 'http://schema-registry:8081' })

// Producer: schema validated and registered before first publish
const { id } = await schemaRegistry.register({
  type: SchemaType.AVRO,
  schema: JSON.stringify(OrderPlacedV2),
}, { subject: 'order.event-store-value' })

await producer.send({
  topic: 'order.event-store',
  messages: [{ key: orderId, value: await schemaRegistry.encode(id, event) }]
})

// Consumer: schema fetched from registry and used to decode
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = await schemaRegistry.decode(message.value!)
    // event is correctly typed and decoded regardless of which schema version was used at publish
  }
})

// ❌ Never modify or delete events — they are the source of truth
await kafkaAdmin.deleteRecords([{
  topic: 'order.event-store',
  partitions: [{ partition: 0, offset: '42' }] // ← NEVER do this in production
}])
// This corrupts the aggregate history and breaks all projections that read past that offset

// ✅ Correct: append a compensating event
await eventStore.append(orderId, [{
  type: 'DataCorrectionApplied',
  orderId,
  correctedField: 'totalCents',
  oldValue: 4999,
  newValue: 4499,
  reason: 'Pricing bug on 2026-09-01',
  correctedAt: new Date().toISOString(),
  version: currentVersion + 1,
}])

// ❌ Anemic event — missing context, forces consumer to re-query state
{ type: 'OrderUpdated', orderId: 'abc123' } // what was updated? what did it change to?

// ✅ Self-contained event — all required context embedded
{
  type: 'OrderShipped',
  orderId: 'abc123',
  trackingNumber: 'UPS-123456',
  carrier: 'UPS',
  estimatedDelivery: '2026-09-08',
  shippedFrom: 'warehouse-us-east-1',
  shippedAt: '2026-09-05T14:30:00Z',
  version: 4
}