Payment Channel Management at Scale

-- Primary channel tracking table
CREATE TABLE payment_channels (
    channel_id VARCHAR(64) PRIMARY KEY,
    account_source VARCHAR(34) NOT NULL,
    account_destination VARCHAR(34) NOT NULL,
    amount_funded BIGINT NOT NULL,
    amount_claimed BIGINT DEFAULT 0,
    amount_reserved BIGINT DEFAULT 0,
    state ENUM('opening', 'active', 'settling', 'closed') NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_activity TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    settle_delay INTEGER NOT NULL,
    expiration TIMESTAMP,
    close_signature VARCHAR(128),
    INDEX idx_state_activity (state, last_activity),
    INDEX idx_destination_state (account_destination, state),
    INDEX idx_expiration (expiration)
);

-- Channel activity log for monitoring and analytics
CREATE TABLE channel_activities (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    channel_id VARCHAR(64) NOT NULL,
    activity_type ENUM('payment', 'claim', 'close_request', 'state_change') NOT NULL,
    amount BIGINT,
    previous_state VARCHAR(20),
    new_state VARCHAR(20),
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    metadata JSON,
    INDEX idx_channel_time (channel_id, timestamp),
    INDEX idx_type_time (activity_type, timestamp),
    FOREIGN KEY (channel_id) REFERENCES payment_channels(channel_id)
);

class ChannelStateMachine:
    VALID_TRANSITIONS = {
        'opening': ['active', 'closed'],  # Success or failure
        'active': ['settling', 'closed'],  # Normal close or force close
        'settling': ['closed'],            # Only forward progression
        'closed': []                       # Terminal state
    }
    
    def transition_channel(self, channel_id, new_state, context=None):
        current_state = self.get_channel_state(channel_id)
        
        if new_state not in self.VALID_TRANSITIONS[current_state]:
            raise InvalidTransitionError(
                f"Cannot transition from {current_state} to {new_state}"
            )
        
        # Execute state-specific logic
        if new_state == 'settling':
            self._initiate_settlement(channel_id, context)
        elif new_state == 'closed':
            self._finalize_closure(channel_id, context)
            
        self._update_channel_state(channel_id, new_state)
        self._log_state_change(channel_id, current_state, new_state)

class SettlementBatcher:
    def __init__(self, max_batch_size=100, max_wait_time=3600):
        self.max_batch_size = max_batch_size
        self.max_wait_time = max_wait_time
        self.pending_settlements = []
        
    def queue_settlement(self, channel_id, priority='normal'):
        settlement = {
            'channel_id': channel_id,
            'queued_at': time.time(),
            'priority': priority,
            'estimated_value': self._calculate_settlement_value(channel_id)
        }
        
        self.pending_settlements.append(settlement)
        
        # Check if batch should be processed immediately
        if (len(self.pending_settlements) >= self.max_batch_size or
            self._has_high_priority_settlements() or
            self._oldest_settlement_expired()):
            self.process_batch()
    
    def process_batch(self):
        if not self.pending_settlements:
            return
            
        # Sort by priority and value
        batch = sorted(self.pending_settlements, 
                      key=lambda x: (x['priority'] == 'high', x['estimated_value']), 
                      reverse=True)[:self.max_batch_size]
        
        # Execute batch settlement
        self._execute_settlement_batch(batch)
        
        # Remove processed settlements
        self.pending_settlements = [s for s in self.pending_settlements 
                                  if s not in batch]

class ChannelRecyclingManager:
    def __init__(self, recycling_threshold=0.1, min_remaining_capacity=0.2):
        self.recycling_threshold = recycling_threshold  # 10% of original capacity
        self.min_remaining_capacity = min_remaining_capacity  # 20% remaining
        
    def evaluate_recycling_candidate(self, channel_id):
        channel = self.get_channel(channel_id)
        
        # Calculate utilization metrics
        utilization_rate = channel.amount_claimed / channel.amount_funded
        remaining_capacity = (channel.amount_funded - channel.amount_claimed) / channel.amount_funded
        
        # Check recycling criteria
        can_recycle = (
            channel.state == 'active' and
            utilization_rate >= self.recycling_threshold and
            remaining_capacity >= self.min_remaining_capacity and
            self._counterparty_is_active(channel.account_destination) and
            self._no_pending_disputes(channel_id)
        )
        
        return {
            'recyclable': can_recycle,
            'utilization_rate': utilization_rate,
            'remaining_capacity': remaining_capacity,
            'estimated_value': remaining_capacity * channel.amount_funded
        }
    
    def recycle_channel(self, channel_id, new_session_context):
        if not self.evaluate_recycling_candidate(channel_id)['recyclable']:
            raise ChannelNotRecyclableError("Channel does not meet recycling criteria")
        
        # Reset channel for new session
        self._clear_session_state(channel_id)
        self._initialize_new_session(channel_id, new_session_context)
        self._log_recycling_event(channel_id, new_session_context)
        
        return channel_id

