Audit Log System.

# Audit Event Schema
# The foundation of the entire system

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


class ActionResult(Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    DENIED = "denied"   # Authorization failure


class ActorType(Enum):
    USER = "user"
    SERVICE = "service"
    SYSTEM = "system"   # Automated processes


@dataclass
class Actor:
    """Who performed the action."""
    id: str
    type: ActorType
    email: Optional[str] = None
    name: Optional[str] = None
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    
    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "type": self.type.value,
            "email": self.email,
            "name": self.name,
            "ip_address": self.ip_address,
            "user_agent": self.user_agent
        }


@dataclass
class Resource:
    """What was acted upon."""
    type: str           # e.g., "user", "document", "account"
    id: str             # Resource identifier
    name: Optional[str] = None  # Human-readable name
    
    def to_dict(self) -> Dict:
        return {
            "type": self.type,
            "id": self.id,
            "name": self.name
        }


@dataclass
class AuditEvent:
    """
    A single audit log entry.
    
    Captures: WHO did WHAT to WHICH RESOURCE, WHEN, and HOW.
    """
    # Core required fields
    actor: Actor
    action: str              # e.g., "document.create", "user.login"
    resource: Resource
    result: ActionResult
    
    # Auto-generated
    event_id: str = field(default_factory=lambda: f"evt_{uuid.uuid4().hex[:12]}")
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    
    # Optional details
    changes: Optional[Dict[str, Any]] = None    # For updates: old/new values
    reason: Optional[str] = None                 # Why action was taken
    error_message: Optional[str] = None          # If result is failure
    
    # Context
    request_id: Optional[str] = None
    session_id: Optional[str] = None
    correlation_id: Optional[str] = None        # For tracing across services
    
    # For tamper evidence
    previous_hash: Optional[str] = None         # Hash of previous event
    
    def to_dict(self) -> Dict:
        return {
            "event_id": self.event_id,
            "timestamp": self.timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
            "actor": self.actor.to_dict(),
            "action": self.action,
            "resource": self.resource.to_dict(),
            "result": self.result.value,
            "changes": self.changes,
            "reason": self.reason,
            "error_message": self.error_message,
            "context": {
                "request_id": self.request_id,
                "session_id": self.session_id,
                "correlation_id": self.correlation_id
            },
            "integrity": {
                "previous_hash": self.previous_hash
            }
        }
    
    def to_json(self) -> str:
        return json.dumps(self.to_dict(), default=str)


# Example usage
def example():
    event = AuditEvent(
        actor=Actor(
            id="usr_456",
            type=ActorType.USER,
            email="alice@company.com",
            ip_address="192.168.1.100"
        ),
        action="document.update",
        resource=Resource(
            type="document",
            id="doc_789",
            name="Q4 Financial Report"
        ),
        result=ActionResult.SUCCESS,
        changes={
            "title": {"old": "Draft Report", "new": "Q4 Financial Report"},
            "status": {"old": "draft", "new": "published"}
        },
        request_id="req_abc123"
    )
    
    print(event.to_json())

# Simple Audit Client
# WARNING: Not production-ready - for learning only

from typing import Optional
from contextlib import contextmanager


class SimpleAuditClient:
    """
    Basic audit client that writes to outbox.
    Application code uses this to log audit events.
    """
    
    def __init__(self, db_session):
        self.db = db_session
    
    def log(self, event: AuditEvent) -> None:
        """
        Log an audit event.
        
        Writes to outbox table in the current transaction.
        """
        self.db.execute(
            """
            INSERT INTO audit_outbox (
                event_id, 
                event_type,
                payload, 
                created_at
            ) VALUES (%s, %s, %s, %s)
            """,
            (
                event.event_id,
                event.action,
                event.to_json(),
                event.timestamp
            )
        )
    
    @contextmanager
    def capture(
        self,
        actor: Actor,
        action: str,
        resource: Resource,
        request_id: Optional[str] = None
    ):
        """
        Context manager that automatically logs success or failure.
        
        Usage:
            with audit.capture(actor, "document.delete", resource) as event:
                delete_document(doc_id)
                event.changes = {"deleted": True}
        """
        event = AuditEvent(
            actor=actor,
            action=action,
            resource=resource,
            result=ActionResult.SUCCESS,  # Optimistic
            request_id=request_id
        )
        
        try:
            yield event
            # If we get here, action succeeded
            event.result = ActionResult.SUCCESS
        except PermissionError as e:
            event.result = ActionResult.DENIED
            event.error_message = str(e)
            raise
        except Exception as e:
            event.result = ActionResult.FAILURE
            event.error_message = str(e)
            raise
        finally:
            # Always log, regardless of outcome
            self.log(event)


