Event-Driven Food Delivery Order System.

# Transactional Outbox Implementation for Order Service

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


@dataclass
class OrderEvent:
    """Event to be published via outbox."""
    event_id: str
    event_type: str
    order_id: str
    payload: Dict[str, Any]
    created_at: datetime


class OrderService:
    """
    Order service with transactional outbox.
    Applies Day 2: Transactional Outbox Pattern.
    """
    
    def __init__(self, db_pool, audit_client):
        self.db = db_pool
        self.audit = audit_client
    
    async def place_order(
        self,
        customer_id: str,
        restaurant_id: str,
        items: list,
        delivery_address: dict,
        payment_method_id: str
    ) -> dict:
        """
        Place a new order.
        
        CRITICAL: Order creation and event publishing happen in
        the same database transaction. This guarantees:
        - If order is saved, event WILL be published
        - If transaction fails, neither is persisted
        """
        order_id = f"ord_{uuid.uuid4().hex[:12]}"
        
        # Calculate totals
        subtotal = sum(item['price'] * item['quantity'] for item in items)
        delivery_fee = 4.99
        total = subtotal + delivery_fee
        
        async with self.db.transaction() as tx:
            # 1. Create the order
            await tx.execute("""
                INSERT INTO orders (
                    id, customer_id, restaurant_id, items,
                    delivery_address, payment_method_id,
                    subtotal, delivery_fee, total,
                    status, created_at
                ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
            """, 
                order_id, customer_id, restaurant_id, json.dumps(items),
                json.dumps(delivery_address), payment_method_id,
                subtotal, delivery_fee, total,
                'placed', datetime.now(timezone.utc)
            )
            
            # 2. Write event to outbox (SAME TRANSACTION)
            event = OrderEvent(
                event_id=f"evt_{uuid.uuid4().hex[:12]}",
                event_type="order.placed",
                order_id=order_id,
                payload={
                    "order_id": order_id,
                    "customer_id": customer_id,
                    "restaurant_id": restaurant_id,
                    "items": items,
                    "total": total,
                    "payment_method_id": payment_method_id,
                    "delivery_address": delivery_address
                },
                created_at=datetime.now(timezone.utc)
            )
            
            await tx.execute("""
                INSERT INTO outbox (
                    event_id, event_type, aggregate_id, 
                    aggregate_type, payload, created_at
                ) VALUES ($1, $2, $3, $4, $5, $6)
            """,
                event.event_id, event.event_type, order_id,
                'order', json.dumps(event.payload), event.created_at
            )
            
            # 3. Write audit event (SAME TRANSACTION)
            await self.audit.log(
                conn=tx,  # same connection = same transaction
                action="order.placed",
                resource_type="order",
                resource_id=order_id,
                actor_id=customer_id,
                details={"total": total, "item_count": len(items)}
            )
        
        # Transaction committed — order and event are guaranteed persisted
        return {"order_id": order_id, "status": "placed", "total": total}


class OutboxPublisher:
    """
    Background process that publishes outbox events to Kafka.
    
    Uses CDC (Debezium) for low-latency, or polling as fallback.
    """
    
    def __init__(self, db_pool, kafka_producer):
        self.db = db_pool
        self.kafka = kafka_producer
    
    async def publish_pending(self, batch_size: int = 100) -> int:
        """Publish pending outbox events."""
        
        # Row locks from FOR UPDATE SKIP LOCKED last only as long as the transaction
        async with self.db.acquire() as conn, conn.transaction():
            # Fetch unpublished events
            rows = await conn.fetch("""
                SELECT id, event_id, event_type, aggregate_id, payload
                FROM outbox
                WHERE published_at IS NULL
                ORDER BY created_at
                LIMIT $1
                FOR UPDATE SKIP LOCKED
            """, batch_size)

            if not rows:
                return 0

            published_ids = []

            for row in rows:
                # Partition by order_id for ordering guarantee
                partition_key = row['aggregate_id'].encode('utf-8')

                await self.kafka.send_and_wait(
                    'orders.events',
                    key=partition_key,
                    value=row['payload'].encode('utf-8'),
                    headers=[
                        ('event_type', row['event_type'].encode()),
                        ('event_id', row['event_id'].encode())
                    ]
                )

                published_ids.append(row['id'])

            # Mark as published
            await conn.execute("""
                UPDATE outbox
                SET published_at = NOW()
                WHERE id = ANY($1)
            """, published_ids)

            return len(published_ids)

