siftlode/backend/app/sysconfig.py

"""Admin-editable system configuration: a DB-override layer over the env/config defaults.

The registry (SPECS) is the single source of truth for which config keys can be overridden
from the admin Configuration page, plus each key's type, admin-UI group, env/config default,
bounds, and whether it's a secret. The resolver returns the DB override if one is set, else
``settings.<default_attr>`` (same DB-override-or-default pattern as app.state). Secrets are
stored Fernet-encrypted at rest (requires TOKEN_ENCRYPTION_KEY; without it, secrets can't be
stored in the DB and stay env-only).
"""

from dataclasses import dataclass

from sqlalchemy.orm import Session

from app import logs, security
from app.config import Settings, canonical_choice, choices_of, settings
from app.models import SystemConfig

log = logs.get_logger(__name__)
# Config keys whose stored ciphertext failed to decrypt — tracked so the warning is logged once per
# key, not once per request (bounded by the number of secret config keys).
_decrypt_warned: set[str] = set()
# The same for a stored value that no longer coerces (a choice list that shrank in a release): it
# fails on EVERY read of a hot knob until someone saves it again. Keyed with the value, so saving a
# different bad value is reported afresh (bounded by the keys times the values ever saved).
_coerce_warned: set[tuple[str, str]] = set()


@dataclass(frozen=True)
class ConfigSpec:
    key: str
    type: str  # "int" | "str" | "bool" | "choice"
    group: str  # admin-UI grouping (logically related keys stay together)
    default_attr: str  # attribute on `settings` holding the env/config default
    secret: bool = False
    min: int | None = None
    max: int | None = None
    allow_empty: bool = False  # str keys: a stored "" is an explicit empty value, not "unset"

    @property
    def choices(self) -> tuple[str, ...]:
        """A choice key's allowed values, read from its setting's Literal type in config.py — the one
        place the list is written. The admin page renders them as a picker."""
        return choices_of(Settings.model_fields[self.default_attr].annotation)


