diff --git a/screen_locker/_poke_credit.py b/screen_locker/_poke_credit.py index 7c4ea43..9e118c8 100644 --- a/screen_locker/_poke_credit.py +++ b/screen_locker/_poke_credit.py @@ -30,6 +30,7 @@ from screen_locker._day import today_str from screen_locker._earned import hhmm from screen_locker._extra_benefits import process_week_transition from screen_locker._grace_floor import tonight_minutes +from screen_locker._manual_sync import ingest_manual_records from screen_locker._session_sync import ingest_session_records, validate_session from screen_locker._shutdown import read_shutdown_config from screen_locker._shutdown_base import reset_to_base_if_new_day @@ -45,6 +46,7 @@ if TYPE_CHECKING: __all__ = [ "BatchOutcome", "PokeOutcome", + "credit_manual_workouts", "credit_session", "credit_sessions", "current_gaming_minutes", @@ -201,11 +203,40 @@ def credit_sessions( results.append(locker.credit_written(entry, prior_entries)) ingested = ingest_session_records(locker.log_file, records, on_ingested=on_ingested) + return _outcome(ingested, results, "session") + + +def credit_manual_workouts( + records: Iterable[tuple[str, Mapping[str, object]]], + *, + today: str | None = None, +) -> BatchOutcome: + """``credit_sessions`` for manual workouts, via ``ingest_manual_records``.""" + locker = fresh_locker() + locker.start_the_day() + results: list[WorkoutCreditResult] = [] + + def on_ingested( + entry: dict[str, str], prior_entries: list[dict[str, object]] + ) -> None: + results.append(locker.credit_written(entry, prior_entries)) + + ingested = ingest_manual_records( + locker.log_file, records, today=today, on_ingested=on_ingested + ) + return _outcome(ingested, results, "manual workout") + + +def _outcome( + ingested: list[str], results: list[WorkoutCreditResult], what: str +) -> BatchOutcome: + """Count what earned time; warn about logged copies that earned none.""" credited = sum(not result.already_counted_today for result in results) if len(results) > credited: _logger.warning( - "%d streamed session(s) were logged but earned no time: today's slot " + "%d streamed %s(s) were logged but earned no time: today's slot " "was already paid by another copy of the same workout", len(results) - credited, + what, ) return BatchOutcome(ingested=tuple(ingested), credited=credited) diff --git a/screen_locker/_stream_cache.py b/screen_locker/_stream_cache.py index 8490ea6..7459aec 100644 --- a/screen_locker/_stream_cache.py +++ b/screen_locker/_stream_cache.py @@ -8,7 +8,8 @@ tombstones ``_manual_push`` re-publishes every tick, and dropping it would let a retracted session be credited again -- but an event that touched only this PC's log never triggers a pass. -Crediting is :func:`screen_locker._poke_credit.credit_sessions` under the +Crediting is :func:`screen_locker._poke_credit.credit_sessions` (and +``credit_manual_workouts`` for phone-logged manual workouts) under the same in-process lock as the LAN poke handler. ``write_signed_entry`` already dedups by workout id; the ``seen`` set here only spares the file I/O for records this daemon has already handed to the ingest. @@ -23,11 +24,19 @@ from typing import TYPE_CHECKING, Final from screen_locker import _poke_credit from screen_locker._rtdb_stream import LOG_LEAF, device_updates -from screen_locker._sync_records import _session_records, merge_device_texts +from screen_locker._sync_records import ( + _manual_records, + _session_records, + merge_device_texts, +) if TYPE_CHECKING: + from collections.abc import Callable import threading + from crdt_sync import Hlc + + from screen_locker._poke_credit import BatchOutcome from screen_locker._rtdb_stream import SseEvent __all__ = ["DeviceLogCache", "EventResult"] @@ -78,15 +87,17 @@ class DeviceLogCache: self._texts[dev] = text return changed - def _pending(self) -> list[tuple[str, dict[str, object], str]]: - """Union records this daemon has not yet handed to the ingest.""" + def _pending( + self, extract: Callable[[str], dict[str, tuple[dict, Hlc]]], what: str + ) -> list[tuple[str, dict[str, object], str]]: + """Union ``extract`` records this daemon has not yet handed to the ingest.""" union = merge_device_texts( ( (f"{self._prefix}/{dev}/{LOG_LEAF}", text) for dev, text in sorted(self._texts.items()) ), - _session_records, - "sessions", + extract, + what, ) return [ (rid, payload, str(hlc)) @@ -115,39 +126,62 @@ class DeviceLogCache: snapshot=False, summary=f"only this PC's own log changed ({devices}); no credit", ) - pending = self._pending() - if not pending: + sessions = self._pending(_session_records, "sessions") + manuals = self._pending(_manual_records, "manual workouts") + if not sessions and not manuals: return EventResult( snapshot=full, summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; " - f"every session already handed to the ingest -- no-op, no file I/O " - f"({_ms(started)} ms)", + f"every session and manual workout already handed to the ingest " + f"-- no-op, no file I/O ({_ms(started)} ms)", ) try: with self._gate: - outcome = _poke_credit.credit_sessions( - [(rid, payload) for rid, payload, _hlc in pending] - ) + done = [ + _credit(pending, credit, what) + for pending, credit, what in ( + (sessions, _poke_credit.credit_sessions, "session"), + ( + manuals, + _poke_credit.credit_manual_workouts, + "manual workout", + ), + ) + if pending + ] except Exception: # pylint: disable=broad-exception-caught # one bad event must not drop the stream # Not a stream problem: keep the connection (a reconnect would # re-download the snapshot and fail the same way). Nothing is # marked seen, so the next event retries. _logger.exception( - "Crediting %d streamed session(s) failed; the next stream event " - "retries, and the 15-min workout-sync still covers them", - len(pending), + "Crediting %d streamed session(s) and %d manual workout(s) failed; " + "the next stream event retries, and the 15-min workout-sync still " + "covers them", + len(sessions), + len(manuals), ) return EventResult(snapshot=full, summary="crediting failed (see above)") - self._seen.update((rid, hlc) for rid, _payload, hlc in pending) + self._seen.update((rid, hlc) for rid, _payload, hlc in [*sessions, *manuals]) return EventResult( snapshot=full, summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; " - f"{len(pending)} session(s) checked, {len(outcome.ingested)} newly " - f"logged {list(outcome.ingested)}, {outcome.credited} credited " - f"({_ms(started)} ms)", + f"{'; '.join(done)} ({_ms(started)} ms)", ) +def _credit( + pending: list[tuple[str, dict[str, object], str]], + credit: Callable[[list[tuple[str, dict[str, object]]]], BatchOutcome], + what: str, +) -> str: + """Run one batch credit step; describe what it did.""" + outcome = credit([(rid, payload) for rid, payload, _hlc in pending]) + return ( + f"{len(pending)} {what}(s) checked, {len(outcome.ingested)} newly " + f"logged {list(outcome.ingested)}, {outcome.credited} credited" + ) + + def _ms(started: float) -> int: """Milliseconds since ``started`` (a ``perf_counter`` reading).""" return round((time.perf_counter() - started) * 1000) diff --git a/screen_locker/tests/test_stream_cache.py b/screen_locker/tests/test_stream_cache.py index 5a3b60d..6e5dbcd 100644 --- a/screen_locker/tests/test_stream_cache.py +++ b/screen_locker/tests/test_stream_cache.py @@ -154,7 +154,9 @@ class TestFailures: failures = [r for r in caplog.records if r.levelno >= logging.ERROR] assert failures assert failures[0].exc_info is not None - assert "Crediting 1 streamed session(s) failed" in failures[0].getMessage() + assert ( + "Crediting 1 streamed session(s) and 0 manual" in failures[0].getMessage() + ) credit.error = None cache.handle(snapshot({"phone": log_blob("s1")})) assert credit.batches == [["s1"]] diff --git a/screen_locker/tests/test_stream_cache_manual.py b/screen_locker/tests/test_stream_cache_manual.py new file mode 100644 index 0000000..36716cb --- /dev/null +++ b/screen_locker/tests/test_stream_cache_manual.py @@ -0,0 +1,148 @@ +"""The Firebase stream credits phone-logged manual workouts on arrival. + +Until 2026-11-02 the stream only handed StrongLifts sessions to the ingest, so +a manual workout logged on the phone sat in Firebase until the next 15-minute +``workout-sync`` tick -- long enough that it looked lost and was re-logged by +hand on the PC. +""" + +from __future__ import annotations + +import json +import threading +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pytest + +from screen_locker import _poke_credit, _stream_cache +from screen_locker._poke_credit import BatchOutcome, PokeLocker +from screen_locker._stream_cache import DeviceLogCache +from screen_locker._workout_credit import WorkoutCreditResult +from screen_locker.tests.test_manual_sync import _OTHER_DRAFT, _manual_record +from screen_locker.tests.test_stream_cache import ( + OWN, + Credit, + log_blob, + patch_event, + snapshot, +) + +if TYPE_CHECKING: + from pathlib import Path + +_DATE = "2026-07-13" +_RID, _PAYLOAD = _manual_record(_OTHER_DRAFT, _DATE) + + +@pytest.fixture(name="steps") +def steps_fixture(monkeypatch: pytest.MonkeyPatch) -> tuple[Credit, Credit]: + """Record both batch credit steps instead of building a locker.""" + sessions, manuals = Credit(), Credit() + monkeypatch.setattr(_stream_cache._poke_credit, "credit_sessions", sessions) + monkeypatch.setattr(_stream_cache._poke_credit, "credit_manual_workouts", manuals) + return sessions, manuals + + +@pytest.fixture(name="cache") +def cache_fixture(steps: tuple[Credit, Credit]) -> DeviceLogCache: + """A cache whose own device dir is ``pc-own``, sharing one gate.""" + gate = threading.Lock() + for credit in steps: + credit.gate = gate + return DeviceLogCache("sync/devices", frozenset({OWN}), gate) + + +class TestStreamedManual: + """Which streamed manual records reach the manual ingest.""" + + def test_a_phone_manual_workout_is_credited_on_arrival( + self, cache: DeviceLogCache, steps: tuple[Credit, Credit] + ) -> None: + """A patch carrying a new manual record is handed over under the gate.""" + sessions, manuals = steps + cache.handle(snapshot({"phone": log_blob("s1")})) + result = cache.handle( + patch_event({"phone": log_blob(_RID, wall=2, payload=_PAYLOAD)}) + ) + assert manuals.batches == [[_RID]] + assert manuals.gate_held == [True] + assert sessions.batches == [["s1"]] + assert "1 manual workout(s) checked, 1 newly logged" in result.summary + + def test_already_handed_manual_records_are_a_no_op( + self, cache: DeviceLogCache, steps: tuple[Credit, Credit] + ) -> None: + """Re-streaming the same manual record costs no second ingest.""" + _sessions, manuals = steps + blob = log_blob(_RID, payload=_PAYLOAD) + cache.handle(snapshot({"phone": blob})) + result = cache.handle(snapshot({"phone": blob})) + assert manuals.batches == [[_RID]] + assert "no-op, no file I/O" in result.summary + + def test_tombstoned_manual_records_are_not_handed_over( + self, cache: DeviceLogCache, steps: tuple[Credit, Credit] + ) -> None: + """A deleted record must not be credited by the stream either.""" + _sessions, manuals = steps + cache.handle( + snapshot({"phone": log_blob(_RID, deleted=True, payload=_PAYLOAD)}) + ) + assert manuals.batches == [] + + def test_a_session_only_event_does_not_call_the_manual_ingest( + self, cache: DeviceLogCache, steps: tuple[Credit, Credit] + ) -> None: + """Each kind goes only to its own ingest.""" + sessions, manuals = steps + cache.handle(snapshot({"phone": log_blob("s1")})) + assert sessions.batches == [["s1"]] + assert manuals.batches == [] + + +@pytest.fixture(name="log_file") +def log_file_fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """A fresh PokeLocker on a tmp log; day reset, local ledger and credit stubbed.""" + log = tmp_path / "poke_log.json" + + def headless(locker_cls: type[PokeLocker]) -> PokeLocker: + """Build a PokeLocker without a display, pointed at the temporary log.""" + locker = object.__new__(locker_cls) + locker.log_file = log + locker.workout_data = {} + return locker + + monkeypatch.setattr(_poke_credit, "_headless_locker", headless) + monkeypatch.setattr(PokeLocker, "start_the_day", lambda _self: None) + monkeypatch.setattr( + PokeLocker, + "_apply_credit_for_written_entry", + MagicMock( + return_value=WorkoutCreditResult( + shutdown_adjusted=True, + extra_bonus_delta=0, + weekly_count=1, + already_counted_today=False, + ) + ), + ) + return log + + +class TestCreditManualWorkouts: + """The real manual ingest, run the way the stream runs it.""" + + def test_logs_and_credits_a_streamed_manual_workout(self, log_file: Path) -> None: + """The record lands in log.json under its own id and earns its credit.""" + outcome = _poke_credit.credit_manual_workouts([(_RID, _PAYLOAD)], today=_DATE) + assert outcome == BatchOutcome(ingested=(_RID,), credited=1) + logged = json.loads(log_file.read_text())[_DATE] + assert [e["workout_id"] for e in logged] == [_RID] + + def test_a_second_pass_is_a_duplicate(self, log_file: Path) -> None: + """Re-crediting the same record logs nothing new.""" + _poke_credit.credit_manual_workouts([(_RID, _PAYLOAD)], today=_DATE) + outcome = _poke_credit.credit_manual_workouts([(_RID, _PAYLOAD)], today=_DATE) + assert outcome == BatchOutcome(ingested=(), credited=0) + assert len(json.loads(log_file.read_text())[_DATE]) == 1