Food Delivery Order System.

# services/order_service.py
# Applies: Week 3, Day 2 - Transactional Outbox

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional
import uuid
import json


@dataclass
class Order:
    order_id: str
    customer_id: str
    restaurant_id: str
    items: list
    delivery_address: dict
    total_amount: float
    status: str
    created_at: datetime


@dataclass
class OutboxMessage:
    message_id: str
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: dict
    created_at: datetime


class OrderService:
    """
    Order service with transactional outbox pattern.
    
    Guarantees:
    - Order and outbox message written atomically
    - Events eventually published to Kafka
    - No order is ever lost
    """
    
    def __init__(self, db_pool, payment_service):
        self.db = db_pool
        self.payments = payment_service
    
    async def place_order(self, request: dict) -> Order:
        """
        Place an order with guaranteed event publishing.
        
        Uses single transaction for order + outbox.
        """
        
        # The client creates request_id once per checkout and resends it on
        # retries and double-clicks, so every attempt maps to the same
        # order_id and the same authorization key (a server-side uuid4
        # would make each click a new order and a second authorization)
        order_id = str(uuid.uuid5(
            uuid.NAMESPACE_URL,
            f"order:{request['customer_id']}:{request['request_id']}"
        ))
        
        # Authorize payment first (can fail, no order created yet)
        auth_result = await self.payments.authorize(
            customer_id=request["customer_id"],
            amount=request["total_amount"],
            idempotency_key=f"order:{order_id}:auth"
        )
        
        if not auth_result.success:
            raise PaymentAuthorizationError(auth_result.error)
        
        # Single transaction: order + outbox
        async with self.db.transaction() as tx:
            # Insert order
            order = Order(
                order_id=order_id,
                customer_id=request["customer_id"],
                restaurant_id=request["restaurant_id"],
                items=request["items"],
                delivery_address=request["delivery_address"],
                total_amount=request["total_amount"],
                status="placed",
                created_at=datetime.now(timezone.utc)
            )
            
            inserted = await tx.fetchval("""
                INSERT INTO orders (
                    order_id, customer_id, restaurant_id, items,
                    delivery_address, total_amount, status, created_at,
                    payment_auth_id
                ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                ON CONFLICT (order_id) DO NOTHING
                RETURNING order_id
            """, order.order_id, order.customer_id, order.restaurant_id,
                json.dumps(order.items), json.dumps(order.delivery_address),
                order.total_amount, order.status, order.created_at,
                auth_result.authorization_id)
            
            if inserted is None:
                # A retry of an order already placed: same order_id, and
                # no second OrderPlaced event
                return order
            
            # Insert outbox message (same transaction!)
            outbox_message = OutboxMessage(
                message_id=str(uuid.uuid4()),
                aggregate_type="Order",
                aggregate_id=order_id,
                event_type="OrderPlaced",
                payload={
                    "order_id": order_id,
                    "customer_id": order.customer_id,
                    "restaurant_id": order.restaurant_id,
                    "items": order.items,
                    "total_amount": order.total_amount,
                    "payment_auth_id": auth_result.authorization_id,
                },
                created_at=datetime.now(timezone.utc)
            )
            
            await tx.execute("""
                INSERT INTO outbox (
                    message_id, aggregate_type, aggregate_id,
                    event_type, payload, created_at
                ) VALUES ($1, $2, $3, $4, $5, $6)
            """, outbox_message.message_id, outbox_message.aggregate_type,
                outbox_message.aggregate_id, outbox_message.event_type,
                json.dumps(outbox_message.payload), outbox_message.created_at)
        
        # Transaction committed - order is durable
        # Outbox publisher will send to Kafka eventually
        
        return order


