inkwell/src/thoughtsync/settings.py

from __future__ import annotations

import json
import secrets
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Literal

from .models.settings import Setting

SettingType = Literal["string", "bool", "int"]


@dataclass(frozen=True)
class SettingDef:
    key: str
    type: SettingType
    default: Any
    label: str
    description: str
    group: str


# The source of truth for every user-facing setting. Add a row here and it appears
# in the admin Settings UI with a working default — no migration, no env var.
REGISTRY: list[SettingDef] = [
    SettingDef(
        "site_name", "string", "ThoughtSync", "Site name", "Shown in the header and the browser tab.", "General"
    ),
    SettingDef(
        "allow_registration",
        "bool",
        True,
        "Allow new registrations",
        "When off, only existing users can sign in. The first account is always allowed.",
        "Access",
    ),
    SettingDef(
        "session_ttl_days",
        "int",
        30,
        "Session length (days)",
        "How long a signed-in session stays valid before another login is required.",
        "Access",
    ),
    SettingDef(
        "max_attachment_mb",
        "int",
        25,
        "Max attachment size (MB)",
        "Largest single file that can be attached to a note. Capped by the server body limit.",
        "Attachments",
    ),
    SettingDef(
        "enable_url_unfurl",
        "bool",
        True,
        "Link previews",
        "Let the server fetch a page's title/description/image to preview pasted links. "
        "The server contacts the linked site; private/internal addresses are always blocked.",
        "Links",
    ),
]

_BY_KEY: dict[str, SettingDef] = {d.key: d for d in REGISTRY}

# Internal, non-UI reserved key: the persisted cookie-signing secret. Stored in the
# same table but never listed in the registry, so it never shows in the Settings UI.
SECRET_KEY_SETTING = "secret_key"


def _coerce_bool(raw: Any) -> bool:
    if isinstance(raw, bool):
        return raw
    if isinstance(raw, str):
        return raw.strip().lower() in ("1", "true", "yes", "on")
    return bool(raw)


def _coerce(defn: SettingDef, raw: Any) -> Any:
    if defn.type == "bool":
        return _coerce_bool(raw)
    if defn.type == "int":
        try:
            return int(raw)
        except (ValueError, TypeError):
            return defn.default
    return str(raw)


async def _load_raw(db, key: str) -> Any:
    row = await db.get(Setting, key)
    if row is None:
        return None
    try:
        return json.loads(row.value)
    except (ValueError, TypeError):
        return None


async def _upsert(db, key: str, value: Any) -> None:
    row = await db.get(Setting, key)
    payload = json.dumps(value)
    if row is None:
        db.add(Setting(key=key, value=payload))
    else:
        row.value = payload
        row.updated_at = datetime.now(timezone.utc)


async def get_setting(db, key: str) -> Any:
    defn = _BY_KEY.get(key)
    if defn is None:
        raise KeyError(key)
    raw = await _load_raw(db, key)
    return defn.default if raw is None else _coerce(defn, raw)


async def get_public_config(db) -> dict:
    """Non-sensitive settings the unauthenticated login/register screen needs."""
    return {
        "site_name": await get_setting(db, "site_name"),
        "allow_registration": await get_setting(db, "allow_registration"),
        "enable_url_unfurl": await get_setting(db, "enable_url_unfurl"),
    }


async def get_admin_settings(db) -> list[dict]:
    """Every registry setting with its current value + metadata, for the admin UI."""
    result: list[dict] = []
    for d in REGISTRY:
        result.append(
            {
                "key": d.key,
                "type": d.type,
                "value": await get_setting(db, d.key),
                "default": d.default,
                "label": d.label,
                "description": d.description,
                "group": d.group,
            }
        )
    return result


def validate_updates(updates: dict) -> tuple[dict, str | None]:
    """Coerce/validate a {key: value} dict against the registry. Returns
    (clean_values, error_message). An unknown key or a bad int is rejected."""
    clean: dict = {}
    for key, val in updates.items():
        defn = _BY_KEY.get(key)
        if defn is None:
            return {}, f"unknown setting: {key}"
        if defn.type == "int":
            try:
                clean[key] = int(val)
            except (ValueError, TypeError):
                return {}, f"{defn.label} must be a whole number"
        elif defn.type == "bool":
            clean[key] = _coerce_bool(val)
        else:
            clean[key] = str(val)
    return clean, None


async def set_settings(db, updates: dict) -> None:
    for key, value in updates.items():
        await _upsert(db, key, value)


async def load_or_create_secret_key(db) -> str:
    """Return the persisted cookie-signing secret, generating + storing one on first
    run. Keeps sessions valid across restarts with no env var or volume required."""
    row = await db.get(Setting, SECRET_KEY_SETTING)
    if row is not None:
        try:
            val = json.loads(row.value)
            if isinstance(val, str) and val:
                return val
        except (ValueError, TypeError):
            pass
    key = secrets.token_urlsafe(48)
    await _upsert(db, SECRET_KEY_SETTING, key)
    await db.commit()
    return key