Design a Multi-Tenant Enterprise SaaS Platform.

# tenant_isolation.py - Core tenant context management

from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional
from enum import Enum

# Thread-safe tenant context
_tenant_context: ContextVar[Optional['TenantContext']] = ContextVar(
    'tenant_context', default=None
)


class IsolationLevel(Enum):
    """Tenant isolation levels by plan."""
    SHARED = "shared"           # Shared tables, RLS
    SCHEMA = "schema"           # Schema per tenant
    DATABASE = "database"       # Database per tenant
    DEDICATED = "dedicated"     # Dedicated infrastructure


@dataclass(frozen=True)
class TenantContext:
    """Immutable tenant context for request processing."""
    tenant_id: str
    tenant_name: str
    region: str
    plan: str
    isolation_level: IsolationLevel
    encryption_key_id: str
    features: frozenset
    

def get_current_tenant() -> TenantContext:
    """Get current tenant context. Raises if not set."""
    ctx = _tenant_context.get()
    if ctx is None:
        raise TenantContextError("No tenant context set")
    return ctx


class TenantMiddleware:
    """
    Middleware that establishes tenant context for every request.
    
    Tenant is determined from:
    1. JWT token claims (signature-verified) or API key lookup
    2. Subdomain (acme.docplatform.com → acme), which must MATCH (1)
    
    NEVER from request body, query parameters or the Host header alone
    (the client controls all of those).
    """
    
    async def __call__(self, request, call_next):
        # Authoritative: tenant from the verified token
        token = request.headers.get("authorization", "").replace("Bearer ", "")
        claims = decode_jwt(token)  # Verifies signature and expiry
        tenant_id = claims.get("tenant_id")
        
        if not tenant_id:
            return JSONResponse({"error": "Tenant not identified"}, status_code=400)
        
        # The subdomain is only a hint and must agree with the token
        host = request.headers.get("host", "")
        subdomain = host.split(".")[0]
        if subdomain not in ("www", "api") and subdomain != tenant_id:
            return JSONResponse({"error": "Tenant mismatch"}, status_code=403)
        
        # Load tenant configuration
        tenant_config = await self.tenant_service.get_tenant(tenant_id)
        
        if not tenant_config:
            return JSONResponse({"error": "Tenant not found"}, status_code=404)
        
        # Set immutable context
        context = TenantContext(
            tenant_id=tenant_config.id,
            tenant_name=tenant_config.name,
            region=tenant_config.region,
            plan=tenant_config.plan,
            isolation_level=IsolationLevel(tenant_config.isolation_level),
            encryption_key_id=tenant_config.encryption_key_id,
            features=frozenset(tenant_config.features)
        )
        
        # Set context for this request
        token = _tenant_context.set(context)
        
        try:
            response = await call_next(request)
            return response
        finally:
            _tenant_context.reset(token)


class TenantAwareRepository:
    """
    Repository base class that enforces tenant isolation.
    
    ALL database access goes through this class.
    """
    
    def __init__(self, db_pool):
        self.db = db_pool
    
    async def find_by_id(self, table: str, id: str):
        """Find record by ID within current tenant."""
        tenant = get_current_tenant()
        
        # ALWAYS filter by tenant_id
        return await self.db.fetchrow(
            f"SELECT * FROM {table} WHERE id = $1 AND tenant_id = $2",
            id, tenant.tenant_id
        )
    
    async def find_all(self, table: str, filters: dict = None):
        """Find all records within current tenant."""
        tenant = get_current_tenant()
        
        query = f"SELECT * FROM {table} WHERE tenant_id = $1"
        params = [tenant.tenant_id]
        
        if filters:
            for i, (key, value) in enumerate(filters.items(), start=2):
                query += f" AND {key} = ${i}"
                params.append(value)
        
        return await self.db.fetch(query, *params)
    
    async def create(self, table: str, data: dict):
        """Create record with automatic tenant_id."""
        tenant = get_current_tenant()
        
        # Force tenant_id - cannot be overridden
        data["tenant_id"] = tenant.tenant_id
        
        columns = ", ".join(data.keys())
        placeholders = ", ".join(f"${i+1}" for i in range(len(data)))
        
        return await self.db.fetchrow(
            f"INSERT INTO {table} ({columns}) VALUES ({placeholders}) RETURNING *",
            *data.values()
        )