class OutboxPublisher:
    """
    Publishes outbox messages to Kafka.
    
    Runs as background process, polling for unpublished messages.
    """
    
    def __init__(self, db_pool, kafka_producer):
        self.db = db_pool
        self.kafka = kafka_producer
    
    async def poll_and_publish(self, batch_size: int = 100):
        """Poll outbox and publish to Kafka."""
        
        async with self.db.transaction() as tx:
            # Lock and fetch unpublished messages
            messages = await tx.fetch("""
                SELECT * FROM outbox
                WHERE published_at IS NULL
                ORDER BY created_at
                LIMIT $1
                FOR UPDATE SKIP LOCKED
            """, batch_size)
            
            for msg in messages:
                topic = self._get_topic(msg["event_type"])
                
                # Publish to Kafka
                await self.kafka.send(
                    topic=topic,
                    key=msg["aggregate_id"].encode(),
                    value=json.dumps(msg["payload"]).encode(),
                    headers=[("message_id", msg["message_id"].encode())]
                )
                
                # Mark as published
                await tx.execute("""
                    UPDATE outbox 
                    SET published_at = NOW()
                    WHERE message_id = $1
                """, msg["message_id"])
    
    def _get_topic(self, event_type: str) -> str:
        return {
            "OrderPlaced": "orders.created",
            "OrderConfirmed": "orders.status",
            "OrderReady": "orders.status",
            "OrderDelivered": "orders.status",
        }.get(event_type, "orders.events")

# services/payment_service.py
# Applies: Week 2, Day 2 - Idempotency

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
from enum import Enum
import hashlib


class PaymentStatus(Enum):
    PENDING = "pending"
    AUTHORIZED = "authorized"
    CAPTURED = "captured"
    REFUNDED = "refunded"
    FAILED = "failed"


@dataclass 
class PaymentRecord:
    payment_id: str
    order_id: str
    amount: float
    status: PaymentStatus
    idempotency_key: str
    stripe_payment_intent_id: Optional[str]
    created_at: datetime
    updated_at: datetime


class PaymentService:
    """
    Payment service with idempotency guarantees.
    
    Every operation uses idempotency keys to ensure:
    - Authorization happens exactly once
    - Capture happens exactly once
    - Refund happens exactly once
    """
    
    def __init__(self, db_pool, stripe_client, redis_client):
        self.db = db_pool
        self.stripe = stripe_client
        self.redis = redis_client
        
        # Idempotency window (Week 2 concept)
        self.idempotency_ttl = timedelta(hours=24)
    
    async def authorize(
        self,
        customer_id: str,
        amount: float,
        idempotency_key: str
    ) -> AuthorizationResult:
        """
        Authorize payment with idempotency.
        
        Same idempotency_key always returns same result.
        """
        
        # Check if we've seen this key before
        existing = await self._get_idempotent_result(idempotency_key)
        if existing:
            return existing
        
        # Try to acquire lock for this key
        lock_acquired = await self._acquire_idempotency_lock(idempotency_key)
        if not lock_acquired:
            # Another request is processing - wait and return their result
            return await self._wait_for_result(idempotency_key)
        
        try:
            # Create payment intent with Stripe
            # Stripe also accepts idempotency key!
            intent = await self.stripe.create_payment_intent(
                amount=int(amount * 100),  # cents
                currency="usd",
                customer=customer_id,
                capture_method="manual",  # Authorize only
                idempotency_key=idempotency_key
            )
            
            result = AuthorizationResult(
                success=True,
                authorization_id=intent.id,
                amount=amount
            )
            
            # Store result for future idempotent requests
            await self._store_idempotent_result(idempotency_key, result)
            
            return result
            
        except StripeError as e:
            result = AuthorizationResult(
                success=False,
                error=str(e)
            )
            # Only cache deterministic outcomes (e.g. card declined). Transient
            # errors (network, 5xx, rate limit) must stay retryable, or every
            # retry with this key would replay the failure for 24 hours
            if isinstance(e, CardError):
                await self._store_idempotent_result(idempotency_key, result)
            return result
            
        finally:
            await self._release_idempotency_lock(idempotency_key)
    
    async def capture(
        self,
        order_id: str,
        authorization_id: str,
        amount: float
    ) -> CaptureResult:
        """
        Capture authorized payment.
        
        Idempotency key derived from order_id ensures exactly-once capture.
        """
        
        idempotency_key = f"capture:{order_id}"
        
        existing = await self._get_idempotent_result(idempotency_key)
        if existing:
            return existing
        
        lock_acquired = await self._acquire_idempotency_lock(idempotency_key)
        if not lock_acquired:
            return await self._wait_for_result(idempotency_key)
        
        try:
            # Check current state in our database
            payment = await self._get_payment_by_order(order_id)
            
            if payment and payment.status == PaymentStatus.CAPTURED:
                # Already captured - return success
                return CaptureResult(success=True, captured=True)
            
            # Capture with Stripe (Stripe idempotency handles their side)
            intent = await self.stripe.capture_payment_intent(
                authorization_id,
                amount_to_capture=int(amount * 100),
                idempotency_key=idempotency_key
            )
            
            # Update our database
            await self._update_payment_status(
                order_id,
                PaymentStatus.CAPTURED
            )
            
            result = CaptureResult(success=True, captured=True)
            await self._store_idempotent_result(idempotency_key, result)
            
            return result
            
        except StripeError as e:
            result = CaptureResult(success=False, error=str(e))
            if isinstance(e, CardError):  # Cache deterministic failures only
                await self._store_idempotent_result(idempotency_key, result)
            return result
            
        finally:
            await self._release_idempotency_lock(idempotency_key)
    
    async def _acquire_idempotency_lock(self, key: str) -> bool:
        """Acquire distributed lock for idempotency key."""
        lock_key = f"idem_lock:{key}"
        return await self.redis.set(
            lock_key,
            "1",
            nx=True,  # Only if not exists
            ex=30     # 30 second timeout
        )
    
    async def _get_idempotent_result(self, key: str):
        """Get stored result for idempotency key."""
        result_key = f"idem_result:{key}"
        data = await self.redis.get(result_key)
        if data:
            return self._deserialize_result(data)
        return None
    
    async def _store_idempotent_result(self, key: str, result):
        """Store result for future idempotent requests."""
        result_key = f"idem_result:{key}"
        await self.redis.setex(
            result_key,
            int(self.idempotency_ttl.total_seconds()),
            self._serialize_result(result)
        )