# Backpressure-Aware Consumer
# Applies Day 3: Backpressure and Flow Control

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


class Priority(Enum):
    CRITICAL = 1  # Payments, order state changes
    HIGH = 2      # Restaurant notifications, driver assignment
    NORMAL = 3    # Customer notifications
    LOW = 4       # Analytics, marketing


class BackpressureLevel(Enum):
    NONE = "none"
    WARNING = "warning"
    CRITICAL = "critical"
    EMERGENCY = "emergency"


@dataclass
class BackpressureConfig:
    """Configuration for backpressure handling."""
    lag_warning_threshold: int = 10_000
    lag_critical_threshold: int = 50_000
    lag_emergency_threshold: int = 100_000
    
    # Batch sizes by pressure level
    batch_size_normal: int = 10
    batch_size_warning: int = 50
    batch_size_critical: int = 200


class BackpressureAwareConsumer:
    """
    Kafka consumer with backpressure detection and response.
    
    Key behaviors:
    - Monitors consumer lag continuously
    - Adjusts batch size based on pressure
    - Sheds low-priority work under extreme pressure
    - Never sheds critical events
    """
    
    def __init__(
        self,
        kafka_consumer,
        processor,
        config: BackpressureConfig,
        metrics_client
    ):
        self.consumer = kafka_consumer
        self.processor = processor
        self.config = config
        self.metrics = metrics_client
        
        self._current_level = BackpressureLevel.NONE
        self._current_batch_size = config.batch_size_normal
    
    async def start(self):
        """Main consumer loop with backpressure handling."""
        
        # Start background monitoring
        asyncio.create_task(self._monitor_loop())
        
        batch = []
        
        async for message in self.consumer:
            batch.append(message)
            
            if len(batch) >= self._current_batch_size:
                await self._process_batch(batch)
                batch = []
    
    async def _monitor_loop(self):
        """Continuously monitor lag and adjust strategy."""
        while True:
            lag = await self._get_consumer_lag()
            
            # Determine pressure level
            if lag >= self.config.lag_emergency_threshold:
                new_level = BackpressureLevel.EMERGENCY
            elif lag >= self.config.lag_critical_threshold:
                new_level = BackpressureLevel.CRITICAL
            elif lag >= self.config.lag_warning_threshold:
                new_level = BackpressureLevel.WARNING
            else:
                new_level = BackpressureLevel.NONE
            
            # Adjust batch size
            if new_level != self._current_level:
                self._current_level = new_level
                self._adjust_batch_size()
                self.metrics.gauge('consumer.backpressure_level', new_level.value)
            
            await asyncio.sleep(1)
    
    def _adjust_batch_size(self):
        """Adjust batch size based on pressure level."""
        if self._current_level == BackpressureLevel.NONE:
            self._current_batch_size = self.config.batch_size_normal
        elif self._current_level == BackpressureLevel.WARNING:
            self._current_batch_size = self.config.batch_size_warning
        else:  # CRITICAL or EMERGENCY
            self._current_batch_size = self.config.batch_size_critical
    
    async def _process_batch(self, messages: list):
        """Process a batch of messages with priority awareness."""
        
        # Shed by priority: LOW at CRITICAL, NORMAL + LOW at EMERGENCY
        keep_up_to = {
            BackpressureLevel.CRITICAL: Priority.NORMAL,
            BackpressureLevel.EMERGENCY: Priority.HIGH,
        }.get(self._current_level)
        if keep_up_to:
            before = len(messages)
            messages = self._filter_by_priority(messages, min_priority=keep_up_to)
            self.metrics.increment('consumer.messages_shed', before - len(messages))
        
        # Process remaining messages
        for msg in messages:
            try:
                await self.processor.process(msg)
            except Exception as e:
                # Send to DLQ (Day 4)
                await self._send_to_dlq(msg, e)
        
        await self.consumer.commit()
    
    def _filter_by_priority(self, messages: list, min_priority: Priority) -> list:
        """Filter messages by priority. Never filter CRITICAL."""
        return [
            msg for msg in messages
            if self._get_priority(msg).value <= min_priority.value
        ]
    
    def _get_priority(self, message) -> Priority:
        """Determine message priority from event type."""
        # Kafka headers are a list of (str, bytes) tuples
        event_type = dict(message.headers or []).get('event_type', b'').decode()
        
        # CRITICAL: Never shed
        if event_type in ('payment.confirmed', 'payment.failed', 
                          'order.placed', 'order.cancelled'):
            return Priority.CRITICAL
        
        # HIGH: Kept even in EMERGENCY
        if event_type in ('restaurant.notified', 'driver.assigned'):
            return Priority.HIGH
        
        # NORMAL: Shed in EMERGENCY
        if event_type in ('notification.sent', 'driver.location_update'):
            return Priority.NORMAL
        
        # LOW: Shed in CRITICAL and EMERGENCY
        return Priority.LOW
    
    async def _get_consumer_lag(self) -> int:
        """Get current consumer lag."""
        # Implementation depends on Kafka client
        return await self.consumer.get_lag()
    
    async def _send_to_dlq(self, message, error: Exception):
        """Send failed message to dead letter queue (Day 4)."""
        # Implementation in Deep Dive 3
        pass

