beer_data/src/flask/mongo/repository.py

"""Mongo access layer for the beer_data app.

Two collections are used:

``beer_data``
    The historic readings.  One document per gravity reading.  **Never modified or
    removed by the code in this module** apart from the explicit "add or update a
    reading" path used by the brew-process form.

``beer_runs``
    New in this version.  One document per brew run holding the *canonical*
    metadata that could not be derived from the readings themselves: the beer
    name, the tap number it is on, the original gravity, and whether the run is
    still in the fermenter.  The backfill script only ever inserts into this
    collection, so it is safe to run repeatedly.
"""

from __future__ import annotations

import logging
import os
from typing import Any

from pymongo import MongoClient
from pymongo.errors import PyMongoError
from werkzeug.security import check_password_hash, generate_password_hash

from brewing import Run

__all__ = ["BeerRepository", "get_repository"]

logger = logging.getLogger(__name__)


class BeerRepository:
    """Thin, testable wrapper around the two Mongo collections."""

    def __init__(self, client: MongoClient | None = None,
                 db_name: str = "beer_db") -> None:
        self.client = client or _build_client()
        self.db = self.client[db_name]
        self.readings = self.db["beer_data"]
        self.users = self.db["users"]
        # NB: named `run_meta`, not `runs` — an attribute called `runs` would
        # shadow the runs() method on the instance.
        self.run_meta = self.db["beer_runs"]

    # -- users -------------------------------------------------------------- #

    def user_check(self, username: str, password: str) -> bool:
        """Check the login credentials.

        Passwords are stored as a hash produced by
        ``werkzeug.security.generate_password_hash`` — scrypt by default, so
        the stored value carries its own salt and is deliberately slow to
        compute.

        Legacy rows stored the password in plain text.  Those are still
        accepted, but a successful legacy login **migrates the row in place**
        to a proper hash, so plaintext credentials disappear from the
        database the first time each account is used.
        """
        if not username or not password:
            return False

        doc = self.users.find_one({"username": str(username)})
        if not doc:
            return False

        stored = str(doc.get("password") or "")
        if not stored:
            return False

        if _is_hashed(stored):
            return check_password_hash(stored, password)

        # Legacy plaintext password — accept it, then upgrade the row.
        if stored == password:
            self._upgrade_password(doc, password)
            return True

        return False

    def _upgrade_password(self, doc: dict, password: str) -> None:
        """Replace a plaintext password with a proper hash."""
        try:
            self.users.update_one(
                {"_id": doc["_id"]},
                {"$set": {
                    "password": generate_password_hash(password),
                    "password_migrated_at": _now_iso(),
                }},
            )
        except PyMongoError:
            # A failed upgrade must not stop the login succeeding — but the
            # operator needs to know the password is still stored in plain
            # text, so log it rather than swallowing the error silently.
            logger.warning(
                "Could not upgrade the password for user %r to a hashed value; "
                "it is still stored in plain text",
                doc.get("username"),
                exc_info=True,
            )

    # -- readings ----------------------------------------------------------- #

    def all_reading_docs(self) -> list[dict]:
        """Every reading document, oldest first."""
        return list(self.readings.find().sort([("date", 1), ("beer_run_id", 1)]))

    def reading_docs_for_run(self, beer_run_id: str) -> list[dict]:
        return list(
            self.readings.find({"beer_run_id": _as_id(beer_run_id)}).sort("date", 1)
        )

    def runs(self) -> list[Run]:
        """Every run, built from the reading documents, oldest run first."""
        grouped: dict[str, list[dict]] = {}
        for doc in self.all_reading_docs():
            grouped.setdefault(str(doc.get("beer_run_id")), []).append(doc)

        def run_sort_key(run: Run) -> tuple:
            first = run.first_reading()
            return (first.date if (first and first.date) else date_min(), run.beer_run_id)

        runs = [Run.from_docs(run_id, docs) for run_id, docs in grouped.items()]
        runs.sort(key=run_sort_key)
        return runs

    def run(self, beer_run_id: str) -> Run | None:
        docs = self.reading_docs_for_run(beer_run_id)
        if not docs:
            return None
        return Run.from_docs(str(beer_run_id), docs)

    def active_runs(self) -> list[Run]:
        """Runs with no final reading yet."""
        return [r for r in self.runs() if r.is_active()]

    def completed_runs(self) -> list[Run]:
        return [r for r in self.runs() if not r.is_active()]

    def upsert_reading(self, *, beer_run_id: str, beer_type: str, sg: int,
                       reading_date: str, final_reading: bool, comment: str) -> str:
        """Add a reading, or update the matching one if it already exists.

        The unique key is ``(date, beer_run_id)`` which is what the original
        application used, so re-submitting the same day updates in place
        instead of creating a duplicate.  Returns the ``_id`` of the document.
        """
        query = {"date": _as_date(reading_date), "beer_run_id": _as_id(beer_run_id)}
        update = {
            "$set": {
                "beer_run_id": str(beer_run_id),
                "beer_type": beer_type,
                "sg": str(sg),
                "date": _as_date(reading_date),
                "final_reading": "True" if final_reading else "False",
                "comment": str(comment),
            }
        }
        result = self.readings.update_one(query, update, upsert=True)
        if result.upserted_id is not None:
            return str(result.upserted_id)
        existing = self.readings.find_one(query)
        return str(existing.get("_id")) if existing else ""

    def set_final_reading(self, beer_run_id: str, reading_date: str,
                          final: bool) -> int:
        """Flag (or unflag) a reading as the final reading of its run."""
        result = self.readings.update_many(
            {"date": _as_date(reading_date), "beer_run_id": _as_id(beer_run_id)},
            {"$set": {"final_reading": "True" if final else "False"}},
        )
        return result.modified_count

    # -- run metadata ------------------------------------------------------- #

    def run_metadata(self, beer_run_id: str) -> dict:
        """The stored metadata for a run, or an empty dict."""
        doc = self.run_meta.find_one({"beer_run_id": _as_id(beer_run_id)})
        if not doc:
            return {}
        doc.pop("_id", None)
        return doc

    def all_run_metadata(self) -> dict[str, dict]:
        """All stored run metadata, keyed by ``beer_run_id``."""
        result: dict[str, dict] = {}
        for doc in self.run_meta.find():
            doc.pop("_id", None)
            result[_as_id(doc.get("beer_run_id"))] = doc
        return result

    def upsert_run_metadata(self, beer_run_id: str, **fields: Any) -> dict:
        """Create or update the metadata document for a run.

        Only the keys passed in are written; anything already stored is left
        alone.  A field passed as ``None`` is **cleared** — that is how the
        form removes a name or a tap number the operator has deleted.

        Use ``upsert_run_metadata_partial`` to update some fields while
        leaving the rest untouched.
        """
        payload = dict(fields)
        payload["updated_at"] = _now_iso()
        self.run_meta.update_one(
            {"beer_run_id": _as_id(beer_run_id)},
            {"$set": payload, "$setOnInsert": {"created_at": _now_iso()}},
            upsert=True,
        )
        return self.run_metadata(beer_run_id)

    def upsert_run_metadata_partial(self, beer_run_id: str, **fields: Any) -> dict:
        """Update only the given fields, leaving every other field alone.

        Fields passed as ``None`` are ignored rather than cleared.
        """
        payload = {k: v for k, v in fields.items() if v is not None}
        if not payload:
            return self.run_metadata(beer_run_id)
        payload["updated_at"] = _now_iso()
        self.run_meta.update_one(
            {"beer_run_id": _as_id(beer_run_id)},
            {"$set": payload},
            upsert=True,
        )
        return self.run_metadata(beer_run_id)

    def next_free_tap_number(self) -> int:
        """The lowest tap number not already in use.

        Taps are numbered from 1.  A run with no tap number recorded does not
        occupy one.
        """
        used = {
            meta.get("tap_number")
            for meta in self.all_run_metadata().values()
            if meta.get("tap_number") is not None
        }
        candidate = 1
        while candidate in used:
            candidate += 1
        return candidate

    def delete_run_metadata(self, beer_run_id: str) -> bool:
        """Remove the metadata document for a run (readings are untouched)."""
        return self.run_meta.delete_one({"beer_run_id": _as_id(beer_run_id)}).deleted_count > 0

    # -- helpers ------------------------------------------------------------ #

    def next_run_id(self) -> str:
        """The next free ``beer_run_id``.

        Run ids are plain integers in the existing data, so the next id is one
        past the highest numeric id currently in use.
        """
        highest = 0
        for run in self.runs():
            try:
                highest = max(highest, int(run.beer_run_id))
            except (TypeError, ValueError):
                continue
        return str(highest + 1)