# Registry of DB-overridable keys. Add a key here + wire its read through get_*() to move it
# off env into the admin UI; the admin page renders it automatically from this list. The ORDER is the
# admin page's tab order (first appearance of each group), so keep a group's keys together.
SPECS: tuple[ConfigSpec, ...] = (
    # --- Access / registration ---
    ConfigSpec("allow_registration", "bool", "access", "allow_registration"),
    # --- Email / SMTP (one logical group; the password is the only secret) ---
    ConfigSpec("smtp_host", "str", "email", "smtp_host"),
    ConfigSpec("smtp_port", "int", "email", "smtp_port", min=1, max=65535),
    ConfigSpec("smtp_user", "str", "email", "smtp_user", allow_empty=True),
    ConfigSpec("smtp_from", "str", "email", "smtp_from"),
    ConfigSpec("smtp_password", "str", "email", "smtp_password", secret=True),
    # --- YouTube Data API key (secret; optional — public reads prefer it over OAuth) ---
    ConfigSpec("youtube_api_key", "str", "youtube", "youtube_api_key", secret=True),
    # Optional egress proxy for YouTube API calls (fixed-IP host for an IP-restricted key).
    ConfigSpec("youtube_api_proxy", "str", "youtube", "youtube_api_proxy", allow_empty=True),
    # --- Google OAuth client (Sign in with Google; both encrypted, env fallback). Set from the
    #     install wizard / Configuration page so no restart is needed when the creds change. ---
    ConfigSpec("google_client_id", "str", "google", "google_client_id", secret=True),
    ConfigSpec("google_client_secret", "str", "google", "google_client_secret", secret=True),
    # --- Quota (shared YouTube Data API daily budget) ---
    ConfigSpec("quota_daily_budget", "int", "quota", "quota_daily_budget", min=1, max=1_000_000),
    ConfigSpec(
        "backfill_quota_reserve", "int", "quota", "backfill_quota_reserve", min=0, max=1_000_000
    ),
    ConfigSpec(
        "search_daily_limit_per_user",
        "int",
        "quota",
        "search_daily_limit_per_user",
        min=0,
        max=100_000,
    ),
    # Live-search source: "scrape" (InnerTube, zero API quota) or "api" (search.list, 100u/page).
    ConfigSpec("search_source", "choice", "quota", "search_source"),
    # --- Backfill (recent-first first pass per channel) ---
    ConfigSpec(
        "backfill_recent_max_videos",
        "int",
        "backfill",
        "backfill_recent_max_videos",
        min=1,
        max=100_000,
    ),
    ConfigSpec(
        "backfill_recent_max_days", "int", "backfill", "backfill_recent_max_days", min=1, max=36_500
    ),
    ConfigSpec("rss_fallback_hours", "int", "backfill", "rss_fallback_hours", min=0, max=168),
    # --- Shorts probe ---
    ConfigSpec(
        "shorts_probe_max_seconds", "int", "shorts", "shorts_probe_max_seconds", min=1, max=3_600
    ),
    ConfigSpec("shorts_probe_batch", "int", "shorts", "shorts_probe_batch", min=1, max=10_000),
    # --- Batch sizes ---
    ConfigSpec("enrich_batch_size", "int", "batch", "enrich_batch_size", min=1, max=50),
    ConfigSpec("autotag_title_sample", "int", "batch", "autotag_title_sample", min=1, max=1_000),
    # --- Maintenance (rolling re-validation of already-known videos) ---
    # Videos re-checked per maintenance run. Folded here from app_state (B-core-11) so every admin
    # knob lives in one store; read by sync/maintenance.py, tuned from the Scheduler dashboard.
    ConfigSpec(
        "maintenance_revalidate_batch",
        "int",
        "maintenance",
        "maintenance_revalidate_batch",
        min=100,
        max=100_000,
    ),
    # --- Discovery cleanup (search + explore ephemeral content) ---
    # Days an explored-but-unsubscribed channel is kept before the cleanup job purges it.
    ConfigSpec("explore_grace_days", "int", "explore", "explore_grace_days", min=0, max=3_650),
    # Days an un-kept live-search result video is kept before the cleanup job removes it.
    ConfigSpec("search_grace_days", "int", "explore", "search_grace_days", min=0, max=3_650),
    # --- Download center (yt-dlp) — per-user defaults + retention/GC + layout ---
    ConfigSpec("download_total_max_bytes", "int", "downloads", "download_total_max_bytes", min=0),
    ConfigSpec(
        "download_default_max_bytes", "int", "downloads", "download_default_max_bytes", min=0
    ),
    ConfigSpec(
        "download_default_max_concurrent",
        "int",
        "downloads",
        "download_default_max_concurrent",
        min=1,
        max=20,
    ),
    ConfigSpec(
        "download_default_max_jobs",
        "int",
        "downloads",
        "download_default_max_jobs",
        min=1,
        max=100_000,
    ),
    # min=1, NOT 0: the per-file cap is enforced as `if cap and size > cap`, so a stored 0 would read
    # as falsy and SILENTLY disable the limit (an unbounded streamed write to the volume). Floor it at
    # 1 so the cap is always active; the overall footprint is separately bounded by the byte quota.
    # `max` is not cosmetic: `bodylimit.ROUTE_MAX_BODY_BYTES` refuses an upload body over this same
    # number at the edge, before the route's own check ever sees it. An admin allowed to raise the
    # knob past the edge would get 413s from a layer the admin page never mentions, so the two are
    # pinned together here and `tests/test_body_limit.py` holds them that way.
    ConfigSpec(
        "download_upload_max_bytes",
        "int",
        "downloads",
        "download_upload_max_bytes",
        min=1,
        max=settings.download_upload_max_bytes,
    ),
    ConfigSpec(
        "download_retention_days", "int", "downloads", "download_retention_days", min=0, max=3_650
    ),
    ConfigSpec(
        "download_gc_grace_days", "int", "downloads", "download_gc_grace_days", min=0, max=3_650
    ),
    ConfigSpec("download_layout", "choice", "downloads", "download_layout"),
    ConfigSpec(
        "download_worker_concurrency",
        "int",
        "downloads",
        "download_worker_concurrency",
        min=1,
        max=16,
    ),
    ConfigSpec("sponsorblock_default", "bool", "downloads", "sponsorblock_default"),
    # --- Audit log ---
    # Days to keep admin audit-log rows before the audit_gc job prunes them (0 = keep forever).
    ConfigSpec("audit_retention_days", "int", "audit", "audit_retention_days", min=0, max=3650),
    # --- Diagnostics (E22) ---
    # Runtime log level for the API process. Applied without a restart — see routes/config.py and
    # main.py's lifespan.
    ConfigSpec("log_level", "choice", "diagnostics", "log_level"),
    # Days the log_events table keeps a row (the log_gc job prunes). No "0 = forever": stored logs
    # carry account-linked request history, and the privacy policy names a limit.
    ConfigSpec("log_retention_days", "int", "diagnostics", "log_retention_days", min=1, max=90),
    # --- Plex integration (optional; local-file playback + Plex metadata) ---
    ConfigSpec("plex_enabled", "bool", "plex", "plex_enabled"),
    ConfigSpec("plex_server_url", "str", "plex", "plex_server_url"),
    ConfigSpec("plex_server_token", "str", "plex", "plex_server_token", secret=True),
    ConfigSpec("plex_path_map", "str", "plex", "plex_path_map"),
    ConfigSpec("plex_libraries", "str", "plex", "plex_libraries"),
    ConfigSpec("plex_overlay_map", "str", "plex", "plex_overlay_map"),
    # (plex_sync_interval_min is tuned from the Scheduler dashboard, not here — it's a scheduler job.)
    ConfigSpec("plex_max_transcodes", "int", "plex", "plex_max_transcodes", min=0, max=16),
    ConfigSpec("plex_max_sessions", "int", "plex", "plex_max_sessions", min=1, max=64),
    # Transcode re-encode tuning (E6 S1). The libx264 knobs apply only to the software fallback.
    # crf 0–51 (x264 range); max_height 0 = source.
    ConfigSpec("plex_transcode_hw", "choice", "plex", "plex_transcode_hw"),
    ConfigSpec("plex_transcode_preset", "choice", "plex", "plex_transcode_preset"),
    ConfigSpec("plex_transcode_crf", "int", "plex", "plex_transcode_crf", min=0, max=51),
    ConfigSpec(
        "plex_transcode_max_height", "int", "plex", "plex_transcode_max_height", min=0, max=4320
    ),
    # Audio loudness normalisation on the HLS path (remux + transcode alike, not direct play).
    ConfigSpec("plex_audio_loudnorm", "bool", "plex", "plex_audio_loudnorm"),
    # NOTE: plex_watch_sync_interval_min is a config.py default read by the watch-sync scheduler job
    # (scheduler.py); it is intentionally NOT a runtime admin knob, so it stays out of SPECS.
    # --- TMDB enrichment (E5 S2; optional) — the API key is the only secret ---
    ConfigSpec("tmdb_api_key", "str", "tmdb", "tmdb_api_key", secret=True),
    ConfigSpec("tmdb_language", "str", "tmdb", "tmdb_language"),
    # --- Bug reports (E16): the Gitea issue tracker a debug session's issue goes to ---
    ConfigSpec("issues_gitea_url", "str", "issues", "issues_gitea_url"),
    ConfigSpec("issues_gitea_repo", "str", "issues", "issues_gitea_repo"),
    ConfigSpec("issues_gitea_token", "str", "issues", "issues_gitea_token", secret=True),
    # (tmdb_enrich_interval_min is a scheduler job, tuned from the Scheduler dashboard; tmdb_enrich_batch
    # is a config.py default — both stay out of SPECS, like plex_sync_interval_min.)
)

