hermes
4bedd23c24
The guard matches on the absence of a token. `{"$exists": False}` covers
a reading with no token field, but NOT one whose token field is present
and empty.
Verified against the live database, before the fix:
token present but empty -> write matched: 0 <-- missed it
`{"$in": [None, ""]}` covers all three - a missing field, an empty
field, and a null one - because `$in` with null also matches a missing
field.
no token field -> write matched: 1
token present but empty -> write matched: 1
token is null -> write matched: 1
183 tests (up from 182).
557 lines
21 KiB
Python
557 lines
21 KiB
Python
"""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
|
|
import threading
|
|
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"]
|
|
self._ensure_indexes()
|
|
|
|
def close(self) -> None:
|
|
"""Close the underlying Mongo client and release it.
|
|
|
|
A ``MongoClient`` keeps a connection pool and spawns three
|
|
monitoring threads. ``close()`` alone does NOT reap them - they
|
|
go when the client object is garbage collected - so the
|
|
reference is dropped as well.
|
|
|
|
The request-scoped repository is closed by the app's teardown, but
|
|
a caller that builds its own - the advice background thread, for
|
|
instance - has to close it itself or the threads and the pool
|
|
accumulate, three per reading.
|
|
"""
|
|
client = self.client
|
|
try:
|
|
client.close()
|
|
except Exception:
|
|
pass
|
|
# Drop the reference so the monitor threads can be reaped.
|
|
self.client = None
|
|
|
|
def __enter__(self) -> "BeerRepository":
|
|
return self
|
|
|
|
def __exit__(self, *exc_info) -> None:
|
|
self.close()
|
|
|
|
def _ensure_indexes(self) -> None:
|
|
"""Make sure the lookup keys are indexed.
|
|
|
|
Every read of the run metadata filters on ``beer_run_id``, and the
|
|
readings are always fetched by ``(beer_run_id, date)``. Without the
|
|
indexes those are collection scans.
|
|
"""
|
|
try:
|
|
self.run_meta.create_index("beer_run_id", name="beer_run_id_1")
|
|
self.readings.create_index(
|
|
[("beer_run_id", 1), ("date", 1)], name="run_date_1"
|
|
)
|
|
except PyMongoError:
|
|
# An index failure must not stop the app from serving, but it
|
|
# should be visible: without the index the lookup is a collection
|
|
# scan, which gets slow as the data grows.
|
|
logger.warning(
|
|
"Could not create the Mongo indexes; the lookups will be "
|
|
"collection scans until they exist",
|
|
exc_info=True,
|
|
)
|
|
|
|
# -- 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),
|
|
},
|
|
# The advice was written for the reading that was there when
|
|
# it was generated. If the reading has been edited, that
|
|
# advice no longer describes it, so it is cleared and a fresh
|
|
# one is generated. (Advice is only ever cleared here - never
|
|
# the brewer's comment.)
|
|
"$unset": {"advice": "", "advice_ok": "", "advice_at": ""},
|
|
}
|
|
# A fresh token marks this generation of the reading. The advice
|
|
# thread carries the token it was given, so a thread that is
|
|
# still working on an EARLIER version of the reading finds a
|
|
# token that no longer matches and its write is refused. The
|
|
# gravity alone is not enough - a comment-only correction leaves
|
|
# the gravity unchanged.
|
|
update["$set"]["advice_token"] = _new_token()
|
|
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_reading_advice(self, reading_id: str, advice: str,
|
|
ok: bool = True,
|
|
expected_sg: int | None = None,
|
|
expected_token: str | None = None) -> int:
|
|
"""Attach the LLM's advice to a reading.
|
|
|
|
The advice lives in its own field, so the brewer's own comment is
|
|
never overwritten. ``ok`` says whether the advice was actually
|
|
generated or is a note explaining that it could not be.
|
|
|
|
The write is CONDITIONAL, so a slow advice thread cannot
|
|
overwrite a correction:
|
|
|
|
- ``expected_token`` is the editing token of the reading at the
|
|
moment the advice was generated. Every edit writes a fresh
|
|
token, so a late thread's token no longer matches.
|
|
- ``expected_sg`` is the gravity the advice was generated for,
|
|
as a second check.
|
|
|
|
This matters because the advice is generated in the background: if
|
|
the brewer corrects a reading while the advice for the old value
|
|
is still being generated, the slow thread would land afterwards
|
|
and overwrite the fresh advice with advice about a reading that
|
|
no longer exists.
|
|
|
|
The token is needed as well as the gravity, because a
|
|
comment-only correction leaves the gravity unchanged - the token
|
|
is what tells the two generations apart.
|
|
"""
|
|
doc_id = _as_doc_id(reading_id)
|
|
if not doc_id:
|
|
return 0
|
|
|
|
query: dict = {"_id": {"$in": doc_id}}
|
|
if expected_token:
|
|
# The editing token is the PRIMARY guard: it changes on every
|
|
# edit, so it catches a stale generation whatever else
|
|
# changed - the gravity, the comment, or nothing.
|
|
query["advice_token"] = str(expected_token)
|
|
elif expected_token == "":
|
|
# The reading has NEVER been edited, so it has no token at
|
|
# all. Match on the ABSENCE of one, which still tells the
|
|
# generations apart: an edit writes a token, so a stale
|
|
# thread's "" stops matching the moment the reading is
|
|
# corrected.
|
|
#
|
|
# Falling back to the gravity here would re-open the hole
|
|
# this guard exists to close - a comment-only correction
|
|
# leaves the gravity unchanged.
|
|
# `$in` with null also matches a MISSING field, so this
|
|
# covers a reading with no token AND one whose token field
|
|
# is present but empty.
|
|
query["advice_token"] = {"$in": [None, ""]}
|
|
elif expected_sg is not None:
|
|
# A direct call with a gravity to check against.
|
|
query["sg"] = str(expected_sg)
|
|
|
|
# `$in` matches either form. A plain list would be an exact
|
|
# ARRAY match in Mongo, which is not what is wanted here.
|
|
result = self.readings.update_one(
|
|
query,
|
|
{"$set": {
|
|
"advice": str(advice),
|
|
"advice_ok": bool(ok),
|
|
"advice_at": _now_iso(),
|
|
}},
|
|
)
|
|
return result.modified_count
|
|
|
|
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 ""
|
|
|
|
|
|
#: A process-wide repository, for the code that runs outside a request.
|
|
#:
|
|
#: A ``MongoClient`` spawns three monitor threads and pymongo does not
|
|
#: reap them on ``close()`` - they go only when the client is garbage
|
|
#: collected. So building a client per advice write accumulated three
|
|
#: threads and a connection pool for every reading recorded. One shared
|
|
#: client for the life of the process is what the request-scoped
|
|
#: repository does too.
|
|
_shared_repository: "BeerRepository | None" = None
|
|
|
|
#: Guards the check-then-set above.
|
|
#:
|
|
#: Without it, concurrent advice threads each saw ``None`` and each
|
|
#: built their own repository - so the fix reintroduced the very leak it
|
|
#: was meant to remove, just bounded by a race instead of by the number
|
|
#: of readings. Eight threads on a cold process built eight clients and
|
|
#: twenty-five threads.
|
|
_shared_repository_lock = threading.Lock()
|
|
|
|
|
|
def get_shared_repository() -> "BeerRepository":
|
|
"""The process-wide repository, built on first use.
|
|
|
|
Thread-safe: the first caller builds it, the rest wait and get the
|
|
same one.
|
|
"""
|
|
global _shared_repository
|
|
|
|
if _shared_repository is None:
|
|
with _shared_repository_lock:
|
|
# Checked again inside the lock - another thread may have
|
|
# built it while this one was waiting.
|
|
if _shared_repository is None:
|
|
_shared_repository = BeerRepository()
|
|
return _shared_repository
|
|
|
|
|
|
def _as_doc_id(value: object):
|
|
"""Coerce a reading document id to something Mongo will match.
|
|
|
|
The ``_id`` of a reading is not a run id, so the ``_as_id`` coercion
|
|
does not apply - it would turn the id into an empty string and the
|
|
update would silently match nothing.
|
|
|
|
Two shapes exist in this database:
|
|
|
|
- the ORIGINAL app's ids are 24 random CHARACTERS stored as a string
|
|
(``l78ZidBJ3QaYn98tvB88x4uM``), and
|
|
- ids written by Mongo itself are ``ObjectId`` objects.
|
|
|
|
So the id is returned in BOTH forms and the query matches either.
|
|
Accepting only the ObjectId form would mean the advice was never
|
|
stored for any of the existing readings.
|
|
|
|
Only an ObjectId or a plain string is accepted, so a dict can never
|
|
be injected as a query operator.
|
|
"""
|
|
from bson import ObjectId
|
|
|
|
if isinstance(value, ObjectId):
|
|
return [value]
|
|
if isinstance(value, str) and value:
|
|
try:
|
|
# A 24-character hex string may or may not be the stored
|
|
# ObjectId or the stored string, so try both.
|
|
return [ObjectId(value), value]
|
|
except Exception:
|
|
# Not hex - it can only be the original app's string id.
|
|
return [value]
|
|
# Anything else - a dict, a list, an int, None - is not a reading
|
|
# id. An EMPTY LIST is returned rather than None: it is a valid
|
|
# `$in` argument that matches nothing, so the helper cannot be
|
|
# misused into a query error or into matching every document.
|
|
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 _new_token() -> str:
|
|
"""A fresh token, to mark one generation of a reading from the next."""
|
|
import secrets
|
|
|
|
return secrets.token_hex(8)
|
|
|
|
|
|
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
|
|
|