# services/driver_matcher.py
# Applies: Week 5, Day 5 - Leader Election & Coordination

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List
import asyncio
import math


@dataclass
class Driver:
    driver_id: str
    location: tuple  # (lat, lng)
    status: str      # available, assigned, busy
    rating: float
    current_order_id: Optional[str]


@dataclass
class AssignmentResult:
    success: bool
    driver_id: Optional[str] = None
    error: Optional[str] = None


class DriverMatcher:
    """
    Assigns drivers to orders with coordination guarantees.
    
    Uses a row lock (SELECT ... FOR UPDATE) plus a version check
    to prevent double assignment.
    """
    
    def __init__(self, db_pool, redis_client, location_service):
        self.db = db_pool
        self.redis = redis_client
        self.locations = location_service
    
    async def assign_driver(
        self,
        order_id: str,
        restaurant_location: tuple,
        max_distance_km: float = 5.0
    ) -> AssignmentResult:
        """
        Find and assign optimal driver for an order.
        
        Uses row locking + version check to prevent race conditions.
        """
        
        # Find candidate drivers near restaurant
        candidates = await self._find_nearby_drivers(
            restaurant_location,
            max_distance_km
        )
        
        if not candidates:
            return AssignmentResult(
                success=False,
                error="No available drivers nearby"
            )
        
        # Sort by score (distance, rating, etc.)
        ranked = self._rank_drivers(candidates, restaurant_location)
        
        # Try to assign, starting with best candidate
        for driver in ranked:
            result = await self._try_assign(order_id, driver)
            if result.success:
                return result
        
        return AssignmentResult(
            success=False,
            error="All nearby drivers unavailable"
        )
    
    async def _try_assign(
        self,
        order_id: str,
        driver: Driver
    ) -> AssignmentResult:
        """
        Attempt to assign driver (row lock + version check).
        
        Only succeeds if driver is still available.
        """
        
        async with self.db.transaction() as tx:
            # Lock the driver row and check status
            current = await tx.fetchrow("""
                SELECT driver_id, status, version
                FROM drivers
                WHERE driver_id = $1
                FOR UPDATE
            """, driver.driver_id)
            
            if not current or current["status"] != "available":
                # Driver no longer available
                return AssignmentResult(success=False)
            
            # Update driver status atomically
            result = await tx.execute("""
                UPDATE drivers
                SET status = 'assigned',
                    current_order_id = $1,
                    version = version + 1,
                    assigned_at = NOW()
                WHERE driver_id = $2
                  AND version = $3
                  AND status = 'available'
            """, order_id, driver.driver_id, current["version"])
            
            if result == "UPDATE 0":
                # Concurrent modification - another order got them
                return AssignmentResult(success=False)
            
            # Create assignment record
            await tx.execute("""
                INSERT INTO assignments (
                    assignment_id, order_id, driver_id,
                    status, created_at
                ) VALUES (gen_random_uuid(), $1, $2, 'pending', NOW())
            """, order_id, driver.driver_id)
        
        # Successfully assigned!
        return AssignmentResult(
            success=True,
            driver_id=driver.driver_id
        )
    
    async def _find_nearby_drivers(
        self,
        location: tuple,
        max_distance_km: float
    ) -> List[Driver]:
        """Find available drivers within radius."""
        
        # Use Redis GEO for fast spatial query
        lat, lng = location
        
        # Get driver IDs within radius
        nearby_ids = await self.redis.georadius(
            "driver_locations",
            lng, lat,
            max_distance_km,
            unit="km",
            count=50
        )
        
        if not nearby_ids:
            return []
        
        # Fetch driver details (only available ones)
        drivers = await self.db.fetch("""
            SELECT driver_id, status, rating
            FROM drivers
            WHERE driver_id = ANY($1)
              AND status = 'available'
        """, [d.decode() for d in nearby_ids])
        
        return [
            Driver(
                driver_id=d["driver_id"],
                location=await self.locations.get(d["driver_id"]),
                status=d["status"],
                rating=d["rating"],
                current_order_id=None
            )
            for d in drivers
        ]
    
    def _rank_drivers(
        self,
        drivers: List[Driver],
        restaurant_location: tuple
    ) -> List[Driver]:
        """Rank drivers by assignment score."""
        
        def score(driver: Driver) -> float:
            distance = self._haversine(driver.location, restaurant_location)
            # Lower distance is better, higher rating is better
            return distance - (driver.rating * 0.5)
        
        return sorted(drivers, key=score)
    
    @staticmethod
    def _haversine(a: tuple, b: tuple) -> float:
        """Great-circle distance in km between two (lat, lng) points."""
        lat1, lng1, lat2, lng2 = map(math.radians, (*a, *b))
        h = (math.sin((lat2 - lat1) / 2) ** 2
             + math.cos(lat1) * math.cos(lat2) * math.sin((lng2 - lng1) / 2) ** 2)
        return 2 * 6371.0 * math.asin(math.sqrt(h))