-- PostgreSQL Row-Level Security Setup

-- Enable RLS on all tenant tables
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
ALTER TABLE users ENABLE ROW LEVEL SECURITY;
ALTER TABLE folders ENABLE ROW LEVEL SECURITY;
ALTER TABLE comments ENABLE ROW LEVEL SECURITY;
ALTER TABLE audit_logs ENABLE ROW LEVEL SECURITY;

-- Table owners bypass RLS unless it is forced
ALTER TABLE documents FORCE ROW LEVEL SECURITY;
ALTER TABLE users FORCE ROW LEVEL SECURITY;
-- ...same for every tenant table

-- Create policy that filters by current tenant
CREATE POLICY tenant_isolation_policy ON documents
    FOR ALL
    USING (tenant_id = current_setting('app.current_tenant_id')::uuid);

CREATE POLICY tenant_isolation_policy ON users
    FOR ALL
    USING (tenant_id = current_setting('app.current_tenant_id')::uuid);

-- Similar for all tables...

-- Application sets tenant at the start of each transaction.
-- SET LOCAL resets at COMMIT/ROLLBACK, so a pooled connection
-- can't carry one tenant's setting into the next request:
-- BEGIN; SET LOCAL app.current_tenant_id = 'tenant-uuid-here';
-- (Day 1 shows the other safe option: a session-level setting
-- that the connection wrapper clears before release. SET LOCAL
-- needs no cleanup step, which is why this design uses it.)

-- Now even this query is filtered:
-- SELECT * FROM documents; 
-- Only returns current tenant's documents!

# enterprise_isolation.py

class EnterpriseTenantManager:
    """
    Manages dedicated resources for enterprise tenants.
    """
    
    async def provision_enterprise_tenant(
        self,
        tenant_id: str,
        config: EnterpriseConfig
    ):
        """
        Provision dedicated resources for enterprise tenant.
        
        Options:
        - Dedicated database
        - Dedicated Elasticsearch index
        - Dedicated encryption keys
        - Dedicated compute (optional)
        """
        
        # 1. Create dedicated database
        if config.dedicated_database:
            await self._create_tenant_database(tenant_id)
        
        # 2. Create dedicated search index
        await self._create_tenant_search_index(tenant_id)
        
        # 3. Create tenant-specific KMS key
        key_id = await self._create_tenant_kms_key(tenant_id)
        
        # 4. Update tenant config
        await self.tenant_service.update_tenant(
            tenant_id,
            isolation_level=IsolationLevel.DATABASE,
            encryption_key_id=key_id,
            database_name=f"tenant_{tenant_id}",
            search_index=f"documents_{tenant_id}"
        )
    
    async def _create_tenant_database(self, tenant_id: str):
        """Create dedicated PostgreSQL database for tenant."""
        safe_name = tenant_id.replace("-", "_")
        
        await self.admin_db.execute(f"""
            CREATE DATABASE tenant_{safe_name};
        """)
        
        # Run migrations on new database
        await self._run_migrations(f"tenant_{safe_name}")
    
    async def _create_tenant_kms_key(self, tenant_id: str) -> str:
        """Create dedicated KMS key for tenant encryption."""
        response = await self.kms.create_key(
            Description=f"Encryption key for tenant {tenant_id}",
            KeyUsage="ENCRYPT_DECRYPT",
            Tags=[
                {"TagKey": "tenant_id", "TagValue": tenant_id},
                {"TagKey": "purpose", "TagValue": "document_encryption"}
            ]
        )
        return response["KeyMetadata"]["KeyId"]