# Dead Letter Queue Handler for Orders
# Applies Day 4: Dead Letters and Poison Pills

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from enum import Enum
import json
import traceback


class ErrorCategory(Enum):
    TRANSIENT = "transient"      # Retry will likely help
    PERMANENT = "permanent"      # Retry won't help, need action
    POISON = "poison"            # Crashes consumer, quarantine


class FailureType(Enum):
    # Payment failures
    PAYMENT_DECLINED = "payment_declined"
    PAYMENT_TIMEOUT = "payment_timeout"
    PAYMENT_FRAUD = "payment_fraud"
    
    # Restaurant failures  
    RESTAURANT_CLOSED = "restaurant_closed"
    RESTAURANT_REJECTED = "restaurant_rejected"
    ITEM_UNAVAILABLE = "item_unavailable"
    
    # Processing failures
    INVALID_MESSAGE = "invalid_message"
    SERVICE_UNAVAILABLE = "service_unavailable"
    UNKNOWN = "unknown"


@dataclass
class DeadLetter:
    """A failed order event with context for investigation."""
    event_id: str
    order_id: str
    event_type: str
    original_payload: Dict[str, Any]
    
    failure_type: FailureType
    error_category: ErrorCategory
    error_message: str
    stack_trace: Optional[str]
    
    retry_count: int
    first_failed_at: datetime
    last_failed_at: datetime
    
    # For resolution
    resolved: bool = False
    resolution: Optional[str] = None
    resolved_at: Optional[datetime] = None