# Usage in application code
async def delete_document(doc_id: str, user: User, audit: SimpleAuditClient):
    """Example: deleting a document with audit logging."""
    
    # Get document for audit trail
    doc = await db.get_document(doc_id)
    
    actor = Actor(
        id=user.id,
        type=ActorType.USER,
        email=user.email
    )
    
    resource = Resource(
        type="document",
        id=doc_id,
        name=doc.title
    )
    
    with audit.capture(actor, "document.delete", resource) as event:
        # The actual business logic
        await db.delete_document(doc_id)
        
        # Capture what was deleted
        event.changes = {
            "title": doc.title,
            "owner": doc.owner_id,
            "size_bytes": doc.size
        }
    
    # If we get here, both delete and audit succeeded
    return {"status": "deleted"}

-- Audit Outbox Table (for transactional outbox pattern)
CREATE TABLE audit_outbox (
    id BIGSERIAL PRIMARY KEY,
    event_id VARCHAR(64) NOT NULL UNIQUE,
    event_type VARCHAR(128) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at TIMESTAMPTZ,
    
    -- For ordering and deduplication
    CONSTRAINT audit_outbox_event_unique UNIQUE (event_id)
);

CREATE INDEX idx_audit_outbox_unpublished 
    ON audit_outbox (created_at) 
    WHERE published_at IS NULL;


-- Hot Storage: Recent audit events (90 days)
-- Using TimescaleDB for time-series optimization
CREATE TABLE audit_events (
    event_id VARCHAR(64) NOT NULL,
    timestamp TIMESTAMPTZ NOT NULL,
    
    -- Actor
    actor_id VARCHAR(64) NOT NULL,
    actor_type VARCHAR(32) NOT NULL,
    actor_email VARCHAR(256),
    actor_ip VARCHAR(45),
    
    -- Action
    action VARCHAR(128) NOT NULL,
    result VARCHAR(32) NOT NULL,
    
    -- Resource
    resource_type VARCHAR(64) NOT NULL,
    resource_id VARCHAR(64) NOT NULL,
    resource_name VARCHAR(256),
    
    -- Details (flexible)
    changes JSONB,
    error_message TEXT,
    
    -- Context
    request_id VARCHAR(64),
    session_id VARCHAR(64),
    correlation_id VARCHAR(64),
    
    -- Integrity
    previous_hash VARCHAR(64),
    event_hash VARCHAR(64),
    
    -- Full event for archive
    raw_event JSONB NOT NULL,

    -- Hypertables require the time column in every unique key
    PRIMARY KEY (event_id, timestamp)
);

-- Convert to TimescaleDB hypertable
SELECT create_hypertable('audit_events', 'timestamp');

-- Indexes for common queries
CREATE INDEX idx_audit_events_actor ON audit_events (actor_id, timestamp DESC);
CREATE INDEX idx_audit_events_resource ON audit_events (resource_type, resource_id, timestamp DESC);
CREATE INDEX idx_audit_events_action ON audit_events (action, timestamp DESC);

-- Composite index for investigation queries
CREATE INDEX idx_audit_events_investigation 
    ON audit_events (actor_id, resource_type, action, timestamp DESC);


-- DLQ for failed audit events
CREATE TABLE audit_dlq (
    id BIGSERIAL PRIMARY KEY,
    event_id VARCHAR(64) NOT NULL,
    original_payload JSONB NOT NULL,
    error_type VARCHAR(128) NOT NULL,
    error_message TEXT,
    stack_trace TEXT,
    retry_count INT NOT NULL DEFAULT 0,
    first_failed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    last_failed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    resolved_at TIMESTAMPTZ,
    resolution VARCHAR(32),  -- 'replayed', 'discarded', 'manual'
    
    CONSTRAINT audit_dlq_event_unique UNIQUE (event_id)
);