# noisy_neighbor/rate_limiter.py

class TenantResourceManager:
    """
    Manages and enforces resource quotas per tenant.
    """
    
    def __init__(self, redis, quota_config):
        self.redis = redis
        self.quotas = quota_config
    
    async def check_rate_limit(
        self,
        tenant_id: str,
        resource: str,
        cost: int = 1
    ) -> RateLimitResult:
        """
        Check if request is within rate limits.
        
        Uses a fixed-window counter per tenant (simple, but allows up to
        2x the limit across a window boundary; a token bucket or sliding
        window smooths that).
        """
        tenant = await self.get_tenant_config(tenant_id)
        limit = self.quotas[tenant.plan][resource]
        
        key = f"ratelimit:{tenant_id}:{resource}"
        
        # Lua script for atomic check-and-decrement
        script = """
        local key = KEYS[1]
        local limit = tonumber(ARGV[1])
        local cost = tonumber(ARGV[2])
        local window = tonumber(ARGV[3])
        local now = tonumber(ARGV[4])
        
        -- Get current count
        local current = tonumber(redis.call('GET', key) or '0')
        
        if current + cost <= limit then
            redis.call('INCRBY', key, cost)
            if current == 0 then
                -- Start the window once; re-setting EXPIRE on every hit
                -- would keep the counter alive forever under steady load
                redis.call('EXPIRE', key, window)
            end
            return {1, limit - current - cost}
        else
            local ttl = redis.call('TTL', key)
            return {0, ttl}
        end
        """
        
        # redis-py: eval(script, numkeys, *keys_and_args)
        result = await self.redis.eval(
            script, 1, key, limit, cost, 60, time.time()
        )
        
        allowed = result[0] == 1
        
        if not allowed:
            await self._record_throttle(tenant_id, resource)
        
        return RateLimitResult(
            allowed=allowed,
            remaining=result[1] if allowed else 0,
            retry_after=result[1] if not allowed else None
        )
    
    async def check_concurrent_limit(
        self,
        tenant_id: str,
        operation: str
    ) -> bool:
        """
        Check concurrent operation limit.
        
        Prevents one tenant from using all worker capacity.
        """
        tenant = await self.get_tenant_config(tenant_id)
        limit = self.quotas[tenant.plan][f"concurrent_{operation}"]
        
        key = f"concurrent:{tenant_id}:{operation}"
        current = await self.redis.scard(key)  # A set: GET would raise WRONGTYPE
        
        return current < limit
    
    async def acquire_concurrent_slot(
        self,
        tenant_id: str,
        operation: str,
        operation_id: str,
        ttl: int = 300
    ) -> bool:
        """Acquire a concurrent operation slot (atomic check-and-add)."""
        tenant = await self.get_tenant_config(tenant_id)
        limit = self.quotas[tenant.plan][f"concurrent_{operation}"]
        key = f"concurrent:{tenant_id}:{operation}"
        
        # Check and add in one Lua script, or two workers can both see
        # "limit - 1" and both take the last slot
        script = """
        if redis.call('SCARD', KEYS[1]) < tonumber(ARGV[1]) then
            redis.call('SADD', KEYS[1], ARGV[2])
            redis.call('EXPIRE', KEYS[1], ARGV[3])
            return 1
        end
        return 0
        """
        result = await self.redis.eval(
            script, 1, key, limit, operation_id, ttl
        )
        return result == 1
    
    async def release_concurrent_slot(
        self,
        tenant_id: str,
        operation: str,
        operation_id: str
    ):
        """Release a concurrent operation slot."""
        key = f"concurrent:{tenant_id}:{operation}"
        await self.redis.srem(key, operation_id)