_BY_KEY: dict[str, ConfigSpec] = {s.key: s for s in SPECS}


def spec(key: str) -> ConfigSpec | None:
    return _BY_KEY.get(key)


def secrets_manageable() -> bool:
    """Secret values can only be stored when Fernet is configured (TOKEN_ENCRYPTION_KEY)."""
    return bool(settings.token_encryption_key)


def _default(s: ConfigSpec):
    return getattr(settings, s.default_attr)


class InvalidSetting(ValueError):
    """A value `set_value` refuses, as a reason the UI translates (`errors.refused.setting_<reason>`)
    plus what the sentence needs — the English message is for logs only. A ValueError still, so a
    caller that only needs "refused or not" keeps working."""

    def __init__(self, key: str, reason: str, **params: object):
        super().__init__(f"{key}: {reason} {params}")
        self.key = key
        self.reason = reason
        self.params = params


def _choice(s: ConfigSpec, raw: object) -> str:
    """The canonical spelling of `raw` among the key's choices (any letter case), or InvalidSetting."""
    option = canonical_choice(s.choices, raw)
    if option is None:
        raise InvalidSetting(s.key, "not_a_choice", choices=", ".join(s.choices))
    return option


def _coerce(s: ConfigSpec, raw: str):
    if s.type == "int":
        return int(raw)
    if s.type == "bool":
        return raw == "true"
    if s.type == "choice":
        # A stored value that is no longer a choice (the list shrank) reads as the default, like a
        # non-numeric int override does — see get().
        return _choice(s, raw)
    return raw


