Session Store Design.

server_index = hash(client_ip) % num_servers

class StickyLoadBalancer:
    def __init__(self, servers: list[str]):
        self.ring = ConsistentHashRing(servers)
    
    def get_server(self, user_id: str) -> str:
        return self.ring.get_node(user_id)
    
    def add_server(self, server: str):
        # Only 1/N of sessions need to move
        self.ring.add_node(server)
    
    def remove_server(self, server: str):
        # Sessions for this server will be redistributed
        self.ring.remove_node(server)

@app.route('/dashboard')
def dashboard():
    session = get_session()
    if not session:
        # Session lost - redirect to login
        return redirect('/login?reason=session_expired')
    return render_dashboard(session)

@dataclass
class Session:
    session_id: str          # Unique identifier (UUID or secure random)
    user_id: str             # Authenticated user
    created_at: int          # Unix timestamp
    last_accessed: int       # Unix timestamp (for TTL extension)
    expires_at: int          # Unix timestamp
    
    # Authentication data
    roles: list[str]
    permissions: list[str]
    
    # User preferences
    locale: str
    timezone: str
    
    # CSRF protection
    csrf_token: str
    
    # Application-specific data
    data: dict               # Flexible storage for app needs

# Example session
session = Session(
    session_id="sess_a1b2c3d4e5f6",
    user_id="user_123",
    created_at=1700000000,
    last_accessed=1700003600,
    expires_at=1700086400,
    roles=["user", "premium"],
    permissions=["read", "write", "delete"],
    locale="en-US",
    timezone="America/New_York",
    csrf_token="csrf_xyz789",
    data={
        "cart_items": ["item_1", "item_2"],
        "last_page": "/products/shoes"
    }
)

def get_partition(session_id: str, num_partitions: int) -> int:
    return crc16(session_id) % num_partitions

import redis
import json
import secrets
import time
from typing import Optional
from dataclasses import asdict

class SessionStore:
    def __init__(
        self,
        redis_cluster: redis.RedisCluster,
        default_ttl: int = 86400,  # 24 hours
        extend_on_access: bool = True
    ):
        self.redis = redis_cluster
        self.default_ttl = default_ttl
        self.extend_on_access = extend_on_access
    
    def create(self, user_id: str, **kwargs) -> Session:
        """Create a new session for a user."""
        now = int(time.time())
        
        session = Session(
            session_id=f"sess_{secrets.token_urlsafe(32)}",
            user_id=user_id,
            created_at=now,
            last_accessed=now,
            expires_at=now + self.default_ttl,
            roles=kwargs.get('roles', []),
            permissions=kwargs.get('permissions', []),
            locale=kwargs.get('locale', 'en-US'),
            timezone=kwargs.get('timezone', 'UTC'),
            csrf_token=secrets.token_urlsafe(32),
            data=kwargs.get('data', {})
        )
        
        key = f"session:{session.session_id}"
        self.redis.setex(key, self.default_ttl, json.dumps(asdict(session)))
        
        # Index by user_id for "logout all devices"
        self.redis.sadd(f"user_sessions:{user_id}", session.session_id)
        self.redis.expire(f"user_sessions:{user_id}", self.default_ttl)
        
        return session
    
    def get(self, session_id: str) -> Optional[Session]:
        """Retrieve a session by ID."""
        key = f"session:{session_id}"
        data = self.redis.get(key)
        
        if not data:
            return None
        
        session_dict = json.loads(data)
        session = Session(**session_dict)
        
        # Check expiration
        if session.expires_at < time.time():
            self.delete(session_id)
            return None
        
        # Extend TTL on access
        if self.extend_on_access:
            self._extend_ttl(session)
        
        return session
    
    def _extend_ttl(self, session: Session):
        """Extend session TTL on access (at most once a minute, so every read isn't also a write)."""
        now = int(time.time())
        if now - session.last_accessed < 60:
            return
        session.last_accessed = now
        session.expires_at = now + self.default_ttl
        
        key = f"session:{session.session_id}"
        self.redis.setex(key, self.default_ttl, json.dumps(asdict(session)))
    
    def update(self, session_id: str, **updates) -> Optional[Session]:
        """Update session data."""
        session = self.get(session_id)
        if not session:
            return None
        
        # Apply updates
        for key, value in updates.items():
            if hasattr(session, key):
                setattr(session, key, value)
            else:
                session.data[key] = value
        
        session.last_accessed = int(time.time())
        
        key = f"session:{session_id}"
        ttl = self.redis.ttl(key)
        self.redis.setex(key, max(ttl, 60), json.dumps(asdict(session)))
        
        return session
    
    def delete(self, session_id: str) -> bool:
        """Delete a session (logout)."""
        session = self.get(session_id)
        if not session:
            return False
        
        # Remove from user index
        self.redis.srem(f"user_sessions:{session.user_id}", session_id)
        
        # Delete session
        key = f"session:{session_id}"
        return self.redis.delete(key) > 0
    
    def delete_all_for_user(self, user_id: str) -> int:
        """Delete all sessions for a user (logout all devices)."""
        session_ids = self.redis.smembers(f"user_sessions:{user_id}")
        
        if not session_ids:
            return 0
        
        # Delete all sessions
        pipe = self.redis.pipeline()
        for sid in session_ids:
            pipe.delete(f"session:{sid}")
        pipe.delete(f"user_sessions:{user_id}")
        results = pipe.execute()
        
        return sum(1 for r in results[:-1] if r > 0)
    
    def validate(self, session_id: str, csrf_token: Optional[str] = None) -> bool:
        """Validate a session exists and optionally verify CSRF token."""
        session = self.get(session_id)
        if not session:
            return False
        
        if csrf_token and session.csrf_token != csrf_token:
            return False
        
        return True

