AR-Autopilot/arautopilot/core/user_store.py

"""Local user database (JSON file).

Sprint 2.5: persists the list of Users + their hashed PINs to a single
JSON file (per Studio install). On the bridge display the same format
is consumed but typically managed by the Owner via the Studio UI.

Sprint 8 will migrate this to a signed/encrypted store bound to the HWID.
"""

from __future__ import annotations

import json
from pathlib import Path

from pydantic import TypeAdapter

from arautopilot.core.rbac import Role
from arautopilot.core.user import User


class UserStore:
    """Append/overwrite list of :class:`User` persisted to a JSON file."""

    def __init__(self, path: Path | str) -> None:
        self.path = Path(path)
        self._users: dict[str, User] = {}
        if self.path.exists():
            self._load()
        else:
            self.path.parent.mkdir(parents=True, exist_ok=True)

    # ----- Persistence ----------------------------------------------------
    def _load(self) -> None:
        text = self.path.read_text(encoding="utf-8")
        if not text.strip():
            return
        data = json.loads(text)
        if not isinstance(data, list):
            raise ValueError(f"{self.path}: expected a JSON list at the top level")
        adapter = TypeAdapter(list[User])
        users = adapter.validate_python(data)
        self._users = {u.user_id: u for u in users}

    def save(self) -> None:
        adapter = TypeAdapter(list[User])
        data = adapter.dump_python(list(self._users.values()), mode="json")
        self.path.write_text(
            json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8"
        )

    # ----- CRUD -----------------------------------------------------------
    def add(self, user: User) -> None:
        if user.user_id in self._users:
            raise ValueError(f"user_id {user.user_id!r} already exists")
        self._users[user.user_id] = user
        self.save()

    def remove(self, user_id: str) -> None:
        if user_id not in self._users:
            raise KeyError(user_id)
        del self._users[user_id]
        self.save()

    def replace(self, user: User) -> None:
        """Insert or update the user with this user_id."""
        self._users[user.user_id] = user
        self.save()

    def get(self, user_id: str) -> User | None:
        return self._users.get(user_id)

    def find_by_name(self, display_name: str) -> User | None:
        for u in self._users.values():
            if u.display_name == display_name:
                return u
        return None

    def all_users(self) -> list[User]:
        """Return every user, sorted by display_name."""
        return sorted(self._users.values(), key=lambda u: u.display_name.lower())

    def by_role(self, role: Role) -> list[User]:
        return [u for u in self.all_users() if u.role is role]

    def __len__(self) -> int:
        return len(self._users)

    def __contains__(self, user_id: object) -> bool:
        return user_id in self._users


def seed_demo_users(store: UserStore) -> None:
    """Populate a fresh store with one user of each role for first-run UX.

    Demo PINs (well-known, only used in dev/sample stores -- the user is
    expected to change these immediately):

        Super Admin "Alvaro"      PIN 1111
        Engineer    "Eng Demo"    PIN 2222
        Owner       "Captain"     PIN 3333
        User        "Crew"        PIN 4444
    """
    if len(store) > 0:
        return
    store.add(User.create(display_name="Alvaro",   role=Role.SUPER_ADMIN, pin="1111"))
    store.add(User.create(display_name="Eng Demo", role=Role.ENGINEER,    pin="2222"))
    store.add(User.create(display_name="Captain",  role=Role.OWNER,       pin="3333"))
    store.add(User.create(display_name="Crew",     role=Role.USER,        pin="4444"))