screen-locker/screen_locker/_stream_cache.py

"""What the Firebase stream does with one ``put``/``patch``: cache, merge, credit.

Holds every device's last-seen log text, so each event recomputes the
cross-device union with the sync's OWN decoder and tombstone rule
(:func:`screen_locker._sync_records.merge_device_texts`) without a single
extra Firebase read. The PC's own log stays in the union -- it carries the
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` (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.
"""

from __future__ import annotations

from dataclasses import dataclass
import logging
import time
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 (
    _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"]

_logger: Final = logging.getLogger(__name__)


@dataclass(frozen=True)
class EventResult:
    """What one event did, as a sentence for the journal."""

    snapshot: bool
    summary: str


class DeviceLogCache:
    """Every device's last-streamed log text, plus records already ingested."""

    def __init__(
        self, prefix: str, own_ids: frozenset[str], gate: threading.Lock
    ) -> None:
        """Cache logs under ``prefix``; ``own_ids`` are this PC's device dirs."""
        self._prefix = prefix
        self._own_ids = own_ids
        self._gate = gate
        self._texts: dict[str, str] = {}
        self._seen: set[tuple[str, str]] = set()

    def _apply(self, updates: dict[str, str | None], *, full: bool) -> set[str]:
        """Fold ``updates`` into the cache; return the devices that changed."""
        if full:
            fresh = {dev: text for dev, text in updates.items() if text is not None}
            changed = {
                dev
                for dev in fresh.keys() | self._texts.keys()
                if fresh.get(dev) != self._texts.get(dev)
            }
            self._texts = fresh
            return changed
        changed = set()
        for dev, text in updates.items():
            if text == self._texts.get(dev):
                continue  # byte-identical rewrite: nothing to read or credit
            changed.add(dev)
            if text is None:
                self._texts.pop(dev, None)
            else:
                self._texts[dev] = text
        return changed

    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())
            ),
            extract,
            what,
        )
        return [
            (rid, payload, str(hlc))
            for rid, (payload, hlc) in sorted(union.items())
            if (rid, str(hlc)) not in self._seen
        ]

    def handle(self, event: SseEvent) -> EventResult:
        """Apply one ``put``/``patch`` and credit whatever it newly brought.

        Raises:
            ValueError: The event's data is not JSON.
            TypeError: The JSON is not the documented ``{path, data}`` shape.
        """
        started = time.perf_counter()
        full, updates = device_updates(event)
        changed = self._apply(updates, full=full)
        devices = ", ".join(sorted(changed)) or "none"
        if not full and not changed:
            return EventResult(
                snapshot=False, summary="byte-identical rewrite; nothing to do"
            )
        foreign = changed - self._own_ids
        if not full and not foreign:
            return EventResult(
                snapshot=False,
                summary=f"only this PC's own log changed ({devices}); no credit",
            )
        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 and manual workout already handed to the ingest "
                f"-- no-op, no file I/O ({_ms(started)} ms)",
            )
        try:
            with self._gate:
                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) 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 [*sessions, *manuals])
        return EventResult(
            snapshot=full,
            summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; "
            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)