# Redis cluster with connection pooling
redis_cluster = redis.RedisCluster(
    host='redis-cluster.example.com',
    port=6379,
    max_connections=100,
    socket_timeout=1.0,
    socket_connect_timeout=1.0
)

from cachetools import TTLCache
import threading

class CachedSessionStore:
    def __init__(self, session_store: SessionStore, local_ttl: int = 5):
        self.store = session_store
        self.cache = TTLCache(maxsize=10000, ttl=local_ttl)
        self.lock = threading.Lock()
    
    def get(self, session_id: str) -> Optional[Session]:
        # Check local cache
        with self.lock:
            if session_id in self.cache:
                return self.cache[session_id]
        
        # Cache miss - fetch from Redis
        session = self.store.get(session_id)
        
        if session:
            with self.lock:
                self.cache[session_id] = session
        
        return session
    
    def invalidate(self, session_id: str):
        """Invalidate local cache entry."""
        with self.lock:
            self.cache.pop(session_id, None)
    
    # Write operations bypass cache and invalidate
    def update(self, session_id: str, **updates) -> Optional[Session]:
        self.invalidate(session_id)
        return self.store.update(session_id, **updates)
    
    def delete(self, session_id: str) -> bool:
        self.invalidate(session_id)
        return self.store.delete(session_id)

def get_sessions_batch(self, session_ids: list[str]) -> dict[str, Session]:
    """Fetch multiple sessions in one round-trip."""
    pipe = self.redis.pipeline()
    
    for sid in session_ids:
        pipe.get(f"session:{sid}")
    
    results = pipe.execute()
    
    sessions = {}
    for sid, data in zip(session_ids, results):
        if data:
            sessions[sid] = Session(**json.loads(data))
    
    return sessions

import zlib