def calculate_optimal_capacity(user_id, session_type):
    # Analyze historical usage patterns
    historical_sessions = get_user_sessions(user_id, session_type, limit=50)
    
    if len(historical_sessions) < 5:
        # Insufficient data, use conservative defaults
        return get_default_capacity(session_type)
    
    # Calculate usage statistics
    session_values = [s.total_payments for s in historical_sessions]
    mean_usage = np.mean(session_values)
    std_usage = np.std(session_values)
    
    # Provision for 95th percentile usage to minimize friction
    capacity_target = mean_usage + (1.96 * std_usage)
    
    # Apply business constraints
    min_capacity = get_minimum_capacity(session_type)
    max_capacity = get_maximum_capacity(user_id)  # Based on user tier/credit
    
    optimal_capacity = max(min_capacity, min(capacity_target, max_capacity))
    
    return {
        'recommended_capacity': optimal_capacity,
        'confidence_level': len(historical_sessions) / 50.0,
        'expected_utilization': mean_usage / optimal_capacity,
        'over_provision_risk': (optimal_capacity - mean_usage) / optimal_capacity
    }

class ChannelHealthMonitor:
    def __init__(self):
        self.health_thresholds = {
            'response_time_p95': 500,  # 95th percentile response time in ms
            'success_rate_1h': 0.99,   # Success rate over 1 hour window
            'capacity_utilization': 0.8, # Warning threshold for capacity usage
            'inactive_duration': 1800,   # 30 minutes without activity
            'settlement_delay_variance': 0.1  # 10% variance from expected
        }
    
    def assess_channel_health(self, channel_id):
        metrics = self._collect_channel_metrics(channel_id)
        
        health_score = 1.0
        issues = []
        
        # Response time assessment
        if metrics['response_time_p95'] > self.health_thresholds['response_time_p95']:
            health_score *= 0.8
            issues.append(f"High response time: {metrics['response_time_p95']}ms")
        
        # Success rate assessment
        if metrics['success_rate_1h'] < self.health_thresholds['success_rate_1h']:
            health_score *= 0.7
            issues.append(f"Low success rate: {metrics['success_rate_1h']:.2%}")
        
        # Capacity utilization
        if metrics['capacity_utilization'] > self.health_thresholds['capacity_utilization']:
            health_score *= 0.9
            issues.append(f"High capacity usage: {metrics['capacity_utilization']:.2%}")
        
        # Activity monitoring
        if metrics['inactive_duration'] > self.health_thresholds['inactive_duration']:
            health_score *= 0.85
            issues.append(f"Extended inactivity: {metrics['inactive_duration']}s")
        
        return {
            'health_score': health_score,
            'status': self._categorize_health(health_score),
            'issues': issues,
            'metrics': metrics,
            'recommendations': self._generate_recommendations(health_score, issues)
        }
    
    def _categorize_health(self, score):
        if score >= 0.9:
            return 'healthy'
        elif score >= 0.7:
            return 'warning'
        elif score >= 0.5:
            return 'critical'
        else:
            return 'failing'

class PredictiveHealthAnalyzer:
    def __init__(self):
        self.trend_window = 3600  # 1 hour trend analysis
        self.prediction_horizon = 1800  # 30 minute prediction window
        
    def predict_channel_issues(self, channel_id):
        # Collect time-series data
        metrics_history = self._get_metrics_history(channel_id, self.trend_window)
        
        if len(metrics_history) < 10:
            return {'prediction': 'insufficient_data', 'confidence': 0.0}
        
        predictions = {}
        
        # Predict capacity exhaustion
        capacity_trend = self._calculate_trend(metrics_history, 'capacity_utilization')
        if capacity_trend > 0:
            time_to_exhaustion = self._extrapolate_time_to_threshold(
                metrics_history, 'capacity_utilization', 1.0, capacity_trend
            )
            if time_to_exhaustion < self.prediction_horizon:
                predictions['capacity_exhaustion'] = {
                    'probability': 0.8,
                    'estimated_time': time_to_exhaustion,
                    'recommended_action': 'increase_capacity_or_initiate_settlement'
                }
        
        # Predict performance degradation
        response_time_trend = self._calculate_trend(metrics_history, 'response_time')
        if response_time_trend > 0.1:  # 10% degradation trend
            predictions['performance_degradation'] = {
                'probability': 0.6,
                'severity': 'moderate',
                'recommended_action': 'investigate_counterparty_health'
            }
        
        return predictions