# services/order_saga.py
# Applies: Week 5, Day 2-3 - Saga Pattern

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import asyncio
import logging

logger = logging.getLogger(__name__)


class SagaStep(Enum):
    VALIDATE_ORDER = "validate"
    AUTHORIZE_PAYMENT = "authorize_payment"
    NOTIFY_RESTAURANT = "notify_restaurant"
    AWAIT_CONFIRMATION = "await_confirmation"
    ASSIGN_DRIVER = "assign_driver"
    CAPTURE_PAYMENT = "capture_payment"
    COMPLETE = "complete"


class SagaStatus(Enum):
    RUNNING = "running"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"


@dataclass
class OrderSagaState:
    saga_id: str
    order_id: str
    current_step: SagaStep
    status: SagaStatus
    payment_auth_id: Optional[str] = None
    driver_id: Optional[str] = None
    compensation_reason: Optional[str] = None


class OrderSagaOrchestrator:
    """
    Orchestrates order lifecycle as a saga.
    
    Each step has a corresponding compensation action.
    If any step fails, we run compensations in reverse order.
    """
    
    def __init__(
        self,
        order_service,
        payment_service,
        restaurant_service,
        driver_service,
        notification_service,
        saga_repo
    ):
        self.orders = order_service
        self.payments = payment_service
        self.restaurants = restaurant_service
        self.drivers = driver_service
        self.notifications = notification_service
        self.sagas = saga_repo
    
    async def execute(self, order_id: str) -> OrderSagaState:
        """Execute order saga from start to completion."""
        
        state = OrderSagaState(
            saga_id=f"saga:{order_id}",
            order_id=order_id,
            current_step=SagaStep.VALIDATE_ORDER,
            status=SagaStatus.RUNNING
        )
        
        await self.sagas.save(state)
        
        try:
            # Step 1: Validate order
            state = await self._validate_order(state)
            
            # Step 2: Authorize payment
            state = await self._authorize_payment(state)
            
            # Step 3: Notify restaurant
            state = await self._notify_restaurant(state)
            
            # Step 4: Wait for restaurant confirmation
            # (This happens async via webhook/event)
            state.current_step = SagaStep.AWAIT_CONFIRMATION
            await self.sagas.save(state)
            
            return state
            
        except SagaStepError as e:
            logger.error(f"Saga {state.saga_id} failed at {state.current_step}: {e}")
            return await self._compensate(state, str(e))
    
    async def on_restaurant_confirmed(self, order_id: str, prep_time_minutes: int):
        """Called when restaurant confirms order."""
        
        state = await self.sagas.get_by_order(order_id)
        if not state or state.status != SagaStatus.RUNNING:
            return
        
        try:
            # Step 5: Assign driver
            state = await self._assign_driver(state)
            
            # Update order with driver info
            await self.orders.update_status(
                order_id,
                "driver_assigned",
                driver_id=state.driver_id
            )
            
            # Notify customer
            await self.notifications.send(
                user_id=state.customer_id,
                type="driver_assigned",
                template="driver_assigned",
                variables={"driver_id": state.driver_id}
            )
            
            state.current_step = SagaStep.COMPLETE
            state.status = SagaStatus.COMPLETED
            await self.sagas.save(state)
            
        except SagaStepError as e:
            await self._compensate(state, str(e))
    
    async def on_restaurant_rejected(self, order_id: str, reason: str):
        """Called when restaurant rejects order."""
        
        state = await self.sagas.get_by_order(order_id)
        if not state:
            return
        
        await self._compensate(state, f"Restaurant rejected: {reason}")
    
    async def on_driver_picked_up(self, order_id: str):
        """Called when driver picks up order — capture payment."""
        
        state = await self.sagas.get_by_order(order_id)
        if not state or not state.payment_auth_id:
            return
        
        # Capture payment (idempotent)
        result = await self.payments.capture(
            order_id=order_id,
            authorization_id=state.payment_auth_id,
            amount=state.order_amount
        )
        
        if not result.success:
            logger.error(f"Payment capture failed for {order_id}: {result.error}")
            # Don't compensate here — food is already with driver
            # Flag for manual review instead
    
    async def _compensate(
        self,
        state: OrderSagaState,
        reason: str
    ) -> OrderSagaState:
        """
        Run compensation actions in reverse order.
        
        Compensation order:
        1. Release driver (if assigned)
        2. Void payment authorization (if authorized)
        3. Update order status to cancelled
        4. Notify customer
        """
        
        state.status = SagaStatus.COMPENSATING
        state.compensation_reason = reason
        await self.sagas.save(state)
        
        logger.info(f"Starting compensation for saga {state.saga_id}")
        
        # Compensate driver assignment
        if state.driver_id:
            await self._retry_until_done(
                "release driver",
                lambda: self.drivers.release(state.driver_id)
            )
            logger.info(f"Released driver {state.driver_id}")
        
        # Compensate payment authorization
        if state.payment_auth_id:
            await self._retry_until_done(
                "void payment",
                lambda: self.payments.void_authorization(
                    authorization_id=state.payment_auth_id,
                    idempotency_key=f"void:{state.order_id}"
                )
            )
            logger.info(f"Voided payment auth {state.payment_auth_id}")
        
        # Update order status
        await self.orders.update_status(
            state.order_id,
            "cancelled",
            reason=reason
        )
        
        # Notify customer
        await self.notifications.send(
            user_id=state.customer_id,
            type="order_cancelled",
            template="order_cancelled",
            variables={"reason": reason}
        )
        
        state.status = SagaStatus.FAILED
        await self.sagas.save(state)
        
        return state
    
    async def _retry_until_done(self, name: str, action) -> None:
        """
        Compensations must eventually succeed (Week 5), so a failure is
        retried with capped backoff, never just logged. Both actions are
        idempotent, so retrying is safe; the saga stays COMPENSATING in
        the repo meanwhile, so a restarted orchestrator can resume it.
        """
        delay = 1
        while True:
            try:
                await action()
                return
            except Exception as e:
                logger.error(f"Failed to {name}, retrying in {delay}s: {e}")
                await asyncio.sleep(delay)
                delay = min(delay * 2, 60)
    
    async def _validate_order(self, state: OrderSagaState) -> OrderSagaState:
        """Validate order details."""
        order = await self.orders.get(state.order_id)
        
        # Check restaurant is open
        restaurant = await self.restaurants.get(order.restaurant_id)
        if not restaurant.is_open():
            raise SagaStepError("Restaurant is closed")
        
        # Check delivery address is in range
        if not restaurant.delivers_to(order.delivery_address):
            raise SagaStepError("Delivery address out of range")
        
        state.current_step = SagaStep.AUTHORIZE_PAYMENT
        state.customer_id = order.customer_id
        state.order_amount = order.total_amount
        await self.sagas.save(state)
        
        return state
    
    async def _authorize_payment(self, state: OrderSagaState) -> OrderSagaState:
        """Authorize payment."""
        result = await self.payments.authorize(
            customer_id=state.customer_id,
            amount=state.order_amount,
            idempotency_key=f"order:{state.order_id}:auth"
        )
        
        if not result.success:
            raise SagaStepError(f"Payment failed: {result.error}")
        
        state.payment_auth_id = result.authorization_id
        state.current_step = SagaStep.NOTIFY_RESTAURANT
        await self.sagas.save(state)
        
        return state
    
    async def _notify_restaurant(self, state: OrderSagaState) -> OrderSagaState:
        """Send order to restaurant."""
        order = await self.orders.get(state.order_id)
        
        await self.restaurants.send_order(
            restaurant_id=order.restaurant_id,
            order=order
        )
        
        state.current_step = SagaStep.AWAIT_CONFIRMATION
        await self.sagas.save(state)
        
        return state
    
    async def _assign_driver(self, state: OrderSagaState) -> OrderSagaState:
        """Assign driver to order."""
        order = await self.orders.get(state.order_id)
        restaurant = await self.restaurants.get(order.restaurant_id)
        
        result = await self.drivers.assign(
            order_id=state.order_id,
            restaurant_location=restaurant.location
        )
        
        if not result.success:
            raise SagaStepError(f"Driver assignment failed: {result.error}")
        
        state.driver_id = result.driver_id
        state.current_step = SagaStep.COMPLETE
        await self.sagas.save(state)
        
        return state

