diff --git a/alembic/versions/c0d1e2f34567_add_reading_timer_sync.py b/alembic/versions/c0d1e2f34567_add_reading_timer_sync.py new file mode 100644 index 0000000..c85a678 --- /dev/null +++ b/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") diff --git a/docs/reading-timer-sync.md b/docs/reading-timer-sync.md new file mode 100644 index 0000000..fc6aaab --- /dev/null +++ b/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. diff --git a/papyrus/models/__init__.py b/papyrus/models/__init__.py index 76e3807..f492a86 100644 --- a/papyrus/models/__init__.py +++ b/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", diff --git a/papyrus/models/reading_timer.py b/papyrus/models/reading_timer.py new file mode 100644 index 0000000..451c3a7 --- /dev/null +++ b/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) diff --git a/papyrus/schemas/reading_timer.py b/papyrus/schemas/reading_timer.py new file mode 100644 index 0000000..a157004 --- /dev/null +++ b/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 diff --git a/papyrus/schemas/sync.py b/papyrus/schemas/sync.py index f6d887a..07275b9 100644 --- a/papyrus/schemas/sync.py +++ b/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", diff --git a/papyrus/services/reading_timer_sync.py b/papyrus/services/reading_timer_sync.py new file mode 100644 index 0000000..1596386 --- /dev/null +++ b/papyrus/services/reading_timer_sync.py @@ -0,0 +1,185 @@ +"""Resolve timer branches under the upload owner's lock before applying their effects.""" + +from datetime import UTC, datetime +from uuid import UUID, uuid5 + +from pydantic import ValidationError as PayloadError +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from papyrus.core.exceptions import ForbiddenError, ValidationError +from papyrus.models import SyncReadingTimer +from papyrus.schemas.reading_timer import TimerPayload +from papyrus.schemas.sync import PowerSyncCrudMutation +from papyrus.schemas.tracking import Activity +from papyrus.services.library_sync import tombstoned +from papyrus.services.tracking_validation import check_reference + +BOOK_PROGRESS_FIELDS = { + "owner_user_id", + "updated_at", + "current_page", + "current_position", + "last_read_at", + "started_at", + "reading_status", + "completed_at", +} + + +def validate_effects(batch: list[PowerSyncCrudMutation], timer: TimerPayload, status: str) -> None: + """A terminal timer and its ledger/progress writes form one indivisible transaction.""" + effects = [mutation for mutation in batch if mutation.table != "reading_timers"] + + if status != "saved": + if effects: + raise ValidationError("Only saving a timer may write reading history") + + return + + if timer.phase != "finished" or sum((part.end - part.start).total_seconds() for part in timer.intervals) < 1: + raise ValidationError("Saved timers require at least one second of finished reading") + + expected = {uuid5(timer.id, str(index)): part for index, part in enumerate(timer.intervals)} + seen: set[UUID] = set() + + for mutation in effects: + if mutation.op.upper() == "DELETE": + raise ValidationError("Saving a timer cannot delete library data") + + if mutation.table == "books": + if ( + mutation.id != str(timer.book_id) + or mutation.op.upper() != "PATCH" + or (mutation.op_data or {}).keys() - BOOK_PROGRESS_FIELDS + ): + raise ValidationError("Timer save may only update its book's progress") + + continue + + if mutation.table != "reading_activities": + raise ValidationError("Unexpected timer save effect") + + if mutation.op.upper() != "PUT": + raise ValidationError("Timer activities must use PUT") + + activity = Activity.model_validate((mutation.op_data or {}).get("payload")) + + if activity.id in seen or str(activity.id) != mutation.id or activity.book_id != timer.book_id: + raise ValidationError("Timer activity identity does not match its session") + + seen.add(activity.id) + + if activity.kind == "completion" and activity.id == timer.id: + continue + + interval = expected.get(activity.id) + + if ( + interval is None + or activity.session_id != timer.id + or activity.kind != "reading" + or activity.start_time != interval.start + or activity.end_time != interval.end + ): + raise ValidationError("Saved reading intervals must match the timer") + + if not expected.keys() <= seen: + raise ValidationError("Timer save is missing reading intervals") + + +async def apply_timer_transaction( + session: AsyncSession, + user_id: UUID, + batch: list[PowerSyncCrudMutation], +) -> bool: + """First accepted branch wins; terminal rows prevent retries from recording time twice.""" + mutations = [mutation for mutation in batch if mutation.table == "reading_timers"] + + if len(mutations) != 1: + raise ValidationError("A timer transaction must contain exactly one state transition") + + mutation = mutations[0] + values = mutation.op_data or {} + + if mutation.op.upper() == "DELETE": + raise ValidationError("Close timers with a saved or discarded state") + + try: + timer = TimerPayload.model_validate(values.get("payload")) + row_id = UUID(mutation.id) + except (PayloadError, ValueError) as exc: + raise ValidationError(str(exc)) from exc + + if timer.id != row_id: + raise ValidationError("Timer payload id must match row id") + + row = await session.get(SyncReadingTimer, row_id, populate_existing=True) + + if row is not None and row.owner_user_id != user_id: + raise ForbiddenError("Cannot access another user's timer") + + status = values.get("status", row.status if row is not None else None) + base = values.get("base_version") + + if not isinstance(status, str) or status not in {"active", "saved", "discarded"}: + raise ValidationError("Invalid timer status") + + if base is not None and (not isinstance(base, str) or not base or len(base) > 100): + raise ValidationError("Invalid timer base version") + + if row is not None: + previous = TimerPayload.model_validate(row.payload) + + if row.status != "active" or timer.version == previous.version or base != previous.version: + return False + + fixed = ("book_id", "book_title", "started_at", "device_id", "shelf_ids", "start_page") + + if ( + any(getattr(previous, field) != getattr(timer, field) for field in fixed) + or timer.revision <= previous.revision + or timer.intervals[: len(previous.intervals)] != previous.intervals + ): + raise ValidationError("Timer transitions cannot rewrite their identity or recorded intervals") + elif base is not None or status != "active": + return False + + try: + validate_effects(batch, timer, status) + except PayloadError as exc: + raise ValidationError(str(exc)) from exc + + await check_reference(session, user_id, "books", timer.book_id) + + for shelf_id in timer.shelf_ids: + await check_reference(session, user_id, "shelves", shelf_id) + + deleted_book = await tombstoned(session, user_id, "books", timer.book_id) + + if deleted_book and status != "discarded": + status = "discarded" + + accepted = not deleted_book or values.get("status") == "discarded" + + if row is None: + active = await session.scalar( + select(SyncReadingTimer.id).where( + SyncReadingTimer.owner_user_id == user_id, + SyncReadingTimer.status == "active", + ) + ) + + if active is not None: + status = "discarded" + accepted = False + + row = SyncReadingTimer(id=row_id, owner_user_id=user_id) + session.add(row) + + row.payload = timer.model_dump(mode="json") + row.status = status + row.base_version = base + row.updated_at = datetime.now(UTC) + await session.flush() + return accepted diff --git a/papyrus/services/sync.py b/papyrus/services/sync.py index 9681a46..3c28c94 100644 --- a/papyrus/services/sync.py +++ b/papyrus/services/sync.py @@ -11,6 +11,7 @@ from papyrus.schemas.sync import PowerSyncCrudMutation from papyrus.services import media as media_service from papyrus.services.library_sync import apply_library_mutation +from papyrus.services.reading_timer_sync import apply_timer_transaction async def lock_owner_mutations(session: AsyncSession, user_id: UUID) -> None: @@ -35,7 +36,17 @@ async def apply_powersync_upload_batch( try: await lock_owner_mutations(session, user_id) + if any(mutation.table == "reading_timers" for mutation in batch): + if not await apply_timer_transaction(session, user_id, batch): + await session.commit() + return 0 + + applied_count += 1 + for mutation in batch: + if mutation.table == "reading_timers": + continue + applied, deleted_media_paths = await apply_library_mutation(session, user_id, mutation) applied_count += applied media_paths_to_delete.extend(deleted_media_paths) diff --git a/powersync/sync-config.yaml b/powersync/sync-config.yaml index d1b6dbb..6fd0ac7 100644 --- a/powersync/sync-config.yaml +++ b/powersync/sync-config.yaml @@ -195,3 +195,11 @@ streams: created_at::text AS created_at, updated_at::text AS updated_at FROM goal_periods WHERE owner_user_id::text = auth.user_id() + + reading_timers: + auto_subscribe: true + query: | + SELECT id, owner_user_id::text AS owner_user_id, payload::text AS payload, + status, base_version, created_at::text AS created_at, updated_at::text AS updated_at + FROM reading_timers + WHERE owner_user_id::text = auth.user_id() diff --git a/scripts/setup_local_powersync.sh b/scripts/setup_local_powersync.sh index e9b84cc..a3a79e3 100755 --- a/scripts/setup_local_powersync.sh +++ b/scripts/setup_local_powersync.sh @@ -29,15 +29,15 @@ END \$\$; GRANT USAGE ON SCHEMA public TO "${POWERSYNC_SOURCE_ROLE}"; -GRANT SELECT ON TABLE public.books, public.shelves, public.tags, public.notes, public.annotations, public.bookmarks, public.book_shelves, public.book_tags, public.reading_goals, public.reading_activities, public.goal_periods, public.powersync_demo_items TO "${POWERSYNC_SOURCE_ROLE}"; +GRANT SELECT ON TABLE public.books, public.shelves, public.tags, public.notes, public.annotations, public.bookmarks, public.book_shelves, public.book_tags, public.reading_timers, public.reading_goals, public.reading_activities, public.goal_periods, public.powersync_demo_items TO "${POWERSYNC_SOURCE_ROLE}"; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO "${POWERSYNC_SOURCE_ROLE}"; DO \$\$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = 'powersync') THEN - CREATE PUBLICATION powersync FOR TABLE public.books, public.shelves, public.tags, public.notes, public.annotations, public.bookmarks, public.book_shelves, public.book_tags, public.reading_goals, public.reading_activities, public.goal_periods, public.powersync_demo_items; + CREATE PUBLICATION powersync FOR TABLE public.books, public.shelves, public.tags, public.notes, public.annotations, public.bookmarks, public.book_shelves, public.book_tags, public.reading_timers, public.reading_goals, public.reading_activities, public.goal_periods, public.powersync_demo_items; ELSE - ALTER PUBLICATION powersync SET TABLE public.books, public.shelves, public.tags, public.notes, public.annotations, public.bookmarks, public.book_shelves, public.book_tags, public.reading_goals, public.reading_activities, public.goal_periods, public.powersync_demo_items; + ALTER PUBLICATION powersync SET TABLE public.books, public.shelves, public.tags, public.notes, public.annotations, public.bookmarks, public.book_shelves, public.book_tags, public.reading_timers, public.reading_goals, public.reading_activities, public.goal_periods, public.powersync_demo_items; END IF; END \$\$; diff --git a/tests/services/test_reading_timer_sync.py b/tests/services/test_reading_timer_sync.py new file mode 100644 index 0000000..149bb8c --- /dev/null +++ b/tests/services/test_reading_timer_sync.py @@ -0,0 +1,288 @@ +"""Device handoff, competing branches, terminal saves and ownership.""" + +import asyncio +from datetime import UTC, datetime, timedelta +from uuid import uuid4, uuid5 + +import pytest +from sqlalchemy import select + +from papyrus.core.exceptions import ForbiddenError, ValidationError +from papyrus.models import SyncBook, SyncReadingActivity, SyncReadingTimer +from papyrus.schemas.reading_timer import TimerInterval, TimerPayload +from papyrus.schemas.sync import PowerSyncCrudMutation +from papyrus.schemas.tracking import Activity +from papyrus.services.sync import apply_powersync_upload_batch +from tests.services.test_sync import _create_book, _create_user + + +async def setup(session): + user = await _create_user(session, f"{uuid4()}@example.com") + book = await _create_book(session, user) + start = datetime.now(UTC) - timedelta(minutes=5) + + timer = TimerPayload( + id=uuid4(), + book_id=book.book_id, + book_title=book.title, + started_at=start, + running_since=start, + device_id="browser", + shelf_ids=[], + start_page=0, + phase="running", + intervals=[], + revision=0, + version=str(uuid4()), + target_seconds=1800, + ) + + await session.commit() + return user.user_id, timer + + +def transition(timer, seconds=60, **values): + return timer.model_copy( + update={ + "version": str(uuid4()), + "revision": timer.revision + 1, + "phase": "paused", + "running_since": None, + "intervals": [TimerInterval(start=timer.started_at, end=timer.started_at + timedelta(seconds=seconds))], + **values, + } + ) + + +def mutation(timer, previous=None, status="active"): + return PowerSyncCrudMutation( + table="reading_timers", + op="PUT" if previous is None else "PATCH", + id=str(timer.id), + op_data={ + "payload": timer.model_dump(mode="json"), + "status": status, + "base_version": previous.version if previous else None, + }, + ) + + +def saved(timer, previous, page=12): + activity = Activity( + id=uuid5(timer.id, "0"), + session_id=timer.id, + book_id=timer.book_id, + book_title=timer.book_title, + start_time=timer.intervals[0].start, + end_time=timer.intervals[0].end, + created_at=datetime.now(UTC), + pages=page, + ) + + return [ + mutation(timer, previous, "saved"), + PowerSyncCrudMutation( + table="reading_activities", + op="PUT", + id=str(activity.id), + op_data={"payload": activity.model_dump(mode="json")}, + ), + PowerSyncCrudMutation(table="books", op="PATCH", id=str(timer.book_id), op_data={"current_page": page}), + ] + + +async def test_start_pause_resume_and_finish_are_owned_shared_state(test_session_maker): + async with test_session_maker() as browser, test_session_maker() as phone: + user, timer = await setup(browser) + await apply_powersync_upload_batch(browser, user, [mutation(timer)]) + remote = await phone.get(SyncReadingTimer, timer.id) + assert TimerPayload.model_validate(remote.payload) == timer + paused = transition(timer) + await apply_powersync_upload_batch(phone, user, [mutation(paused, timer)]) + await browser.refresh(await browser.get(SyncReadingTimer, timer.id)) + resumed = transition(paused, phase="running", running_since=timer.started_at + timedelta(minutes=2)) + await apply_powersync_upload_batch(browser, user, [mutation(resumed, paused)]) + await phone.refresh(remote) + assert remote.payload["running_since"] == resumed.model_dump(mode="json")["running_since"] + assert remote.payload["intervals"] == paused.model_dump(mode="json")["intervals"] + + +async def test_stale_branch_and_descendants_cannot_replace_accepted_state(test_session_maker): + async with test_session_maker() as session: + user, timer = await setup(session) + await apply_powersync_upload_batch(session, user, [mutation(timer)]) + winner, stale = transition(timer), transition(timer, seconds=90) + assert await apply_powersync_upload_batch(session, user, [mutation(winner, timer)]) == 1 + assert await apply_powersync_upload_batch(session, user, [mutation(stale, timer)]) == 0 + descendant = transition( + stale, seconds=90, phase="running", running_since=timer.started_at + timedelta(minutes=2) + ) + assert await apply_powersync_upload_batch(session, user, [mutation(descendant, stale)]) == 0 + assert (await session.get(SyncReadingTimer, timer.id)).payload["version"] == winner.version + + +@pytest.mark.parametrize("status", ["saved", "discarded"]) +@pytest.mark.parametrize("phase", ["paused", "running", "finished"]) +async def test_sparse_offline_transition_after_terminal_state_does_not_block_uploads(test_session_maker, status, phase): + async with test_session_maker() as session: + user, timer = await setup(session) + await apply_powersync_upload_batch(session, user, [mutation(timer)]) + terminal = transition(timer, phase="finished") + batch = saved(terminal, timer) if status == "saved" else [mutation(terminal, timer, status)] + await apply_powersync_upload_batch(session, user, batch) + + stale = transition( + timer, + seconds=90, + phase=phase, + running_since=timer.started_at + timedelta(minutes=2) if phase == "running" else None, + ) + + patch = mutation(stale, timer) + del patch.op_data["status"] + assert await apply_powersync_upload_batch(session, user, [patch]) == 0 + row = await session.get(SyncReadingTimer, timer.id) + assert row.status == status + assert row.payload["version"] == terminal.version + history = (await session.scalars(select(SyncReadingActivity))).all() + assert len(history) == (1 if status == "saved" else 0) + + if status == "saved": + assert (await session.get(SyncBook, timer.book_id)).current_page == 12 + + next_timer = timer.model_copy(update={"id": uuid4(), "version": str(uuid4())}) + assert await apply_powersync_upload_batch(session, user, [mutation(next_timer)]) == 1 + + +async def test_concurrent_save_and_retries_write_history_and_progress_once(test_session_maker): + async with test_session_maker() as setup_session: + user, timer = await setup(setup_session) + await apply_powersync_upload_batch(setup_session, user, [mutation(timer)]) + + first, second = transition(timer, phase="finished"), transition(timer, seconds=120, phase="finished") + + async def send(batch): + async with test_session_maker() as session: + return await apply_powersync_upload_batch(session, user, batch) + + batches = [saved(first, timer, 12), saved(second, timer, 24)] + results = await asyncio.gather(*(send(batch) for batch in batches)) + assert sorted(results) == [0, 3] + winner = first if results[0] else second + assert await send(batches[0]) == 0 + assert await send(batches[1]) == 0 + assert await send([mutation(timer)]) == 0 + + async with test_session_maker() as session: + rows = (await session.scalars(select(SyncReadingActivity))).all() + assert len(rows) == 1 + assert rows[0].payload["end_time"] == winner.model_dump(mode="json")["intervals"][0]["end"] + assert (await session.get(SyncBook, timer.book_id)).current_page == (12 if results[0] else 24) + assert (await session.get(SyncReadingTimer, timer.id)).status == "saved" + + +async def test_concurrent_starts_converge_and_discard_cannot_be_revived(test_session_maker): + async with test_session_maker() as session: + user, timer = await setup(session) + other = timer.model_copy(update={"id": uuid4(), "version": str(uuid4())}) + await apply_powersync_upload_batch(session, user, [mutation(timer)]) + assert await apply_powersync_upload_batch(session, user, [mutation(other)]) == 0 + assert (await session.get(SyncReadingTimer, other.id)).status == "discarded" + discarded = transition(timer) + assert await apply_powersync_upload_batch(session, user, [mutation(discarded, timer, "discarded")]) == 1 + assert await apply_powersync_upload_batch(session, user, [mutation(transition(timer), timer)]) == 0 + assert await apply_powersync_upload_batch(session, user, [mutation(other)]) == 0 + third = timer.model_copy(update={"id": uuid4(), "version": str(uuid4())}) + assert await apply_powersync_upload_batch(session, user, [mutation(third)]) == 1 + + +async def test_ownership_validates_timer_and_payload_book(test_session_maker): + async with test_session_maker() as session: + owner, timer = await setup(session) + other, other_timer = await setup(session) + await apply_powersync_upload_batch(session, owner, [mutation(timer)]) + + with pytest.raises(ForbiddenError): + await apply_powersync_upload_batch(session, other, [mutation(transition(timer), timer)]) + + with pytest.raises(ForbiddenError): + await apply_powersync_upload_batch( + session, other, [mutation(other_timer.model_copy(update={"book_id": timer.book_id}))] + ) + + +async def test_invalid_save_rolls_back_state_and_history(test_session_maker): + async with test_session_maker() as session: + user, timer = await setup(session) + await apply_powersync_upload_batch(session, user, [mutation(timer)]) + batch = saved(transition(timer, phase="finished"), timer) + batch[-1].op_data["title"] = "Not a progress update" + + with pytest.raises(ValidationError): + await apply_powersync_upload_batch(session, user, batch) + + row = await session.get(SyncReadingTimer, timer.id) + assert row.status == "active" + assert row.payload["version"] == timer.version + assert (await session.scalars(select(SyncReadingActivity))).all() == [] + + +@pytest.mark.parametrize("activity_kind", ["reading", "completion"]) +async def test_missing_activity_patch_rejects_save_without_losing_retry(test_session_maker, activity_kind): + async with test_session_maker() as session: + user, timer = await setup(session) + await apply_powersync_upload_batch(session, user, [mutation(timer)]) + original_page = (await session.get(SyncBook, timer.book_id)).current_page + finished = transition(timer, phase="finished") + batch = saved(finished, timer) + effect = batch[1] + + if activity_kind == "completion": + completion = Activity( + id=timer.id, + book_id=timer.book_id, + book_title=timer.book_title, + kind="completion", + start_time=finished.intervals[0].end, + end_time=finished.intervals[0].end, + created_at=datetime.now(UTC), + ) + + effect = PowerSyncCrudMutation( + table="reading_activities", + op="PUT", + id=str(completion.id), + op_data={"payload": completion.model_dump(mode="json")}, + ) + + batch.append(effect) + + effect.op = "PATCH" + + with pytest.raises(ValidationError, match="Timer activities must use PUT"): + await apply_powersync_upload_batch(session, user, batch) + + row = await session.get(SyncReadingTimer, timer.id) + assert row.status == "active" + assert row.payload["version"] == timer.version + assert (await session.get(SyncBook, timer.book_id)).current_page == original_page + assert (await session.scalars(select(SyncReadingActivity))).all() == [] + effect.op = "PUT" + assert await apply_powersync_upload_batch(session, user, batch) == len(batch) + assert await apply_powersync_upload_batch(session, user, batch) == 0 + assert (await session.get(SyncReadingTimer, timer.id)).status == "saved" + assert (await session.get(SyncBook, timer.book_id)).current_page == 12 + assert len((await session.scalars(select(SyncReadingActivity))).all()) == len(batch) - 2 + + +async def test_deleted_book_closes_timer_without_resurrecting_book(test_session_maker): + async with test_session_maker() as session: + user, timer = await setup(session) + await apply_powersync_upload_batch(session, user, [mutation(timer)]) + await apply_powersync_upload_batch( + session, user, [PowerSyncCrudMutation(table="books", op="DELETE", id=str(timer.book_id))] + ) + assert await apply_powersync_upload_batch(session, user, saved(transition(timer, phase="finished"), timer)) == 0 + assert (await session.get(SyncReadingTimer, timer.id)).status == "discarded" + assert await session.get(SyncBook, timer.book_id) is None + assert (await session.scalars(select(SyncReadingActivity))).all() == [] diff --git a/tests/test_powersync_sync_config.py b/tests/test_powersync_sync_config.py index 0e7750c..c792dc4 100644 --- a/tests/test_powersync_sync_config.py +++ b/tests/test_powersync_sync_config.py @@ -13,7 +13,17 @@ def test_library_streams_filter_owners_and_setup_publishes_tables() -> None: config = (root / "powersync/sync-config.yaml").read_text() setup = (root / "scripts/setup_local_powersync.sh").read_text() - for table in ("books", "shelves", "tags", "notes", "annotations", "bookmarks", "book_shelves", "book_tags"): + for table in ( + "books", + "shelves", + "tags", + "notes", + "annotations", + "bookmarks", + "book_shelves", + "book_tags", + "reading_timers", + ): stream = config.split(f" {table}:\n", 1)[1].split("\n\n", 1)[0] assert "auto_subscribe: true" in stream assert "WHERE owner_user_id::text = auth.user_id()" in stream diff --git a/tests/test_reading_timer_migration.py b/tests/test_reading_timer_migration.py new file mode 100644 index 0000000..c06f0b3 --- /dev/null +++ b/tests/test_reading_timer_migration.py @@ -0,0 +1,66 @@ +"""Tracking tables are additive and preserve an existing owned library.""" + +import importlib.util +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +import pytest +from alembic.autogenerate import compare_metadata +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy import inspect + +from papyrus.models import Base, SyncBook, User + +pytestmark = pytest.mark.migration + + +async def test_timer_upgrade_preserves_library_and_matches_metadata(db_session): + path = Path(__file__).parents[1] / "alembic/versions/c0d1e2f34567_add_reading_timer_sync.py" + spec = importlib.util.spec_from_file_location("tracking_revision", path) + revision = importlib.util.module_from_spec(spec) + spec.loader.exec_module(revision) + + def migrate(session, fn): + with Operations.context(MigrationContext.configure(session.connection())): + fn() + + await db_session.run_sync(lambda session: migrate(session, revision.downgrade)) + + owner = User( + user_id=uuid4(), + display_name="Reader", + primary_email="migration@example.com", + primary_email_verified=True, + last_login_at=datetime.now(UTC), + ) + + db_session.add(owner) + await db_session.flush() + + book = SyncBook( + book_id=uuid4(), + owner_user_id=owner.user_id, + title="Preserved", + author="Author", + added_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + ) + + identifier = book.book_id + db_session.add(book) + await db_session.flush() + await db_session.run_sync(lambda session: migrate(session, revision.upgrade)) + + differences = await db_session.run_sync( + lambda session: compare_metadata(MigrationContext.configure(session.connection()), Base.metadata) + ) + + assert differences == [] + assert (await db_session.get(SyncBook, identifier)).title == "Preserved" + + for table in ("reading_timers",): + assert await db_session.run_sync(lambda session, name=table: inspect(session.connection()).has_table(name)) + + await db_session.commit()