"""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