class SearchQueryGuard:
    """
    Guards search queries against noisy neighbors.
    """
    
    async def guard_search(
        self,
        tenant_id: str,
        query: SearchQuery
    ) -> tuple[bool, Optional[int]]:
        """
        Check if search query should be allowed.
        
        Returns (allowed, timeout_seconds).
        """
        tenant = await self.tenant_service.get_tenant(tenant_id)
        
        # Get tenant's search timeout
        timeout = self.quotas[tenant.plan]["search_timeout"]
        
        # Estimate query cost
        cost = self._estimate_query_cost(query)
        
        if cost > 100:  # High cost query
            # Check if tenant can run expensive queries
            if tenant.plan == "standard":
                return False, None
            
            # Use longer timeout for expensive queries
            timeout = min(timeout * 2, 120)
        
        return True, timeout
    
    def _estimate_query_cost(self, query: SearchQuery) -> int:
        """Estimate query cost based on complexity."""
        cost = 1
        
        # Wildcard queries are expensive
        if "*" in query.text or "?" in query.text:
            cost += 10
        
        # Regex queries are very expensive
        if query.use_regex:
            cost += 50
        
        # Large result sets
        if query.limit > 100:
            cost += query.limit // 100
        
        return cost

# noisy_neighbor/fair_scheduler.py

class TenantFairQueue:
    """
    Fair queue that prevents one tenant from monopolizing workers.
    
    Uses weighted fair queuing based on tenant plan.
    """
    
    PLAN_WEIGHTS = {
        "standard": 1,
        "professional": 3,
        "enterprise": 10
    }
    
    async def enqueue(
        self,
        tenant_id: str,
        job: Job
    ):
        """Add job to tenant's queue."""
        tenant = await self.tenant_service.get_tenant(tenant_id)
        
        # Check queue depth limit
        current_depth = await self.get_queue_depth(tenant_id)
        max_depth = self.quotas[tenant.plan]["max_queue_depth"]
        
        if current_depth >= max_depth:
            raise QueueFullError(
                f"Queue full for tenant {tenant_id}. "
                f"Current: {current_depth}, Max: {max_depth}"
            )
        
        # Add to tenant's queue
        await self.redis.lpush(
            f"queue:{tenant_id}",
            job.serialize()
        )
    
    async def dequeue(self) -> Optional[tuple[str, Job]]:
        """
        Dequeue next job using weighted fair scheduling.
        
        Higher-weight tenants get proportionally more slots.
        """
        # Get all tenant queues with pending jobs
        tenant_queues = await self._get_active_queues()
        
        if not tenant_queues:
            return None
        
        # Calculate weighted selection
        weighted_tenants = []
        for tenant_id, depth in tenant_queues.items():
            tenant = await self.tenant_service.get_tenant(tenant_id)
            weight = self.PLAN_WEIGHTS.get(tenant.plan, 1)
            # Weight decreases as queue depth increases (fairness)
            adjusted_weight = weight / (1 + depth * 0.1)
            weighted_tenants.append((tenant_id, adjusted_weight))
        
        # Select tenant based on weights
        selected_tenant = self._weighted_random_choice(weighted_tenants)
        
        # Pop job from selected tenant's queue
        job_data = await self.redis.rpop(f"queue:{selected_tenant}")
        
        if job_data:
            return selected_tenant, Job.deserialize(job_data)
        
        return None

# data_residency/router.py