CREATE INDEX idx_audit_dlq_unresolved 
    ON audit_dlq (first_failed_at) 
    WHERE resolved_at IS NULL;

# Production Audit Log System
# Combines all Week 3 patterns

import asyncio
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, List
from enum import Enum
import logging

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class AuditConfig:
    """Configuration for the audit system."""
    
    # Kafka
    kafka_topic: str = "audit-events"
    kafka_partitions: int = 32
    kafka_replication: int = 3
    
    # Outbox
    outbox_poll_interval: float = 0.1  # 100ms
    outbox_batch_size: int = 1000
    
    # Consumer
    consumer_batch_size: int = 500
    consumer_batch_timeout: float = 1.0
    
    # Backpressure
    lag_warning_threshold: int = 10_000
    lag_critical_threshold: int = 100_000
    
    # DLQ
    max_retries: int = 3
    
    # Integrity
    enable_hash_chain: bool = True


# =============================================================================
# Audit Client (Application-Side)
# =============================================================================

class AuditClient:
    """
    Production audit client with transactional outbox.
    
    Usage in application code:
        async with db.transaction():
            await business_logic()
            await audit.log(event)  # Same transaction!
    """
    
    def __init__(self, db_pool, config: AuditConfig):
        self.db = db_pool
        self.config = config
        self._actor_context: Dict[str, Actor] = {}  # Per-request context
    
    def set_context(self, request_id: str, actor: Actor):
        """Set actor context for current request (call in middleware)."""
        self._actor_context[request_id] = actor
    
    def clear_context(self, request_id: str):
        """Clear context after request completes."""
        self._actor_context.pop(request_id, None)
    
    def get_actor(self, request_id: str) -> Optional[Actor]:
        """Get actor for current request."""
        return self._actor_context.get(request_id)
    
    async def log(
        self,
        action: str,
        resource: Resource,
        result: ActionResult = ActionResult.SUCCESS,
        changes: Optional[Dict] = None,
        reason: Optional[str] = None,
        error_message: Optional[str] = None,
        request_id: Optional[str] = None,
        actor: Optional[Actor] = None  # Override context
    ) -> str:
        """
        Log an audit event to the outbox.
        
        MUST be called within a database transaction to ensure
        atomicity with the business operation.
        
        Returns: event_id
        """
        # Get actor from context if not provided
        if actor is None and request_id:
            actor = self.get_actor(request_id)
        
        if actor is None:
            raise ValueError("Actor must be provided or set in context")
        
        event = AuditEvent(
            actor=actor,
            action=action,
            resource=resource,
            result=result,
            changes=changes,
            reason=reason,
            error_message=error_message,
            request_id=request_id
        )
        
        # Write to outbox
        await self.db.execute(
            """
            INSERT INTO audit_outbox (event_id, event_type, payload, created_at)
            VALUES ($1, $2, $3, $4)
            """,
            event.event_id,
            event.action,
            event.to_json(),
            event.timestamp
        )
        
        logger.debug(f"Audit event logged to outbox: {event.event_id}")
        return event.event_id
    
    async def log_success(
        self,
        action: str,
        resource: Resource,
        changes: Optional[Dict] = None,
        **kwargs
    ) -> str:
        """Convenience method for successful actions."""
        return await self.log(
            action=action,
            resource=resource,
            result=ActionResult.SUCCESS,
            changes=changes,
            **kwargs
        )
    
    async def log_failure(
        self,
        action: str,
        resource: Resource,
        error_message: str,
        **kwargs
    ) -> str:
        """Convenience method for failed actions."""
        return await self.log(
            action=action,
            resource=resource,
            result=ActionResult.FAILURE,
            error_message=error_message,
            **kwargs
        )


# =============================================================================
# Outbox Publisher (Background Process)
# =============================================================================