class OrderDLQHandler:
    """
    Handles failed order events with intelligent routing.
    
    Key behaviors:
    - Classifies errors into transient/permanent/poison
    - Transient: Retry with exponential backoff
    - Permanent: Route to compensation logic
    - Poison: Quarantine immediately
    """
    
    # Error classification rules
    TRANSIENT_ERRORS = [
        FailureType.PAYMENT_TIMEOUT,
        FailureType.SERVICE_UNAVAILABLE,
    ]
    
    PERMANENT_ERRORS = [
        FailureType.PAYMENT_DECLINED,
        FailureType.PAYMENT_FRAUD,
        FailureType.RESTAURANT_CLOSED,
        FailureType.RESTAURANT_REJECTED,
        FailureType.ITEM_UNAVAILABLE,
    ]
    
    POISON_ERRORS = [
        FailureType.INVALID_MESSAGE,
    ]
    
    def __init__(
        self,
        kafka_producer,
        db_pool,
        notification_service,
        config
    ):
        self.kafka = kafka_producer
        self.db = db_pool
        self.notifications = notification_service
        self.config = config
    
    async def handle_failure(
        self,
        event: Dict[str, Any],
        error: Exception,
        retry_count: int = 0
    ) -> None:
        """
        Handle a failed order event.
        
        Routes to appropriate handling based on error type.
        """
        # Classify the error
        failure_type = self._classify_error(error)
        category = self._get_category(failure_type)
        
        order_id = event.get('order_id', 'unknown')
        
        if category == ErrorCategory.POISON:
            # POISON: Quarantine immediately, never retry
            await self._quarantine_poison(event, error, failure_type)
            await self._alert_oncall(event, error, "Poison pill detected")
            
        elif category == ErrorCategory.PERMANENT:
            # PERMANENT: Don't retry, execute compensation
            await self._handle_permanent_failure(event, error, failure_type)
            
        elif retry_count < self.config.max_retries:
            # TRANSIENT: Retry with backoff
            await self._schedule_retry(event, retry_count + 1)
            
        else:
            # Exhausted retries: Move to DLQ
            await self._send_to_dlq(event, error, failure_type, retry_count)
            await self._notify_customer_failure(order_id)
    
    def _classify_error(self, error: Exception) -> FailureType:
        """Classify exception into failure type."""
        error_str = str(error).lower()
        
        # Payment errors
        if 'card_declined' in error_str:
            return FailureType.PAYMENT_DECLINED
        if 'timeout' in error_str and 'payment' in error_str:
            return FailureType.PAYMENT_TIMEOUT
        if 'fraud' in error_str:
            return FailureType.PAYMENT_FRAUD
        
        # Restaurant errors
        if 'restaurant_closed' in error_str:
            return FailureType.RESTAURANT_CLOSED
        if 'rejected' in error_str:
            return FailureType.RESTAURANT_REJECTED
        if 'out_of_stock' in error_str or 'unavailable' in error_str:
            return FailureType.ITEM_UNAVAILABLE
        
        # Processing errors
        if isinstance(error, (json.JSONDecodeError, KeyError, TypeError)):
            return FailureType.INVALID_MESSAGE
        if 'service unavailable' in error_str or '503' in error_str:
            return FailureType.SERVICE_UNAVAILABLE
        
        return FailureType.UNKNOWN
    
    def _get_category(self, failure_type: FailureType) -> ErrorCategory:
        """Get error category for failure type."""
        if failure_type in self.TRANSIENT_ERRORS:
            return ErrorCategory.TRANSIENT
        if failure_type in self.PERMANENT_ERRORS:
            return ErrorCategory.PERMANENT
        if failure_type in self.POISON_ERRORS:
            return ErrorCategory.POISON
        return ErrorCategory.TRANSIENT  # Default to retry for unknown
    
    async def _schedule_retry(self, event: Dict, retry_count: int):
        """Schedule retry with exponential backoff."""
        # Determine delay based on retry count
        delays = ['1m', '5m', '15m']
        delay = delays[min(retry_count - 1, len(delays) - 1)]
        
        retry_topic = f"orders.retry.{delay}"
        
        # Add retry metadata
        event['_retry_count'] = retry_count
        event['_retry_at'] = datetime.now(timezone.utc).isoformat()
        
        await self.kafka.send(
            retry_topic,
            key=event['order_id'].encode(),
            value=json.dumps(event).encode()
        )
    
    async def _handle_permanent_failure(
        self,
        event: Dict,
        error: Exception,
        failure_type: FailureType
    ):
        """Handle permanent failures with compensation logic."""
        order_id = event.get('order_id')
        
        if failure_type == FailureType.PAYMENT_DECLINED:
            # Notify customer, mark order as payment_failed
            await self._update_order_status(order_id, 'payment_failed')
            await self.notifications.send(
                user_id=event['customer_id'],
                template='payment_declined',
                data={'order_id': order_id}
            )
            
        elif failure_type == FailureType.RESTAURANT_REJECTED:
            # Try alternative restaurant or refund
            alternative = await self._find_alternative_restaurant(event)
            if alternative:
                await self._reroute_order(order_id, alternative)
            else:
                await self._refund_order(order_id)
                
        elif failure_type == FailureType.ITEM_UNAVAILABLE:
            # Notify customer, offer alternatives
            await self.notifications.send(
                user_id=event['customer_id'],
                template='item_unavailable',
                data={'order_id': order_id, 'items': event.get('unavailable_items')}
            )
        
        # Always record in DLQ for tracking
        await self._send_to_dlq(event, error, failure_type, 0, resolved=True)
    
    async def _quarantine_poison(
        self,
        event: Dict,
        error: Exception,
        failure_type: FailureType
    ):
        """Quarantine poison pill - never retry."""
        await self.kafka.send(
            'orders.dlq.poison',
            key=event.get('order_id', 'unknown').encode(),
            value=json.dumps({
                'event': event,
                'error': str(error),
                'stack_trace': traceback.format_exc(),
                'quarantined_at': datetime.now(timezone.utc).isoformat()
            }).encode()
        )
    
    async def _send_to_dlq(
        self,
        event: Dict,
        error: Exception,
        failure_type: FailureType,
        retry_count: int,
        resolved: bool = False
    ):
        """Send to dead letter queue for investigation."""
        dead_letter = DeadLetter(
            event_id=event.get('event_id', 'unknown'),
            order_id=event.get('order_id', 'unknown'),
            event_type=event.get('event_type', 'unknown'),
            original_payload=event,
            failure_type=failure_type,
            error_category=self._get_category(failure_type),
            error_message=str(error),
            stack_trace=traceback.format_exc(),
            retry_count=retry_count,
            first_failed_at=datetime.now(timezone.utc),
            last_failed_at=datetime.now(timezone.utc),
            resolved=resolved
        )
        
        await self.kafka.send(
            'orders.dlq',
            key=event.get('order_id', 'unknown').encode(),
            value=json.dumps(dead_letter.__dict__, default=str).encode()
        )