# services/location_service.py
# Applies: Week 4 - Caching, Week 1 - Hot Keys

from datetime import datetime, timedelta
from typing import Optional, List, Tuple
import asyncio


class LocationService:
    """
    High-frequency location tracking with caching.
    
    Scale: 50K drivers × 1 update/3sec = 17K updates/sec
    (~40-60K/sec with the 120-180K drivers of a dinner peak)
    
    Strategy:
    - Redis GEO for current locations (hot data)
    - Batch writes to PostgreSQL for history (cold data)
    - Customers don't read from here: TrackingPublisher (below) pushes to them
    """
    
    def __init__(self, redis_client, db_pool):
        self.redis = redis_client
        self.db = db_pool
        self._history_buffer = []
        self._buffer_lock = asyncio.Lock()
    
    async def update_location(
        self,
        driver_id: str,
        lat: float,
        lng: float,
        timestamp: datetime
    ):
        """
        Update driver location.
        
        Hot path: Redis GEO (immediate)
        Cold path: Batched PostgreSQL writes (async)
        """
        
        # Update Redis GEO (immediate, for spatial queries)
        await self.redis.geoadd(
            "driver_locations",
            lng, lat, driver_id
        )
        
        # Store current location with timestamp
        await self.redis.hset(
            f"driver:{driver_id}:location",
            mapping={
                "lat": str(lat),
                "lng": str(lng),
                "timestamp": timestamp.isoformat()
            }
        )
        
        # Buffer for batch history write
        async with self._buffer_lock:
            self._history_buffer.append({
                "driver_id": driver_id,
                "lat": lat,
                "lng": lng,
                "timestamp": timestamp
            })
            
            # Flush if buffer is large
            if len(self._history_buffer) >= 1000:
                await self._flush_history()
    
    async def get_driver_location(
        self,
        driver_id: str
    ) -> Optional[Tuple[float, float, datetime]]:
        """Get current location for a driver."""
        
        data = await self.redis.hgetall(f"driver:{driver_id}:location")
        
        if not data:
            return None
        
        return (
            float(data[b"lat"]),
            float(data[b"lng"]),
            datetime.fromisoformat(data[b"timestamp"].decode())
        )
    
    async def find_nearby_drivers(
        self,
        lat: float,
        lng: float,
        radius_km: float,
        limit: int = 50
    ) -> List[str]:
        """Find drivers within radius using Redis GEO."""
        
        results = await self.redis.georadius(
            "driver_locations",
            lng, lat,
            radius_km,
            unit="km",
            count=limit,
            sort="ASC"  # Closest first
        )
        
        return [r.decode() for r in results]
    
    async def _flush_history(self):
        """Batch write location history to PostgreSQL."""
        
        if not self._history_buffer:
            return
        
        buffer = self._history_buffer
        self._history_buffer = []
        
        # Batch insert
        await self.db.executemany("""
            INSERT INTO driver_location_history 
            (driver_id, lat, lng, recorded_at)
            VALUES ($1, $2, $3, $4)
        """, [
            (h["driver_id"], h["lat"], h["lng"], h["timestamp"])
            for h in buffer
        ])

