// ❌ Shared model — optimized for neither reads nor writes
// Write side: needs strong consistency, invariant enforcement, row-level locking
// Read side: needs denormalized, pre-joined data for the dashboard query

// This single query powers a dashboard that joins 6 tables:
const dashboardData = await db.query(`
  SELECT
    u.name, u.email,
    COUNT(o.id)                           AS order_count,
    SUM(o.total_cents)                    AS lifetime_value_cents,
    MAX(o.created_at)                     AS last_order_at,
    json_agg(DISTINCT p.category)         AS categories_purchased,
    AVG(r.rating)                         AS avg_review_rating,
    COUNT(DISTINCT r.product_id)          AS products_reviewed
  FROM users u
  LEFT JOIN orders o ON o.user_id = u.id
  LEFT JOIN order_items oi ON oi.order_id = o.id
  LEFT JOIN products p ON p.id = oi.product_id
  LEFT JOIN reviews r ON r.user_id = u.id
  WHERE u.id = $1
  GROUP BY u.id
`, [userId])
// This query runs on every page load — same database, same connection pool as your write path
// High read traffic → contention on write tables → degraded write P99

// ✅ Write side — Order aggregate enforces business invariants
class OrderAggregate {
  private status: 'pending' | 'confirmed' | 'cancelled' | 'shipped'
  private readonly orderId: string
  private readonly userId: string
  private readonly lineItems: OrderLineItem[]
  private version: number  // used for optimistic concurrency

  static create(cmd: PlaceOrderCommand): { aggregate: OrderAggregate; event: OrderPlacedEvent } {
    if (cmd.lineItems.length === 0) throw new DomainError('Order must contain at least one item')
    if (cmd.lineItems.some(item => item.quantity <= 0)) throw new DomainError('Quantity must be positive')
    if (cmd.totalCents <= 0) throw new DomainError('Order total must be positive')

    const aggregate = new OrderAggregate(cmd.orderId, cmd.userId, cmd.lineItems)
    const event: OrderPlacedEvent = {
      type: 'OrderPlaced',
      orderId: cmd.orderId,
      userId: cmd.userId,
      lineItems: cmd.lineItems,
      totalCents: cmd.totalCents,
      placedAt: new Date().toISOString(),
      version: 1,
    }
    return { aggregate, event }
  }

  cancel(cmd: CancelOrderCommand): OrderCancelledEvent {
    if (this.status === 'shipped') throw new DomainError('Cannot cancel a shipped order')
    if (this.status === 'cancelled') throw new DomainError('Order already cancelled')
    this.status = 'cancelled'
    return { type: 'OrderCancelled', orderId: this.orderId, reason: cmd.reason,
             cancelledAt: new Date().toISOString(), version: this.version + 1 }
  }
}

// ✅ Optimistic concurrency — expected_version check prevents concurrent write collisions
async function handleCancelOrder(cmd: CancelOrderCommand): Promise<void> {
  const { aggregate, currentVersion } = await orderRepository.load(cmd.orderId)
  const event = aggregate.cancel(cmd)

  await db.transaction(async (trx) => {
    // Fail if another command modified this aggregate since we loaded it
    const updated = await trx('orders')
      .where({ id: cmd.orderId, version: currentVersion }) // ← expected version
      .update({ status: 'cancelled', version: currentVersion + 1 })

    if (updated === 0) {
      throw new ConcurrencyError(`Order ${cmd.orderId} was modified concurrently — retry`)
    }

    await eventBus.publish('order.cancelled', event, { trx })
  })
}

// ✅ Projection handler — subscribes to the event stream and maintains the read model
// This runs as a separate process from the command handlers

class OrderDashboardProjection {
  async handle(event: DomainEvent): Promise<void> {
    switch (event.type) {
      case 'OrderPlaced':
        await redis.hset(`user:${event.userId}:dashboard`, {
          order_count:         await redis.hincrby(`user:${event.userId}:dashboard`, 'order_count', 1),
          last_order_at:       event.placedAt,
          lifetime_value_cents: await redis.hincrbyfloat(`user:${event.userId}:dashboard`, 'lifetime_value_cents', event.totalCents),
        })
        break

      case 'OrderCancelled':
        await redis.hset(`user:${event.userId}:dashboard`, {
          order_count: await redis.hincrby(`user:${event.userId}:dashboard`, 'order_count', -1),
        })
        break
    }
  }
}

// ✅ Query handler — reads directly from the pre-built read model (O(1) Redis lookup)
async function getOrderDashboard(userId: string): Promise<OrderDashboard> {
  const data = await redis.hgetall(`user:${userId}:dashboard`)
  return {
    orderCount:          Number(data.order_count ?? 0),
    lifetimeValueCents:  Number(data.lifetime_value_cents ?? 0),
    lastOrderAt:         data.last_order_at ?? null,
  }
  // No JOIN, no aggregation, no database — pure cache read
}

// ✅ Rebuild — replay all events from offset 0 to reconstruct a new or changed projection
async function rebuildOrderDashboardProjection(): Promise<void> {
  // Step 1: Clear the existing read model
  await redis.flushdb()

  // Step 2: Subscribe from the beginning and replay all events
  await kafkaConsumer.subscribe({ topic: 'order.events', fromBeginning: true })
  await kafkaConsumer.run({
    eachMessage: async ({ message }) => {
      const event = JSON.parse(message.value!.toString()) as DomainEvent
      await projection.handle(event)
      // No commits during rebuild — processing is synchronous and sequential
    }
  })
}