class RegionalDataRouter:
    """
    Routes data operations to the correct region based on tenant.
    
    EU tenant data NEVER touches US infrastructure.
    """
    
    REGION_CONFIGS = {
        "us": {
            "database": "postgres://db-us.internal:5432/app",
            "elasticsearch": "https://es-us.internal:9200",
            "s3_bucket": "documents-us-east-1",
            "redis": "redis://cache-us.internal:6379"
        },
        "eu": {
            "database": "postgres://db-eu.internal:5432/app",
            "elasticsearch": "https://es-eu.internal:9200",
            "s3_bucket": "documents-eu-central-1",
            "redis": "redis://cache-eu.internal:6379"
        },
        "apac": {
            "database": "postgres://db-apac.internal:5432/app",
            "elasticsearch": "https://es-apac.internal:9200",
            "s3_bucket": "documents-ap-southeast-1",
            "redis": "redis://cache-apac.internal:6379"
        }
    }
    
    async def get_database_connection(self, tenant_id: str):
        """Get database connection for tenant's region."""
        tenant = await self.tenant_service.get_tenant(tenant_id)
        config = self.REGION_CONFIGS[tenant.region]
        
        return await self.connection_pools[tenant.region].acquire()
    
    async def get_storage_bucket(self, tenant_id: str) -> str:
        """Get S3 bucket for tenant's region."""
        tenant = await self.tenant_service.get_tenant(tenant_id)
        return self.REGION_CONFIGS[tenant.region]["s3_bucket"]
    
    async def upload_document(
        self,
        tenant_id: str,
        document_id: str,
        content: bytes
    ) -> str:
        """Upload document to tenant's regional storage."""
        tenant = await self.tenant_service.get_tenant(tenant_id)
        bucket = self.REGION_CONFIGS[tenant.region]["s3_bucket"]
        
        # Path includes tenant for isolation
        key = f"tenants/{tenant_id}/documents/{document_id}"
        
        # Encrypt with tenant's key before upload
        encrypted = await self.encrypt_for_tenant(tenant_id, content)
        
        await self.s3.put_object(
            Bucket=bucket,
            Key=key,
            Body=encrypted,
            ServerSideEncryption="aws:kms",
            SSEKMSKeyId=tenant.encryption_key_id
        )
        
        return f"s3://{bucket}/{key}"

# gdpr/consent.py
from datetime import datetime, timezone


class GDPRConsentService:
    """
    Manages user consent for GDPR compliance.
    """
    
    CONSENT_PURPOSES = [
        "service_delivery",      # Required for service
        "analytics",             # Product analytics
        "marketing_email",       # Marketing communications
        "third_party_sharing",   # Sharing with partners
    ]
    
    async def record_consent(
        self,
        user_id: str,
        tenant_id: str,
        purpose: str,
        granted: bool,
        ip_address: str,
        consent_text: str
    ) -> ConsentRecord:
        """
        Record a consent decision.
        
        Creates immutable audit record.
        """
        record = ConsentRecord(
            id=str(uuid.uuid4()),
            user_id=user_id,
            tenant_id=tenant_id,
            purpose=purpose,
            status="granted" if granted else "denied",
            granted_at=datetime.now(timezone.utc) if granted else None,
            ip_address=ip_address,
            consent_text=consent_text,
            consent_version="2024-01"
        )
        
        # Store in regional database
        await self.db.execute(
            """
            INSERT INTO consent_records 
            (id, user_id, tenant_id, purpose, status, granted_at,
             ip_address, consent_text, consent_version, created_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
            """,
            record.id, record.user_id, record.tenant_id,
            record.purpose, record.status, record.granted_at,
            record.ip_address, record.consent_text,
            record.consent_version, datetime.now(timezone.utc)
        )
        
        # Publish event for downstream systems
        await self.events.publish("consent", {
            "type": "consent.recorded",
            "user_id": user_id,
            "purpose": purpose,
            "granted": granted
        })
        
        return record
    
    async def check_consent(
        self,
        user_id: str,
        tenant_id: str,
        purpose: str
    ) -> bool:
        """Check if user has consented to a purpose."""
        result = await self.db.fetchrow(
            """
            SELECT status FROM consent_records
            WHERE user_id = $1 AND tenant_id = $2 AND purpose = $3
            ORDER BY created_at DESC
            LIMIT 1
            """,
            user_id, tenant_id, purpose
        )
        
        return result and result["status"] == "granted"

# gdpr/export.py
from datetime import datetime, timezone