# services/tracking_publisher.py
# Applies: Week 6, Day 3 - registry, one pub/sub channel per gateway, resume by sequence number

class TrackingPublisher:
    """
    Pushes tracking events to the gateway holding each customer's app.

    - Location is coalesced: the gateway keeps only the newest point per order
      in each connection's outbound queue (coalesce_key), so a slow client
      never builds a backlog.
    - Status changes are written to the order's event log with a sequence
      number before anyone is told; the socket message is only a hint.
    """

    def __init__(self, registry, pubsub, order_events):
        self.registry = registry          # user -> [(device, gateway_id, epoch)]
        self.pubsub = pubsub              # sharded pub/sub, one channel per gateway
        self.events = order_events        # durable, (order_id, seq) unique

    async def on_location(self, order, lat, lng, eta_minutes):
        frame = {
            "type": "location",
            "order_id": order.order_id,
            "lat": lat, "lng": lng, "eta_minutes": eta_minutes,
            "coalesce_key": f"loc:{order.order_id}",   # newest replaces any queued point
        }
        await self._publish(order.customer_id, frame)

    async def on_status(self, order, status):
        seq = await self.events.append(order.order_id, status)    # durable first
        frame = {"type": "status", "order_id": order.order_id,
                 "seq": seq, "status": status}
        await self._publish(order.customer_id, frame)
        return seq

    async def resume(self, order_id, after_seq):
        """A reconnecting app sends its last seq; it gets what it missed."""
        return await self.events.read_after(order_id, after_seq)

    async def _publish(self, user_id, frame):
        for conn in await self.registry.live_connections(user_id):  # skips dead leases
            await self.pubsub.spublish(f"gw:{conn.gateway_id}", {"user": user_id, **frame})

