-- ❌ Normalized read — 6-table join, runs on every dashboard page load
-- At 1,000 concurrent users: 1,000 concurrent 6-table scans on the same write database
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,
  AVG(r.rating)                AS avg_rating
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
-- Query time: 40–200ms per call, holds read locks on 4 tables

// ✅ Materialized view — O(1) lookup, pre-computed at write time
async function getUserDashboard(userId: string): Promise<UserDashboard> {
  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,
    categories:         JSON.parse(data.categories ?? '[]'),
    avgRating:          Number(data.avg_rating ?? 0),
  }
  // Query time: 1–2ms, no JOIN, no lock, no database connection
}

// ✅ Synchronous — projection updated in the same DB transaction as the write
// Guarantees read-your-own-writes; couples view store availability to write transaction

async function placeOrder(cmd: PlaceOrderCommand): Promise<void> {
  await db.transaction(async (trx) => {
    // Write 1: persist the order
    await trx('orders').insert({
      id: cmd.orderId, user_id: cmd.userId, total_cents: cmd.totalCents,
      status: 'pending', created_at: new Date()
    })

    // Write 2: update the user dashboard projection — IN THE SAME TRANSACTION
    await trx.raw(`
      INSERT INTO user_dashboard_projection (user_id, order_count, lifetime_value_cents, last_order_at)
      VALUES ($1, 1, $2, NOW())
      ON CONFLICT (user_id)
      DO UPDATE SET
        order_count          = user_dashboard_projection.order_count + 1,
        lifetime_value_cents = user_dashboard_projection.lifetime_value_cents + $2,
        last_order_at        = NOW()
    `, [cmd.userId, cmd.totalCents])

    // Both commits atomically — or both roll back
    // Cost: write latency += projection write time; projection DB must be available for writes to succeed
  })
}

// ✅ Asynchronous — projection updated by consuming events after the write commits
// Decouples write throughput from projection latency; introduces eventual consistency window

// Step 1: Write side — publish event, do not update projection
async function placeOrder(cmd: PlaceOrderCommand): Promise<void> {
  await db('orders').insert({ id: cmd.orderId, user_id: cmd.userId, /* ... */ })
  await eventBus.publish('order.placed', {
    type: 'OrderPlaced', orderId: cmd.orderId, userId: cmd.userId, totalCents: cmd.totalCents,
    placedAt: new Date().toISOString()
  })
  // Returns immediately — projection will update asynchronously
}

// Step 2: Projection service — consumes events and updates the view
class UserDashboardProjection {
  async handle(event: OrderPlacedEvent): Promise<void> {
    await redis.hincrbyfloat(`user:${event.userId}:dashboard`, 'lifetime_value_cents', event.totalCents)
    await redis.hincrby(`user:${event.userId}:dashboard`, 'order_count', 1)
    await redis.hset(`user:${event.userId}:dashboard`, 'last_order_at', event.placedAt)
    // This runs milliseconds to seconds after the order was placed
  }
}

// ✅ Multi-source projection — merge Order and Review events into a single user dashboard
// Problem: order.placed may arrive before review.submitted for the same user, or vice versa

class UserDashboardMultiSourceProjection {
  async handleOrderPlaced(event: OrderPlacedEvent): Promise<void> {
    await redis.hincrbyfloat(`user:${event.userId}:dashboard`, 'lifetime_value_cents', event.totalCents)
    await redis.hincrby(`user:${event.userId}:dashboard`, 'order_count', 1)
    // avg_rating may not exist yet if no reviews — that is fine; projection is partial until first review
  }

  async handleReviewSubmitted(event: ReviewSubmittedEvent): Promise<void> {
    const current = await redis.hgetall(`user:${event.userId}:dashboard`)
    const currentTotal = Number(current.rating_total ?? 0) + event.rating
    const currentCount = Number(current.review_count ?? 0) + 1

    await redis.hset(`user:${event.userId}:dashboard`, {
      rating_total:  currentTotal,
      review_count:  currentCount,
      avg_rating:    (currentTotal / currentCount).toFixed(2)
    })
    // Projection is correct regardless of event arrival order
    // avg_rating field is absent until at least one review event arrives — consumers must handle null
  }
}

// ✅ Blue-green rebuild orchestration
async function rebuildUserDashboardV2(): Promise<void> {
  const NEW_KEY_PREFIX = 'user:v2' // write to a new key prefix — does not affect live traffic

  // Step 1: Replay all events from the beginning into the new projection
  await kafkaConsumer.subscribe({ topic: 'order.placed', fromBeginning: true })
  await kafkaConsumer.run({
    eachMessage: async ({ message }) => {
      const event = JSON.parse(message.value!.toString()) as OrderPlacedEvent
      await redis.hincrbyfloat(`${NEW_KEY_PREFIX}:${event.userId}:dashboard`, 'lifetime_value_cents', event.totalCents)
      // ... all other fields including new 'categories_purchased'
    }
  })

  // Step 2: Atomically swap the key prefix in the routing config
  await featureFlags.set('dashboard_projection_prefix', 'user:v2') // cut over reads

  // Step 3: Keep old 'user:' keys for 24h rollback window
  // Step 4: Decommission old keys after confidence window
  await redis.rename('user:*:dashboard', 'deprecated:user:*:dashboard') // or flushdb namespace
}

// ✅ Track projection lag as a first-class metric
class ProjectionLagMonitor {
  async measureLag(topic: string, groupId: string): Promise<LagReport> {
    const admin = kafka.admin()
    const offsets = await admin.fetchTopicOffsets(topic)
    const committed = await admin.fetchOffsets({ groupId, topics: [topic] })

    const lagPerPartition = offsets.map(partition => {
      const committedOffset = committed[0]?.partitions
        .find(p => p.partition === partition.partition)?.offset ?? '0'
      return {
        partition: partition.partition,
        lag: Number(partition.offset) - Number(committedOffset),
      }
    })

    const totalLag = lagPerPartition.reduce((sum, p) => sum + p.lag, 0)
    metrics.gauge('projection.consumer_lag', totalLag, { topic, group_id: groupId })
    if (totalLag > 10_000) {
      alerts.fire('ProjectionLagCritical', { topic, groupId, lag: totalLag })
    }
    return { topic, groupId, totalLag, lagPerPartition }
  }
}