diora-web/tui/diora_tui/progress.py

"""Reading position per book: the place last read wins.

This is the contract in diora-web docs/progress-sync.md, the same rules the server
(books/views.py save_progress) and the other clients apply (diora-android Ops.kt,
diora-desktop src/main/ops.js):

* A position carries `read_at`, the moment the reader arrived there. Local times are the
  server's clock: `offset` (server - this machine, ms, learnt from the server's clock on every
  answer) turns this machine's `read_at` into it. Saving an unchanged position again carries the
  old time, so an idle reader can never beat a newer position from elsewhere.
* A write is stored when its time is >= the stored one (`force`: always, time = now). The
  furthest place ever reported only grows.
* A write is also queued (`pending`, one per book: the later `read_at` wins, `force` sticks)
  until the server has it. After a pull the pending write is applied again on top of the
  server's state with the same rule.

State lives in progress.json next to the book list; the file also holds the clock offset.
"""

from __future__ import annotations

import json
import functools
import os
import threading
import time
from dataclasses import asdict, dataclass, replace
from pathlib import Path

from . import paths
from .blocks import parse_anchor

DEVICE_NAME = "diora TUI"

_DATA_DIR = paths.data_dir()
_PROGRESS_FILE = _DATA_DIR / "progress.json"
_VERSION = 2


def now_ms() -> int:
    return int(time.time() * 1000)


def plausible_ms(v) -> int | None:
    """A time (ms since epoch) from the server or a body, or None (as Ops.ms)."""
    if isinstance(v, bool) or not isinstance(v, (int, float)) or v != v or v in (float("inf"), float("-inf")):
        return None
    t = int(v)
    return t if 10**12 <= t < 10**14 else None


def cmp_progress(a_anchor: str, a_frac: float, b_anchor: str, b_frac: float) -> float:
    """> 0 if A is further into the book than B, < 0 if behind (books/views.py _cmp_progress)."""
    a, b = parse_anchor(a_anchor or ""), parse_anchor(b_anchor or "")
    if a is not None and b is not None:
        return (a[0] - b[0]) if a[0] != b[0] else a[1] - b[1]
    return a_frac - b_frac


@dataclass
class BookState:
    anchor: str = ""
    fraction: float = 0.0
    read_at: int | None = None  # server clock, ms
    device: str = ""
    furthest_anchor: str = ""
    furthest_fraction: float = 0.0
    # The write the server hasn't got yet: {"anchor", "fraction", "read_at" (this machine's
    # clock), "force"}.
    pending: dict | None = None
    # The server's read_at for the last write of ours that it stored: tells our own position
    # coming back from a pull from another device's.
    echo_read_at: int | None = None


def _locked(fn):
    @functools.wraps(fn)
    def wrapper(self, *a, **kw):
        with self._lock:
            return fn(self, *a, **kw)
    return wrapper


