diff --git a/papyrus/schemas/tracking.py b/papyrus/schemas/tracking.py index d07f659..9e24e38 100644 --- a/papyrus/schemas/tracking.py +++ b/papyrus/schemas/tracking.py @@ -127,6 +127,9 @@ def ordered(self) -> "PageCoverage": class Activity(TrackingPayload): session_id: UUID | None = None + start_position: float | None = Field(default=None, ge=0, le=1) + end_position: float | None = Field(default=None, ge=0, le=1) + device_type: str | None = None book_id: UUID book_title: str = Field(min_length=1, max_length=500) start_time: AwareDatetime diff --git a/papyrus/services/progress.py b/papyrus/services/progress.py index 8b87706..d980ada 100644 --- a/papyrus/services/progress.py +++ b/papyrus/services/progress.py @@ -1,5 +1,6 @@ """Real activity-backed sessions and cross-device statistics.""" +from collections import defaultdict from datetime import UTC, date, datetime, time, timedelta from uuid import UUID, uuid4 @@ -27,7 +28,7 @@ from papyrus.services.goal_progress import effective_activities, project_goal, union_length from papyrus.services.goals import activities from papyrus.services.library_sync import owned_row -from papyrus.services.sync import apply_powersync_upload_batch +from papyrus.services.sync import apply_powersync_upload_batch, lock_owner_mutations def session_response(activity: Activity) -> ReadingSession: @@ -39,7 +40,9 @@ def session_response(activity: Activity) -> ReadingSession: end_time=activity.end_time, pages_read=activity.pages, duration_minutes=int((activity.end_time - activity.start_time).total_seconds()) // 60, - device_type=activity.source, + start_position=activity.start_position, + end_position=activity.end_position, + device_type=activity.device_type if activity.device_type is not None else activity.source, device_name=activity.device_id, created_at=activity.created_at, ) @@ -55,9 +58,11 @@ def grouped_sessions(entries: list[Activity]) -> list[ReadingSession]: for group in groups.values(): first = min(group, key=lambda entry: entry.start_time) - end = max(entry.end_time for entry in group) + last = max(group, key=lambda entry: entry.end_time) + end = last.end_time result = session_response(first) result.end_time = end + result.end_position = last.end_position result.duration_minutes = ( int(union_length([(entry.start_time.timestamp(), entry.end_time.timestamp()) for entry in group])) // 60 @@ -116,6 +121,8 @@ async def list_sessions( async def create_session(session: AsyncSession, user_id: UUID, request: CreateReadingSessionRequest) -> ReadingSession: + await lock_owner_mutations(session, user_id) + book = await owned_row( session, user_id, @@ -146,6 +153,9 @@ async def create_session(session: AsyncSession, user_id: UUID, request: CreateRe original.book_id != request.book_id or original.start_time != start or (requested_end is not None and original.end_time != requested_end) + or original.start_position != request.start_position + or original.end_position != request.end_position + or original.device_type != request.device_type or original.pages != (request.pages_read or 0) or original.device_id != (request.device_name or "manual") ): @@ -173,6 +183,9 @@ async def create_session(session: AsyncSession, user_id: UUID, request: CreateRe start_time=start, end_time=end, created_at=datetime.now(UTC), + start_position=request.start_position, + end_position=request.end_position, + device_type=request.device_type, pages=request.pages_read or 0, device_id=request.device_name or "manual", shelf_ids=shelves, @@ -211,6 +224,61 @@ def totals_definition(start: datetime, end: datetime, book_id: UUID | None = Non ) +def overlaps_period(activity: Activity, start: datetime, end: datetime) -> bool: + if activity.start_time == activity.end_time: + return start <= activity.start_time < end + + return activity.start_time < end and activity.end_time > start + + +def current_reading_streak(ledger: list[Activity], now: datetime) -> int: + """Union complete day ranges and partial boundary days without walking long activities.""" + boundaries: dict[date, list[tuple[float, float]]] = defaultdict(list) + qualified_ranges: list[tuple[int, int]] = [] + + for activity in ledger: + if activity.kind != "reading": + continue + + start = activity.start_time.astimezone(UTC) + end = min(activity.end_time, now).astimezone(UTC) + + if start >= end: + continue + + if start.date() == end.date(): + boundaries[start.date()].append((start.timestamp(), end.timestamp())) + continue + + next_day = datetime.combine(start.date() + timedelta(days=1), time.min, UTC) + last_day = datetime.combine(end.date(), time.min, UTC) + boundaries[start.date()].append((start.timestamp(), next_day.timestamp())) + boundaries[end.date()].append((last_day.timestamp(), end.timestamp())) + first_complete, last_complete = start.date().toordinal() + 1, end.date().toordinal() - 1 + + if first_complete <= last_complete: + qualified_ranges.append((first_complete, last_complete)) + + for day, intervals in boundaries.items(): + if union_length(intervals) >= 300: + qualified_ranges.append((day.toordinal(), day.toordinal())) + + merged: list[tuple[int, int]] = [] + + for first, last in sorted(qualified_ranges): + if merged and first <= merged[-1][1] + 1: + merged[-1] = (merged[-1][0], max(merged[-1][1], last)) + else: + merged.append((first, last)) + + today = now.astimezone(UTC).date().toordinal() + + if merged and merged[-1][1] in {today, today - 1}: + return merged[-1][1] - merged[-1][0] + 1 + + return 0 + + async def statistics( session: AsyncSession, user_id: UUID, start_date: date | None, end_date: date | None ) -> ReadingStatistics: @@ -227,12 +295,9 @@ async def statistics( total = project_goal(totals_definition(start, end), ledger, now) real = effective_activities(ledger) - counted = [ - entry for entry in real if entry.kind == "reading" and entry.start_time < end and entry.end_time >= start - ] + counted = [entry for entry in real if entry.kind == "reading" and overlaps_period(entry, start, end)] daily = [] - reading_dates = [] longest = streak = 0 for index in range((last - first).days + 1): @@ -240,9 +305,7 @@ async def statistics( next_day = day + timedelta(days=1) values = project_goal(totals_definition(day, next_day), ledger, now) - count = len( - {entry.session_id or entry.id for entry in counted if entry.start_time < next_day and entry.end_time > day} - ) + count = len({entry.session_id or entry.id for entry in counted if overlaps_period(entry, day, next_day)}) daily.append( DailyBreakdown( @@ -254,19 +317,11 @@ async def statistics( ) if values["seconds"] >= 300: - reading_dates.append(day.date()) streak += 1 longest = max(longest, streak) else: streak = 0 - current = 0 - cursor = today if today in reading_dates else today - timedelta(days=1) - - while cursor in reading_dates: - current += 1 - cursor -= timedelta(days=1) - books = {entry.book_id: entry.book_title for entry in counted} breakdown = [] @@ -294,7 +349,7 @@ async def statistics( sessions_count=sessions_count, average_session_minutes=total["seconds"] / 60 / sessions_count if sessions_count else 0, reading_days=total["days"], - current_streak=current, + current_streak=current_reading_streak(real, now), longest_streak=longest, ), daily_breakdown=daily, diff --git a/papyrus/services/sync.py b/papyrus/services/sync.py index 87d283c..b7d4bd5 100644 --- a/papyrus/services/sync.py +++ b/papyrus/services/sync.py @@ -12,6 +12,11 @@ from papyrus.services.library_sync import apply_library_mutation +async def lock_owner_mutations(session: AsyncSession, user_id: UUID) -> None: + """Hold the owner's mutation lock until the surrounding transaction ends.""" + await session.execute(select(User.user_id).where(User.user_id == user_id).with_for_update()) + + async def apply_powersync_upload_batch( session: AsyncSession, user_id: UUID, @@ -22,7 +27,7 @@ async def apply_powersync_upload_batch( media_paths_to_delete: list[Path] = [] try: - await session.execute(select(User.user_id).where(User.user_id == user_id).with_for_update()) + await lock_owner_mutations(session, user_id) for mutation in batch: applied, deleted_media_paths = await apply_library_mutation(session, user_id, mutation) diff --git a/tests/api/routes/test_progress.py b/tests/api/routes/test_progress.py index ccae388..c4fc869 100644 --- a/tests/api/routes/test_progress.py +++ b/tests/api/routes/test_progress.py @@ -46,6 +46,7 @@ async def test_create_reading_session(client: AsyncClient, auth_headers: dict[st "start_time": now.isoformat(), "start_position": 0.25, "end_position": 0.35, + "device_type": "android", "pages_read": 20, }, ) @@ -53,6 +54,11 @@ async def test_create_reading_session(client: AsyncClient, auth_headers: dict[st assert response.status_code == 200 data = response.json() assert data["book_id"] == book_id + assert data["start_position"] == 0.25 + assert data["end_position"] == 0.35 + assert data["device_type"] == "android" + sessions = (await client.get("/v1/progress/sessions", headers=auth_headers)).json() + assert sessions["sessions"][0] == data async def test_get_reading_statistics(client: AsyncClient, auth_headers: dict[str, str]): @@ -134,6 +140,9 @@ async def test_statistics_group_reader_checkpoints_as_one_session(client, auth_h "book_id": book_id, "book_title": "Reading book", "source": "reader", + "start_position": index / 10, + "end_position": (index + 1) / 10, + "device_type": "linux", "start_time": (end - timedelta(minutes=20 - index * 10)).isoformat(), "end_time": (end - timedelta(minutes=10 - index * 10)).isoformat(), "created_at": end.isoformat(), @@ -149,6 +158,9 @@ async def test_statistics_group_reader_checkpoints_as_one_session(client, auth_h sessions = (await client.get("/v1/progress/sessions", headers=auth_headers)).json() assert sessions["pagination"]["total"] == 1 assert sessions["sessions"][0]["duration_minutes"] == 20 + assert sessions["sessions"][0]["start_position"] == 0 + assert sessions["sessions"][0]["end_position"] == 0.2 + assert sessions["sessions"][0]["device_type"] == "linux" result = await client.get( "/v1/progress/statistics", @@ -161,3 +173,119 @@ async def test_statistics_group_reader_checkpoints_as_one_session(client, auth_h assert statistics["totals"]["sessions_count"] == 1 assert statistics["totals"]["average_session_minutes"] == 20 assert statistics["books_breakdown"][0]["sessions_count"] == 1 + + +async def test_concurrent_session_retry_reuses_generated_timestamps(client, auth_headers, book_id, monkeypatch): + import asyncio + + from papyrus.services import progress + + upload = progress.apply_powersync_upload_batch + + async def delayed_upload(*args, **kwargs): + await asyncio.sleep(0.1) + return await upload(*args, **kwargs) + + monkeypatch.setattr(progress, "apply_powersync_upload_batch", delayed_upload) + + payload = { + "session_id": str(uuid4()), + "book_id": book_id, + "start_time": (datetime.now(UTC) - timedelta(minutes=10)).isoformat(), + "start_position": 0.2, + "end_position": 0.4, + "device_type": "android", + "pages_read": 12, + } + + first, second = await asyncio.gather( + client.post("/v1/progress/sessions", headers=auth_headers, json=payload), + client.post("/v1/progress/sessions", headers=auth_headers, json=payload), + ) + + assert first.status_code == second.status_code == 200 + assert first.json() == second.json() + + for field, value in [("start_position", 0.3), ("end_position", 0.5), ("device_type", "windows")]: + response = await client.post("/v1/progress/sessions", headers=auth_headers, json={**payload, field: value}) + + assert response.status_code == 400 + + +async def test_statistics_excludes_sessions_ending_at_period_start(client, auth_headers, book_id): + boundary = datetime(2026, 1, 2, tzinfo=UTC) + + response = await client.post( + "/v1/progress/sessions", + headers=auth_headers, + json={ + "book_id": book_id, + "start_time": (boundary - timedelta(minutes=10)).isoformat(), + "end_time": boundary.isoformat(), + "pages_read": 5, + }, + ) + + assert response.status_code == 200 + params = {"start_date": "2026-01-02", "end_date": "2026-01-02"} + result = await client.get("/v1/progress/statistics", headers=auth_headers, params=params) + statistics = result.json() + assert statistics["totals"]["sessions_count"] == 0 + assert statistics["totals"]["pages_read"] == 0 + assert statistics["totals"]["reading_time_minutes"] == 0 + assert statistics["books_breakdown"] == [] + assert statistics["daily_breakdown"][0]["sessions_count"] == 0 + + response = await client.post( + "/v1/progress/sessions", + headers=auth_headers, + json={ + "book_id": book_id, + "start_time": boundary.isoformat(), + "end_time": boundary.isoformat(), + "pages_read": 3, + }, + ) + + assert response.status_code == 200 + result = await client.get("/v1/progress/statistics", headers=auth_headers, params=params) + statistics = result.json() + assert statistics["totals"]["sessions_count"] == 1 + assert statistics["totals"]["pages_read"] == 3 + assert statistics["daily_breakdown"][0]["sessions_count"] == 1 + + +async def test_current_streak_is_independent_of_statistics_period(client, auth_headers, book_id, monkeypatch): + from papyrus.services import progress + + now = datetime(2026, 3, 1, 12, tzinfo=UTC) + + class FrozenDateTime(datetime): + @classmethod + def now(cls, tz=None): + return now + + monkeypatch.setattr(progress, "datetime", FrozenDateTime) + + for days_ago in range(3): + end = now - timedelta(days=days_ago) + + response = await client.post( + "/v1/progress/sessions", + headers=auth_headers, + json={ + "book_id": book_id, + "start_time": (end - timedelta(minutes=10)).isoformat(), + "end_time": end.isoformat(), + }, + ) + + assert response.status_code == 200 + + for params in [{}, {"start_date": "2026-02-01", "end_date": "2026-02-01"}]: + response = await client.get("/v1/progress/statistics", headers=auth_headers, params=params) + assert response.status_code == 200 + assert response.json()["totals"]["current_streak"] == 3 + + current = (await client.get("/v1/progress/statistics", headers=auth_headers)).json() + assert current["totals"]["longest_streak"] == 1 diff --git a/tests/services/test_progress.py b/tests/services/test_progress.py new file mode 100644 index 0000000..27226e8 --- /dev/null +++ b/tests/services/test_progress.py @@ -0,0 +1,63 @@ +"""Statistics streaks union activity boundaries without expanding every calendar day.""" + +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +from papyrus.schemas.tracking import Activity +from papyrus.services import progress + + +def activity(start: datetime, end: datetime) -> Activity: + return Activity( + id=uuid4(), + book_id=uuid4(), + book_title="Book", + start_time=start, + end_time=end, + created_at=end, + ) + + +def test_streak_unions_partial_days_and_does_not_double_count_overlaps(): + today = datetime(2026, 3, 1, tzinfo=UTC) + now = today + timedelta(hours=12) + + entries = [ + activity(today - timedelta(days=2), today - timedelta(days=2) + timedelta(minutes=5)), + activity(today - timedelta(minutes=2), today + timedelta(minutes=2)), + activity(today - timedelta(minutes=2), today + timedelta(minutes=2)), + activity(today - timedelta(minutes=5), today - timedelta(minutes=2)), + activity(today + timedelta(hours=1), today + timedelta(hours=1, minutes=3)), + ] + + assert progress.current_reading_streak(entries, now) == 3 + assert progress.current_reading_streak(entries[:-1], now) == 2 + assert progress.current_reading_streak(entries, now + timedelta(days=2)) == 0 + + +def test_streak_clips_future_time_and_ignores_non_reading_records(): + today = datetime(2026, 3, 1, tzinfo=UTC) + entry = activity(today, today + timedelta(days=5)) + assert progress.current_reading_streak([entry], today + timedelta(minutes=4)) == 0 + assert progress.current_reading_streak([entry], today + timedelta(minutes=5)) == 1 + completion = entry.model_copy(update={"kind": "completion"}) + assert progress.current_reading_streak([completion], today + timedelta(days=5)) == 0 + + +def test_streak_cost_does_not_scale_with_activity_duration(monkeypatch): + now = datetime(2026, 3, 1, 12, tzinfo=UTC) + start = datetime(1, 1, 1, tzinfo=UTC) + combines = 0 + + class CountingDateTime(datetime): + @classmethod + def combine(cls, *args, **kwargs): + nonlocal combines + combines += 1 + assert combines <= 4 + return datetime.combine(*args, **kwargs) + + monkeypatch.setattr(progress, "datetime", CountingDateTime) + entries = [activity(start, now), activity(start, now)] + assert progress.current_reading_streak(entries, now) == now.date().toordinal() + assert combines == 4