def _raw_override(db: Session, key: str) -> str | None:
    """Stored (decrypted) override string for `key`, or None if no override is set.

    A decrypt failure — a rotated or removed ``TOKEN_ENCRYPTION_KEY`` vs the key the value was written
    with — must NOT crash the caller. `_raw_override` sits on hot paths (`/api/me` via `google_enabled`,
    login, every YouTube scheduler job, account purge), so a raw ``InvalidToken`` here would 500 the
    whole app and the SPA wouldn't even boot. Fall back to the env/config default instead, and warn once
    per key so the operator gets a real signal without a per-request traceback. Recovery is documented in
    docs/self-hosting.md: the key must never change once secrets are stored. (B-core-2)
    """
    row = db.get(SystemConfig, key)
    if row is None or row.value is None:
        return None
    if not row.encrypted:
        return row.value
    try:
        return security.decrypt(row.value)
    except Exception:  # noqa: BLE001 — Fernet InvalidToken / malformed ciphertext (key mismatch)
        if key not in _decrypt_warned:
            _decrypt_warned.add(key)
            log.warning(
                "config %r could not be decrypted (TOKEN_ENCRYPTION_KEY changed or missing?) — falling "
                "back to the env/config default; the key must not change once secrets are stored",
                key,
            )
        return None


def get(db: Session, key: str):
    """Effective value: the DB override (typed) if set, else the env/config default."""
    s = _BY_KEY[key]
    raw = _raw_override(db, key)
    if raw is None:
        return _default(s)
    if raw == "" and not s.allow_empty:
        # For most keys an empty override means "fall back to the env default". Keys that opt
        # into allow_empty (e.g. smtp_user, youtube_api_proxy) treat "" as a real value — so an
        # admin can blank one to DISABLE it rather than being stuck with the env default.
        return _default(s)
    try:
        return _coerce(s, raw)
    except (TypeError, ValueError) as exc:
        # The admin set this in the UI and it is quietly not in force: the app runs on the env
        # default while the settings page goes on showing the value that was saved.
        if (s.key, raw) not in _coerce_warned:
            _coerce_warned.add((s.key, raw))
            log.warning(
                "a stored setting could not be coerced; the default is used instead",
                extra=logs.event(
                    "sysconfig.override_uncoercible", key=s.key, reason=type(exc).__name__
                ),
            )
        return _default(s)


def get_str(db: Session, key: str) -> str:
    v = get(db, key)
    return "" if v is None else str(v)


def get_int(db: Session, key: str) -> int:
    return int(get(db, key))


def get_bool(db: Session, key: str) -> bool:
    return bool(get(db, key))


def set_value(db: Session, key: str, value) -> None:
    """Validate, coerce, and stage a DB override for `key` (secrets are encrypted). Flushes but does
    NOT commit — the caller owns the transaction, so the change lands together with its audit row
    (B-core-10). The staged value is visible to a same-session `get()` before the commit."""
    s = _BY_KEY[key]
    if s.type == "int":
        try:
            v = int(value)
        except (TypeError, ValueError):
            raise InvalidSetting(key, "not_a_number") from None
        if s.min is not None and v < s.min:
            raise InvalidSetting(key, "too_low", min=s.min)
        if s.max is not None and v > s.max:
            raise InvalidSetting(key, "too_high", max=s.max)
        raw = str(v)
    elif s.type == "bool":
        raw = "true" if value else "false"
    elif s.type == "choice":
        raw = _choice(s, value)
    else:
        raw = "" if value is None else str(value)

    stored, encrypted = raw, False
    if s.secret:
        if not secrets_manageable():
            raise InvalidSetting(key, "no_encryption_key")
        stored, encrypted = security.encrypt(raw) or "", True

    row = db.get(SystemConfig, key)
    if row is None:
        row = SystemConfig(key=key)
        db.add(row)
    row.value = stored
    row.encrypted = encrypted
    db.flush()


def reset(db: Session, key: str) -> None:
    """Remove the DB override for `key` (fall back to the env/config default). Flushes but does NOT
    commit — the caller owns the transaction (B-core-10)."""
    row = db.get(SystemConfig, key)
    if row is not None:
        db.delete(row)
        db.flush()


def describe(db: Session) -> dict:
    """Admin-UI metadata + current values, grouped. Secret values are never echoed —
    only whether an override is set and whether an env default exists."""
    groups: dict[str, list[dict]] = {}
    for s in SPECS:
        is_set = db.get(SystemConfig, s.key) is not None
        item: dict = {
            "key": s.key,
            "type": s.type,
            "group": s.group,
            "secret": s.secret,
            "min": s.min,
            "max": s.max,
            "is_set": is_set,  # an override row exists
            "allow_empty": s.allow_empty,  # if true, the UI stores "" rather than deleting the row
            "choices": list(s.choices),  # the picker's options; [] for any other type
        }
        if s.secret:
            item["default_is_set"] = bool(_default(s))
        else:
            item["value"] = get(db, s.key)
            item["default"] = _default(s)
        groups.setdefault(s.group, []).append(item)
    return {"groups": groups, "secrets_manageable": secrets_manageable()}