class GDPRDataExporter:
    """
    Exports user data for GDPR portability requests.
    """
    
    async def export_user_data(
        self,
        user_id: str,
        tenant_id: str
    ) -> DataExportResult:
        """
        Export all user's personal data.
        
        GDPR Article 20: Right to data portability
        """
        export_id = str(uuid.uuid4())
        
        # Collect data from all sources
        data = {
            "export_metadata": {
                "export_id": export_id,
                "exported_at": datetime.now(timezone.utc).isoformat(),
                "user_id": user_id,
                "tenant_id": tenant_id
            },
            "profile": await self._export_profile(user_id, tenant_id),
            "documents": await self._export_documents(user_id, tenant_id),
            "comments": await self._export_comments(user_id, tenant_id),
            "activity_history": await self._export_activity(user_id, tenant_id),
            "consent_records": await self._export_consent(user_id, tenant_id),
        }
        
        # Package as JSON
        export_json = json.dumps(data, indent=2, default=str)
        
        # Also create ZIP with actual files
        zip_buffer = await self._create_export_zip(user_id, tenant_id, data)
        
        # Upload to tenant's regional storage
        bucket = await self.router.get_storage_bucket(tenant_id)
        export_key = f"exports/{tenant_id}/{user_id}/{export_id}.zip"
        
        await self.s3.put_object(
            Bucket=bucket,
            Key=export_key,
            Body=zip_buffer.getvalue()
        )
        
        # Generate download link (7 days is the SigV4 maximum; if signed
        # with temporary role credentials it dies when they expire)
        download_url = await self.s3.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket, "Key": export_key},
            ExpiresIn=604800  # 7 days
        )
        
        return DataExportResult(
            export_id=export_id,
            download_url=download_url,
            expires_at=datetime.now(timezone.utc) + timedelta(days=7),
            size_bytes=len(zip_buffer.getvalue())
        )

# gdpr/deletion.py
from datetime import datetime, timezone


class UserDeletionService:
    """
    Orchestrates user data deletion across all systems.
    """
    
    # Systems in deletion order (dependencies first)
    DELETION_TARGETS = [
        ("search", "elasticsearch", 2),   # Remove from search
        ("storage", "s3", 3),             # Delete files
        ("analytics", "bigquery", 4),     # Remove from analytics
        ("database", "postgresql", 10),   # Primary DB after the copies
        ("cache", "redis", 11),           # Cache last, so reads can't refill it
    ]
    
    async def process_deletion_request(
        self,
        user_id: str,
        tenant_id: str,
        requested_by: str
    ) -> DeletionRequest:
        """
        Process a GDPR deletion request.
        
        Must complete within one month per GDPR (30-day internal target).
        """
        request = DeletionRequest(
            id=str(uuid.uuid4()),
            user_id=user_id,
            tenant_id=tenant_id,
            requested_at=datetime.now(timezone.utc),
            requested_by=requested_by,
            deadline=datetime.now(timezone.utc) + timedelta(days=30),
            status="pending"
        )
        
        # Store request
        await self._save_request(request)
        
        # Execute deletion workflow
        try:
            await self._execute_deletion(request)
            
            # Verify deletion
            verification = await self._verify_deletion(request)
            
            if verification.all_verified:
                request.status = "completed"
                request.completed_at = datetime.now(timezone.utc)
            else:
                request.status = "partial"
                request.issues = verification.issues
                
        except Exception as e:
            request.status = "failed"
            request.error = str(e)
            raise
        
        finally:
            await self._save_request(request)
            await self._notify_user(request)
        
        return request
    
    async def _execute_deletion(self, request: DeletionRequest):
        """Execute deletion across all systems."""
        
        for system_type, system_name, priority in sorted(
            self.DELETION_TARGETS, key=lambda x: x[2]
        ):
            target = DeletionTarget(
                system_name=system_name,
                system_type=system_type,
                status="pending"
            )
            
            try:
                executor = self.executors[system_name]
                result = await executor.delete_user_data(
                    request.user_id,
                    request.tenant_id
                )
                
                target.status = "completed"
                target.records_deleted = result.get("records_deleted", 0)
                
                await self.audit.log(
                    action="deletion_executed",
                    system=system_name,
                    user_id=request.user_id,
                    records_deleted=target.records_deleted
                )
                
            except Exception as e:
                target.status = "failed"
                target.error = str(e)
                raise
            
            request.targets.append(target)
    
    async def _verify_deletion(
        self,
        request: DeletionRequest
    ) -> VerificationResult:
        """Verify that deletion was successful."""
        issues = []
        
        for target in request.targets:
            executor = self.executors[target.system_name]
            
            still_exists = await executor.check_user_exists(
                request.user_id,
                request.tenant_id
            )
            
            if still_exists:
                issues.append(f"Data still exists in {target.system_name}")
        
        return VerificationResult(
            all_verified=len(issues) == 0,
            issues=issues
        )