class ProgressStore:
    """Shared by the UI thread (the reader writes) and the sync workers (they pull and push)."""

    def __init__(self, path: Path | None = None) -> None:
        self.path = path or _PROGRESS_FILE
        self._lock = threading.RLock()
        self.offset = 0
        self.books: dict[str, BookState] = {}
        self._load()

    # ---- persistence

    def _load(self) -> None:
        try:
            raw = json.loads(self.path.read_text())
        except (OSError, ValueError):
            return
        if not isinstance(raw, dict):
            return
        if raw.get("version") == _VERSION:
            self.offset = int(raw.get("offset") or 0)
            for key, entry in (raw.get("books") or {}).items():
                try:
                    self.books[key] = BookState(**entry)
                except TypeError:
                    continue
            return
        # version 1: {book_id: {anchor, updated_at (s)}}, furthest-wins, nothing queued
        for key, entry in raw.items():
            if isinstance(entry, dict) and entry.get("anchor"):
                self.books[key] = BookState(
                    anchor=entry["anchor"], read_at=int(entry.get("updated_at", 0) * 1000) or None,
                    furthest_anchor=entry["anchor"],
                )

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        data = {"version": _VERSION, "offset": self.offset,
                "books": {k: asdict(v) for k, v in self.books.items()}}
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(data, indent=1))
        os.replace(tmp, self.path)

    # ---- clock

    @_locked
    def set_server_now(self, server_now: int | None, device_now: int | None = None) -> None:
        t = plausible_ms(server_now)
        if t is not None:
            self.offset = t - (device_now if device_now is not None else now_ms())

    # ---- reads

    @_locked
    def get(self, key: str) -> BookState | None:
        """A copy: the sync workers change the live state."""
        st = self.books.get(key)
        return replace(st, pending=dict(st.pending) if st.pending else None) if st else None

    @_locked
    def pending_keys(self) -> list[str]:
        return [k for k, v in self.books.items() if v.pending]

    # ---- writes

    @_locked
    def write(self, key: str, anchor: str, fraction: float, read_at: int, *, force: bool = False,
              queue: bool = True, device: str = DEVICE_NAME, now: int | None = None) -> bool:
        """Apply a position the reader reached (`read_at`: this machine's clock) locally and
        queue it for the server (`queue` False: the book isn't one of the server's). Returns
        whether it is the continue point now."""
        st = self.books.setdefault(key, BookState())
        fraction = max(0.0, min(1.0, fraction))
        took = self._apply(st, anchor, fraction, read_at, force, device, now)
        if queue:
            new = {"anchor": anchor, "fraction": fraction, "read_at": read_at, "force": force}
            st.pending = _merge_pending(st.pending, new)
        self.save()
        return took

    def _apply(self, st: BookState, anchor: str, fraction: float, read_at: int | None, force: bool,
               device: str, now: int | None) -> bool:
        now_server = (now if now is not None else now_ms()) + self.offset
        if force:
            when, take = now_server, True
        else:
            when = min((read_at or 0) + self.offset, now_server)
            take = st.read_at is None or when >= st.read_at
        if take:
            st.anchor, st.fraction, st.read_at, st.device = anchor, fraction, when, device[:40]
        if cmp_progress(anchor, fraction, st.furthest_anchor, st.furthest_fraction) > 0:
            st.furthest_anchor, st.furthest_fraction = anchor, fraction
        return take

    @_locked
    def apply_server(self, key: str, entry: dict, *, keep_pending: bool = True) -> None:
        """Take the server's state for the book (an entry of /books/ or a progress answer), then
        apply the pending write on top by the same rule."""
        st = self.books.setdefault(key, BookState())
        st.anchor = entry.get("position_anchor") or ""
        st.fraction = float(entry.get("scroll_fraction") or 0.0)
        st.read_at = plausible_ms(entry.get("read_at"))
        st.device = entry.get("device") or ""
        st.furthest_anchor = entry.get("furthest_anchor") or ""
        st.furthest_fraction = float(entry.get("furthest_fraction") or 0.0)
        if st.pending and keep_pending:
            p = st.pending
            self._apply(st, p["anchor"], p["fraction"], p["read_at"], bool(p.get("force")), DEVICE_NAME, None)
        self.save()

    @_locked
    def sent(self, key: str, body: dict, answer: dict | None) -> None:
        """The server took `body` (the pending write as it was sent): forget it unless a newer one
        queued meanwhile, and take the server's state from the answer."""
        st = self.books.get(key)
        if st is None:
            return
        if st.pending and st.pending.get("read_at") == body.get("read_at") and bool(st.pending.get("force")) == bool(body.get("force")):
            st.pending = None
        if answer:
            self.set_server_now(answer.get("server_now"))
            if answer.get("stored") is True:
                st.echo_read_at = plausible_ms(answer.get("read_at"))
            self.apply_server(key, answer)
        else:
            self.save()


def _merge_pending(old: dict | None, new: dict) -> dict:
    """Two queued writes for the book: the place read later is the one to send; a forced place
    stays forced (Ops.merge)."""
    if old is None:
        return new
    keep = new if new["read_at"] >= old["read_at"] else old
    keep = dict(keep)
    if old.get("force") or new.get("force"):
        keep["force"] = True
    return keep


_default: ProgressStore | None = None


def default() -> ProgressStore:
    global _default
    if _default is None:
        _default = ProgressStore()
    return _default