import functools
import json
import os
import threading
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Union

from loguru import logger
from sqlalchemy import func, or_
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from .. import defaults
from ..__version__ import __version__ as package_version
from ..database.models import Setting, SettingType
from ..web.models.settings import (
    AppSetting,
    BaseSetting,
    ChatSetting,
    LLMSetting,
    ReportSetting,
    SearchSetting,
)
from ..utilities.type_utils import to_bool
from .base import ISettingsManager
from .env_registry import registry as env_registry


def parse_boolean(value: Any) -> bool:
    """
    Convert various representations to boolean using HTML checkbox semantics.

    This function handles form values, JSON booleans, and environment variables,
    ensuring consistent behavior across client and server.

    **HTML Checkbox Semantics** (INTENTIONAL DESIGN):
    - **Any value present (except explicit false) = checked = True**
    - This matches standard HTML form behavior where checkbox presence indicates checked state
    - In HTML forms, checkboxes send a value when checked, nothing when unchecked

    **Examples**:
        parse_boolean("on")         # True  - standard HTML checkbox value
        parse_boolean("true")       # True  - explicit true
        parse_boolean("1")          # True  - numeric true
        parse_boolean("enabled")    # True  - any non-empty string
        parse_boolean("disabled")   # True  - INTENTIONAL: any string = checkbox was checked!
        parse_boolean("custom")     # True  - custom checkbox value

        parse_boolean("false")      # False - explicit false
        parse_boolean("off")        # False - explicit false
        parse_boolean("0")          # False - explicit false
        parse_boolean("")           # False - empty string = unchecked
        parse_boolean(None)         # False - missing = unchecked

    **Why "disabled" returns True**:
    This is NOT a bug! If a checkbox sends the value "disabled", it means the checkbox
    was checked (present in form data). The actual string content doesn't matter for
    HTML checkboxes - only presence vs absence matters.

    Args:
        value: Value to convert to boolean. Accepts strings, booleans, or None.

    Returns:
        bool: True for truthy values (any non-empty string except explicit false);
              False for falsy values ('off', 'false', '0', '', 'no', False, None)

    Note:
        This function implements HTML form semantics, NOT generic boolean parsing.
        See tests/settings/test_boolean_parsing.py for comprehensive test coverage.
    """
    # Constants for boolean value parsing
    FALSY_VALUES = ("off", "false", "0", "", "no")

    # Handle already-boolean values
    if isinstance(value, bool):
        return value

    # Handle None (missing values)
    if value is None:
        return False

    # Handle string values
    if isinstance(value, str):
        value_lower = value.lower().strip()
        # Explicitly falsy values (empty string, false-like values)
        if value_lower in FALSY_VALUES:
            return False
        # Any other non-empty string = True (HTML checkbox semantics)
        return True

    # For other types (numbers, lists, etc.), use Python's bool conversion
    return bool(value)


def _parse_number(x):
    """Parse number, returning int if it's a whole number, otherwise float."""
    f = float(x)
    if f.is_integer():
        return int(f)
    return f


def _parse_json_value(x):
    """Parse JSON ui_element values.

    DB values (via SQLAlchemy JSON column) arrive as Python objects already.
    Form POST and env var overrides arrive as raw strings and need parsing.
    For example, a textarea containing ``["general"]`` arrives as the string
    ``'[\\r\\n  "general"\\r\\n]'`` which must be decoded into a list.
    """
    if isinstance(x, str):
        stripped = x.strip()
        if stripped:
            try:
                return json.loads(stripped)
            except (json.JSONDecodeError, ValueError, RecursionError):
                logger.warning("Failed to parse JSON value, returning raw")
                return x
    return x


def _parse_multiselect(x):
    """Parse multiselect value, handling both lists and strings.

    DB values (via SQLAlchemy JSON column) arrive as Python lists already.
    Env var overrides arrive as strings and need parsing — either as JSON
    arrays (e.g. '["markdown","latex"]') or comma-separated values
    (e.g. 'markdown,latex').
    """
    if isinstance(x, list):
        return x
    if isinstance(x, str):
        stripped = x.strip()
        if stripped.startswith("["):
            try:
                parsed = json.loads(stripped)
                if isinstance(parsed, list):
                    return parsed
            except (json.JSONDecodeError, ValueError):
                pass
        # Comma-separated fallback
        return [item.strip() for item in stripped.split(",") if item.strip()]
    return x


def _filter_setting_columns(data: dict) -> dict:
    """Filter a dict to only keys that are valid Setting model columns.

    Prevents crashes when default_settings.json contains keys not present
    as columns on the Setting model (e.g. future flags).
    """
    valid_columns = {c.name for c in Setting.__table__.columns}
    return {k: v for k, v in data.items() if k in valid_columns}


_POLICY_AUDIT_KEYS = frozenset(
    {
        "llm.require_local_endpoint",
        "llm.allowed_local_hostnames",
        "embeddings.require_local",
    }
)


def _is_policy_setting(key: str) -> bool:
    """Return True for security-relevant setting keys that need an
    audit-log entry on change. Scope is intentionally narrow so this
    audit hook doesn't widen into general settings-change logging.
    """
    if key.startswith("policy."):
        return True
    return key in _POLICY_AUDIT_KEYS


