ocean_cloud_database.py

#!/usr/bin/env python3
"""
🌊 OCEAN CLOUD DATABASE - Neon PostgreSQL Integration
====================================================

This implements the cloud database layer for OCEAN Chess:
- Neon PostgreSQL connection
- AWS S3 model storage
- Redis caching layer
- Background learning jobs
- Real-time analytics
"""

import os
import json
import logging
import asyncio
import asyncpg
import redis
import boto3
from datetime import datetime, timezone
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass
import uuid

logger = logging.getLogger(__name__)

@dataclass
class GameData:
    """Game data structure."""
    game_id: str
    player_id: str
    start_time: datetime
    end_time: Optional[datetime] = None
    outcome: Optional[str] = None
    move_count: int = 0
    difficulty_level: float = 0.0

@dataclass
class MoveData:
    """Move data structure."""
    game_id: str
    move_number: int
    player: str
    move_uci: str
    fen_before: str
    fen_after: str
    evaluation: float = 0.0
    game_phase: str = "middlegame"
    time_spent: int = 0

class OceanCloudDatabase:
    """
    🌊 OCEAN Cloud Database - Neon PostgreSQL + AWS S3 + Redis
    
    Features:
    - Neon PostgreSQL for persistent data
    - AWS S3 for model storage
    - Redis for caching and sessions
    - Background learning jobs
    - Real-time analytics
    """
    
    def __init__(self):
        """Initialize cloud database connections."""
        self.pg_pool = None
        self.redis_client = None
        self.s3_client = None
        self._initialize_connections()
    
    def _initialize_connections(self):
        """Initialize database connections."""
        # Neon PostgreSQL
        self.neon_url = os.getenv('NEON_DATABASE_URL', 'postgresql://user:pass@localhost/ocean_chess')
        
        # Redis (Upstash)
        self.redis_url = os.getenv('REDIS_URL', 'redis://localhost:6379')
        
        # AWS S3
        self.aws_access_key = os.getenv('AWS_ACCESS_KEY_ID')
        self.aws_secret_key = os.getenv('AWS_SECRET_ACCESS_KEY')
        self.s3_bucket = os.getenv('S3_BUCKET', 'ocean-chess-models')
        
        logger.info("🌊 Initializing cloud database connections...")
    
    async def connect(self):
        """Connect to all database services."""
        try:
            # Connect to Neon PostgreSQL
            self.pg_pool = await asyncpg.create_pool(
                self.neon_url,
                min_size=1,
                max_size=10,
                command_timeout=60
            )
            logger.info("✅ Connected to Neon PostgreSQL")
            
            # Connect to Redis
            self.redis_client = redis.from_url(self.redis_url, decode_responses=True)
            self.redis_client.ping()
            logger.info("✅ Connected to Redis")
            
            # Connect to AWS S3
            if self.aws_access_key and self.aws_secret_key:
                self.s3_client = boto3.client(
                    's3',
                    aws_access_key_id=self.aws_access_key,
                    aws_secret_access_key=self.aws_secret_key
                )
                logger.info("✅ Connected to AWS S3")
            else:
                logger.warning("⚠️ AWS credentials not found - S3 features disabled")
            
            # Initialize database schema
            await self._initialize_schema()
            
        except Exception as e:
            logger.error(f"❌ Failed to connect to cloud databases: {e}")
            raise
    
    async def _initialize_schema(self):
        """Initialize database schema."""
        async with self.pg_pool.acquire() as conn:
            # Create tables if they don't exist
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS games (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    game_id VARCHAR(50) UNIQUE NOT NULL,
                    player_id VARCHAR(50) NOT NULL,
                    start_time TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
                    end_time TIMESTAMP WITH TIME ZONE,
                    outcome VARCHAR(20) CHECK (outcome IN ('ocean_win', 'player_win', 'draw')),
                    move_count INTEGER DEFAULT 0,
                    game_phase VARCHAR(20) DEFAULT 'opening',
                    difficulty_level REAL DEFAULT 0.0,
                    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
                    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
                )
            ''')
            
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS moves (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    game_id VARCHAR(50) NOT NULL,
                    move_number INTEGER NOT NULL,
                    player VARCHAR(10) CHECK (player IN ('ocean', 'human')),
                    move_uci VARCHAR(10) NOT NULL,
                    fen_before TEXT NOT NULL,
                    fen_after TEXT NOT NULL,
                    evaluation REAL DEFAULT 0.0,
                    game_phase VARCHAR(20) DEFAULT 'middlegame',
                    time_spent INTEGER DEFAULT 0,
                    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
                )
            ''')
            
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS player_profiles (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    player_id VARCHAR(50) UNIQUE NOT NULL,
                    total_games INTEGER DEFAULT 0,
                    wins INTEGER DEFAULT 0,
                    losses INTEGER DEFAULT 0,
                    draws INTEGER DEFAULT 0,
                    preferred_openings JSONB DEFAULT '[]',
                    playing_style JSONB DEFAULT '{}',
                    weakness_patterns JSONB DEFAULT '[]',
                    strength_patterns JSONB DEFAULT '[]',
                    elo_rating INTEGER DEFAULT 1200,
                    last_played TIMESTAMP WITH TIME ZONE,
                    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
                    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
                )
            ''')
            
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS learning_patterns (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    pattern_type VARCHAR(50) NOT NULL,
                    pattern_data JSONB NOT NULL,
                    success_rate REAL DEFAULT 0.0,
                    usage_count INTEGER DEFAULT 0,
                    confidence_level REAL DEFAULT 0.0,
                    last_used TIMESTAMP WITH TIME ZONE,
                    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
                    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
                )
            ''')
            
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS model_versions (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    version_name VARCHAR(50) UNIQUE NOT NULL,
                    model_type VARCHAR(50) NOT NULL,
                    s3_path TEXT NOT NULL,
                    performance_metrics JSONB DEFAULT '{}',
                    is_active BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
                )
            ''')
            
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS learning_metrics (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    metric_name VARCHAR(100) NOT NULL,
                    metric_value REAL NOT NULL,
                    metric_type VARCHAR(50) DEFAULT 'performance',
                    player_id VARCHAR(50),
                    timestamp TIMESTAMP WITH TIME ZONE DEFAULT NOW()
                )
            ''')
            
            logger.info("📊 Database schema initialized")
    
    async def start_game(self, player_id: str = "anonymous") -> str:
        """Start a new game."""
        game_id = f"game_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}"
        
        async with self.pg_pool.acquire() as conn:
            await conn.execute('''
                INSERT INTO games (game_id, player_id, start_time)
                VALUES ($1, $2, $3)
            ''', game_id, player_id, datetime.now(timezone.utc))
        
        # Cache game start
        await self._cache_game_data(game_id, {"status": "started", "player_id": player_id})
        
        logger.info(f"🎮 Started new game: {game_id}")
        return game_id
    
    async def record_move(self, move_data: MoveData):
        """Record a move in the database."""
        async with self.pg_pool.acquire() as conn:
            await conn.execute('''
                INSERT INTO moves (game_id, move_number, player, move_uci, fen_before, fen_after, evaluation, game_phase, time_spent)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            ''', move_data.game_id, move_data.move_number, move_data.player, 
                move_data.move_uci, move_data.fen_before, move_data.fen_after,
                move_data.evaluation, move_data.game_phase, move_data.time_spent)
        
        # Update game move count
        await self._update_game_move_count(move_data.game_id)
        
        # Cache move for real-time analysis
        await self._cache_move_data(move_data)
    
    async def end_game(self, game_id: str, outcome: str, final_fen: str):
        """End a game and update statistics."""
        async with self.pg_pool.acquire() as conn:
            # Update game record
            await conn.execute('''
                UPDATE games 
                SET end_time = $1, outcome = $2, updated_at = $3
                WHERE game_id = $4
            ''', datetime.now(timezone.utc), outcome, datetime.now(timezone.utc), game_id)
            
            # Get game data
            game_data = await conn.fetchrow('''
                SELECT player_id, move_count FROM games WHERE game_id = $1
            ''', game_id)
            
            if game_data:
                player_id = game_data['player_id']
                move_count = game_data['move_count']
                
                # Update player profile
                await self._update_player_profile(player_id, outcome)
                
                # Update learning metrics
                await self._update_learning_metrics(player_id, outcome, move_count)
                
                # Trigger background learning job
                await self._trigger_learning_job(game_id, outcome)
        
        # Clear game cache
        await self._clear_game_cache(game_id)
        
        logger.info(f"🏁 Game ended: {game_id} - {outcome}")
    
    async def _update_player_profile(self, player_id: str, outcome: str):
        """Update player profile statistics."""
        async with self.pg_pool.acquire() as conn:
            # Check if profile exists
            profile = await conn.fetchrow('''
                SELECT id FROM player_profiles WHERE player_id = $1
            ''', player_id)
            
            if profile:
                # Update existing profile
                if outcome == "player_win":
                    await conn.execute('''
                        UPDATE player_profiles 
                        SET wins = wins + 1, total_games = total_games + 1, 
                            last_played = $1, updated_at = $1
                        WHERE player_id = $2
                    ''', datetime.now(timezone.utc), player_id)
                elif outcome == "ocean_win":
                    await conn.execute('''
                        UPDATE player_profiles 
                        SET losses = losses + 1, total_games = total_games + 1,
                            last_played = $1, updated_at = $1
                        WHERE player_id = $2
                    ''', datetime.now(timezone.utc), player_id)
                else:  # draw
                    await conn.execute('''
                        UPDATE player_profiles 
                        SET draws = draws + 1, total_games = total_games + 1,
                            last_played = $1, updated_at = $1
                        WHERE player_id = $2
                    ''', datetime.now(timezone.utc), player_id)
            else:
                # Create new profile
                wins = 1 if outcome == "player_win" else 0
                losses = 1 if outcome == "ocean_win" else 0
                draws = 1 if outcome == "draw" else 0
                
                await conn.execute('''
                    INSERT INTO player_profiles (player_id, total_games, wins, losses, draws, last_played)
                    VALUES ($1, 1, $2, $3, $4, $5)
                ''', player_id, wins, losses, draws, datetime.now(timezone.utc))
    
    async def _update_learning_metrics(self, player_id: str, outcome: str, move_count: int):
        """Update learning metrics."""
        async with self.pg_pool.acquire() as conn:
            # Calculate adaptation level based on recent performance
            recent_games = await conn.fetch('''
                SELECT outcome FROM games 
                WHERE player_id = $1 
                ORDER BY start_time DESC 
                LIMIT 10
            ''', player_id)
            
            if recent_games:
                wins = sum(1 for game in recent_games if game['outcome'] == 'player_win')
                adaptation_level = wins / len(recent_games)
                
                await conn.execute('''
                    INSERT INTO learning_metrics (metric_name, metric_value, metric_type, player_id)
                    VALUES ($1, $2, $3, $4)
                ''', 'adaptation_level', adaptation_level, 'performance', player_id)
    
    async def _trigger_learning_job(self, game_id: str, outcome: str):
        """Trigger background learning job."""
        # This would typically use a job queue like AWS SQS or Upstash QStash
        # For now, we'll just log it
        logger.info(f"🧠 Triggering learning job for game {game_id} with outcome {outcome}")
        
        # In production, this would:
        # 1. Add job to queue
        # 2. Background worker processes the game
        # 3. Updates learning patterns
        # 4. Retrains models if needed
    
    async def get_player_profile(self, player_id: str) -> Optional[Dict]:
        """Get player profile."""
        # Try cache first
        cached = await self._get_cached_profile(player_id)
        if cached:
            return cached
        
        async with self.pg_pool.acquire() as conn:
            profile = await conn.fetchrow('''
                SELECT * FROM player_profiles WHERE player_id = $1
            ''', player_id)
            
            if profile:
                profile_dict = dict(profile)
                # Cache the result
                await self._cache_profile(player_id, profile_dict)
                return profile_dict
        
        return None
    
    async def get_learning_statistics(self, player_id: str = None) -> Dict:
        """Get comprehensive learning statistics."""
        async with self.pg_pool.acquire() as conn:
            if player_id:
                # Player-specific stats
                stats = await conn.fetchrow('''
                    SELECT 
                        COUNT(*) as total_games,
                        SUM(CASE WHEN outcome = 'player_win' THEN 1 ELSE 0 END) as wins,
                        SUM(CASE WHEN outcome = 'ocean_win' THEN 1 ELSE 0 END) as losses,
                        SUM(CASE WHEN outcome = 'draw' THEN 1 ELSE 0 END) as draws
                    FROM games 
                    WHERE player_id = $1
                ''', player_id)
                
                # Get recent adaptation level
                adaptation = await conn.fetchrow('''
                    SELECT metric_value FROM learning_metrics 
                    WHERE player_id = $1 AND metric_name = 'adaptation_level'
                    ORDER BY timestamp DESC LIMIT 1
                ''', player_id)
                
                return {
                    'player_id': player_id,
                    'total_games': stats['total_games'] or 0,
                    'wins': stats['wins'] or 0,
                    'losses': stats['losses'] or 0,
                    'draws': stats['draws'] or 0,
                    'win_rate': (stats['wins'] or 0) / max(stats['total_games'] or 1, 1),
                    'adaptation_level': adaptation['metric_value'] if adaptation else 0.0
                }
            else:
                # Global stats
                stats = await conn.fetchrow('''
                    SELECT 
                        COUNT(*) as total_games,
                        COUNT(DISTINCT player_id) as unique_players,
                        AVG(move_count) as avg_moves_per_game
                    FROM games
                ''')
                
                return {
                    'total_games': stats['total_games'] or 0,
                    'unique_players': stats['unique_players'] or 0,
                    'avg_moves_per_game': float(stats['avg_moves_per_game'] or 0)
                }
    
    async def save_model(self, model_data: bytes, version: str, model_type: str) -> str:
        """Save model to S3."""
        if not self.s3_client:
            raise Exception("S3 client not initialized")
        
        s3_key = f"models/{version}/{model_type}.pkl"
        
        self.s3_client.put_object(
            Bucket=self.s3_bucket,
            Key=s3_key,
            Body=model_data,
            ContentType='application/octet-stream'
        )
        
        # Record model version in database
        async with self.pg_pool.acquire() as conn:
            await conn.execute('''
                INSERT INTO model_versions (version_name, model_type, s3_path)
                VALUES ($1, $2, $3)
            ''', version, model_type, s3_key)
        
        logger.info(f"💾 Saved model {model_type} version {version} to S3")
        return s3_key
    
    async def load_model(self, version: str, model_type: str) -> bytes:
        """Load model from S3."""
        if not self.s3_client:
            raise Exception("S3 client not initialized")
        
        s3_key = f"models/{version}/{model_type}.pkl"
        
        response = self.s3_client.get_object(Bucket=self.s3_bucket, Key=s3_key)
        return response['Body'].read()
    
    # Caching methods
    async def _cache_game_data(self, game_id: str, data: Dict):
        """Cache game data in Redis."""
        if self.redis_client:
            self.redis_client.setex(f"game:{game_id}", 3600, json.dumps(data))
    
    async def _cache_move_data(self, move_data: MoveData):
        """Cache move data for real-time analysis."""
        if self.redis_client:
            key = f"moves:{move_data.game_id}"
            self.redis_client.lpush(key, json.dumps(move_data.__dict__))
            self.redis_client.expire(key, 3600)
    
    async def _cache_profile(self, player_id: str, profile: Dict):
        """Cache player profile."""
        if self.redis_client:
            self.redis_client.setex(f"profile:{player_id}", 1800, json.dumps(profile))
    
    async def _get_cached_profile(self, player_id: str) -> Optional[Dict]:
        """Get cached player profile."""
        if self.redis_client:
            cached = self.redis_client.get(f"profile:{player_id}")
            if cached:
                return json.loads(cached)
        return None
    
    async def _clear_game_cache(self, game_id: str):
        """Clear game cache."""
        if self.redis_client:
            self.redis_client.delete(f"game:{game_id}")
            self.redis_client.delete(f"moves:{game_id}")
    
    async def _update_game_move_count(self, game_id: str):
        """Update game move count."""
        async with self.pg_pool.acquire() as conn:
            await conn.execute('''
                UPDATE games 
                SET move_count = move_count + 1, updated_at = $1
                WHERE game_id = $2
            ''', datetime.now(timezone.utc), game_id)
    
    async def close(self):
        """Close all database connections."""
        if self.pg_pool:
            await self.pg_pool.close()
        if self.redis_client:
            self.redis_client.close()
        logger.info("🔒 Cloud database connections closed")

# Global database instance
_cloud_db = None

async def get_cloud_database():
    """Get the global cloud database instance."""
    global _cloud_db
    if _cloud_db is None:
        _cloud_db = OceanCloudDatabase()
        await _cloud_db.connect()
    return _cloud_db

async def initialize_cloud_database():
    """Initialize the cloud database."""
    return await get_cloud_database()

if __name__ == "__main__":
    # Test the cloud database
    async def test():
        db = await initialize_cloud_database()
        
        # Test game
        game_id = await db.start_game("test_player")
        
        # Test move
        move = MoveData(
            game_id=game_id,
            move_number=1,
            player="human",
            move_uci="e2e4",
            fen_before="rnbqkbnr/pppppppp/8/8/8/8/PPPPPPPP/RNBQKBNR w KQkq - 0 1",
            fen_after="rnbqkbnr/pppppppp/8/8/4P3/8/PPPP1PPP/RNBQKBNR b KQkq e3 0 1"
        )
        await db.record_move(move)
        
        # Test end game
        await db.end_game(game_id, "player_win", "final_fen")
        
        # Test statistics
        stats = await db.get_learning_statistics("test_player")
        print(f"Learning Statistics: {stats}")
        
        await db.close()
    
    asyncio.run(test())