class OutboxPublisher:
    """
    Publishes audit events from outbox to Kafka.
    
    Day 2 pattern: Transactional Outbox
    """
    
    def __init__(
        self,
        db_pool,
        kafka_producer,
        config: AuditConfig
    ):
        self.db = db_pool
        self.kafka = kafka_producer
        self.config = config
        self._running = False
        
        # Metrics
        self.events_published = 0
        self.publish_errors = 0
    
    async def start(self):
        """Start the publisher loop."""
        self._running = True
        logger.info("Outbox publisher started")
        
        while self._running:
            try:
                count = await self._publish_batch()
                if count == 0:
                    await asyncio.sleep(self.config.outbox_poll_interval)
            except Exception as e:
                logger.error(f"Publisher error: {e}")
                self.publish_errors += 1
                await asyncio.sleep(1)  # Back off on error
    
    async def stop(self):
        """Stop the publisher."""
        self._running = False
        logger.info(f"Outbox publisher stopped. Published: {self.events_published}")
    
    async def _publish_batch(self) -> int:
        """Publish a batch of events from outbox."""
        # FOR UPDATE SKIP LOCKED only holds its row locks inside a transaction
        async with self.db.acquire() as conn, conn.transaction():
            # Fetch unpublished events
            rows = await conn.fetch(
                """
                SELECT id, event_id, event_type, payload
                FROM audit_outbox
                WHERE published_at IS NULL
                ORDER BY created_at
                LIMIT $1
                FOR UPDATE SKIP LOCKED
                """,
                self.config.outbox_batch_size
            )

            if not rows:
                return 0

            # Publish to Kafka
            published_ids = []
            for row in rows:
                try:
                    # Partition by actor_id for ordering
                    payload = json.loads(row['payload'])
                    actor_id = payload.get('actor', {}).get('id', '')
                    partition_key = actor_id.encode('utf-8')

                    await self.kafka.send_and_wait(
                        self.config.kafka_topic,
                        key=partition_key,
                        value=row['payload'].encode('utf-8')
                    )

                    published_ids.append(row['id'])
                    self.events_published += 1

                except Exception as e:
                    logger.error(f"Failed to publish event {row['event_id']}: {e}")
                    self.publish_errors += 1

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

            return len(published_ids)


# =============================================================================
# Audit Consumer (Writes to Storage)
# =============================================================================