def _infer_ui_element(value: Any, current: str = "text") -> str:
    """Infer the appropriate ui_element string from a Python value's type.

    Args:
        value: The value to infer the ui_element from.
        current: The existing ui_element. If it is already something more
            specific than ``"text"``, it is kept as-is.
    """
    if current != "text":
        return current
    if isinstance(value, bool):
        return "checkbox"
    if isinstance(value, (int, float)):
        return "number"
    if isinstance(value, (list, dict)):
        return "json"
    return "text"


# Default categories for each typed setting prefix, used by the self-heal
# block in set_setting() when a row's type column doesn't match its key
# prefix. The values mirror the canonical strings in
# web/routes/settings_routes.py: a legacy chat.* row with type=APP and a
# stale category gets repointed to type=CHAT + category="chat" on next
# save. We use the most-general category per prefix here — sub-classifying
# llm_general vs llm_parameters depends on the specific key, but the
# self-heal only fires when the row was already mis-typed, so over-
# generalizing the category is preferable to leaving it stale.
_INFERRED_CATEGORY: Dict[str, str] = {
    "llm.": "llm_general",
    "search.": "search_general",
    "report.": "report_parameters",
    "database.": "database_parameters",
    "chat.": "chat",
}


UI_ELEMENT_TO_SETTING_TYPE: Dict[str, Callable[..., Any]] = {
    "text": str,
    "json": _parse_json_value,
    "password": str,
    "select": str,
    "number": _parse_number,
    "range": _parse_number,  # Same behavior as number for consistency
    "checkbox": parse_boolean,
    "textarea": str,
    "multiselect": _parse_multiselect,
}


def get_typed_setting_value(
    key: str,
    value: Any,
    ui_element: str,
    default: Any = None,
    check_env: bool = True,
) -> Any:
    """
    Extracts the value for a particular setting, ensuring that it has the
    correct type.

    Args:
        key: The setting key.
        value: The setting value from the database.
        ui_element: The setting UI element ID.
        default: Default value to return if the value of the setting is
            invalid.
        check_env: If true, it will check the environment variable for
            this setting before reading from the DB.

    Returns:
        The value of the setting.

    """
    setting_type = UI_ELEMENT_TO_SETTING_TYPE.get(ui_element, None)
    if setting_type is None:
        logger.warning(
            "Got unknown type {} for setting {}, returning default value.",
            ui_element,
            key,
        )
        return default

    # Check environment variable first (highest priority).
    if check_env:
        env_value = check_env_setting(key)
        if env_value is not None:
            try:
                return setting_type(env_value)
            except ValueError:
                logger.warning(
                    "Setting {} has invalid value {}. Falling back to DB.",
                    key,
                    env_value,
                )

    # If value is None (not in database), return default.
    if value is None:
        return default

    # Read from the database.
    try:
        return setting_type(value)
    except (ValueError, TypeError):
        logger.warning(
            "Setting {} has invalid value {}. Returning default.",
            key,
            value,
        )
        return default


def check_env_setting(key: str) -> str | None:
    """
    Checks environment variables for a particular setting.

    Args:
        key: The database key for the setting.

    Returns:
        The setting from the environment variables, or None if the variable
        is not set or is empty.

    Note:
        Empty environment variables ("") are treated as unset. This is standard
        practice across the ecosystem — see CPython's official docs (PYTHON*
        env vars require "a non-empty string"), botocore PR #1687, Pallets/Click
        PR #2223, and Vercel Turborepo PR #6929. Orchestration tools like Unraid,
        Terraform, and Kubernetes manifests often cannot conditionally omit env
        var declarations, so they pass "" for unconfigured values. Treating ""
        as unset prevents these empty strings from overriding database defaults.
        See: https://github.com/LearningCircuit/local-deep-research/pull/3362

    """
    env_variable_name = f"LDR_{'_'.join(key.split('.')).upper()}"
    env_value = os.getenv(env_variable_name)
    # Treat empty string as unset — orchestration tools (Unraid, Terraform, K8s)
    # often cannot omit env var declarations and pass "" for unconfigured values.
    if env_value is not None and env_value != "":
        logger.debug(f"Overriding {key} setting from environment variable.")
        return env_value
    if env_value == "":
        logger.warning(
            "Environment variable {} is set but empty — "
            "ignoring it and falling back to DB/default for setting '{}'. "
            "This is expected on Unraid or Docker templates that create "
            "all variables even when left blank. To suppress this warning, "
            "remove the variable from your environment or set a value.",
            env_variable_name,
            key,
        )
    return None