class OptimizedChannelDatabase:
    def __init__(self, pool_size=50):
        self.connection_pool = self._create_connection_pool(pool_size)
        self.query_cache = LRUCache(maxsize=10000)
        self.prepared_statements = self._prepare_common_statements()
    
    def get_channels_by_state_batch(self, states, limit=1000):
        # Use prepared statement with IN clause for efficiency
        cache_key = f"channels_by_state_{hash(tuple(sorted(states)))}"
        
        if cache_key in self.query_cache:
            return self.query_cache[cache_key]
        
        with self.connection_pool.get_connection() as conn:
            cursor = conn.cursor(prepared=True)
            placeholders = ','.join(['%s'] * len(states))
            query = self.prepared_statements['channels_by_state'].format(
                placeholders=placeholders
            )
            cursor.execute(query, states + [limit])
            results = cursor.fetchall()
        
        self.query_cache[cache_key] = results
        return results
    
    def bulk_update_channel_states(self, updates):
        # Batch multiple updates into single transaction
        with self.connection_pool.get_connection() as conn:
            cursor = conn.cursor()
            try:
                conn.start_transaction()
                
                # Use bulk update with CASE statement for efficiency
                update_query = """
                    UPDATE payment_channels 
                    SET state = CASE channel_id
                        {}
                    END,
                    last_activity = NOW()
                    WHERE channel_id IN ({})
                """.format(
                    ' '.join([f"WHEN %s THEN %s" for _ in updates]),
                    ','.join(['%s'] * len(updates))
                )
                
                # Flatten update parameters
                params = []
                channel_ids = []
                for channel_id, new_state in updates:
                    params.extend([channel_id, new_state])
                    channel_ids.append(channel_id)
                
                cursor.execute(update_query, params + channel_ids)
                conn.commit()
                
            except Exception as e:
                conn.rollback()
                raise DatabaseUpdateError(f"Bulk update failed: {e}")

class MemoryEfficientChannelTracker:
    def __init__(self, max_memory_channels=5000):
        self.max_memory_channels = max_memory_channels
        self.active_channels = {}  # Hot channels in memory
        self.channel_access_times = {}  # LRU tracking
        
    def get_channel_state(self, channel_id):
        # Check memory cache first
        if channel_id in self.active_channels:
            self.channel_access_times[channel_id] = time.time()
            return self.active_channels[channel_id]
        
        # Load from database and cache if space available
        channel_state = self._load_from_database(channel_id)
        
        if len(self.active_channels) < self.max_memory_channels:
            self.active_channels[channel_id] = channel_state
            self.channel_access_times[channel_id] = time.time()
        else:
            # Evict least recently used channel
            lru_channel = min(self.channel_access_times, 
                            key=self.channel_access_times.get)
            del self.active_channels[lru_channel]
            del self.channel_access_times[lru_channel]
            
            # Add new channel
            self.active_channels[channel_id] = channel_state
            self.channel_access_times[channel_id] = time.time()
        
        return channel_state

import asyncio
import aioredis
from concurrent.futures import ThreadPoolExecutor

class AsyncChannelProcessor:
    def __init__(self, max_workers=20):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.redis_pool = None
        self.processing_queue = asyncio.Queue(maxsize=10000)
        
    async def initialize(self):
        self.redis_pool = await aioredis.create_redis_pool(
            'redis://localhost', encoding='utf-8'
        )
        
        # Start background processing tasks
        for _ in range(5):
            asyncio.create_task(self._process_channel_updates())
    
    async def queue_channel_update(self, channel_id, update_type, data):
        update_item = {
            'channel_id': channel_id,
            'update_type': update_type,
            'data': data,
            'timestamp': time.time()
        }
        
        try:
            await self.processing_queue.put(update_item, timeout=1.0)
        except asyncio.TimeoutError:
            # Queue full, process synchronously as fallback
            await self._process_update_sync(update_item)
    
    async def _process_channel_updates(self):
        while True:
            try:
                update_item = await self.processing_queue.get()
                await self._process_single_update(update_item)
                self.processing_queue.task_done()
            except Exception as e:
                logger.error(f"Error processing channel update: {e}")
                await asyncio.sleep(0.1)  # Brief pause on error
    
    async def _process_single_update(self, update_item):
        # Process update based on type
        if update_item['update_type'] == 'payment':
            await self._process_payment_update(update_item)
        elif update_item['update_type'] == 'state_change':
            await self._process_state_change(update_item)
        elif update_item['update_type'] == 'settlement':
            await self._process_settlement_update(update_item)