class CompressedSessionStore:
    COMPRESSION_THRESHOLD = 1024  # bytes
    
    def _serialize(self, session: Session) -> bytes:
        data = json.dumps(asdict(session)).encode()
        if len(data) > self.COMPRESSION_THRESHOLD:
            return b'Z' + zlib.compress(data)
        return b'R' + data
    
    def _deserialize(self, data: bytes) -> Session:
        if data[0:1] == b'Z':
            data = zlib.decompress(data[1:])
        else:
            data = data[1:]
        return Session(**json.loads(data))

from enum import Enum
from typing import Optional
import re

class Region(Enum):
    US_EAST = "use1"
    US_WEST = "usw2"
    EU_WEST = "euw1"
    AP_SOUTH = "aps1"

class MultiRegionSessionStore:
    def __init__(
        self,
        local_region: Region,
        redis_clients: dict[Region, redis.RedisCluster]
    ):
        self.local_region = local_region
        self.clients = redis_clients
        self.local_store = SessionStore(redis_clients[local_region])
    
    def create(self, user_id: str, **kwargs) -> Session:
        """Create session in local region."""
        session = self.local_store.create(user_id, **kwargs)
        
        # Encode region in session ID
        session.session_id = f"sess_{self.local_region.value}_{session.session_id[5:]}"
        
        # Re-save with region-prefixed ID
        key = f"session:{session.session_id}"
        self.clients[self.local_region].setex(
            key, 
            self.local_store.default_ttl, 
            json.dumps(asdict(session))
        )
        
        return session
    
    def get(self, session_id: str) -> Optional[Session]:
        """Get session from appropriate region."""
        region = self._extract_region(session_id)
        
        if region == self.local_region:
            return self.local_store.get(session_id)
        
        # Remote region - need cross-region call
        remote_client = self.clients.get(region)
        if not remote_client:
            return None
        
        key = f"session:{session_id}"
        data = remote_client.get(key)
        
        if data:
            return Session(**json.loads(data))
        return None
    
    def _extract_region(self, session_id: str) -> Region:
        """Extract region from session ID."""
        # Format: sess_{region}_{random}
        match = re.match(r'sess_([a-z0-9]+)_', session_id)
        if match:
            region_code = match.group(1)
            for region in Region:
                if region.value == region_code:
                    return region
        return self.local_region  # Default to local
    
    def should_redirect(self, session_id: str) -> Optional[str]:
        """
        Check if request should be redirected to session's home region.
        Returns redirect URL or None.
        """
        region = self._extract_region(session_id)
        if region != self.local_region:
            return self._get_region_url(region)
        return None
    
    def _get_region_url(self, region: Region) -> str:
        return {
            Region.US_EAST: "https://us-east.example.com",
            Region.US_WEST: "https://us-west.example.com",
            Region.EU_WEST: "https://eu-west.example.com",
            Region.AP_SOUTH: "https://ap-south.example.com"
        }[region]

class ResilientSessionStore:
    """
    Session store with cross-region backup for critical sessions.
    """
    
    def __init__(
        self,
        local_region: Region,
        backup_region: Region,
        redis_clients: dict[Region, redis.RedisCluster]
    ):
        self.local = redis_clients[local_region]
        self.backup = redis_clients[backup_region]
        self.local_region = local_region
        self.backup_region = backup_region
    
    def create(self, user_id: str, **kwargs) -> Session:
        session = self._create_session(user_id, **kwargs)
        
        # Write to local
        self._save(self.local, session)
        
        # Async replicate to backup
        threading.Thread(
            target=self._save,
            args=(self.backup, session),
            daemon=True
        ).start()
        
        return session
    
    def get(self, session_id: str) -> Optional[Session]:
        # Try local first
        session = self._load(self.local, session_id)
        if session:
            return session
        
        # Try backup (maybe we're recovering from local failure)
        session = self._load(self.backup, session_id)
        if session:
            # Restore to local
            self._save(self.local, session)
        
        return session
    
    def _save(self, client, session: Session):
        key = f"session:{session.session_id}"
        client.setex(key, 86400, json.dumps(asdict(session)))
    
    def _load(self, client, session_id: str) -> Optional[Session]:
        try:
            key = f"session:{session_id}"
            data = client.get(key)
            return Session(**json.loads(data)) if data else None
        except redis.RedisError:
            return None