class PostgreSQLDeletionExecutor:
    """
    Deletes user data from PostgreSQL.
    """
    
    async def delete_user_data(
        self,
        user_id: str,
        tenant_id: str
    ) -> dict:
        """Delete user and related data."""
        records_deleted = 0
        
        async with self.db.transaction():
            # Delete from leaf tables first
            
            # Comments (anonymize, keep content)
            result = await self.db.execute(
                """
                UPDATE comments 
                SET user_id = NULL, author_name = 'Deleted User'
                WHERE user_id = $1 AND tenant_id = $2
                """,
                user_id, tenant_id
            )
            records_deleted += int(result.split()[-1])
            
            # Activity logs (anonymize)
            result = await self.db.execute(
                """
                UPDATE activity_logs
                SET user_id = 'DELETED', ip_address = 'DELETED'
                WHERE user_id = $1 AND tenant_id = $2
                """,
                user_id, tenant_id
            )
            records_deleted += int(result.split()[-1])
            
            # Documents (reassign ownership to admin or delete)
            result = await self.db.execute(
                """
                UPDATE documents
                SET owner_id = (
                    SELECT id FROM users 
                    WHERE tenant_id = $2 AND 'admin' = ANY(roles)
                    LIMIT 1
                )
                WHERE owner_id = $1 AND tenant_id = $2
                """,
                user_id, tenant_id
            )
            records_deleted += int(result.split()[-1])
            
            # Consent records (keep anonymized for audit)
            result = await self.db.execute(
                """
                UPDATE consent_records
                SET user_id = 'DELETED', ip_address = 'DELETED'
                WHERE user_id = $1 AND tenant_id = $2
                """,
                user_id, tenant_id
            )
            records_deleted += int(result.split()[-1])
            
            # Finally, delete user
            result = await self.db.execute(
                "DELETE FROM users WHERE id = $1 AND tenant_id = $2",
                user_id, tenant_id
            )
            records_deleted += int(result.split()[-1])
        
        return {"records_deleted": records_deleted}

# security/auth.py
from datetime import datetime, timezone


