python/src/server/services/credentials/manager.py

import os
import re
import time
from typing import Any

from supabase import Client, create_client

from ...config.logfire_config import get_logger
from .crypto_utils import CryptoUtils
from .models import CredentialItem

logger = get_logger(__name__)


class CredentialManager:
    """
    Service for managing application credentials and configuration.

    Handles loading, storing, and accessing credentials with encryption
    for sensitive values. Credentials include API keys, service credentials,
    and application configuration stored in the archon_settings table.
    """

    def __init__(self):
        self._supabase: Client | None = None
        self._repository: Any | None = None
        self._cache: dict[str, Any] = {}
        self._cache_initialized = False
        self._rag_settings_cache: dict[str, Any] | None = None
        self._rag_cache_timestamp: float | None = None
        self._rag_cache_ttl = 300  # 5 minutes TTL for RAG settings cache
        self._active_tier = 1

    def get_active_tier(self) -> int:
        """Get the currently active model tier (1, 2, or 3)."""
        return getattr(self, "_active_tier", 1)

    def set_active_tier(self, tier: int) -> None:
        """Set the active model tier (1, 2, or 3)."""
        self._active_tier = tier

    def _get_repository(self) -> Any:
        """Get or create the CredentialRepository."""
        if self._repository is None:
            from .repository import CredentialRepository
            self._repository = CredentialRepository(self._get_supabase_client())
        return self._repository

    def _get_supabase_client(self) -> Client:
        """
        Get or create a properly configured Supabase client using environment variables.
        Uses the standard Supabase client initialization.

        Returns:
            A configured Supabase Client instance.

        Raises:
            ValueError: If required environment variables are missing.
        """
        if self._supabase is None:
            url = os.getenv("SUPABASE_URL")
            key = os.getenv("SUPABASE_SERVICE_KEY")

            if not url or not key:
                raise ValueError("SUPABASE_URL and SUPABASE_SERVICE_KEY must be set in environment variables")

            try:
                # Initialize with standard Supabase client - no need for custom headers
                self._supabase = create_client(url, key)

                # Extract project ID from URL for logging purposes only
                match = re.match(r"https://([^.]+)\.supabase\.co", url)
                if match:
                    project_id = match.group(1)
                    logger.debug(f"Supabase client initialized for project: {project_id}")
                else:
                    logger.debug("Supabase client initialized successfully")

            except Exception as e:
                logger.error(f"Error initializing Supabase client: {e}")
                raise

        return self._supabase

    async def load_all_credentials(self) -> dict[str, Any]:
        """
        Load all credentials from database and cache them.
        This is typically called at system startup via initialize_credentials.

        Returns:
            A dictionary of all loaded credentials.
        """
        try:
            repository = self._get_repository()
            data = repository.fetch_all()

            credentials = {}
            for item in data:
                key = item["key"]
                if item["is_encrypted"] and item["encrypted_value"]:
                    # For encrypted values, we store the encrypted version
                    # Decryption happens when the value is actually needed via get_credential
                    credentials[key] = {
                        "encrypted_value": item["encrypted_value"],
                        "is_encrypted": True,
                        "category": item["category"],
                        "description": item["description"],
                    }
                else:
                    # Plain text values are stored directly
                    credentials[key] = item["value"]

            self._cache = credentials
            self._cache_initialized = True
            logger.info(f"Loaded {len(credentials)} credentials from database")

            return credentials

        except Exception as e:
            logger.error(f"Error loading credentials: {e}")
            raise

    async def get_credential(self, key: str, default: Any = None, decrypt: bool = True) -> Any:
        """
        Get a credential value by key.

        This method checks the internal cache first, and falls back to OS
        environment variables if the key is not found or empty in the database.

        Args:
            key: The configuration key to look up.
            default: Value to return if the key is not found.
            decrypt: Whether to automatically decrypt encrypted values.

        Returns:
            The credential value, or the default value if not found.
        """
        if not self._cache_initialized:
            await self.load_all_credentials()

        value = self._cache.get(key, default)

        # If it's an encrypted value and we want to decrypt it
        if isinstance(value, dict) and value.get("is_encrypted") and decrypt:
            encrypted_value = value.get("encrypted_value")
            if encrypted_value:
                try:
                    return CryptoUtils.decrypt_value(encrypted_value)
                except Exception as e:
                    logger.error(f"Failed to decrypt credential {key}: {e}")
                    return default

        if isinstance(value, str):
            stripped = value.strip()
            if stripped:
                return stripped

        # PHYSICAL FALLBACK: If not in cache or is empty string, check OS environment
        # This allows environment variables to override or provide defaults for missing settings
        env_value = os.getenv(key) or os.getenv(key.upper())
        if env_value:
            return env_value

        return value

    async def get_encrypted_credential_raw(self, key: str) -> str | None:
        """
        Get the raw encrypted value for a credential (without decryption).

        Args:
            key: The configuration key.

        Returns:
            The raw encrypted string or None.
        """
        if not self._cache_initialized:
            await self.load_all_credentials()

        value = self._cache.get(key)
        if isinstance(value, dict) and value.get("is_encrypted"):
            return value.get("encrypted_value")

        return None

    async def set_credential(
        self,
        key: str,
        value: str,
        is_encrypted: bool = False,
        category: str | None = None,
        description: str | None = None,
    ) -> bool:
        """
        Set a credential/setting in the database.

        Args:
            key: Setting key
            value: Plain text value
            is_encrypted: Whether to encrypt the value before storing
            category: Optional category for grouping
            description: Optional description

        Returns:
            True if successful, False otherwise
        """
        try:
            data: dict[str, Any] = {
                "key": key,
                "is_encrypted": is_encrypted,
                "category": category,
                "description": description,
            }

            if is_encrypted:
                data["encrypted_value"] = CryptoUtils.encrypt_value(value)
                data["value"] = None
            else:
                data["value"] = value
                data["encrypted_value"] = None

            # Upsert to database via repository
            repository = self._get_repository()
            repository.upsert(data)

            # Invalidate RAG settings cache if this is a rag_strategy setting
            if category == "rag_strategy":
                self._rag_settings_cache = None
                self._rag_cache_timestamp = None
                logger.debug(f"Invalidated RAG settings cache due to update of {key}")

            logger.info(f"Successfully {'encrypted and ' if is_encrypted else ''}stored credential: {key}")
            return True

        except Exception as e:
            logger.error(f"Error setting credential {key}: {e}")
            return False

    async def delete_credential(self, key: str) -> bool:
        """
        Delete a credential from database and cache.

        Args:
            key: The key to delete.

        Returns:
            True if successful.
        """
        try:
            # Execute delete via repository
            repository = self._get_repository()
            repository.delete(key)

            # Remove from local cache
            if key in self._cache:
                del self._cache[key]

            # Invalidate RAG settings cache if needed
            if self._rag_settings_cache is not None and key in self._rag_settings_cache:
                self._rag_settings_cache = None
                self._rag_cache_timestamp = None
                logger.debug(f"Invalidated RAG settings cache due to deletion of {key}")

            logger.info(f"Successfully deleted credential: {key}")
            return True

        except Exception as e:
            logger.error(f"Error deleting credential {key}: {e}")
            return False

    async def get_credentials_by_category(self, category: str) -> dict[str, Any]:
        """
        Get all credentials for a specific category.

        Special caching is applied for the 'rag_strategy' category to
        reduce redundant database calls during high-frequency RAG operations.
        """
        if not self._cache_initialized:
            await self.load_all_credentials()

        # Special caching for rag_strategy category
        if category == "rag_strategy":
            current_time = time.time()
            if (
                self._rag_settings_cache is not None
                and self._rag_cache_timestamp is not None
                and current_time - self._rag_cache_timestamp < self._rag_cache_ttl
            ):
                logger.debug("Using cached RAG settings")
                return self._rag_settings_cache

        try:
            repository = self._get_repository()
            data = repository.fetch_by_category(category)

            credentials = {}
            for item in data:
                key = item["key"]
                if item["is_encrypted"]:
                    credentials[key] = {
                        "value": "[ENCRYPTED]",
                        "is_encrypted": True,
                        "category": item["category"],
                        "description": item["description"],
                    }
                else:
                    credentials[key] = item["value"]

            # Cache rag_strategy results
            if category == "rag_strategy":
                self._rag_settings_cache = credentials
                self._rag_cache_timestamp = time.time()
                logger.debug(f"Cached RAG settings with {len(credentials)} items")

            return credentials

        except Exception as e:
            logger.error(f"Error getting credentials for category {category}: {e}")
            return {}

    async def list_all_credentials(self) -> list[CredentialItem]:
        """Get all credentials as a list of CredentialItem objects (for Settings UI)."""
        try:
            repository = self._get_repository()
            data = repository.fetch_non_system_protected()

            credentials = []
            for item in data:
                if item["is_encrypted"] and item["encrypted_value"]:
                    cred = CredentialItem(
                        key=item["key"],
                        value="[ENCRYPTED]",
                        encrypted_value=None,
                        is_encrypted=item["is_encrypted"],
                        category=item["category"],
                        description=item["description"],
                    )
                else:
                    cred = CredentialItem(
                        key=item["key"],
                        value=item["value"],
                        encrypted_value=None,
                        is_encrypted=item["is_encrypted"],
                        category=item["category"],
                        description=item["description"],
                    )
                credentials.append(cred)

            return credentials

        except Exception as e:
            logger.error(f"Error listing credentials: {e}")
            return []

    async def check_credentials_exist(self, keys: list[str]) -> dict[str, dict[str, Any]]:
        from .helpers import check_credentials_exist

        return await check_credentials_exist(self, keys)

    async def get_active_provider(self, service_type: str = "llm") -> dict[str, Any]:
        from .provider_configs import get_active_provider

        return await get_active_provider(self, service_type)

    async def get_embedding_provider_configs(self) -> list[dict[str, Any]]:
        from .provider_configs import get_embedding_provider_configs

        return await get_embedding_provider_configs(self)

    async def _get_provider_api_key(self, provider: str) -> str | None:
        from .provider_configs import _get_provider_api_key

        return await _get_provider_api_key(self, provider)

    def _get_provider_base_url(self, provider: str, rag_settings: dict) -> str | None:
        from .provider_configs import _get_provider_base_url

        return _get_provider_base_url(provider, rag_settings)

    async def set_active_provider(self, provider: str, service_type: str = "llm") -> bool:
        from .provider_configs import set_active_provider

        return await set_active_provider(self, provider, service_type)