# Session Store Technology Decision

## Context
We need a session store for 10M concurrent users with <5ms P99 latency
and 99.99% availability.

## Decision
We will use **Redis Cluster** via AWS ElastiCache.

## Rationale

### Why Redis
1. **Latency**: Sub-millisecond reads meet our <5ms requirement with margin
2. **TTL**: Native TTL support matches session lifecycle
3. **Proven**: One of the most widely used session stores in production

### Why Not DynamoDB
1. **Latency**: 5-10ms baseline doesn't leave margin for our 5ms target
2. **Cost**: At 3M reads/sec, on-demand pricing is prohibitive
3. **Complexity**: DAX required to match Redis latency, adding another layer

### Why Not Custom
1. **Operational burden**: Managing PostgreSQL + Redis is more work than managed Redis
2. **Not needed**: We don't require SQL capabilities for sessions

## Architecture
- 6 shards with 1 replica each (12 nodes total)
- ElastiCache (Redis OSS 7.x or Valkey engine)
- Multi-AZ deployment for high availability
- Local application caching (5-second TTL) to reduce load

## Multi-Region Strategy
- Primary region: US-East
- Sessions are region-affine (encoded in session ID)
- Cross-region reads for mobile users who travel
- Async backup replication to US-West for DR

## Failure Handling
- Redis node failure: Automatic failover to replica (<30 seconds)
- Datacenter failure: Users re-authenticate (acceptable UX)
- Full region failure: DR failover to US-West (manual, 15-minute RTO)

## Cost Estimate
- ElastiCache: ~$1,800/month (on-demand; reserved nodes cut this substantially)
- Data transfer: ~$200/month
- Total: ~$2,000/month

## Monitoring
- Latency: Alert if P99 > 3ms
- Memory: Alert if > 80% utilization
- Replication lag: Alert if > 1 second
- Connection count: Alert if > 80% of max

## Review Date
Re-evaluate in 6 months or when reaching 50M concurrent sessions.

import secrets

def generate_session_id() -> str:
    # 256 bits of randomness = practically unguessable
    return f"sess_{secrets.token_urlsafe(32)}"

# Good: sess_7Hj2kL9mNpQrStUvWxYz1234567890abcdef
# Bad:  sess_1, sess_2, sess_3 (sequential = guessable)

def login(username: str, password: str, old_session_id: Optional[str]) -> Session:
    user = authenticate(username, password)
    
    if old_session_id:
        # Destroy old session to prevent fixation
        session_store.delete(old_session_id)
    
    # Create new session with new ID
    return session_store.create(user.id)

response.set_cookie(
    'session_id',
    session.session_id,
    httponly=True,      # JavaScript can't access
    secure=True,        # HTTPS only
    samesite='Lax',     # CSRF protection
    max_age=86400,      # 24 hours
    domain='.example.com'  # Subdomain sharing if needed
)

from cryptography.fernet import Fernet

class EncryptedSessionStore:
    def __init__(self, session_store: SessionStore, encryption_key: bytes):
        self.store = session_store
        self.cipher = Fernet(encryption_key)
    
    def create(self, user_id: str, **kwargs) -> Session:
        session = self.store.create(user_id, **kwargs)
        
        # Encrypt sensitive fields
        if 'data' in kwargs:
            encrypted = self.cipher.encrypt(json.dumps(kwargs['data']).encode())
            session.data = {'_encrypted': encrypted.decode()}
            self.store.update(session.session_id, data=session.data)
        
        return session
    
    def get(self, session_id: str) -> Optional[Session]:
        session = self.store.get(session_id)
        if session and '_encrypted' in session.data:
            decrypted = self.cipher.decrypt(session.data['_encrypted'].encode())
            session.data = json.loads(decrypted)
        return session