class AuditConsumer:
    """
    Consumes audit events from Kafka and writes to storage.
    
    Day 3 pattern: Backpressure handling
    Day 4 pattern: Dead letter queue
    """
    
    def __init__(
        self,
        kafka_consumer,
        db_pool,
        dlq_handler,
        config: AuditConfig,
        metrics_client
    ):
        self.kafka = kafka_consumer
        self.db = db_pool
        self.dlq = dlq_handler
        self.config = config
        self.metrics = metrics_client
        
        self._running = False
        # NOTE: one chain per consumer only works with a single writer.
        # With N partitions/consumers, keep one chain per partition and
        # verify each chain separately.
        self._last_hash: Optional[str] = None
        
        # Metrics
        self.events_processed = 0
        self.events_failed = 0
    
    async def start(self):
        """Start consuming."""
        self._running = True
        logger.info("Audit consumer started")
        
        # Load last hash for chain integrity
        if self.config.enable_hash_chain:
            self._last_hash = await self._get_last_hash()
        
        batch = []
        batch_start = datetime.now(timezone.utc)
        
        async for msg in self.kafka:
            if not self._running:
                break
            
            batch.append(msg)
            
            # Process batch if full or timeout
            should_flush = (
                len(batch) >= self.config.consumer_batch_size or
                (datetime.now(timezone.utc) - batch_start).total_seconds() > self.config.consumer_batch_timeout
            )
            
            if should_flush and batch:
                await self._process_batch(batch)
                batch = []
                batch_start = datetime.now(timezone.utc)
    
    async def _process_batch(self, messages: List) -> None:
        """Process a batch of messages."""
        events_to_insert = []
        
        for msg in messages:
            try:
                event_dict = json.loads(msg.value.decode('utf-8'))
                
                # Add hash chain
                if self.config.enable_hash_chain:
                    event_dict['integrity']['previous_hash'] = self._last_hash
                    event_hash = self._compute_hash(event_dict)
                    event_dict['integrity']['event_hash'] = event_hash
                    self._last_hash = event_hash
                
                events_to_insert.append(event_dict)
                
            except Exception as e:
                logger.error(f"Failed to process event: {e}")
                await self.dlq.handle_failure(msg, e)
                self.events_failed += 1
        
        # Batch insert
        if events_to_insert:
            try:
                await self._batch_insert(events_to_insert)
                self.events_processed += len(events_to_insert)
                
                # Commit offsets
                await self.kafka.commit()
                
            except Exception as e:
                logger.error(f"Batch insert failed: {e}")
                # Send all to DLQ
                for msg in messages:
                    await self.dlq.handle_failure(msg, e)
                self.events_failed += len(messages)
    
    async def _batch_insert(self, events: List[Dict]) -> None:
        """Insert batch of events to database."""
        # Build batch insert
        values = []
        for e in events:
            values.append((
                e['event_id'],
                datetime.fromisoformat(e['timestamp'].replace('Z', '+00:00')),
                e['actor']['id'],
                e['actor']['type'],
                e['actor'].get('email'),
                e['actor'].get('ip_address'),
                e['action'],
                e['result'],
                e['resource']['type'],
                e['resource']['id'],
                e['resource'].get('name'),
                json.dumps(e.get('changes')) if e.get('changes') else None,
                e.get('error_message'),
                e['context'].get('request_id'),
                e['context'].get('session_id'),
                e['context'].get('correlation_id'),
                e['integrity'].get('previous_hash'),
                e['integrity'].get('event_hash'),
                json.dumps(e)
            ))
        
        await self.db.executemany(
            """
            INSERT INTO audit_events (
                event_id, timestamp, actor_id, actor_type, actor_email, actor_ip,
                action, result, resource_type, resource_id, resource_name,
                changes, error_message, request_id, session_id, correlation_id,
                previous_hash, event_hash, raw_event
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
            ON CONFLICT (event_id, timestamp) DO NOTHING
            """,
            values
        )
    
    def _compute_hash(self, event: Dict) -> str:
        """Compute hash for tamper evidence."""
        # Canonical JSON for consistent hashing
        canonical = json.dumps(event, sort_keys=True, separators=(',', ':'))
        return hashlib.sha256(canonical.encode()).hexdigest()
    
    async def _get_last_hash(self) -> Optional[str]:
        """Get hash of last event for chain integrity."""
        row = await self.db.fetchrow(
            """
            SELECT event_hash
            FROM audit_events
            ORDER BY timestamp DESC
            LIMIT 1
            """
        )
        return row['event_hash'] if row else None


# =============================================================================
# Archive Writer (Cold Storage)
# =============================================================================

class ArchiveWriter:
    """
    Archives old audit events to S3 in Parquet format.
    Runs as a daily job.
    """
    
    def __init__(self, db_pool, s3_client, config: AuditConfig):
        self.db = db_pool
        self.s3 = s3_client
        self.config = config
    
    async def archive_day(self, date: datetime) -> Dict[str, Any]:
        """Archive all events from a specific day."""
        import pyarrow as pa
        import pyarrow.parquet as pq
        
        # Fetch events for the day
        start = date.replace(hour=0, minute=0, second=0, microsecond=0)
        end = start + timedelta(days=1)
        
        rows = await self.db.fetch(
            """
            SELECT raw_event
            FROM audit_events
            WHERE timestamp >= $1 AND timestamp < $2
            ORDER BY timestamp
            """,
            start, end
        )
        
        if not rows:
            return {"date": date.isoformat(), "events": 0, "status": "empty"}
        
        # Convert to Parquet
        events = [json.loads(row['raw_event']) for row in rows]
        table = pa.Table.from_pylist(events)
        
        # Write to S3
        key = f"audit-archive/{date.year}/{date.month:02d}/{date.day:02d}/events.parquet"
        
        buffer = pa.BufferOutputStream()
        pq.write_table(table, buffer, compression='snappy')
        
        await self.s3.put_object(
            Bucket='audit-archive',
            Key=key,
            Body=buffer.getvalue().to_pybytes(),
            # Object Lock for immutability
            ObjectLockMode='COMPLIANCE',  # GOVERNANCE can be bypassed with a special permission
            ObjectLockRetainUntilDate=datetime.now(timezone.utc) + timedelta(days=2555)  # 7 years
        )
        
        logger.info(f"Archived {len(events)} events for {date.date()}")
        
        return {
            "date": date.isoformat(),
            "events": len(events),
            "s3_key": key,
            "status": "archived"
        }