class AuthenticationService:
    """
    Multi-tenant authentication service.
    """
    
    async def authenticate(
        self,
        email: str,
        password: str,
        tenant_id: str,
        ip_address: str
    ) -> AuthResult:
        """
        Authenticate user within their tenant.
        """
        # Rate limiting per IP + tenant
        if await self._is_rate_limited(ip_address, tenant_id):
            raise AuthError("Too many attempts")
        
        # Find user in tenant
        user = await self.db.fetchrow(
            """
            SELECT id, email, password_hash, roles, mfa_enabled, status
            FROM users
            WHERE email = $1 AND tenant_id = $2
            """,
            email.lower(), tenant_id
        )
        
        if not user:
            await self._record_failed_attempt(ip_address, tenant_id)
            raise AuthError("Invalid credentials")
        
        # Verify password before revealing anything about the account
        if not bcrypt.checkpw(password.encode(), user["password_hash"].encode()):
            await self._record_failed_attempt(ip_address, tenant_id)
            raise AuthError("Invalid credentials")
        
        if user["status"] != "active":
            raise AuthError("Account disabled")
        
        # Create session
        session_id = secrets.token_urlsafe(32)
        
        result = AuthResult(
            user_id=user["id"],
            tenant_id=tenant_id,
            roles=user["roles"],
            mfa_required=user["mfa_enabled"],
            session_id=session_id
        )
        
        # Audit log
        await self.audit.log(
            action="login_success",
            user_id=user["id"],
            tenant_id=tenant_id,
            ip_address=ip_address
        )
        
        return result
    
    async def create_jwt(self, auth_result: AuthResult) -> str:
        """Create JWT with tenant claims."""
        signing_key = await self.secrets.get_secret("jwt/signing_key")
        
        payload = {
            "sub": auth_result.user_id,
            "tenant_id": auth_result.tenant_id,
            "roles": auth_result.roles,
            "session_id": auth_result.session_id,
            "iat": datetime.now(timezone.utc),
            "exp": datetime.now(timezone.utc) + timedelta(minutes=15)
        }
        
        return jwt.encode(payload, signing_key.value, algorithm="RS256")


class AuthorizationService:
    """
    Multi-tenant authorization with RBAC.
    """
    
    async def check_document_access(
        self,
        user_id: str,
        tenant_id: str,
        document_id: str,
        required_permission: str
    ) -> bool:
        """
        Check if user can access document.
        
        Enforces:
        1. Tenant isolation (user's tenant == document's tenant)
        2. Role-based permission
        3. Document-level permission
        """
        # Get document
        document = await self.db.fetchrow(
            """
            SELECT tenant_id, owner_id, permissions
            FROM documents
            WHERE id = $1
            """,
            document_id
        )
        
        if not document:
            return False
        
        # CRITICAL: Tenant isolation check
        if document["tenant_id"] != tenant_id:
            await self.audit.log(
                action="access_denied",
                reason="tenant_mismatch",
                user_id=user_id,
                document_id=document_id
            )
            return False
        
        # Check ownership
        if document["owner_id"] == user_id:
            return True
        
        # Check document permissions
        permissions = document["permissions"] or {}
        user_permission = permissions.get(user_id)
        
        if user_permission:
            return self._has_permission(user_permission, required_permission)
        
        # Check folder permissions (inherited)
        # ... folder permission check logic ...
        
        return False

# security/audit.py
from datetime import datetime, timezone


class AuditService:
    """
    Comprehensive audit logging for compliance.
    """
    
    async def log(
        self,
        action: str,
        **context
    ):
        """
        Log an audit event.
        
        All access, modifications, and security events are logged.
        """
        # Not get_current_tenant(): it raises when unset (e.g. failed logins),
        # and those events must still be audited
        tenant = _tenant_context.get()
        
        event = AuditEvent(
            id=str(uuid.uuid4()),
            timestamp=datetime.now(timezone.utc),
            tenant_id=tenant.tenant_id if tenant else context.get("tenant_id"),
            action=action,
            actor_id=context.get("user_id"),
            actor_type=context.get("actor_type", "user"),
            resource_type=context.get("resource_type"),
            resource_id=context.get("resource_id"),
            ip_address=context.get("ip_address"),
            user_agent=context.get("user_agent"),
            details=context
        )
        
        # Write to regional audit log (immutable)
        await self.db.execute(
            """
            INSERT INTO audit_logs
            (id, timestamp, tenant_id, action, actor_id, actor_type,
             resource_type, resource_id, ip_address, user_agent, details)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
            """,
            event.id, event.timestamp, event.tenant_id, event.action,
            event.actor_id, event.actor_type, event.resource_type,
            event.resource_id, event.ip_address, event.user_agent,
            json.dumps(event.details)
        )
        
        # Also stream to Kafka for real-time monitoring
        await self.kafka.produce(
            topic="audit_events",
            key=event.tenant_id,
            value=event.to_dict()
        )