# --------------------------------------------------------------------------- #
# Connection helpers
# --------------------------------------------------------------------------- #

def _build_client() -> MongoClient:
    user = os.getenv("MONGO_USER", "root")
    password = os.getenv("MONGO_PASS", "")
    host = os.getenv("MONGO_HOST", "127.0.0.1")
    uri = os.getenv("MONGO_URI") or f"mongodb://{user}:{password}@{host}/beer_db?authSource=admin"
    return MongoClient(uri, serverSelectionTimeoutMS=5000)


def _as_id(value: object) -> str:
    """Coerce a run id to a plain string.

    Every value that reaches a Mongo query is coerced first.  Without this, a
    caller passing a dict (e.g. ``{"$ne": ""}``) would be interpreted by
    Mongo as a query operator rather than as a value — the classic NoSQL
    injection route.
    """
    if isinstance(value, str):
        return value
    if isinstance(value, (int, float)):
        return str(value)
    return ""


def _as_date(value: object) -> str:
    """Coerce a reading date to a plain ``YYYY-MM-DD`` string."""
    if isinstance(value, str):
        return value
    return ""


def _is_hashed(stored: str) -> bool:
    """True when *stored* looks like a werkzeug password hash.

    Werkzeug hashes are of the form ``method$salt$hash``, e.g.
    ``scrypt:32768:8:1$abcdef...$0123...``.  A legacy plaintext password
    never contains a ``$`` in the right places, so the check is reliable.
    """
    return "$" in stored and len(stored.split("$")) >= 3


def hash_password(password: str) -> str:
    """Hash a password for storage (scrypt via werkzeug)."""
    return generate_password_hash(password)


def _now_iso() -> str:
    from datetime import datetime, timezone

    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def date_min():
    from datetime import date

    return date.min


def get_repository() -> BeerRepository:
    """Build a repository against the configured Mongo."""
    return BeerRepository()


def is_alive(repo: BeerRepository) -> bool:
    """Cheap health check used by the status page."""
    try:
        repo.client.admin.command("ping")
        return True
    except PyMongoError:
        return False