# services/rate_limiter.py
# Applies: Week 1, Day 3 - Rate Limiting; Week 3, Day 3 - Backpressure

from datetime import datetime, timezone
from typing import Tuple
import uuid


class MultiLayerRateLimiter:
    """
    Rate limiting at multiple layers.
    
    Layers:
    1. Per-user: 10 orders/hour
    2. Per-IP: 100 requests/minute  
    3. Per-restaurant: 500 orders/hour (prevent overwhelming kitchen)
    4. Global: System capacity protection
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
        
        self.limits = {
            "user_orders": {"count": 10, "window": 3600},      # 10/hour
            "ip_requests": {"count": 100, "window": 60},       # 100/min
            "restaurant_orders": {"count": 500, "window": 3600}, # 500/hour
            "global_orders": {"count": 1000, "window": 1},     # 1000/sec
        }
    
    async def check_order_allowed(
        self,
        user_id: str,
        ip_address: str,
        restaurant_id: str
    ) -> Tuple[bool, str]:
        """
        Check if order is allowed by all rate limits.
        
        Returns (allowed, reason).
        """
        
        # Check user limit
        if not await self._check_limit(f"ratelimit:user:{user_id}", "user_orders"):
            return (False, "Too many orders. Please try again later.")
        
        # Check IP limit
        if not await self._check_limit(f"ratelimit:ip:{ip_address}", "ip_requests"):
            return (False, "Too many requests. Please slow down.")
        
        # Check restaurant limit
        if not await self._check_limit(f"ratelimit:restaurant:{restaurant_id}", "restaurant_orders"):
            return (False, "Restaurant is very busy. Please try again soon.")
        
        # Check global limit
        if not await self._check_limit("ratelimit:global:orders", "global_orders"):
            return (False, "System is busy. Please try again in a moment.")
        
        return (True, "")
    
    # Atomic sliding window: trim, count, and record the attempt only if
    # it is under the limit
    SLIDING_WINDOW = """
    redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, ARGV[2])
    if redis.call('ZCARD', KEYS[1]) >= tonumber(ARGV[3]) then
        return 0
    end
    redis.call('ZADD', KEYS[1], ARGV[1], ARGV[4])
    redis.call('EXPIRE', KEYS[1], ARGV[5])
    return 1
    """
    
    async def _check_limit(self, key: str, limit_type: str) -> bool:
        """Check if under rate limit using sliding window."""
        
        config = self.limits[limit_type]
        now = datetime.now(timezone.utc).timestamp()
        window_start = now - config["window"]
        
        # Check before ZADD: if rejected attempts were added too, a client
        # that keeps retrying would stay over the limit indefinitely
        allowed = await self.redis.eval(
            self.SLIDING_WINDOW, 1, key,
            now, window_start, config["count"],
            f"{now}:{uuid.uuid4().hex}",  # unique member per attempt
            config["window"]
        )
        
        return allowed == 1


class BackpressureManager:
    """
    Handles system backpressure during overload.
    
    Applies: Week 3, Day 3 - Backpressure
    """
    
    def __init__(self, redis_client, kafka_admin):
        self.redis = redis_client
        self.kafka = kafka_admin
    
    async def get_system_pressure(self) -> float:
        """
        Calculate current system pressure (0-1).
        
        Based on:
        - Kafka consumer lag
        - Worker queue depth
        - Database connection usage
        """
        
        # Get Kafka lag
        lag = await self._get_kafka_lag()
        max_acceptable_lag = 100000
        lag_pressure = min(1.0, lag / max_acceptable_lag)
        
        # Get worker queue depth
        queue_depth = await self.redis.llen("order_processing_queue")
        max_queue = 10000
        queue_pressure = min(1.0, queue_depth / max_queue)
        
        # Weighted average
        return (lag_pressure * 0.6) + (queue_pressure * 0.4)
    
    async def should_accept_order(self) -> Tuple[bool, str]:
        """
        Determine if we should accept new orders.
        
        Implements graceful degradation:
        - pressure < 0.7: Accept all
        - pressure 0.7-0.9: Accept, in degraded mode (skip non-essential features)
        - pressure > 0.9: Emergency mode
        """
        
        pressure = await self.get_system_pressure()
        
        if pressure < 0.7:
            return (True, "normal")
        
        if pressure < 0.9:
            return (True, "degraded")  # Maybe skip some features
        
        return (False, "System at capacity. Please try again shortly.")

# services/order_notifications.py
# Applies: Week 6, Days 1-3 - lanes, idempotency keys, inbox first, ack-timeout fallback to push

from datetime import timedelta

LANES = {
    "otp":                  "critical",
    "driver_offer":         "critical",       # worthless once the offer window has passed
    "order_confirmed":      "transactional",
    "driver_assigned":      "transactional",
    "driver_arriving":      "transactional",
    "delivered":            "transactional",
    "restaurant_new_order": "transactional",
    "promo":                "bulk",
}


async def on_order_event(evt, notify):
    lane = LANES[evt.kind]
    await notify.send(
        # named after the business event, so a retried event is the same notification
        idempotency_key=f"orders:{evt.order_id}:{evt.kind}",
        user_id=evt.recipient_id,
        priority=lane,
        template=evt.kind,
        variables=evt.variables,
        # one lock-screen entry per order: "arriving" replaces "assigned"
        collapse_key=f"order:{evt.order_id}" if lane == "transactional" else None,
        expires_at=evt.at + timedelta(seconds=15) if evt.kind == "driver_offer" else None,
    )
    # The platform writes the inbox item with the next per-user sequence number,
    # hints every live connection through the registry, and pushes via APNs or
    # FCM only if no device acks within a few seconds.