The Ultimate System Design Interview.

-- Lua script for atomic claim
local inventory_key = KEYS[1]  -- inventory:{sale_id}
local claim_key = KEYS[2]      -- claim:{sale_id}:{user_id}
local expiry_key = KEYS[3]     -- claims:expiry:{sale_id} (all keys share the {sale_id} hash tag)
local user_id = ARGV[1]
local claim_id = ARGV[2]
local ttl = ARGV[3]            -- 180 seconds

-- Check if already claimed
if redis.call('EXISTS', claim_key) == 1 then
    return redis.call('GET', claim_key)  -- Return existing claim
end

-- Try to decrement inventory
local remaining = redis.call('DECR', inventory_key)

if remaining < 0 then
    -- No inventory, restore counter
    redis.call('INCR', inventory_key)
    return nil  -- Sold out
end

-- Success! Store claim with TTL
local claim_data = cjson.encode({
    claim_id = claim_id,
    user_id = user_id,
    claimed_at = redis.call('TIME')[1],
    expires_at = redis.call('TIME')[1] + ttl
})

redis.call('SET', claim_key, claim_data, 'EX', ttl)

-- Add to expiry tracking set. Score = expiry + 60s grace, so a checkout
-- that started just before expiry (25s budget) finishes before the unit
-- can be returned to the pool
redis.call('ZADD', expiry_key,
           redis.call('TIME')[1] + ttl + 60, claim_key)

return claim_data

{
    "event_type": "claimed",
    "claim_id": "clm_abc123",
    "sale_id": "sale_xyz",
    "user_id": "usr_456",
    "expires_at": "2024-01-15T10:03:00Z",
    "remaining_inventory": 9523
}

{
    "status": "claimed",
    "claim_id": "clm_abc123",
    "expires_at": "2024-01-15T10:03:00Z",
    "seconds_remaining": 180,
    "checkout_url": "/checkout/clm_abc123"
}