class SettingsManager(ISettingsManager):
    """
    Manager for handling application settings with database storage and file fallback.
    Provides methods to get and set settings, with the ability to override settings in memory.
    """

    def __init__(
        self,
        db_session: Optional[Session] = None,
        owns_session: bool = False,
    ):
        """
        Initialize the settings manager

        Args:
            db_session: SQLAlchemy session for database operations
            owns_session: If True, close() will close the session.
                Defaults to False (safe for borrowed sessions).  Set to True
                only when this manager created/owns the session — currently
                only get_settings_manager() in db_utils.py does this.
        """
        self.db_session = db_session
        self._owns_session = owns_session
        self._closed = False
        self.db_first = True  # Always prioritize DB settings

        # Store the thread ID this instance was created in
        self._creation_thread_id = threading.get_ident()

        # Initialize settings lock as None - will be checked lazily
        self.__settings_locked: Optional[bool] = None

        # Auto-initialize settings if database is empty
        if self.db_session:
            self._ensure_settings_initialized()

    def close(self):
        """Close the DB session if this manager owns it.

        Borrowed sessions (owns_session=False) are left open for their
        owner to close (e.g. Flask teardown closes g.db_session).
        Safe to call multiple times — subsequent calls are no-ops.
        """
        if self._owns_session and self.db_session is not None:
            try:
                logger.debug("Closing owned DB session in SettingsManager")
                self.db_session.close()
            except Exception:
                logger.warning(
                    "Failed to close SettingsManager DB session — "
                    "connection may leak",
                )
        self._closed = True
        self.db_session = None

    def _ensure_settings_initialized(self):
        """Ensure settings are initialized in the database."""
        # Check if we have any settings at all
        from ..database.models import Setting

        if self.db_session is None:
            raise RuntimeError("Database session is not initialized")
        settings_count = self.db_session.query(Setting).count()

        if settings_count == 0:
            logger.info("No settings found in database, loading defaults")
            self.load_from_defaults_file(commit=True)
            logger.info("Default settings loaded successfully")

    def _check_thread_safety(self):
        """Check if this instance is being used in the same thread it was created in."""
        current_thread_id = threading.get_ident()
        if self.db_session and current_thread_id != self._creation_thread_id:
            raise RuntimeError(
                f"SettingsManager instance created in thread {self._creation_thread_id} "
                f"is being used in thread {current_thread_id}. This is not thread-safe! "
                f"Create a new SettingsManager instance within the current thread context."
            )

    @property
    def settings_locked(self) -> bool:
        """Check if settings are locked (lazy evaluation)."""
        if self.__settings_locked is None:
            try:
                self.__settings_locked = self.get_setting(
                    "app.lock_settings", False
                )
                if self.settings_locked:
                    logger.info(
                        "Settings are locked. Disabling all settings changes."
                    )
            except Exception:
                logger.warning(
                    "Failed to check settings lock status, assuming not locked"
                )
                self.__settings_locked = False
        return bool(self.__settings_locked)

    @functools.cached_property
    def default_settings(self) -> Dict[str, Any]:
        """
        Returns:
            The default settings, loaded from JSON files and merged.
            Automatically discovers and loads all .json files in the defaults
            directory and its subdirectories.
            Theme options are dynamically injected from the theme registry.

        """
        settings: Dict[str, Any] = {}

        try:
            # Get the defaults package path
            defaults_path = Path(defaults.__file__).parent

            # Find all JSON files recursively in the defaults directory
            json_files = sorted(defaults_path.rglob("*.json"))

            logger.debug(f"Found {len(json_files)} JSON settings files")

            # Load and merge all JSON files
            for json_file in json_files:
                try:
                    with open(json_file, "r", encoding="utf-8-sig") as f:
                        file_settings = json.load(f)

                    # Get relative path for logging
                    relative_path = json_file.relative_to(defaults_path)

                    # Warn about key conflicts
                    conflicts = set(settings.keys()) & set(file_settings.keys())
                    if conflicts:
                        logger.warning(
                            f"Keys {conflicts} from {relative_path} "
                            f"override existing values"
                        )

                    settings.update(file_settings)
                    logger.debug(f"Loaded {relative_path}")

                except json.JSONDecodeError:
                    logger.exception(f"Invalid JSON in {json_file}")
                except Exception:
                    logger.warning(f"Could not load {json_file}")

        except Exception:
            logger.warning("Error loading settings files")

        # Inject dynamic theme options from theme registry
        if "app.theme" in settings:
            try:
                from local_deep_research.web.themes import theme_registry

                settings["app.theme"]["options"] = (
                    theme_registry.get_settings_options()
                )
            except ImportError:
                # Theme registry not available, use static options from JSON
                pass

        # Inject search strategy options from code (single source of truth)
        if "search.search_strategy" in settings:
            from local_deep_research.constants import get_available_strategies

            strategies = get_available_strategies()
            settings["search.search_strategy"]["options"] = [
                {"label": s["label"], "value": s["name"]} for s in strategies
            ]

        logger.debug(f"Loaded {len(settings)} total settings")
        return settings

    def __get_typed_setting_value(
        self,
        setting: Setting,
        default: Any = None,
        check_env: bool = True,
    ) -> Any:
        """
        Extracts the value for a particular setting, ensuring that it has the
        correct type.

        Args:
            setting: The setting to get the value for.
            default: Default value to return if the value of the setting is
                invalid.
            check_env: If true, it will check the environment variable for
                this setting before reading from the DB.

        Returns:
            The value of the setting.

        """
        return get_typed_setting_value(
            str(setting.key),
            setting.value,
            str(setting.ui_element),
            default=default,
            check_env=check_env,
        )

    def __query_settings(self, key: str | None = None) -> List[Setting]:
        """
        Abstraction for querying settings that also transparently handles
        reading the default settings file if the DB is not enabled.

        Args:
            key: The key to read. If None, it will read everything.

        Returns:
            The settings it queried.

        """
        if self.db_session:
            self._check_thread_safety()
            query = self.db_session.query(Setting)
            if key is not None:
                # This will find exact matches and any subkeys.
                query = query.filter(
                    or_(
                        Setting.key == key,
                        Setting.key.startswith(f"{key}."),
                    )
                )
            return query.all()

        logger.debug(
            "DB is disabled, reading setting '{}' from defaults file.", key
        )

        settings = []
        for candidate_key, setting in self.default_settings.items():
            if key is None or (
                candidate_key == key or candidate_key.startswith(f"{key}.")
            ):
                settings.append(
                    Setting(
                        key=candidate_key,  # gitleaks:allow
                        **_filter_setting_columns(setting),
                    )
                )

        return settings

    def get_setting(
        self, key: str, default: Any = None, check_env: bool = True
    ) -> Any:
        """
        Get a setting value

        Args:
            key: Setting key
            default: Default value if setting is not found
            check_env: If true, it will check the environment variable for
                this setting before reading from the DB.

        Returns:
            Setting value or default if not found
        """
        if self._closed:
            logger.error(
                "SettingsManager.get_setting('{}') called after close() — "
                "this is a bug; the caller should not reuse a closed manager",
                key,
            )
            raise RuntimeError(
                "SettingsManager has been closed. "
                "Create a new instance or call close() only at end of lifecycle."
            )

        # First check if this is an env-only setting
        if env_registry.is_env_only(key):
            return env_registry.get(key, default)

        # If using database first approach and session available, check database
        try:
            settings = self.__query_settings(key)
            if len(settings) == 1:
                # This is a bottom-level key.
                return self.__get_typed_setting_value(
                    settings[0], default, check_env
                )
                # Cache the result
            if len(settings) > 1:
                # This is a higher-level key.
                settings_map = {}
                for setting in settings:
                    output_key = str(setting.key).removeprefix(f"{key}.")
                    settings_map[output_key] = self.__get_typed_setting_value(
                        setting, default, check_env
                    )
                return settings_map
        except SQLAlchemyError:
            logger.exception(f"Error retrieving setting {key} from database")

        # Check env var before returning default (setting not in DB)
        if check_env:
            env_value = check_env_setting(key)
            if env_value is not None:
                default_meta = self.default_settings.get(key)
                if default_meta and isinstance(default_meta, dict):
                    ui_element = default_meta.get("ui_element", "text")
                    return get_typed_setting_value(
                        key,
                        None,
                        ui_element,
                        default=default,
                        check_env=True,
                    )
                logger.warning(
                    "Setting '{}' has env var override but is not in "
                    "defaults — returning raw string without type "
                    "conversion. Add this setting to a defaults JSON "
                    "file with a ui_element type to enable proper "
                    "type conversion.",
                    key,
                )
                return env_value

        # Return default if not found
        return default

    def get_bool_setting(
        self, key: str, default: bool = False, check_env: bool = True
    ) -> bool:
        """
        Get a setting value as a boolean, handling string conversion.

        Args:
            key: Setting key
            default: Default boolean value if setting is not found
            check_env: If true, it will check the environment variable for
                this setting before reading from the DB.

        Returns:
            Boolean value of the setting
        """
        value = self.get_setting(key, default, check_env)
        return to_bool(value, default)

    def set_setting(self, key: str, value: Any, commit: bool = True) -> bool:
        """
        Set a setting value

        Args:
            key: Setting key
            value: Setting value
            commit: Whether to commit the change

        Returns:
            True if successful, False otherwise
        """
        if self._closed:
            logger.error(
                "SettingsManager.set_setting('{}') called after close() — "
                "this is a bug; the caller should not reuse a closed manager",
                key,
            )
            raise RuntimeError(
                "SettingsManager has been closed. "
                "Create a new instance or call close() only at end of lifecycle."
            )
        if not self.db_session:
            logger.error(
                "Cannot edit setting {} because no DB was provided.", key
            )
            return False
        if self.settings_locked:
            logger.error("Cannot edit setting {} because they are locked.", key)
            return False

        # Always update database if available
        try:
            self._check_thread_safety()
            setting = (
                self.db_session.query(Setting)
                .filter(Setting.key == key)
                .first()
            )
            # Capture old value for the policy-change audit log below.
            old_value = setting.value if setting is not None else None
            if setting:
                if not setting.editable:
                    logger.error(
                        "Cannot change setting '{}' because it "
                        "is marked as non-editable.",
                        key,
                    )
                    return False

                setting.value = value  # type: ignore[assignment]
                setting.updated_at = (  # type: ignore[assignment]
                    func.now()
                )  # Explicitly set the current timestamp

                # Self-heal stale ui_element from before inference was added
                setting.ui_element = _infer_ui_element(
                    value, setting.ui_element
                )

                # Self-heal stale type from before the prefix dispatch was
                # added (e.g. legacy chat.* rows created with type=APP).
                # Also re-points category to the canonical per-prefix
                # value, since a row with the wrong type column was
                # almost certainly created before category dispatch was
                # in place either.
                inferred_type: Optional[SettingType] = None
                inferred_category: Optional[str] = None
                for prefix, category in _INFERRED_CATEGORY.items():
                    if key.startswith(prefix):
                        if prefix == "llm.":
                            inferred_type = SettingType.LLM
                        elif prefix == "search.":
                            inferred_type = SettingType.SEARCH
                        elif prefix == "report.":
                            inferred_type = SettingType.REPORT
                        elif prefix == "database.":
                            inferred_type = SettingType.DATABASE
                        elif prefix == "chat.":
                            inferred_type = SettingType.CHAT
                        inferred_category = category
                        break
                # Only self-heal when the key matches a known prefix. Keys
                # outside the dispatch map (e.g. focused_iteration.*,
                # langgraph_agent.* which ship as type=SEARCH) must keep their
                # shipped type — defaulting to APP here would wrongly demote
                # them on every edit.
                if inferred_type is not None and setting.type != inferred_type:
                    setting.type = inferred_type  # type: ignore[assignment]
                    if inferred_category is not None:
                        setting.category = inferred_category  # type: ignore[assignment]
            else:
                # Determine setting type from key
                setting_type = SettingType.APP
                if key.startswith("llm."):
                    setting_type = SettingType.LLM
                elif key.startswith("search."):
                    setting_type = SettingType.SEARCH
                elif key.startswith("report."):
                    setting_type = SettingType.REPORT
                elif key.startswith("database."):
                    setting_type = SettingType.DATABASE
                elif key.startswith("chat."):
                    setting_type = SettingType.CHAT

                # Infer ui_element from the value type
                ui_element = _infer_ui_element(value)

                # Create a new setting
                new_setting = Setting(
                    key=key,
                    value=value,
                    type=setting_type,
                    name=key.split(".")[-1].replace("_", " ").title(),
                    ui_element=ui_element,
                    description=f"Setting for {key}",
                )
                self.db_session.add(new_setting)

            if commit:
                self.db_session.commit()
                # Emit WebSocket event for settings change
                self._emit_settings_changed([key])

            # N16: audit log on policy.* / llm.require_local_endpoint /
            # embeddings.require_local changes. Targeted scope — only
            # security-relevant settings are logged, to avoid widening
            # this PR into a general audit-log refactor.
            if _is_policy_setting(key):
                logger.bind(policy_audit=True).warning(
                    "policy setting changed | key={} old={} new={}",
                    key,
                    old_value,
                    value,
                )

            return True
        except SQLAlchemyError:
            logger.exception(f"Error setting value for key: {key}")
            self.db_session.rollback()
            return False

    def clear_cache(self):
        """Clear the settings cache."""
        self.__dict__.pop("default_settings", None)
        logger.debug("Settings cache cleared")

    def get_all_settings(self, bypass_cache: bool = False) -> Dict[str, Any]:
        """
        Get all settings, merging defaults with database values.

        This ensures that new settings added to defaults.json automatically
        appear in the UI without requiring a database reset.

        Args:
            bypass_cache: If True, bypass the cache and read directly from database

        Returns:
            Dictionary of all settings
        """
        if self._closed:
            logger.error(
                "SettingsManager.get_all_settings() called after close() — "
                "this is a bug; the caller should not reuse a closed manager",
            )
            raise RuntimeError(
                "SettingsManager has been closed. "
                "Create a new instance or call close() only at end of lifecycle."
            )

        result = {}

        # Start with defaults so new settings are always included
        for key, default_setting in self.default_settings.items():
            result[key] = dict(default_setting)

            # Check env var override for defaults not yet in DB
            env_value = check_env_setting(key)
            if env_value is not None:
                ui_element = default_setting.get("ui_element", "text")
                typed_value = get_typed_setting_value(
                    key,
                    None,
                    ui_element,
                    default=env_value,
                    check_env=True,
                )
                result[key]["value"] = typed_value
                result[key]["editable"] = False

        # Override with database settings
        try:
            db_settings = self.__query_settings()
        except (SQLAlchemyError, LookupError):
            # LookupError fires when a row's `type` column holds an enum
            # value that's no longer in `SettingType` (e.g. legacy 'CHAT'-
            # typed rows from removed features). The previous handler only
            # caught SQLAlchemyError, so a single stale row would crash the
            # whole snapshot — and every caller of get_all_settings (the
            # /settings/api endpoint, benchmark start, research start, MCP
            # entry points, …) downstream of it. Falling back to defaults-
            # only is strictly safer than crashing.
            logger.exception(
                "Error querying settings from database in get_all_settings"
            )
            db_settings = []

        for setting in db_settings:
            # Handle type field - it might be a string or an enum
            setting_type = setting.type
            if hasattr(setting_type, "name"):
                setting_type = setting_type.name

            # Log if this is a custom setting not in defaults
            if str(setting.key) not in result:
                logger.debug(
                    f"Database contains custom setting not in "
                    f"defaults: {setting.key} (type={setting_type}, "
                    f"category={setting.category})"
                )

            # Override default with database value
            result[str(setting.key)] = {
                "value": setting.value,
                "type": setting_type,
                "name": setting.name,
                "description": setting.description,
                "category": setting.category,
                "ui_element": setting.ui_element,
                "options": setting.options,
                "min_value": setting.min_value,
                "max_value": setting.max_value,
                "step": setting.step,
                "visible": setting.visible,
                "editable": False if self.settings_locked else setting.editable,
            }

            # Override from the environment variables if needed.
            env_value = check_env_setting(str(setting.key))
            if env_value is not None:
                ui_element = result[str(setting.key)].get(
                    "ui_element", setting.ui_element
                )
                typed_value = get_typed_setting_value(
                    str(setting.key),
                    None,
                    ui_element,
                    default=env_value,
                    check_env=True,
                )
                result[str(setting.key)]["value"] = typed_value
                # Mark it as non-editable, because changes to the DB
                # value have no effect as long as the environment
                # variable is set.
                result[str(setting.key)]["editable"] = False

        # Re-inject search strategy options from code after DB merge,
        # since the DB stores options=null for this setting.
        if "search.search_strategy" in result:
            from local_deep_research.constants import get_available_strategies

            strategies = get_available_strategies()
            result["search.search_strategy"]["options"] = [
                {"label": s["label"], "value": s["name"]} for s in strategies
            ]

        return result

    def get_settings_snapshot(self) -> Dict[str, Any]:
        """
        Get a simplified settings snapshot with just key-value pairs.
        This is useful for passing settings to background threads or storing in metadata.

        Returns:
            Dictionary with setting keys mapped to their values
        """
        if self._closed:
            logger.error(
                "SettingsManager.get_settings_snapshot() called after close() — "
                "this is a bug; the caller should not reuse a closed manager",
            )
            raise RuntimeError(
                "SettingsManager has been closed. "
                "Create a new instance or call close() only at end of lifecycle."
            )

        all_settings = self.get_all_settings()
        settings_snapshot = {}

        for key, setting in all_settings.items():
            if isinstance(setting, dict) and "value" in setting:
                settings_snapshot[key] = setting["value"]
            else:
                settings_snapshot[key] = setting

        return settings_snapshot

    def create_or_update_setting(
        self, setting: Union[BaseSetting, Dict[str, Any]], commit: bool = True
    ) -> Optional[Setting]:
        """
        Create or update a setting

        Args:
            setting: Setting object or dictionary
            commit: Whether to commit the change

        Returns:
            The created or updated Setting model, or None if failed
        """
        if not self.db_session:
            logger.warning(
                "No database session available, cannot create/update setting"
            )
            return None
        if self.settings_locked:
            logger.error("Cannot edit settings because they are locked.")
            return None

        # Convert dict to BaseSetting if needed
        if isinstance(setting, dict):
            # Determine type from key if not specified
            if "type" not in setting and "key" in setting:
                setting_obj: BaseSetting
                key = setting["key"]
                if key.startswith("llm."):
                    setting_obj = LLMSetting(**setting)
                elif key.startswith("search."):
                    setting_obj = SearchSetting(**setting)
                elif key.startswith("report."):
                    setting_obj = ReportSetting(**setting)
                elif key.startswith("chat."):
                    setting_obj = ChatSetting(**setting)
                elif key.startswith("app."):
                    setting_obj = AppSetting(**setting)
                else:
                    # Keys outside the four buckets (e.g. local_search_*,
                    # embeddings.*, rag.*) live in their own namespaces.
                    # Use BaseSetting so the key is written verbatim —
                    # AppSetting's validator would otherwise prepend
                    # `app.` and silently relocate the row away from
                    # where every reader looks it up.  See #4208.
                    setting_obj = BaseSetting(type=SettingType.APP, **setting)
            else:
                # Use generic BaseSetting
                setting_obj = BaseSetting(**setting)
        else:
            setting_obj = setting

        try:
            # Check if setting exists
            db_setting = (
                self.db_session.query(Setting)
                .filter(Setting.key == setting_obj.key)
                .first()
            )

            if db_setting:
                # Update existing setting
                if not db_setting.editable:
                    logger.error(
                        "Cannot change setting '{}' because it "
                        "is marked as non-editable.",
                        setting_obj.key,
                    )
                    return None

                db_setting.value = setting_obj.value  # type: ignore[assignment]
                db_setting.name = setting_obj.name  # type: ignore[assignment]
                db_setting.description = setting_obj.description  # type: ignore[assignment]
                db_setting.category = setting_obj.category  # type: ignore[assignment]
                db_setting.type = setting_obj.type  # type: ignore[assignment]
                db_setting.ui_element = setting_obj.ui_element  # type: ignore[assignment]
                db_setting.options = setting_obj.options  # type: ignore[assignment]
                db_setting.min_value = setting_obj.min_value  # type: ignore[assignment]
                db_setting.max_value = setting_obj.max_value  # type: ignore[assignment]
                db_setting.step = setting_obj.step  # type: ignore[assignment]
                db_setting.visible = setting_obj.visible  # type: ignore[assignment]
                db_setting.editable = setting_obj.editable  # type: ignore[assignment]
                db_setting.updated_at = (  # type: ignore[assignment]
                    func.now()
                )  # Explicitly set the current timestamp
            else:
                # Create new setting
                db_setting = Setting(
                    key=setting_obj.key,
                    value=setting_obj.value,
                    type=setting_obj.type,
                    name=setting_obj.name,
                    description=setting_obj.description,
                    category=setting_obj.category,
                    ui_element=setting_obj.ui_element,
                    options=setting_obj.options,
                    min_value=setting_obj.min_value,
                    max_value=setting_obj.max_value,
                    step=setting_obj.step,
                    visible=setting_obj.visible,
                    editable=setting_obj.editable,
                )
                self.db_session.add(db_setting)

            if commit:
                self.db_session.commit()
                # Emit WebSocket event for settings change
                self._emit_settings_changed([setting_obj.key])

            return db_setting

        except SQLAlchemyError:
            logger.exception(
                f"Error creating/updating setting {setting_obj.key}"
            )
            self.db_session.rollback()
            return None

    def delete_setting(self, key: str, commit: bool = True) -> bool:
        """
        Delete a setting

        Args:
            key: Setting key
            commit: Whether to commit the change

        Returns:
            True if successful, False otherwise
        """
        if not self.db_session:
            logger.warning(
                "No database session available, cannot delete setting"
            )
            return False

        try:
            # Remove from database
            result = (
                self.db_session.query(Setting)
                .filter(Setting.key == key)
                .delete()
            )

            if commit:
                self.db_session.commit()

            return result > 0
        except SQLAlchemyError:
            logger.exception("Error deleting setting")
            self.db_session.rollback()
            return False

    def load_from_defaults_file(
        self, commit: bool = True, **kwargs: Any
    ) -> None:
        """
        Import settings from the defaults settings file.

        Args:
            commit: Whether to commit changes to database. The post-login
                atomic block in `web/auth/routes.py` passes ``commit=False``
                and combines this call with ``update_db_version(commit=False)``
                under a single terminal ``db_session.commit()`` — preserving
                the all-or-nothing invariant is what prevents the sticky-loop
                bug where `app.version` is missing after a partial write.
            **kwargs: Will be passed to `import_settings`.

        """
        start = time.perf_counter()
        row_count = len(self.default_settings)
        self.import_settings(self.default_settings, commit=commit, **kwargs)
        elapsed_ms = (time.perf_counter() - start) * 1000
        if elapsed_ms > 100:
            logger.info(
                f"load_from_defaults_file imported {row_count} settings "
                f"in {elapsed_ms:.0f}ms (commit={commit})"
            )
        else:
            logger.debug(
                f"load_from_defaults_file imported {row_count} settings "
                f"in {elapsed_ms:.0f}ms (commit={commit})"
            )

    def db_version_matches_package(self) -> bool:
        """
        Returns:
            True if the version saved in the DB matches the package version.

        """
        db_version = self.get_setting("app.version")
        logger.debug(
            f"App version saved in DB is {db_version}, have package "
            f"settings from version {package_version}."
        )

        return bool(db_version == package_version)

    def update_db_version(self, commit: bool = True) -> None:
        """
        Updates the version saved in the DB based on the package version.

        Args:
            commit: Whether to commit the version write to the database.
                Callers that want to combine this with other writes into
                a single atomic transaction should pass commit=False and
                commit the session themselves. The post-login block in
                `web/auth/routes.py` relies on this to bundle the defaults
                import and the `app.version` write into one SQLite
                transaction; splitting them risks the sticky-loop state
                where `app.version` never gets written.
        """
        logger.debug(f"Updating saved DB version to {package_version}.")

        self.delete_setting("app.version", commit=False)
        version = Setting(
            key="app.version",
            value=package_version,
            description="Version of the app this database is associated with.",
            editable=False,
            name="App Version",
            type=SettingType.APP,
            ui_element="text",
            visible=False,
        )

        if self.db_session is None:
            raise RuntimeError("Database session is not initialized")
        self.db_session.add(version)
        if commit:
            self.db_session.commit()

    def import_settings(
        self,
        settings_data: Dict[str, Any],
        commit: bool = True,
        overwrite: bool = True,
        delete_extra: bool = False,
    ) -> None:
        """
        Import settings directly from the export format. This can be used to
        re-import settings that have been exported with `get_all_settings()`.

        Args:
            settings_data: The raw settings data to import.
            commit: Whether to commit the DB after loading the settings.
            overwrite: If true, it will overwrite the value of settings that
                are already in the database.
            delete_extra: If true, it will delete any settings that are in
                the database but don't have a corresponding entry in
                `settings_data`.

        """
        if self.db_session is None:
            raise RuntimeError("Database session is not initialized")
        logger.debug(f"Importing {len(settings_data)} settings")

        for key, setting_values in settings_data.items():
            setting_values = dict(setting_values)
            if not overwrite:
                existing_value = self.get_setting(key)
                if existing_value is not None:
                    # Preserve the value from this setting.
                    setting_values["value"] = existing_value

            # Delete any existing setting so we can completely overwrite it.
            self.delete_setting(key, commit=False)

            # Convert type string to SettingType enum if needed
            if "type" in setting_values and isinstance(
                setting_values["type"], str
            ):
                setting_values["type"] = SettingType[setting_values["type"]]

            setting = Setting(
                key=key, **_filter_setting_columns(setting_values)
            )
            self.db_session.add(setting)

        if commit or delete_extra:
            self.db_session.commit()
            logger.info(f"Successfully imported {len(settings_data)} settings")
            # Emit WebSocket event for all imported settings
            self._emit_settings_changed(list(settings_data.keys()))

        if delete_extra:
            all_settings = self.get_all_settings()
            for key in all_settings:
                if key not in settings_data:
                    logger.debug(f"Deleting extraneous setting: {key}")
                    self.delete_setting(key, commit=False)

    def _emit_settings_changed(self, changed_keys: Optional[List[Any]] = None):
        """
        Emit WebSocket event when settings change

        Args:
            changed_keys: List of setting keys that changed
        """
        try:
            # Import here to avoid circular imports
            from ..web.services.socket_service import SocketIOService

            try:
                socket_service = SocketIOService()
            except ValueError:
                logger.debug(
                    "Not emitting socket event because server is not initialized."
                )
                return

            # settings_changed carries raw setting values (including plaintext
            # API keys), so it must reach only the owning user's own browser
            # tabs — never every connected client on this shared, single-user-
            # agnostic Socket.IO server. Resolve the user from the request that
            # triggered the change and scope the emit to that user's room.
            # A change made outside a request context (app start-up defaults,
            # background workers, migrations) has no user tab to notify, so we
            # skip the emit entirely rather than fall back to a broadcast.
            from flask import has_request_context, session

            if not has_request_context():
                return
            username = session.get("username")
            if not username:
                return

            # Get the changed settings
            settings_data = {}
            if changed_keys:
                for key in changed_keys:
                    setting_value = self.get_setting(key)
                    if setting_value is not None:
                        settings_data[key] = {"value": setting_value}

            # Emit the settings change event
            from datetime import datetime, UTC

            socket_service.emit_socket_event(
                "settings_changed",
                {
                    "changed_keys": changed_keys or [],
                    "settings": settings_data,
                    "timestamp": datetime.now(UTC).isoformat(),
                },
                room=socket_service.user_room(username),
            )

            logger.debug(
                f"Emitted settings_changed event for keys: {changed_keys}"
            )

        except Exception:
            logger.exception("Failed to emit settings change event")
            # Don't let WebSocket emission failures break settings saving

    @staticmethod
    def get_bootstrap_env_vars() -> Dict[str, str]:
        """
        Get environment variables that must be available before database access.
        These are critical for system initialization.

        Returns:
            Dict mapping env var names to their descriptions
        """
        # Get bootstrap vars from env registry
        return env_registry.get_bootstrap_vars()

    @staticmethod
    def is_bootstrap_env_var(env_var: str) -> bool:
        """
        Check if an environment variable is a bootstrap variable (needed before DB access).

        Args:
            env_var: Environment variable name

        Returns:
            True if this is a bootstrap variable
        """
        bootstrap_vars = SettingsManager.get_bootstrap_env_vars()
        return env_var in bootstrap_vars

    @staticmethod
    def is_env_only_setting(key: str) -> bool:
        """
        Check if a setting key is environment-only.

        Args:
            key: Setting key to check

        Returns:
            True if it's an env-only setting, False otherwise
        """
        return env_registry.is_env_only(key)

    @staticmethod
    def get_env_var_for_setting(setting_key: str) -> str:
        """
        Get the environment variable name for a given setting key.

        Args:
            setting_key: Setting key (e.g., "app.host")

        Returns:
            Environment variable name (e.g., "LDR_APP_HOST")
        """
        # Use the same logic as check_env_setting for consistency
        return f"LDR_{'_'.join(setting_key.split('.')).upper()}"

    @staticmethod
    def get_setting_key_for_env_var(env_var: str) -> Optional[str]:
        """
        Get the setting key for a given environment variable.

        Args:
            env_var: Environment variable name (e.g., "LDR_APP_HOST")

        Returns:
            Setting key (e.g., "app.host") or None if not a valid LDR env var
        """
        if not env_var.startswith("LDR_"):
            return None

        # Remove LDR_ prefix and convert to lowercase
        without_prefix = env_var[4:]
        parts = without_prefix.split("_")

        return ".".join(part.lower() for part in parts)


class SnapshotSettingsContext:
    """Read-only settings context backed by a snapshot dict.

    Unwraps {"value": x} setting objects into plain values and provides
    get_setting(key, default) for thread-safe snapshot access.
    """

    def __init__(
        self, snapshot=None, username=None, missing_key_log_level="DEBUG"
    ):
        self.snapshot = snapshot or {}
        self.username = username
        self._missing_key_log_level = missing_key_log_level
        self.values = {}
        for key, setting in self.snapshot.items():
            if isinstance(setting, dict) and "value" in setting:
                self.values[key] = setting["value"]
            else:
                self.values[key] = setting

    def get_setting(self, key, default=None):
        """Return the setting value for *key*, or *default* if absent."""
        if key in self.values:
            return self.values[key]
        logger.log(
            self._missing_key_log_level,
            "Setting '{}' not found in snapshot, using default",
            key,
        )
        return default
