redis_client.py

"""Redis client module for caching and storage."""

import json
import os
from typing import Optional, Any
import redis.asyncio as redis
from functools import wraps
import hashlib


# Redis configuration from environment variables
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
REDIS_DB = int(os.getenv("REDIS_DB", 0))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
CACHE_TTL = int(os.getenv("CACHE_TTL", 300))  # Default 5 minutes

# Global Redis client instance
_redis_client: Optional[redis.Redis] = None


def get_redis_client() -> redis.Redis:
    """Get or create the Redis client instance."""
    global _redis_client
    if _redis_client is None:
        _redis_client = redis.Redis(
            host=REDIS_HOST,
            port=REDIS_PORT,
            db=REDIS_DB,
            password=REDIS_PASSWORD,
            decode_responses=True,
        )
    return _redis_client


async def init_redis() -> None:
    """Initialize Redis connection."""
    global _redis_client
    _redis_client = redis.Redis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        db=REDIS_DB,
        password=REDIS_PASSWORD,
        decode_responses=True,
    )
    try:
        await _redis_client.ping()
        print(f"✓ Connected to Redis at {REDIS_HOST}:{REDIS_PORT}")
    except redis.ConnectionError as e:
        print(f"⚠ Redis connection failed: {e}")
        print("⚠ Running without Redis caching")


async def close_redis() -> None:
    """Close Redis connection."""
    global _redis_client
    if _redis_client:
        await _redis_client.close()
        _redis_client = None


def _generate_cache_key(prefix: str, *args, **kwargs) -> str:
    """Generate a cache key from arguments."""
    key_parts = [prefix]
    for arg in args:
        key_parts.append(str(arg))
    for k, v in sorted(kwargs.items()):
        key_parts.append(f"{k}={v}")
    key_string = ":".join(key_parts)
    return f"cache:{hashlib.md5(key_string.encode()).hexdigest()}"


def cache_it(prefix: str, ttl: int = CACHE_TTL):
    """Decorator to cache function results in Redis.
    
    Args:
        prefix: Key prefix for the cache
        ttl: Time to live in seconds
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            client = get_redis_client()
            cache_key = _generate_cache_key(prefix, *args, **kwargs)
            
            # Try to get from cache
            try:
                cached = await client.get(cache_key)
                if cached:
                    return json.loads(cached)
            except Exception:
                pass  # Cache miss or error, proceed to fetch
            
            # Fetch fresh data
            result = func(*args, **kwargs)
            
            # Store in cache
            try:
                await client.setex(cache_key, ttl, json.dumps(result))
            except Exception:
                pass  # Cache write failed, but return the result anyway
            
            return result
        return wrapper
    return decorator


# ============== Cache helper functions ==============

async def cache_get(key: str) -> Optional[Any]:
    """Get a value from cache."""
    try:
        client = get_redis_client()
        value = await client.get(key)
        return json.loads(value) if value else None
    except Exception:
        return None


async def cache_set(key: str, value: Any, ttl: int = CACHE_TTL) -> bool:
    """Set a value in cache."""
    try:
        client = get_redis_client()
        await client.setex(key, ttl, json.dumps(value))
        return True
    except Exception:
        return False


async def cache_delete(key: str) -> bool:
    """Delete a key from cache."""
    try:
        client = get_redis_client()
        await client.delete(key)
        return True
    except Exception:
        return False


async def cache_delete_pattern(pattern: str) -> bool:
    """Delete keys matching a pattern."""
    try:
        client = get_redis_client()
        keys = []
        async for key in client.scan_iter(match=pattern):
            keys.append(key)
        if keys:
            await client.delete(*keys)
        return True
    except Exception:
        return False


# ============== Entity-specific cache functions ==============

# Songs cache
async def cache_song(song_id: int, song_data: dict) -> bool:
    """Cache a song by ID."""
    return await cache_set(f"song:{song_id}", song_data)


async def get_cached_song(song_id: int) -> Optional[dict]:
    """Get a cached song by ID."""
    return await cache_get(f"song:{song_id}")


async def invalidate_song_cache(song_id: int) -> None:
    """Invalidate song cache."""
    await cache_delete(f"song:{song_id}")


# Users cache
async def cache_user(user_id: int, user_data: dict) -> bool:
    """Cache a user by ID."""
    return await cache_set(f"user:{user_id}", user_data)


async def get_cached_user(user_id: int) -> Optional[dict]:
    """Get a cached user by ID."""
    return await cache_get(f"user:{user_id}")


async def invalidate_user_cache(user_id: int) -> None:
    """Invalidate user cache."""
    await cache_delete(f"user:{user_id}")


# Playlists cache
async def cache_playlist(playlist_id: int, playlist_data: dict) -> bool:
    """Cache a playlist by ID."""
    return await cache_set(f"playlist:{playlist_id}", playlist_data)


async def get_cached_playlist(playlist_id: int) -> Optional[dict]:
    """Get a cached playlist by ID."""
    return await cache_get(f"playlist:{playlist_id}")


async def invalidate_playlist_cache(playlist_id: int) -> None:
    """Invalidate playlist cache."""
    await cache_delete(f"playlist:{playlist_id}")


# Lists cache (for collections)
async def cache_list(key: str, data: list) -> bool:
    """Cache a list of items."""
    return await cache_set(f"list:{key}", data, ttl=60)  # Shorter TTL for lists


async def get_cached_list(key: str) -> Optional[list]:
    """Get a cached list."""
    return await cache_get(f"list:{key}")


async def invalidate_list_cache(key: str) -> None:
    """Invalidate a list cache."""
    await cache_delete(f"list:{key}")


# ============== Play History in Redis (hot storage) ==============

async def store_play_event(user_id: int, song_id: int, context_id: Optional[int] = None) -> dict:
    """Store a play event in Redis (fast write)."""
    from datetime import datetime
    played_at = datetime.utcnow().isoformat()
    event = {
        "user_id": user_id,
        "song_id": song_id,
        "context_id": context_id,
        "played_at": played_at
    }
    
    try:
        client = get_redis_client()
        # Add to sorted set for this user (score = timestamp)
        await client.zadd(
            f"play_history:user:{user_id}",
            {json.dumps(event): datetime.utcnow().timestamp()}
        )
        # Also add to global history
        await client.zadd(
            "play_history:global",
            {json.dumps(event): datetime.utcnow().timestamp()}
        )
    except Exception:
        pass  # Best effort, SQLite is the source of truth
    
    return event


async def get_user_play_history(user_id: int, limit: int = 50) -> list[dict]:
    """Get play history for a user from Redis."""
    try:
        client = get_redis_client()
        # Get latest entries from sorted set
        events = await client.zrevrange(
            f"play_history:user:{user_id}",
            0, limit - 1
        )
        return [json.loads(e) for e in events]
    except Exception:
        return []


async def get_global_play_history(limit: int = 50) -> list[dict]:
    """Get global play history from Redis."""
    try:
        client = get_redis_client()
        events = await client.zrevrange(
            "play_history:global",
            0, limit - 1
        )
        return [json.loads(e) for e in events]
    except Exception:
        return []