from prometheus_client import Counter, Histogram, Gauge

# Operation counts
session_operations = Counter(
    'session_operations_total',
    'Session store operations',
    ['operation', 'status']  # operation: create/get/update/delete, status: success/failure
)

# Latency
session_latency = Histogram(
    'session_operation_seconds',
    'Session operation latency',
    ['operation'],
    buckets=[.001, .0025, .005, .01, .025, .05, .1, .25, .5, 1]
)

# Active sessions
active_sessions = Gauge(
    'active_sessions_total',
    'Number of active sessions'
)

# Cache hit rate
cache_hits = Counter(
    'session_cache_hits_total',
    'Local cache hits'
)
cache_misses = Counter(
    'session_cache_misses_total',
    'Local cache misses'
)

async def session_store_health() -> dict:
    """Health check for session store."""
    results = {
        'status': 'healthy',
        'checks': {}
    }
    
    # Check Redis connectivity
    try:
        start = time.time()
        await redis.ping()
        latency = (time.time() - start) * 1000
        
        results['checks']['redis_ping'] = {
            'status': 'pass',
            'latency_ms': latency
        }
    except Exception as e:
        results['status'] = 'unhealthy'
        results['checks']['redis_ping'] = {
            'status': 'fail',
            'error': str(e)
        }
    
    # Check read/write
    try:
        test_key = f"health_check_{uuid.uuid4()}"
        await redis.setex(test_key, 10, 'test')
        value = await redis.get(test_key)
        await redis.delete(test_key)
        
        results['checks']['redis_readwrite'] = {'status': 'pass'}
    except Exception as e:
        results['status'] = 'unhealthy'
        results['checks']['redis_readwrite'] = {
            'status': 'fail',
            'error': str(e)
        }
    
    return results

groups:
  - name: session_store
    rules:
      - alert: SessionStoreHighLatency
        expr: histogram_quantile(0.99, rate(session_operation_seconds_bucket[5m])) > 0.005
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Session store P99 latency > 5ms"
      
      - alert: SessionStoreErrors
        expr: |
          sum(rate(session_operations_total{status="failure"}[5m]))
            / sum(rate(session_operations_total[5m])) > 0.01
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Session store error rate > 1%"
      
      - alert: SessionStoreCacheLow
        expr: rate(session_cache_hits_total[5m]) / (rate(session_cache_hits_total[5m]) + rate(session_cache_misses_total[5m])) < 0.8
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Session cache hit rate below 80%"

"""
Production session store implementation tying together all Week 1 concepts.
"""

import json
import secrets
import time
import threading
import logging
from dataclasses import dataclass, asdict, field
from typing import Optional, Dict, Set, Any
from enum import Enum
import redis
from cachetools import TTLCache
from prometheus_client import Counter, Histogram, Gauge

# ============================================================================
# Metrics
# ============================================================================

session_ops = Counter(
    'session_operations_total',
    'Session store operations',
    ['operation', 'status']
)

session_latency = Histogram(
    'session_operation_seconds',
    'Session operation latency',
    ['operation'],
    buckets=[.0005, .001, .0025, .005, .01, .025, .05, .1]
)

cache_stats = Counter(
    'session_cache_total',
    'Session cache statistics',
    ['result']  # hit, miss
)

active_sessions = Gauge(
    'active_sessions_total',
    'Estimated active sessions'
)

# ============================================================================
# Data Models
# ============================================================================

@dataclass
class Session:
    session_id: str
    user_id: str
    created_at: int
    last_accessed: int
    expires_at: int
    roles: list = field(default_factory=list)
    permissions: list = field(default_factory=list)
    locale: str = 'en-US'
    timezone: str = 'UTC'
    csrf_token: str = ''
    data: dict = field(default_factory=dict)
    
    def is_expired(self) -> bool:
        return time.time() > self.expires_at
    
    def to_dict(self) -> dict:
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: dict) -> 'Session':
        return cls(**data)