# =============================================================================
# Query Service (Investigation)
# =============================================================================

class AuditQueryService:
    """
    Query interface for audit log investigation.
    """
    
    def __init__(self, db_pool, s3_client):
        self.db = db_pool
        self.s3 = s3_client
    
    async def search(
        self,
        actor_id: Optional[str] = None,
        resource_type: Optional[str] = None,
        resource_id: Optional[str] = None,
        action: Optional[str] = None,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
        limit: int = 100,
        offset: int = 0
    ) -> List[Dict]:
        """
        Search audit events with filters.
        """
        conditions = ["1=1"]
        params = []
        param_idx = 1
        
        if actor_id:
            conditions.append(f"actor_id = ${param_idx}")
            params.append(actor_id)
            param_idx += 1
        
        if resource_type:
            conditions.append(f"resource_type = ${param_idx}")
            params.append(resource_type)
            param_idx += 1
        
        if resource_id:
            conditions.append(f"resource_id = ${param_idx}")
            params.append(resource_id)
            param_idx += 1
        
        if action:
            conditions.append(f"action LIKE ${param_idx}")
            params.append(f"{action}%")
            param_idx += 1
        
        if start_time:
            conditions.append(f"timestamp >= ${param_idx}")
            params.append(start_time)
            param_idx += 1
        
        if end_time:
            conditions.append(f"timestamp <= ${param_idx}")
            params.append(end_time)
            param_idx += 1
        
        params.extend([limit, offset])
        
        query = f"""
            SELECT raw_event
            FROM audit_events
            WHERE {' AND '.join(conditions)}
            ORDER BY timestamp DESC
            LIMIT ${param_idx} OFFSET ${param_idx + 1}
        """
        
        rows = await self.db.fetch(query, *params)
        return [json.loads(row['raw_event']) for row in rows]
    
    async def get_actor_activity(
        self,
        actor_id: str,
        days: int = 30
    ) -> List[Dict]:
        """Get all activity for a specific actor."""
        start_time = datetime.now(timezone.utc) - timedelta(days=days)
        return await self.search(actor_id=actor_id, start_time=start_time, limit=1000)
    
    async def get_resource_history(
        self,
        resource_type: str,
        resource_id: str
    ) -> List[Dict]:
        """Get complete history of a resource."""
        return await self.search(
            resource_type=resource_type,
            resource_id=resource_id,
            limit=1000
        )
    
    async def verify_integrity(
        self,
        start_time: datetime,
        end_time: datetime
    ) -> Dict[str, Any]:
        """
        Verify hash chain integrity for a time range.
        
        Returns any gaps or inconsistencies found.
        """
        rows = await self.db.fetch(
            """
            SELECT event_id, timestamp, previous_hash, event_hash, raw_event
            FROM audit_events
            WHERE timestamp >= $1 AND timestamp <= $2
            ORDER BY timestamp
            """,
            start_time, end_time
        )
        
        issues = []
        prev_hash = None
        
        for row in rows:
            # Check chain continuity
            if prev_hash and row['previous_hash'] != prev_hash:
                issues.append({
                    "type": "chain_break",
                    "event_id": row['event_id'],
                    "expected_prev": prev_hash,
                    "actual_prev": row['previous_hash']
                })
            
            # Verify hash
            event = json.loads(row['raw_event'])
            event['integrity']['previous_hash'] = row['previous_hash']
            event['integrity'].pop('event_hash', None)  # Hash was computed before this key existed
            
            canonical = json.dumps(event, sort_keys=True, separators=(',', ':'))
            computed_hash = hashlib.sha256(canonical.encode()).hexdigest()
            
            if computed_hash != row['event_hash']:
                issues.append({
                    "type": "hash_mismatch",
                    "event_id": row['event_id'],
                    "expected_hash": computed_hash,
                    "actual_hash": row['event_hash']
                })
            
            prev_hash = row['event_hash']
        
        return {
            "verified_events": len(rows),
            "issues_found": len(issues),
            "issues": issues,
            "integrity": "valid" if not issues else "compromised"
        }