# Audit Log System for Order Disputes
# Applies Day 5: Audit Log System

from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List
from enum import Enum
import json
import hashlib


class AuditAction(Enum):
    # Order lifecycle
    ORDER_PLACED = "order.placed"
    ORDER_CANCELLED = "order.cancelled"
    ORDER_COMPLETED = "order.completed"
    
    # Payment
    PAYMENT_INITIATED = "payment.initiated"
    PAYMENT_CONFIRMED = "payment.confirmed"
    PAYMENT_FAILED = "payment.failed"
    PAYMENT_REFUNDED = "payment.refunded"
    
    # Restaurant
    RESTAURANT_NOTIFIED = "restaurant.notified"
    RESTAURANT_ACCEPTED = "restaurant.accepted"
    RESTAURANT_REJECTED = "restaurant.rejected"
    
    # Driver
    DRIVER_ASSIGNED = "driver.assigned"
    DRIVER_PICKED_UP = "driver.picked_up"
    DRIVER_DELIVERED = "driver.delivered"
    
    # Support
    SUPPORT_INTERVENTION = "support.intervention"
    DISPUTE_OPENED = "dispute.opened"
    DISPUTE_RESOLVED = "dispute.resolved"


@dataclass
class AuditEvent:
    """
    Immutable audit event for order tracking.
    
    Contains all context needed to understand what happened,
    who did it, and when.
    """
    # Identity
    event_id: str
    timestamp: datetime
    
    # What happened
    action: AuditAction
    order_id: str
    
    # Who did it
    actor_type: str  # 'customer', 'system', 'restaurant', 'driver', 'support'
    actor_id: str
    
    # State change
    previous_state: Optional[str]
    new_state: str
    
    # Details
    details: Dict[str, Any]
    
    # Context for debugging
    service_name: str
    correlation_id: str
    
    # Integrity
    previous_hash: Optional[str] = None
    event_hash: Optional[str] = None
    
    def compute_hash(self) -> str:
        """Compute hash for tamper detection."""
        content = json.dumps({
            'event_id': self.event_id,
            'timestamp': self.timestamp.isoformat(),
            'action': self.action.value,
            'order_id': self.order_id,
            'actor_type': self.actor_type,
            'actor_id': self.actor_id,
            'previous_state': self.previous_state,
            'new_state': self.new_state,
            'details': self.details,  # hash every field a tamperer might change
            'previous_hash': self.previous_hash
        }, sort_keys=True, default=str)
        return hashlib.sha256(content.encode()).hexdigest()


