wastream-mirror/wastream/utils/cache.py

import asyncio
import json
import time
from typing import Optional, List, Dict, Tuple

from wastream.config.settings import settings
from wastream.utils.helpers import build_cache_key
from wastream.utils.logger import cache_logger
from wastream.utils.database import (
    cache_stats_updates_suppressed,
    run_database_operation,
    update_cache_stats_on_set,
)
from wastream.utils.urls import canonicalize_results


# ===========================
# Cache Retrieval
# ===========================
async def get_cache(database, cache_type: str, title: str, year: Optional[str] = None) -> Optional[List[Dict]]:
    cache_key = build_cache_key(cache_type, title, year)

    try:
        current_time = int(time.time())
        result = await database.fetch_one(
            "SELECT content FROM content_cache WHERE cache_key = :cache_key AND (expires_at = -1 OR expires_at > :current_time)",
            {"cache_key": cache_key, "current_time": current_time}
        )

        if not result:
            cache_logger.debug(f"Miss: {cache_type} {title} ({year})")
            return None

        cached_data = json.loads(result["content"])
        cache_logger.debug(f"Hit: {cache_type} {title} ({year}) - {len(cached_data)} results")
        return cached_data
    except json.JSONDecodeError as e:
        cache_logger.error(f"Corrupted cache: {type(e).__name__}: {e}")
        return None
    except Exception as e:
        cache_logger.error(f"Cache read failed: {type(e).__name__}: {e}")
        return None


async def get_cache_with_status(database, cache_type: str, title: str, year: Optional[str] = None) -> Tuple[Optional[List[Dict]], bool]:
    cache_key = build_cache_key(cache_type, title, year)

    try:
        current_time = int(time.time())
        result = await database.fetch_one(
            "SELECT content, expires_at FROM content_cache WHERE cache_key = :cache_key",
            {"cache_key": cache_key}
        )

        if not result:
            cache_logger.debug(f"Miss: {cache_type} {title} ({year})")
            return None, False

        cached_data = json.loads(result["content"])
        is_valid = result["expires_at"] == -1 or result["expires_at"] > current_time

        if is_valid:
            cache_logger.debug(f"Hit (valid): {cache_type} {title} ({year}) - {len(cached_data)} results")
        else:
            cache_logger.debug(f"Hit (expired): {cache_type} {title} ({year}) - {len(cached_data)} results")

        return cached_data, is_valid
    except json.JSONDecodeError as e:
        cache_logger.error(f"Corrupted cache: {type(e).__name__}: {e}")
        return None, False
    except Exception as e:
        cache_logger.error(f"Cache read failed: {type(e).__name__}: {e}")
        return None, False


# ===========================
# Cache Storage
# ===========================
async def set_cache(database, cache_type: str, title: str, year: Optional[str] = None,
                    results: Optional[List] = None, ttl: int = 3600):
    cache_key = build_cache_key(cache_type, title, year)

    try:
        old_row = await database.fetch_one(
            "SELECT content FROM content_cache WHERE cache_key = :cache_key",
            {"cache_key": cache_key}
        )
        old_results = None
        if old_row:
            try:
                old_results = json.loads(old_row["content"])
            except (json.JSONDecodeError, TypeError):
                pass

        if ttl == -1:
            expires_at = -1
        else:
            current_time = int(time.time())
            expires_at = current_time + ttl
        content = json.dumps(canonicalize_results(results or []))

        if settings.DATABASE_TYPE == "sqlite":
            query = """INSERT OR REPLACE INTO content_cache (cache_key, content, expires_at)
                       VALUES (:cache_key, :content, :expires_at)"""
        else:
            query = """INSERT INTO content_cache (cache_key, content, expires_at)
                       VALUES (:cache_key, :content, :expires_at)
                       ON CONFLICT (cache_key) DO UPDATE
                       SET content = :content, expires_at = :expires_at"""

        async def save_cache_row() -> None:
            await database.execute(query, {
                "cache_key": cache_key,
                "content": content,
                "expires_at": expires_at
            })

        await run_database_operation(save_cache_row, "content cache save")

        if not cache_stats_updates_suppressed():
            asyncio.create_task(update_cache_stats_on_set(cache_key, results or [], old_results))

        ttl_str = "permanent" if ttl == -1 else f"{ttl}s"
        cache_logger.debug(f"Saved: {cache_type} {title} ({year}) - {len(results or [])} results ({ttl_str})")
    except Exception as e:
        cache_logger.error(f"Cache save failed: {type(e).__name__}: {e}")


async def set_cache_if_not_exists(
        database, cache_type: str, title: str, year: Optional[str] = None,
        results: Optional[List] = None, ttl: int = 3600) -> bool:
    cache_key = build_cache_key(cache_type, title, year)

    try:
        existing = await database.fetch_one(
            "SELECT cache_key FROM content_cache WHERE cache_key = :cache_key",
            {"cache_key": cache_key}
        )

        if existing:
            cache_logger.debug(f"Skip (exists): {cache_type} {title} ({year})")
            return False

        if ttl == -1:
            expires_at = -1
        else:
            current_time = int(time.time())
            expires_at = current_time + ttl
        content = json.dumps(canonicalize_results(results or []))

        async def insert_cache_row() -> None:
            await database.execute(
                "INSERT INTO content_cache (cache_key, content, expires_at) VALUES (:cache_key, :content, :expires_at)",
                {"cache_key": cache_key, "content": content, "expires_at": expires_at}
            )

        await run_database_operation(insert_cache_row, "new content cache save")

        if not cache_stats_updates_suppressed():
            asyncio.create_task(update_cache_stats_on_set(cache_key, results or []))

        ttl_str = "permanent" if ttl == -1 else f"{ttl}s"
        cache_logger.debug(f"Saved (new): {cache_type} {title} ({year}) - {len(results or [])} results ({ttl_str})")
        return True
    except Exception as e:
        cache_logger.error(f"Cache save failed: {type(e).__name__}: {e}")
        return False


# ===========================
# Parallel Cache Lookup (Local + Remote)
# ===========================
async def get_cache_parallel(database, cache_type: str, title: str, year: Optional[str] = None) -> Tuple[Optional[List[Dict]], bool]:
    from wastream.services.remote import fetch_remote_cache

    async def check_local():
        return await get_cache(database, cache_type, title, year)

    async def check_remote():
        return await fetch_remote_cache(cache_type, title, year)

    try:
        local_result, remote_result = await asyncio.gather(
            check_local(),
            check_remote(),
            return_exceptions=True
        )

        if isinstance(local_result, list) and local_result is not None:
            return local_result, False

        if local_result is not None and not isinstance(local_result, Exception):
            return local_result, False

        if isinstance(remote_result, tuple):
            remote_data, should_store = remote_result
            if remote_data:
                cache_logger.debug(f"Remote hit: {cache_type} {title} ({year}) - {len(remote_data)} results")
                return remote_data, should_store

        cache_logger.debug(f"Remote miss: {cache_type} {title} ({year})")
        return None, False

    except Exception as e:
        cache_logger.error(f"Parallel cache lookup failed: {type(e).__name__}: {e}")
        return None, False