app/store_config.py

"""Persist bootstrap credentials and session-version state.

Configuration updates are serialized and atomically replace `config.json`.
"""

import json
from datetime import datetime, timezone

from app.fsutil import atomic_write_text, file_lock, read_json
from app.models import StoredConfig
from app.paths import Paths
from app.security_passwords import hash_password, verify_password


def utc_now() -> datetime:
    """Return an aware UTC timestamp."""

    return datetime.now(timezone.utc)


class ConfigStore:
    """Read and mutate the durable single-user configuration."""

    def __init__(self, paths: Paths) -> None:
        self.paths = paths

    def bootstrap(self, initial_password: str | None) -> StoredConfig | None:
        """Create initial configuration only when a password is available."""

        with file_lock(self.paths.config):
            existing = read_json(self.paths.config)
            if existing is not None:
                return StoredConfig.model_validate(existing)
            if not initial_password:
                return None
            now = utc_now()
            config = StoredConfig(
                password_hash=hash_password(initial_password),
                session_version=1,
                created_at=now,
                updated_at=now,
            )
            self._write_unlocked(config)
            return config

    def load(self) -> StoredConfig | None:
        """Load configuration, or None when setup has not occurred."""

        value = read_json(self.paths.config)
        return StoredConfig.model_validate(value) if value is not None else None

    def password_matches(self, password: str) -> bool:
        """Check a candidate password against current configuration."""

        config = self.load()
        return bool(config and verify_password(password, config.password_hash))

    def change_password(self, old_password: str, new_password: str) -> bool:
        """Rotate the password and invalidate all prior sessions."""

        with file_lock(self.paths.config):
            value = read_json(self.paths.config)
            if value is None:
                return False
            config = StoredConfig.model_validate(value)
            if not verify_password(old_password, config.password_hash):
                return False
            updated = config.model_copy(
                update={
                    "password_hash": hash_password(new_password),
                    "session_version": config.session_version + 1,
                    "updated_at": utc_now(),
                }
            )
            self._write_unlocked(updated)
            return True

    def _write_unlocked(self, config: StoredConfig) -> None:
        payload = json.dumps(
            config.model_dump(mode="json"),
            ensure_ascii=False,
            indent=2,
        )
        atomic_write_text(self.paths.config, payload + "\n")