class Region(Enum):
    US_EAST = "use1"
    US_WEST = "usw2"
    EU_WEST = "euw1"

# ============================================================================
# Session Store Implementation
# ============================================================================

class ProductionSessionStore:
    """
    Production session store with:
    - Redis Cluster backend
    - Local caching
    - Multi-region support
    - Comprehensive metrics
    - Security features
    """
    
    def __init__(
        self,
        redis_cluster: redis.RedisCluster,
        region: Region = Region.US_EAST,
        default_ttl: int = 86400,
        local_cache_ttl: int = 5,
        local_cache_size: int = 10000,
        extend_on_access: bool = True
    ):
        self.redis = redis_cluster
        self.region = region
        self.default_ttl = default_ttl
        self.extend_on_access = extend_on_access
        
        # Local cache
        self.local_cache = TTLCache(maxsize=local_cache_size, ttl=local_cache_ttl)
        self.cache_lock = threading.Lock()
        
        # Invalidation via pub/sub
        self._start_invalidation_listener()
        
        self.logger = logging.getLogger('session_store')
    
    # -------------------------------------------------------------------------
    # Core Operations
    # -------------------------------------------------------------------------
    
    def create(
        self,
        user_id: str,
        roles: list = None,
        permissions: list = None,
        **kwargs
    ) -> Session:
        """Create a new session."""
        start = time.time()
        
        try:
            now = int(time.time())
            session = Session(
                session_id=self._generate_session_id(),
                user_id=user_id,
                created_at=now,
                last_accessed=now,
                expires_at=now + self.default_ttl,
                roles=roles or [],
                permissions=permissions or [],
                csrf_token=secrets.token_urlsafe(32),
                **kwargs
            )
            
            # Store in Redis
            self._store_session(session)
            
            # Add to user index
            self._add_to_user_index(session)
            
            session_ops.labels(operation='create', status='success').inc()
            return session
            
        except Exception as e:
            session_ops.labels(operation='create', status='failure').inc()
            self.logger.error(f"Failed to create session: {e}")
            raise
        
        finally:
            session_latency.labels(operation='create').observe(time.time() - start)
    
    def get(self, session_id: str) -> Optional[Session]:
        """Retrieve a session by ID."""
        start = time.time()
        
        try:
            # Check local cache first
            session = self._check_local_cache(session_id)
            if session:
                cache_stats.labels(result='hit').inc()
                session_ops.labels(operation='get', status='success').inc()
                return session
            
            cache_stats.labels(result='miss').inc()
            
            # Fetch from Redis
            session = self._fetch_from_redis(session_id)
            
            if session:
                if session.is_expired():
                    self.delete(session_id)
                    return None
                
                # Extend TTL if configured
                if self.extend_on_access:
                    self._extend_ttl(session)
                
                # Update local cache
                self._update_local_cache(session)
                
                session_ops.labels(operation='get', status='success').inc()
            else:
                session_ops.labels(operation='get', status='not_found').inc()
            
            return session
            
        except Exception as e:
            session_ops.labels(operation='get', status='failure').inc()
            self.logger.error(f"Failed to get session: {e}")
            return None
        
        finally:
            session_latency.labels(operation='get').observe(time.time() - start)
    
    def update(self, session_id: str, **updates) -> Optional[Session]:
        """Update session data."""
        start = time.time()
        
        try:
            session = self.get(session_id)
            if not session:
                return None
            
            # Apply updates
            for key, value in updates.items():
                if key == 'data':
                    session.data.update(value)
                elif hasattr(session, key):
                    setattr(session, key, value)
            
            session.last_accessed = int(time.time())
            
            # Store and invalidate cache
            self._store_session(session)
            self._invalidate_cache(session_id)
            
            session_ops.labels(operation='update', status='success').inc()
            return session
            
        except Exception as e:
            session_ops.labels(operation='update', status='failure').inc()
            self.logger.error(f"Failed to update session: {e}")
            raise
        
        finally:
            session_latency.labels(operation='update').observe(time.time() - start)
    
    def delete(self, session_id: str) -> bool:
        """Delete a session."""
        start = time.time()
        
        try:
            # Get session to find user_id
            session = self._fetch_from_redis(session_id)
            
            if session:
                # Remove from user index
                self.redis.srem(f"user_sessions:{session.user_id}", session_id)
            
            # Delete session
            result = self.redis.delete(f"session:{session_id}") > 0
            
            # Invalidate cache
            self._invalidate_cache(session_id)
            
            session_ops.labels(operation='delete', status='success').inc()
            return result
            
        except Exception as e:
            session_ops.labels(operation='delete', status='failure').inc()
            self.logger.error(f"Failed to delete session: {e}")
            return False
        
        finally:
            session_latency.labels(operation='delete').observe(time.time() - start)
    
    def delete_all_for_user(self, user_id: str) -> int:
        """Delete all sessions for a user (logout all devices)."""
        session_ids = self.redis.smembers(f"user_sessions:{user_id}")
        
        if not session_ids:
            return 0
        
        pipe = self.redis.pipeline()
        for sid in session_ids:
            sid_str = sid.decode() if isinstance(sid, bytes) else sid
            pipe.delete(f"session:{sid_str}")
            self._invalidate_cache(sid_str)
        
        pipe.delete(f"user_sessions:{user_id}")
        results = pipe.execute()
        
        return sum(1 for r in results[:-1] if r > 0)
    
    # -------------------------------------------------------------------------
    # Validation
    # -------------------------------------------------------------------------
    
    def validate(
        self,
        session_id: str,
        csrf_token: Optional[str] = None,
        required_roles: Optional[list] = None
    ) -> tuple[bool, Optional[str]]:
        """
        Validate a session.
        Returns (is_valid, error_message).
        """
        session = self.get(session_id)
        
        if not session:
            return False, "Session not found"
        
        if session.is_expired():
            return False, "Session expired"
        
        if csrf_token and session.csrf_token != csrf_token:
            return False, "Invalid CSRF token"
        
        if required_roles:
            if not set(required_roles).issubset(set(session.roles)):
                return False, "Insufficient permissions"
        
        return True, None
    
    # -------------------------------------------------------------------------
    # Internal Methods
    # -------------------------------------------------------------------------
    
    def _generate_session_id(self) -> str:
        """Generate a secure session ID with region prefix."""
        random_part = secrets.token_urlsafe(32)
        return f"sess_{self.region.value}_{random_part}"
    
    def _store_session(self, session: Session):
        """Store session in Redis."""
        key = f"session:{session.session_id}"
        ttl = session.expires_at - int(time.time())
        self.redis.setex(key, max(ttl, 1), json.dumps(session.to_dict()))
    
    def _fetch_from_redis(self, session_id: str) -> Optional[Session]:
        """Fetch session from Redis."""
        key = f"session:{session_id}"
        data = self.redis.get(key)
        
        if data:
            return Session.from_dict(json.loads(data))
        return None
    
    def _add_to_user_index(self, session: Session):
        """Add session to user's session index."""
        key = f"user_sessions:{session.user_id}"
        self.redis.sadd(key, session.session_id)
        self.redis.expire(key, self.default_ttl)
    
    def _extend_ttl(self, session: Session):
        """Extend session TTL."""
        now = int(time.time())
        session.last_accessed = now
        session.expires_at = now + self.default_ttl
        self._store_session(session)
    
    def _check_local_cache(self, session_id: str) -> Optional[Session]:
        """Check local cache for session."""
        with self.cache_lock:
            return self.local_cache.get(session_id)
    
    def _update_local_cache(self, session: Session):
        """Update local cache."""
        with self.cache_lock:
            self.local_cache[session.session_id] = session
    
    def _invalidate_cache(self, session_id: str):
        """Invalidate local cache and broadcast to other servers."""
        with self.cache_lock:
            self.local_cache.pop(session_id, None)
        
        # Broadcast invalidation
        try:
            self.redis.publish('session_invalidation', session_id)
        except Exception as e:
            self.logger.warning(f"Failed to broadcast invalidation: {e}")
    
    def _start_invalidation_listener(self):
        """Start pub/sub listener for cache invalidation."""
        def listen():
            try:
                pubsub = self.redis.pubsub()
                pubsub.subscribe('session_invalidation')
                
                for message in pubsub.listen():
                    if message['type'] == 'message':
                        session_id = message['data']
                        if isinstance(session_id, bytes):
                            session_id = session_id.decode()
                        
                        with self.cache_lock:
                            self.local_cache.pop(session_id, None)
            
            except Exception as e:
                self.logger.error(f"Invalidation listener error: {e}")
                # Reconnect after delay
                time.sleep(5)
                self._start_invalidation_listener()
        
        thread = threading.Thread(target=listen, daemon=True)
        thread.start()
    
    # -------------------------------------------------------------------------
    # Health Check
    # -------------------------------------------------------------------------
    
    def health_check(self) -> dict:
        """Check session store health."""
        result = {
            'status': 'healthy',
            'checks': {}
        }
        
        # Redis ping
        try:
            start = time.time()
            self.redis.ping()
            latency = (time.time() - start) * 1000
            
            result['checks']['redis_ping'] = {
                'status': 'pass',
                'latency_ms': round(latency, 2)
            }
        except Exception as e:
            result['status'] = 'unhealthy'
            result['checks']['redis_ping'] = {
                'status': 'fail',
                'error': str(e)
            }
        
        # Read/write test
        try:
            test_id = f"health_{secrets.token_hex(8)}"
            test_session = self.create(user_id='health_check')
            retrieved = self.get(test_session.session_id)
            self.delete(test_session.session_id)
            
            if retrieved and retrieved.user_id == 'health_check':
                result['checks']['read_write'] = {'status': 'pass'}
            else:
                result['status'] = 'unhealthy'
                result['checks']['read_write'] = {'status': 'fail'}
        except Exception as e:
            result['status'] = 'unhealthy'
            result['checks']['read_write'] = {
                'status': 'fail',
                'error': str(e)
            }
        
        return result