class ClaimExpiryWorker:
    """
    Background worker that returns expired claims to inventory.
    Runs continuously, processing expired claims.
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def run(self):
        """Main worker loop."""
        while True:
            await self.process_expired_claims()
            await asyncio.sleep(1)  # Check every second
    
    async def process_expired_claims(self):
        """Find and process expired claims."""
        
        now = int(time.time())
        
        # Get all active sales
        active_sales = await self.redis.smembers('active_sales')
        
        for sale_id in active_sales:
            expiry_key = f'claims:expiry:{sale_id}'
            inventory_key = f'inventory:{sale_id}'
            
            # Get claims that have expired
            expired = await self.redis.zrangebyscore(
                expiry_key,
                min=0,
                max=now
            )
            
            for claim_key in expired:
                # Atomically return inventory and remove claim
                await self.return_claim_to_inventory(
                    sale_id, claim_key, inventory_key, expiry_key
                )
    
    async def return_claim_to_inventory(
        self, sale_id, claim_key, inventory_key, expiry_key
    ):
        """Return a single expired claim to inventory."""
        
        # Lua script for atomic return
        script = """
        local claim_key = KEYS[1]
        local inventory_key = KEYS[2]
        local expiry_key = KEYS[3]
        
        -- Checkout removes the claim from this set, so if it is still here
        -- it was never checked out. (Don't test EXISTS: the claim key has its
        -- own TTL and has usually expired already, which would look like a
        -- checkout and leak the unit forever.)
        if redis.call('ZREM', expiry_key, claim_key) == 0 then
            return 0  -- already checked out or already returned
        end
        
        redis.call('DEL', claim_key)  -- may already be gone via its TTL
        redis.call('INCR', inventory_key)
        
        return 1
        """
        
        # redis-py signature: eval(script, numkeys, *keys_and_args)
        returned = await self.redis.eval(
            script, 3,
            claim_key, inventory_key, expiry_key
        )
        
        if returned:
            # Publish event
            await self.publish_claim_expired(sale_id, claim_key)

claim_data = redis.get(f"claim:{sale_id}:{user_id}")

if not claim_data:
    return 410 Gone "Your claim has expired"

# Mark checkout started (prevent concurrent checkouts). Atomic SET NX on a
# separate key: the claim itself is a JSON string, so HSET on it would fail
# (WRONGTYPE), and a separate check-then-set would race.
if not redis.set(f"checkout:{claim_id}", timestamp, nx=True, ex=180):
    return 409 Conflict "Checkout already in progress"

existing = await idempotency_store.get(claim_id)
if existing:
    if existing.status == "completed":
        return existing.response  # Return same successful response
    if existing.status == "processing":
        return 409 "Payment in progress"
    if existing.status == "failed":
        # Allow retry
        pass

# Record processing started
await idempotency_store.set(claim_id, {
    status: "processing",
    started_at: now
}, ttl=3600)

# Check circuit breaker first (Day 3 pattern)
if stripe_circuit.is_open:
    await idempotency_store.set(claim_id, {status: "failed"})
    return 503 "Payment service temporarily unavailable"

# Call Stripe with timeout (Day 1 pattern)
# Note: Stripe has deprecated the Charges API for creating payments; new code
# would use PaymentIntents, with the same idempotency_key.
try:
    payment = await asyncio.wait_for(
        stripe.charges.create(
            amount=sale.discounted_price,
            currency="usd",
            customer=user.stripe_customer_id,
            idempotency_key=f"checkout_{claim_id}",  # Day 2 pattern
            metadata={"sale_id": sale_id, "claim_id": claim_id}
        ),
        timeout=10.0  # 10 second timeout
    )
    stripe_circuit.record_success()

except asyncio.TimeoutError:
    stripe_circuit.record_failure()
    # Don't fail yet - payment might have succeeded!
    payment = await verify_payment_status(claim_id)
    if not payment:
        await idempotency_store.set(claim_id, {status: "failed"})
        return 504 "Payment timeout - please try again"

except StripeError as e:
    stripe_circuit.record_failure()
    await idempotency_store.set(claim_id, {status: "failed", error: e})
    return 400 f"Payment failed: {e.message}"

async with database.transaction():
    # Create order record
    order = await Order.create(
        id=generate_order_id(),
        user_id=user_id,
        sale_id=sale_id,
        claim_id=claim_id,
        amount=payment.amount,
        payment_id=payment.id,
        status="confirmed"
    )

    # Record in idempotency store
    await idempotency_store.set(claim_id, {
        status: "completed",
        order_id: order.id,
        response: {order_id: order.id, status: "confirmed"}
    }, ttl=86400 * 7)  # Keep for 7 days

    # Delete claim (inventory already decremented)
    await redis.delete(f"claim:{sale_id}:{user_id}")
    await redis.zrem(f"claims:expiry:{sale_id}",
                     f"claim:{sale_id}:{user_id}")

# Publish to Kafka - don't wait for confirmation
await kafka.send_async("order_events", {
    "event_type": "order.created",
    "order_id": order.id,
    "user_id": user_id,
    "sale_id": sale_id,
    "amount": payment.amount
})

# Queue confirmation email (Day 4 - webhook pattern)
await notification_queue.send({
    "type": "email",
    "template": "order_confirmation",
    "user_id": user_id,
    "order_id": order.id
})

{
    "status": "confirmed",
    "order_id": "ord_xyz789",
    "message": "Congratulations! Your order is confirmed.",
    "receipt_url": "/orders/ord_xyz789/receipt"
}

from datetime import datetime, timezone

class CheckoutService:
    """
    Checkout service implementing all Week 1-2 patterns:
    - Timeouts (Day 1)
    - Idempotency (Day 2)
    - Circuit Breakers (Day 3)
    - Async Events (Day 4)
    """
    
    def __init__(
        self,
        redis: Redis,
        db: Database,
        stripe_client: Stripe,
        kafka: KafkaProducer,
        idempotency_store: IdempotencyStore,
        circuit_breaker: CircuitBreaker
    ):
        self.redis = redis
        self.db = db
        self.stripe = stripe_client
        self.kafka = kafka
        self.idempotency = idempotency_store
        self.circuit = circuit_breaker
        
        # Timeout configuration (Day 1)
        self.payment_timeout = 10.0
        self.db_timeout = 5.0
        self.total_timeout = 25.0  # Budget for entire checkout
    
    async def checkout(
        self, 
        user_id: str, 
        sale_id: str, 
        claim_id: str
    ) -> CheckoutResult:
        """
        Process checkout with full reliability guarantees.
        """
        
        # Start timeout budget (Day 1)
        deadline = time.time() + self.total_timeout
        
        # Step 1: Validate claim
        claim = await self._validate_claim(user_id, sale_id, claim_id)
        if not claim:
            return CheckoutResult(
                success=False,
                error="Claim expired or invalid"
            )
        
        # Step 2: Idempotency check (Day 2)
        existing = await self.idempotency.get(claim_id)
        if existing:
            if existing['status'] == 'completed':
                return CheckoutResult(
                    success=True,
                    order_id=existing['order_id'],
                    idempotent=True
                )
            elif existing['status'] == 'processing':
                return CheckoutResult(
                    success=False,
                    error="Checkout already in progress"
                )
        
        # Mark as processing
        await self.idempotency.set(claim_id, {
            'status': 'processing',
            'started_at': time.time()
        })
        
        try:
            # Step 3: Process payment (Day 1 timeout + Day 3 circuit breaker)
            payment = await self._process_payment(
                user_id, sale_id, claim_id, deadline
            )
            
            # Step 4: Create order (with remaining timeout budget)
            remaining_time = deadline - time.time()
            if remaining_time <= 0:
                raise TimeoutError("Checkout timeout exceeded")
            
            order = await asyncio.wait_for(
                self._create_order(user_id, sale_id, claim_id, payment),
                timeout=remaining_time
            )
            
            # Step 5: Finalize idempotency record
            await self.idempotency.set(claim_id, {
                'status': 'completed',
                'order_id': order.id,
                'completed_at': time.time()
            }, ttl=86400 * 7)
            
            # Step 6: Publish events (async, non-blocking) (Day 4)
            asyncio.create_task(self._publish_order_events(order))
            
            return CheckoutResult(
                success=True,
                order_id=order.id
            )
        
        except PaymentError as e:
            await self.idempotency.set(claim_id, {
                'status': 'failed',
                'error': str(e)
            })
            return CheckoutResult(success=False, error=str(e))
        
        except TimeoutError:
            # Don't mark as failed - payment might have succeeded
            # Let user retry, idempotency will handle it
            return CheckoutResult(
                success=False,
                error="Request timeout - please try again"
            )
    
    async def _process_payment(
        self, 
        user_id: str, 
        sale_id: str, 
        claim_id: str,
        deadline: float
    ) -> PaymentResult:
        """Process payment with circuit breaker and timeout."""
        
        # Check circuit breaker (Day 3)
        if self.circuit.is_open():
            raise PaymentError("Payment service temporarily unavailable")
        
        # Calculate remaining timeout
        remaining = deadline - time.time()
        timeout = min(self.payment_timeout, remaining)
        
        if timeout <= 0:
            raise TimeoutError("No time remaining for payment")
        
        # Charges API shown for brevity; Stripe has deprecated it for new payments (use PaymentIntents)
        charge_params = dict(
            amount=await self._get_sale_price(sale_id),
            currency='usd',
            customer=await self._get_stripe_customer(user_id),
            idempotency_key=f"checkout_{claim_id}",  # Day 2
            metadata={
                'sale_id': sale_id,
                'claim_id': claim_id,
                'user_id': user_id
            }
        )
        
        try:
            # Call Stripe with idempotency key (Day 2)
            payment = await asyncio.wait_for(
                self.stripe.charges.create(**charge_params),
                timeout=timeout
            )
            
            self.circuit.record_success()
            return payment
        
        except asyncio.TimeoutError:
            self.circuit.record_failure()
            
            # Payment might have succeeded - verify by replaying the SAME
            # request with the SAME idempotency key: Stripe returns the
            # original charge if it went through, and charges once if not
            payment = await self._verify_payment(charge_params)
            if payment:
                return payment
            
            raise TimeoutError("Payment timeout")
        
        except StripeError as e:
            self.circuit.record_failure()
            raise PaymentError(str(e))
    
    async def _verify_payment(self, charge_params: dict) -> Optional[PaymentResult]:
        """
        Timeout recovery: replay the charge with the same idempotency key.
        (charges.list can't filter by metadata, and the Search API is
        eventually consistent, lagging up to about a minute.)
        """
        
        try:
            return await asyncio.wait_for(
                self.stripe.charges.create(**charge_params),
                timeout=5.0
            )
        except Exception:
            return None
    
    async def _create_order(
        self, 
        user_id: str, 
        sale_id: str, 
        claim_id: str,
        payment: PaymentResult
    ) -> Order:
        """Create order in database."""
        
        async with self.db.transaction():
            order = await self.db.orders.create(
                id=generate_order_id(),
                user_id=user_id,
                sale_id=sale_id,
                claim_id=claim_id,
                payment_id=payment.id,
                amount=payment.amount,
                status='confirmed',
                created_at=datetime.now(timezone.utc)
            )
            
            # Clean up claim, and take it out of the expiry set so the
            # expiry worker never returns this (sold) unit to the pool
            await self.redis.delete(f"claim:{sale_id}:{user_id}")
            await self.redis.zrem(f"claims:expiry:{sale_id}",
                                  f"claim:{sale_id}:{user_id}")
            
            return order
    
    async def _publish_order_events(self, order: Order):
        """Publish events for order (non-blocking)."""
        
        # Order created event
        await self.kafka.send('order_events', {
            'event_type': 'order.created',
            'order_id': order.id,
            'user_id': order.user_id,
            'sale_id': order.sale_id,
            'amount': order.amount,
            'timestamp': datetime.now(timezone.utc).isoformat()
        })
        
        # Queue confirmation email
        await self.kafka.send('notification_events', {
            'type': 'email',
            'template': 'order_confirmation',
            'recipient_id': order.user_id,
            'data': {
                'order_id': order.id,
                'amount': order.amount
            }
        })

SELECT user_id, email, push_token, notification_preferences
FROM users
WHERE (user_id IN (SELECT user_id FROM product_watchers
                   WHERE product_id = 'xyz')
       OR subscribed_to_deals = true)
AND notification_enabled = true

{
    "sale_id": "xyz",
    "batch_count": 24,
    "trigger_time": "2024-01-15T10:00:00Z"
}

class NotificationWorker:
    """
    Notification worker implementing Day 4 patterns:
    - At-least-once delivery
    - Retry with backoff
    - Dead letter queue
    - Circuit breaker for external services
    """
    
    def __init__(
        self,
        redis: Redis,
        kafka_consumer: KafkaConsumer,
        push_service: PushNotificationService,
        email_service: EmailService,
        circuit_breakers: Dict[str, CircuitBreaker]
    ):
        self.redis = redis
        self.consumer = kafka_consumer
        self.push = push_service
        self.email = email_service
        self.circuits = circuit_breakers
    
    async def process_batch(self, trigger: NotificationTrigger):
        """Process a notification batch for a sale."""
        
        batch_key = f"notification_batch:{trigger.sale_id}:{trigger.batch_id}"
        
        # Get users from pre-computed batch
        users = await self.redis.lrange(batch_key, 0, -1)
        
        for user_data in users:
            user = json.loads(user_data)
            
            try:
                await self._send_notification(user, trigger)
            except Exception as e:
                # Log and continue - don't let one failure stop batch
                logger.error(
                    "Notification failed",
                    user_id=user['id'],
                    error=str(e)
                )
    
    async def _send_notification(self, user: dict, trigger: NotificationTrigger):
        """Send notification to a single user."""
        
        # Idempotency check (Day 2)
        idem_key = f"notif_sent:{trigger.sale_id}:{user['id']}"
        if await self.redis.exists(idem_key):
            return  # Already sent
        
        # Send based on preferences
        if user.get('push_enabled') and user.get('push_token'):
            await self._send_push(user, trigger)
        
        if user.get('email_enabled') and user.get('email'):
            await self._queue_email(user, trigger)
        
        # Mark as sent (idempotency)
        await self.redis.set(idem_key, '1', ex=3600)  # 1 hour TTL
    
    async def _send_push(self, user: dict, trigger: NotificationTrigger):
        """Send push notification with circuit breaker."""
        
        # Check circuit breaker (Day 3)
        provider = 'fcm' if user['platform'] == 'android' else 'apns'
        
        if self.circuits[provider].is_open():
            # Queue for retry later
            await self._queue_for_retry(user, trigger, 'push')
            return
        
        try:
            await asyncio.wait_for(
                self.push.send(
                    token=user['push_token'],
                    title="⚡ Lightning Deal Live!",
                    body=f"The {trigger.product_name} deal is live! Tap to claim.",
                    data={'sale_id': trigger.sale_id}
                ),
                timeout=2.0  # Day 1: Timeout
            )
            
            self.circuits[provider].record_success()
            
        except asyncio.TimeoutError:
            self.circuits[provider].record_failure()
            await self._queue_for_retry(user, trigger, 'push')
        
        except Exception as e:
            self.circuits[provider].record_failure()
            logger.error("Push failed", error=str(e))
    
    async def _queue_email(self, user: dict, trigger: NotificationTrigger):
        """Queue email for async sending (Day 4 pattern)."""
        
        await self.kafka.send('email_queue', {
            'recipient': user['email'],
            'template': 'flash_sale_live',
            'data': {
                'user_name': user['name'],
                'product_name': trigger.product_name,
                'sale_url': f"https://megamart.com/sale/{trigger.sale_id}"
            },
            'idempotency_key': f"sale_email:{trigger.sale_id}:{user['id']}"
        })

from datetime import datetime, timezone

class FlashSaleScheduler:
    """
    Distributed flash sale scheduler using Day 5 patterns:
    - Leader election
    - Fencing tokens
    - Exactly-once scheduling
    - Heartbeat monitoring
    """
    
    def __init__(
        self,
        etcd_client,
        redis: Redis,
        db: Database,
        kafka: KafkaProducer,
        notification_service: NotificationService
    ):
        self.etcd = etcd_client
        self.redis = redis
        self.db = db
        self.kafka = kafka
        self.notifications = notification_service
        
        self.is_leader = False
        self.fencing_token = 0
        self.leader_election = LeaderElection(
            etcd_client,
            election_name="sale-scheduler"
        )
    
    async def start(self):
        """Start the scheduler."""
        
        # Start leader election
        asyncio.create_task(
            self.leader_election.campaign(
                on_elected=self._on_elected,
                on_demoted=self._on_demoted
            )
        )
        
        # Main scheduler loop
        while True:
            if self.is_leader:
                await self._scheduler_tick()
            await asyncio.sleep(1)
    
    def _on_elected(self):
        """Called when this instance becomes leader."""
        self.is_leader = True
        # Strictly increasing across elections (e.g. the etcd revision of our
        # leader key). Not wall-clock time: clocks can go backwards.
        self.fencing_token = self.leader_election.leader_revision
        logger.info("Became sale scheduler leader", token=self.fencing_token)
    
    def _on_demoted(self):
        """Called when this instance loses leadership."""
        self.is_leader = False
        logger.info("Lost sale scheduler leadership")
    
    async def _scheduler_tick(self):
        """Main scheduling loop tick."""
        
        now = datetime.now(timezone.utc)
        
        # Find sales that need action
        await self._prepare_upcoming_sales(now)
        await self._start_due_sales(now)
        await self._end_expired_sales(now)
    
    async def _prepare_upcoming_sales(self, now: datetime):
        """Prepare sales starting in next 10 minutes."""
        
        upcoming = await self.db.query("""
            SELECT * FROM flash_sales
            WHERE status = 'scheduled'
            AND start_time BETWEEN $1 AND $2
            AND preparation_started = false
        """, now, now + timedelta(minutes=10))
        
        for sale in upcoming:
            # Mark as preparing (idempotent)
            await self.db.execute("""
                UPDATE flash_sales
                SET preparation_started = true,
                    preparation_token = $1
                WHERE id = $2 AND preparation_started = false
            """, self.fencing_token, sale.id)
            
            # Pre-compute notifications (async)
            asyncio.create_task(
                self.notifications.prepare_recipients(sale)
            )
            
            # Warm up Redis
            await self._warm_up_sale_data(sale)
    
    async def _start_due_sales(self, now: datetime):
        """Start sales that are due."""
        
        due_sales = await self.db.query("""
            SELECT * FROM flash_sales
            WHERE status = 'scheduled'
            AND start_time <= $1
            AND preparation_started = true
        """, now)
        
        for sale in due_sales:
            await self._start_sale(sale)
    
    async def _start_sale(self, sale):
        """
        Start a single sale.
        Uses fencing token to prevent double-start.
        """
        
        # Atomically claim sale start with fencing token
        claimed = await self.db.execute("""
            UPDATE flash_sales
            SET status = 'active',
                started_at = NOW(),
                started_by_token = $1
            WHERE id = $2
            AND status = 'scheduled'
            AND (started_by_token IS NULL OR started_by_token < $1)
            RETURNING id
        """, self.fencing_token, sale.id)
        
        if not claimed:
            logger.warning("Could not claim sale start", sale_id=sale.id)
            return
        
        logger.info("Starting sale", sale_id=sale.id, token=self.fencing_token)
        
        # Initialize inventory in Redis
        await self.redis.set(
            f"inventory:{sale.id}",
            sale.total_inventory
        )
        
        # Add to active sales set
        await self.redis.sadd('active_sales', sale.id)
        
        # Trigger notifications
        await self.kafka.send('notification_triggers', {
            'type': 'sale_started',
            'sale_id': sale.id,
            'product_name': sale.product_name,
            'fencing_token': self.fencing_token
        })
        
        # Schedule sale end
        asyncio.create_task(
            self._schedule_sale_end(sale.id, sale.duration_minutes)
        )
    
    async def _schedule_sale_end(self, sale_id: str, duration_minutes: int):
        """Schedule the end of a sale."""
        
        await asyncio.sleep(duration_minutes * 60)
        
        if self.is_leader:  # Only end if still leader
            await self._end_sale(sale_id)
    
    async def _end_sale(self, sale_id: str):
        """End a sale and clean up."""
        
        # Update database with fencing token
        ended = await self.db.execute("""
            UPDATE flash_sales
            SET status = 'ended',
                ended_at = NOW(),
                ended_by_token = $1
            WHERE id = $2
            AND status = 'active'
            RETURNING id
        """, self.fencing_token, sale_id)
        
        if not ended:
            return
        
        logger.info("Ending sale", sale_id=sale_id)
        
        # Remove from active sales
        await self.redis.srem('active_sales', sale_id)
        
        # Return unclaimed inventory to report
        remaining = await self.redis.get(f"inventory:{sale_id}")
        
        # Publish sale ended event
        await self.kafka.send('sale_events', {
            'type': 'sale_ended',
            'sale_id': sale_id,
            'remaining_inventory': int(remaining) if remaining else 0,
            'fencing_token': self.fencing_token
        })
        
        # Clean up Redis (keep for a while for debugging)
        await self.redis.expire(f"inventory:{sale_id}", 3600)
    
    async def _warm_up_sale_data(self, sale):
        """Pre-load sale data into Redis."""
        
        await self.redis.hset(f"sale:{sale.id}", mapping={
            'product_id': sale.product_id,
            'product_name': sale.product_name,
            'original_price': sale.original_price,
            'sale_price': sale.sale_price,
            'total_inventory': sale.total_inventory,
            'start_time': sale.start_time.isoformat(),
            'end_time': sale.end_time.isoformat()
        })
        
        # Set TTL beyond sale end
        await self.redis.expire(
            f"sale:{sale.id}",
            sale.duration_minutes * 60 + 3600
        )

# Prometheus Alert Rules

groups:
  - name: flash_sale_critical
    rules:
      # INVENTORY ALERTS
      - alert: InventoryOversold
        expr: flash_sale_inventory < 0
        for: 0s
        labels:
          severity: critical
        annotations:
          summary: "CRITICAL: Inventory oversold for sale {{ $labels.sale_id }}"
          
      - alert: InventoryNotDecreasing
        expr: deriv(flash_sale_inventory[1m]) >= 0 and flash_sale_active == 1  # deriv() for gauges; PromQL operators are lowercase
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Inventory not moving during active sale"

      # LATENCY ALERTS
      - alert: ClaimLatencyHigh
        expr: histogram_quantile(0.99, rate(claim_request_duration_seconds_bucket[1m])) > 0.5
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "P99 claim latency {{ $value }}s exceeds 500ms"
          
      - alert: CheckoutLatencyHigh
        expr: histogram_quantile(0.99, rate(checkout_request_duration_seconds_bucket[1m])) > 3
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "P99 checkout latency {{ $value }}s exceeds 3s"

      # ERROR RATE ALERTS
      - alert: HighErrorRate
        expr: sum(rate(http_requests_total{status=~"5.."}[1m])) / sum(rate(http_requests_total[1m])) > 0.01
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Error rate {{ $value | humanizePercentage }} exceeds 1%"

      # CIRCUIT BREAKER ALERTS
      - alert: CircuitBreakerOpen
        expr: circuit_breaker_state == 2  # 2 = OPEN
        for: 0s
        labels:
          severity: critical
        annotations:
          summary: "Circuit breaker {{ $labels.name }} is OPEN"

      # INFRASTRUCTURE ALERTS
      - alert: RedisHighLatency
        expr: redis_command_duration_seconds_p99 > 0.01
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "Redis P99 latency {{ $value }}s exceeds 10ms"
          
      - alert: KafkaLag
        expr: kafka_consumer_lag > 10000
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Kafka consumer lag {{ $value }} exceeds 10k"