src/densefeed/web/auth.py

"""Web auth helpers — token resolution, username lookup, DB hydration."""

from __future__ import annotations

import logging
import threading

import gradio as gr

from densefeed.auth import hf_token, install_token
from densefeed.config import DenseFeedConfig, load_config
from densefeed.storage.duckdb_backend import DuckDBBackend
from densefeed.storage import sync as storage_sync

logger = logging.getLogger(__name__)

# Shared mutable state — initialized once in create_ui(), accessed via get_config/get_backend
_backend: DuckDBBackend | None = None
_config: DenseFeedConfig | None = None
_config_lock = threading.Lock()
_db_loaded_for_repo: str | None = None


def default_user_repos(username: str) -> tuple[str, str]:
    """Return default private config and public audio dataset repo IDs for a user."""
    import os

    app_name = (os.environ.get("SPACE_ID", "DenseFeed").split("/")[-1] or "DenseFeed")
    slug = app_name.lower().replace("_", "-")
    return f"{username}/{slug}-data", f"{username}/{slug}-audio"


def set_storage_repos(hf_dataset_repo: str, hf_audio_repo: str) -> None:
    """Update shared storage config and force the next backend access to hydrate it."""
    global _backend, _db_loaded_for_repo
    cfg = get_config()
    changed = (
        cfg.storage.hf_dataset_repo != hf_dataset_repo
        or cfg.storage.hf_audio_repo != hf_audio_repo
    )
    cfg.storage.hf_dataset_repo = hf_dataset_repo
    cfg.storage.hf_audio_repo = hf_audio_repo
    if changed:
        if _backend is not None:
            _backend.close()
        _backend = None
        _db_loaded_for_repo = None


def space_default_config(cfg: DenseFeedConfig) -> DenseFeedConfig:
    """Apply production defaults when running on Hugging Face Spaces."""
    import os

    if os.environ.get("SPACE_ID"):
        cfg.storage.backend = "duckdb"
        try:
            import spaces  # noqa: F401

            cfg.llm.backend = "zerogpu"
            logger.info("[config] ZeroGPU detected — backend=zerogpu")
        except ImportError:
            logger.info(
                "[config] SPACE_ID but no ZeroGPU — backend=%s",
                cfg.llm.backend,
            )
    return cfg


def get_config() -> DenseFeedConfig:
    """Return the module-level config, loading fresh only if not yet initialized."""
    global _config
    with _config_lock:
        if _config is None:
            _config = space_default_config(load_config())
        return _config


def get_config_copy() -> DenseFeedConfig:
    """Return a deep copy of the module-level config for per-request mutation."""
    return get_config().model_copy(deep=True)


def get_backend() -> DuckDBBackend:
    """Lazily initialize and return the DuckDB backend."""
    global _backend, _config, _db_loaded_for_repo

    with _config_lock:
        if _backend is None:
            cfg = space_default_config(_config or load_config())
            _config = cfg
            if (
                cfg.storage.backend == "duckdb"
                and cfg.storage.hf_dataset_repo
                and hf_token.get()
                and _db_loaded_for_repo != cfg.storage.hf_dataset_repo
            ):
                storage_sync.download_db(
                    cfg.storage.hf_dataset_repo, cfg.storage.duckdb_path
                )
            _backend = DuckDBBackend(
                cfg.storage.duckdb_path, hf_audio_repo=cfg.storage.hf_audio_repo
            )
            if cfg.storage.hf_dataset_repo:
                _db_loaded_for_repo = cfg.storage.hf_dataset_repo
        return _backend


def set_initial_state(config: DenseFeedConfig, backend: DuckDBBackend | None) -> None:
    """Set the module-level state during app startup."""
    global _backend, _config
    _config = config
    _backend = backend


def init_backend() -> DuckDBBackend:
    """Create a backend from current config without hydration (for startup)."""
    cfg = get_config()
    return DuckDBBackend(
        cfg.storage.duckdb_path, hf_audio_repo=cfg.storage.hf_audio_repo
    )


def request_bearer_token(request: gr.Request | None) -> str | None:
    """Return an HF token from API Authorization headers, if present."""
    if request is None:
        return None
    auth = request.headers.get("authorization") or request.headers.get("Authorization")
    if not auth:
        return None
    scheme, _, value = auth.partition(" ")
    if scheme.lower() == "bearer" and value:
        return value
    return None


def active_token(
    token: gr.OAuthToken | None = None, request: gr.Request | None = None
) -> str | None:
    """Return the current HF token.

    Browser users authenticate via Space OAuth. API/debug callers can provide an
    explicit Authorization bearer token, which must take precedence because
    Gradio may still inject a Space OAuth token with narrower scopes.
    """
    return request_bearer_token(request) or (token.token if token else None)


def install_active_token(token_str: str) -> None:
    """Install the session token for all code paths that need HF auth.

    Delegates to auth.install_token for the ContextVar + env write,
    then triggers DuckDB hydration so the backend is ready for use.
    """
    install_token(token_str)
    ensure_remote_db_loaded()


def ensure_remote_db_loaded() -> None:
    """Download the persisted DuckDB once auth is available, even after startup."""
    global _backend, _db_loaded_for_repo

    cfg = get_config()
    if cfg.storage.backend != "duckdb" or not cfg.storage.hf_dataset_repo:
        return
    token = hf_token.get()
    if not token:
        logger.warning("[sync] authenticated DB load skipped: no token")
        return
    if _db_loaded_for_repo == cfg.storage.hf_dataset_repo:
        return
    with _config_lock:
        if _db_loaded_for_repo == cfg.storage.hf_dataset_repo:
            return
        downloaded = storage_sync.download_db(
            cfg.storage.hf_dataset_repo, cfg.storage.duckdb_path
        )
        if _backend is not None:
            _backend.close()
        _backend = DuckDBBackend(
            cfg.storage.duckdb_path, hf_audio_repo=cfg.storage.hf_audio_repo
        )
        _db_loaded_for_repo = cfg.storage.hf_dataset_repo
        logger.info(
            "[sync] DB hydration complete repo=%s downloaded=%s path=%s",
            cfg.storage.hf_dataset_repo,
            downloaded,
            cfg.storage.duckdb_path,
        )


def get_username(
    token: gr.OAuthToken | None = None, request: gr.Request | None = None
) -> str:
    """Safely retrieve HF username from OAuthToken or API bearer token."""
    if token and hasattr(token, "username") and token.username:
        return token.username
    tok = active_token(token, request)
    if not tok:
        return ""
    try:
        from huggingface_hub import HfApi

        api = HfApi(token=tok)
        return api.whoami().get("name", "")
    except Exception:
        return ""


def merge_prefs(cfg: DenseFeedConfig, prefs: dict) -> None:
    """Merge user preferences into pipeline config."""
    import json

    cfg.pipeline.min_relevance = int(
        prefs.get("min_relevance", cfg.pipeline.min_relevance)
    )
    cfg.pipeline.top_n_items = int(prefs.get("top_n_items", cfg.pipeline.top_n_items))
    cfg.storage.hf_dataset_repo = prefs.get(
        "hf_dataset_repo", cfg.storage.hf_dataset_repo
    )
    cfg.storage.hf_audio_repo = prefs.get("hf_audio_repo", cfg.storage.hf_audio_repo)
    excluded = prefs.get("excluded_sources", [])
    if isinstance(excluded, str):
        excluded = json.loads(excluded)
    for src in excluded:
        attr = getattr(cfg.sources, src, None)
        if attr and hasattr(attr, "enabled"):
            attr.enabled = False