# ============================================================================
# Multi-Region Wrapper
# ============================================================================

class MultiRegionSessionStore:
    """
    Session store supporting multiple regions with session affinity.
    """
    
    def __init__(
        self,
        local_region: Region,
        stores: Dict[Region, ProductionSessionStore]
    ):
        self.local_region = local_region
        self.stores = stores
        self.local_store = stores[local_region]
    
    def create(self, user_id: str, **kwargs) -> Session:
        """Create session in local region."""
        return self.local_store.create(user_id, **kwargs)
    
    def get(self, session_id: str) -> Optional[Session]:
        """Get session from appropriate region."""
        region = self._extract_region(session_id)
        store = self.stores.get(region, self.local_store)
        return store.get(session_id)
    
    def _extract_region(self, session_id: str) -> Region:
        """Extract region from session ID."""
        # Format: sess_{region}_{random}
        parts = session_id.split('_')
        if len(parts) >= 2:
            region_code = parts[1]
            for region in Region:
                if region.value == region_code:
                    return region
        return self.local_region
    
    def get_redirect_url(self, session_id: str) -> Optional[str]:
        """Get redirect URL if session is in different region."""
        region = self._extract_region(session_id)
        if region != self.local_region:
            return self._region_urls.get(region)
        return None
    
    _region_urls = {
        Region.US_EAST: "https://us-east.example.com",
        Region.US_WEST: "https://us-west.example.com",
        Region.EU_WEST: "https://eu-west.example.com"
    }