class AuditService:
    """
    Service for recording and querying audit events.
    
    Key guarantees:
    - Events are immutable once written
    - Hash chain provides tamper evidence
    - Fast queries by order_id for support
    """
    
    def __init__(self, kafka_producer, db_pool):
        self.kafka = kafka_producer
        self.db = db_pool
        # One chain per order, so verify_integrity(order_id) can check it.
        # (In production, load an order's last hash from the store.)
        self._last_hash: Dict[str, str] = {}
    
    async def record(
        self,
        action: AuditAction,
        order_id: str,
        actor_type: str,
        actor_id: str,
        previous_state: Optional[str],
        new_state: str,
        details: Dict[str, Any],
        service_name: str,
        correlation_id: str
    ) -> str:
        """
        Record an audit event.
        
        Returns the event_id.
        """
        import uuid
        
        event = AuditEvent(
            event_id=f"aud_{uuid.uuid4().hex[:12]}",
            timestamp=datetime.now(timezone.utc),
            action=action,
            order_id=order_id,
            actor_type=actor_type,
            actor_id=actor_id,
            previous_state=previous_state,
            new_state=new_state,
            details=details,
            service_name=service_name,
            correlation_id=correlation_id,
            previous_hash=self._last_hash.get(order_id)
        )
        
        # Compute hash for chain
        event.event_hash = event.compute_hash()
        self._last_hash[order_id] = event.event_hash
        
        # Publish to Kafka
        await self.kafka.send(
            'audit.events',
            key=order_id.encode(),
            value=json.dumps(asdict(event), default=str).encode()
        )
        
        return event.event_id
    
    async def get_order_timeline(self, order_id: str) -> List[Dict]:
        """
        Get complete timeline for an order.
        
        Used by customer support for dispute resolution.
        """
        rows = await self.db.fetch("""
            SELECT 
                event_id,
                timestamp,
                action,
                actor_type,
                actor_id,
                previous_state,
                new_state,
                details
            FROM audit_events
            WHERE order_id = $1
            ORDER BY timestamp ASC
        """, order_id)
        
        return [dict(row) for row in rows]
    
    async def verify_integrity(self, order_id: str) -> Dict[str, Any]:
        """
        Verify hash chain integrity for an order.
        
        Used for dispute verification - proves logs weren't tampered.
        """
        rows = await self.db.fetch("""
            SELECT event_id, event_hash, previous_hash
            FROM audit_events
            WHERE order_id = $1
            ORDER BY timestamp ASC
        """, order_id)
        
        issues = []
        prev_hash = None
        
        for row in rows:
            if row['previous_hash'] != prev_hash:
                issues.append({
                    'event_id': row['event_id'],
                    'issue': 'Chain break detected',
                    'expected': prev_hash,
                    'actual': row['previous_hash']
                })
            prev_hash = row['event_hash']
        
        return {
            'order_id': order_id,
            'events_checked': len(rows),
            'integrity': 'valid' if not issues else 'compromised',
            'issues': issues
        }


class SupportDashboard:
    """
    Support dashboard for dispute resolution.
    
    Provides easy access to order history and investigation tools.
    """
    
    def __init__(self, audit_service: AuditService, order_service):
        self.audit = audit_service
        self.orders = order_service
    
    async def investigate_order(self, order_id: str) -> Dict[str, Any]:
        """
        Get complete investigation package for an order.
        
        Used by support agents during disputes.
        """
        # Get current order state
        order = await self.orders.get_order(order_id)
        
        # Get complete timeline
        timeline = await self.audit.get_order_timeline(order_id)
        
        # Verify integrity
        integrity = await self.audit.verify_integrity(order_id)
        
        # Get related DLQ entries if any
        dlq_entries = await self._get_dlq_entries(order_id)
        
        return {
            'order': order,
            'timeline': timeline,
            'integrity': integrity,
            'failures': dlq_entries,
            'summary': self._generate_summary(timeline)
        }
    
    def _generate_summary(self, timeline: List[Dict]) -> str:
        """Generate human-readable summary of order events."""
        if not timeline:
            return "No events found for this order."
        
        summary_lines = []
        for event in timeline:
            ts = event['timestamp'].strftime('%H:%M:%S')
            action = event['action'].replace('.', ' ').title()
            actor = f"{event['actor_type']}:{event['actor_id'][:8]}"
            summary_lines.append(f"{ts} - {action} by {actor}")
        
        return "\n".join(summary_lines)

-- "I never got my food but was charged"
SELECT 
    timestamp,
    action,
    actor_type,
    new_state,
    details
FROM audit_events
WHERE order_id = 'ord_abc123'
ORDER BY timestamp;

-- Result shows:
-- 18:30:01 - order.placed (customer)
-- 18:30:02 - payment.confirmed (system) ← Payment went through
-- 18:30:05 - restaurant.notified (system)
-- 18:31:00 - restaurant.accepted (restaurant)
-- 18:35:00 - driver.assigned (system)
-- 18:50:00 - driver.picked_up (driver)
-- 19:15:00 - ??? (no delivery event!)
-- 
-- Gap after pickup suggests driver issue


-- "The restaurant says they never got my order"
SELECT * FROM audit_events
WHERE order_id = 'ord_xyz789'
AND action = 'restaurant.notified';

-- Shows notification was sent at 18:30:05
-- Check details for delivery confirmation