Add active reading timer synchronization through PowerSync

alembic/versions/c0d1e2f34567_add_reading_timer_sync.py

@@ -0,0 +1,42 @@
+"""Add account-owned active timer state without changing reading history."""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "c0d1e2f34567"
+down_revision: str | Sequence[str] | None = "b9c0d1e2f345"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "reading_timers",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("payload", postgresql.JSONB(), nullable=False),
+        sa.Column("status", sa.Text(), nullable=False),
+        sa.Column("base_version", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.CheckConstraint("status IN ('active', 'saved', 'discarded')", name="reading_timers_status"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_reading_timers_owner_user_id", "reading_timers", ["owner_user_id"])
+
+    op.create_index(
+        "uq_reading_timers_active_owner",
+        "reading_timers",
+        ["owner_user_id"],
+        unique=True,
+        postgresql_where=sa.text("status = 'active'"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("reading_timers")

docs/reading-timer-sync.md

@@ -0,0 +1,65 @@
+# Reading timer synchronization
+
+Signed-in devices share one active reading timer per account through the
+`reading_timers` PowerSync stream. Guest timers remain local. The selected book,
+stopwatch/countdown mode, recorded intervals, running timestamp and paused/finished
+state are in the payload. Devices compute elapsed time from UTC timestamps locally.
+There is no network write for each tick. Countdown elapsed time is capped at its
+goal even when no device is open.
+
+## Concurrent and offline actions
+
+Every transition carries a unique payload version and the preceding `base_version`.
+The upload handler locks the owner before checking that base version. The first
+accepted transition wins. A stale branch and its descendants are acknowledged
+without applying their changes, then PowerSync reconciles them to server state.
+This follows PowerSync's [server-authoritative conflict resolution](https://docs.powersync.com/usage/lifecycle-maintenance/handling-update-conflicts).
+
+A partial unique index permits one active timer per owner. If two disconnected
+devices start independent timers, the first uploaded start wins. The other session
+is marked discarded. Offline edits remain usable locally until reconnection, but
+cannot replace a more recent accepted branch. Device clocks should be synchronized.
+
+Finishing and saving writes the terminal timer, its reading activities and optional
+book progress in one SQLite transaction and one upload batch. The server validates
+these effects before committing. Conflicting or repeated saves skip the entire
+batch, including progress updates. Terminal rows remain durable so later offline
+writes cannot revive a saved or discarded session. Stable activity IDs use the
+session UUID and interval index.
+
+Save batches must use `PUT` for every reading or completion activity. An activity
+`PATCH` is rejected because a missing target would otherwise be silently skipped,
+allowing the timer to become terminal without its history. Rejection leaves the
+timer and book progress unchanged so a corrected batch can be retried.
+
+The old local `reading_timer` draft is moved to the new table once when its profile
+opens. Guest and authenticated databases remain separate. Account/server switches
+invalidate the old repository and subscription.
+
+## Rollout
+
+Deploy the server and replication changes before shipping the new client:
+
+1. Apply Alembic revision `c0d1e2f34567` (`uv run alembic upgrade head`). This adds a
+   table and indexes, leaving books and reading history untouched.
+2. Add `public.reading_timers` to the PowerSync publication and grant the source
+   role SELECT access. The local setup script includes both changes.
+3. Apply the updated `powersync/sync-config.yaml` and run the updated API.
+4. Release the client that writes the new table.
+
+Older servers do not accept timer mutations. A new client must use the updated
+server, otherwise its queued timer transactions remain pending. Timer save batches
+are never split by tracking-capability discovery, because separating the ledger
+from the terminal state would defeat exactly-once save handling.
+
+## Verification
+
+- `tests/services/test_reading_timer_sync.py` covers handoff, competing transitions,
+  simultaneous saves, retries, ownership, atomic rollback and deleted books.
+- `tests/test_reading_timer_migration.py` verifies additive migration and metadata.
+- Client persistence tests cover queued transaction boundaries, migration, guest
+  isolation, recovery and retained terminal states.
+- `integration_test/powersync_timer_integration_test.dart` uses three isolated local
+  databases and two disposable accounts against live API and PowerSync services.
+  Run with `RUN_POWERSYNC_INTEGRATION=true` on macOS. It checks countdown handoff,
+  pause/resume, offline conflicts, competing saves, reset and account isolation.

papyrus/models/__init__.py

@@ -20,11 +20,13 @@
 )
 from papyrus.models.media import AccountMediaCleanup, MediaAsset
 from papyrus.models.powersync_demo import PowerSyncDemoItem
+from papyrus.models.reading_timer import SyncReadingTimer
 from papyrus.models.sync import SyncBook
 from papyrus.models.tracking import SyncGoalPeriod, SyncReadingActivity, SyncReadingGoal
 from papyrus.models.user import User
 
 __all__ = [
+    "SyncReadingTimer",
     "SyncReadingGoal",
     "SyncReadingActivity",
     "SyncGoalPeriod",

papyrus/models/reading_timer.py

@@ -0,0 +1,20 @@
+"""Owned timer state, including terminal records that prevent replayed saves."""
+
+from sqlalchemy import CheckConstraint, Index, Text, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from papyrus.core.database import Base
+from papyrus.models.tracking import TrackingRecord
+
+
+class SyncReadingTimer(TrackingRecord, Base):
+    __tablename__ = "reading_timers"
+    __table_args__ = (
+        CheckConstraint("status IN ('active', 'saved', 'discarded')", name="reading_timers_status"),
+        Index(
+            "uq_reading_timers_active_owner", "owner_user_id", unique=True, postgresql_where=text("status = 'active'")
+        ),
+    )
+
+    status: Mapped[str] = mapped_column(Text)
+    base_version: Mapped[str | None] = mapped_column(Text)

papyrus/schemas/reading_timer.py

@@ -0,0 +1,60 @@
+"""Timestamp-based timer payloads shared with the offline client."""
+
+from datetime import UTC, datetime, timedelta
+from typing import Literal
+from uuid import UUID
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator
+
+
+class TimerInterval(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    start: AwareDatetime
+    end: AwareDatetime
+
+
+class TimerPayload(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    id: UUID
+    book_id: UUID
+    book_title: str = Field(min_length=1, max_length=500)
+    started_at: AwareDatetime
+    device_id: str = Field(min_length=1, max_length=255)
+    shelf_ids: list[UUID] = Field(max_length=1000)
+    start_page: int = Field(ge=0)
+    phase: Literal["running", "paused", "finished"]
+    target_seconds: int | None = Field(default=None, ge=60, le=86400)
+    running_since: AwareDatetime | None = None
+    intervals: list[TimerInterval] = Field(max_length=10000)
+    revision: int = Field(ge=0)
+    version: str = Field(min_length=1, max_length=100)
+
+    @model_validator(mode="after")
+    def valid_timeline(self) -> "TimerPayload":
+        latest = self.started_at
+        recorded = timedelta()
+        future = datetime.now(UTC) + timedelta(minutes=5)
+
+        for interval in self.intervals:
+            if interval.start < latest or interval.end <= interval.start:
+                raise ValueError("Timer intervals must be positive and ordered without overlap")
+
+            latest = interval.end
+            recorded += interval.end - interval.start
+
+        if (self.phase == "running") != (self.running_since is not None):
+            raise ValueError("Only running timers require running_since")
+
+        if self.running_since is not None:
+            if self.running_since < latest:
+                raise ValueError("Running time cannot overlap recorded intervals")
+
+            latest = self.running_since
+
+        if latest > future:
+            raise ValueError("Timer cannot begin in the future")
+
+        if self.target_seconds is not None and recorded > timedelta(seconds=self.target_seconds):
+            raise ValueError("Recorded time exceeds the countdown goal")
+
+        return self

papyrus/schemas/sync.py

@@ -55,6 +55,7 @@
 ENTITY_FIELDS = frozenset({"owner_user_id", "created_at", "updated_at"})
 
 UPLOAD_FIELDS = {
+    "reading_timers": ENTITY_FIELDS | {"payload", "status", "base_version"},
     "reading_goals": ENTITY_FIELDS | {"payload"},
     "reading_activities": ENTITY_FIELDS | {"payload"},
     "goal_periods": ENTITY_FIELDS | {"payload"},
@@ -97,6 +98,7 @@ class PowerSyncCrudMutation(BaseModel):
         "bookmarks",
         "book_shelves",
         "book_tags",
+        "reading_timers",
         "reading_goals",
         "reading_activities",
         "goal_periods",