feat: add durable sync state transitions

Add state-transition methods to ActivityRepository for advancing activity
stages (mark_downloaded, mark_converted, mark_imported, mark_duplicate,
mark_failed) with proper retention of last_completed_stage on failure.
Add list_pending_for_user to filter activities for processing. Implement
SyncRunRepository for creating and finalizing sync runs with counts and
summary errors.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

app/db/repositories.py

@@ -1,10 +1,10 @@
 from datetime import datetime
 
-from sqlalchemy import select
+from sqlalchemy import and_, or_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.db.models import Activity, ActivityStatus, SyncUser
+from app.db.models import Activity, ActivityStatus, SyncRun, SyncRunStatus, SyncUser, utcnow
 
 
 class UserRepository:
@@ -37,6 +37,15 @@ class ActivityRepository:
     def __init__(self, session: Session) -> None:
         self.session = session
 
+    def _require(self, activity_id: int) -> Activity:
+        activity = self.session.get(Activity, activity_id)
+        if activity is None:
+            raise ValueError(f"activity {activity_id} not found")
+        return activity
+
+    def get(self, activity_id: int) -> Activity | None:
+        return self.session.get(Activity, activity_id)
+
     def get_or_create_discovered(
         self,
         *,
@@ -76,3 +85,101 @@ class ActivityRepository:
                 raise
             return existing, False
         return activity, True
+
+    def mark_downloaded(self, activity_id: int, path: str) -> Activity:
+        activity = self._require(activity_id)
+        activity.source_fit_path = path
+        activity.status = ActivityStatus.DOWNLOADED
+        activity.last_completed_stage = ActivityStatus.DOWNLOADED
+        activity.last_error = None
+        activity.retryable = True
+        self.session.commit()
+        return activity
+
+    def mark_converted(self, activity_id: int, path: str) -> Activity:
+        activity = self._require(activity_id)
+        activity.converted_fit_path = path
+        activity.status = ActivityStatus.CONVERTED
+        activity.last_completed_stage = ActivityStatus.CONVERTED
+        activity.last_error = None
+        activity.retryable = True
+        self.session.commit()
+        return activity
+
+    def mark_imported(self, activity_id: int, garmin_activity_id: str | None) -> Activity:
+        activity = self._require(activity_id)
+        activity.status = ActivityStatus.IMPORTED
+        activity.last_completed_stage = ActivityStatus.IMPORTED
+        activity.garmin_activity_id = garmin_activity_id
+        activity.last_error = None
+        activity.retryable = False
+        self.session.commit()
+        return activity
+
+    def mark_duplicate(self, activity_id: int) -> Activity:
+        activity = self._require(activity_id)
+        activity.status = ActivityStatus.DUPLICATE
+        activity.last_completed_stage = ActivityStatus.DUPLICATE
+        activity.last_error = None
+        activity.retryable = False
+        self.session.commit()
+        return activity
+
+    def mark_failed(self, activity_id: int, error: str, *, retryable: bool) -> Activity:
+        activity = self._require(activity_id)
+        activity.status = ActivityStatus.FAILED
+        activity.last_error = error[:2000]
+        activity.retryable = retryable
+        self.session.commit()
+        return activity
+
+    def list_pending_for_user(self, user_id: int) -> list[Activity]:
+        return list(
+            self.session.scalars(
+                select(Activity).where(
+                    Activity.user_id == user_id,
+                    or_(
+                        Activity.status.in_([ActivityStatus.DISCOVERED, ActivityStatus.DOWNLOADED, ActivityStatus.CONVERTED]),
+                        and_(Activity.status == ActivityStatus.FAILED, Activity.retryable.is_(True)),
+                    ),
+                ).order_by(Activity.id)
+            )
+        )
+
+
+class SyncRunRepository:
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def start(self, user_id: int) -> SyncRun:
+        sync_run = SyncRun(user_id=user_id, status=SyncRunStatus.RUNNING)
+        self.session.add(sync_run)
+        self.session.commit()
+        return sync_run
+
+    def get(self, sync_run_id: int) -> SyncRun | None:
+        return self.session.get(SyncRun, sync_run_id)
+
+    def finish(
+        self,
+        sync_run_id: int,
+        *,
+        status: SyncRunStatus,
+        discovered: int,
+        imported: int,
+        skipped: int,
+        failed: int,
+        summary_error: str | None = None,
+    ) -> SyncRun:
+        sync_run = self.session.get(SyncRun, sync_run_id)
+        if sync_run is None:
+            raise ValueError(f"sync_run {sync_run_id} not found")
+        sync_run.finished_at = utcnow()
+        sync_run.status = status
+        sync_run.discovered_count = discovered
+        sync_run.imported_count = imported
+        sync_run.skipped_count = skipped
+        sync_run.failed_count = failed
+        sync_run.summary_error = summary_error[:2000] if summary_error else None
+        self.session.commit()
+        return sync_run

tests/conftest.py

@@ -8,8 +8,8 @@ from sqlalchemy.orm import Session, sessionmaker
 from sqlalchemy.pool import StaticPool
 
 from app.config import Settings
-from app.db.models import Base
+from app.db.models import Activity, Base, HealthState
-from app.db.repositories import ActivityRepository, UserRepository
+from app.db.repositories import ActivityRepository, SyncRunRepository, UserRepository
 from app.main import create_app
 
 
@@ -39,6 +39,11 @@ def activity_repository(db_session: Session) -> ActivityRepository:
     return ActivityRepository(db_session)
 
 
+@pytest.fixture
+def sync_run_repository(db_session: Session) -> SyncRunRepository:
+    return SyncRunRepository(db_session)
+
+
 @pytest.fixture
 def client(tmp_path: Path) -> TestClient:
     settings = Settings(
@@ -54,3 +59,23 @@ def client(tmp_path: Path) -> TestClient:
         yield TestClient(app)
     finally:
         app.state.db_engine.dispose()
+
+
+@pytest.fixture
+def seeded_activity(db_session: Session, user_repository: UserRepository, activity_repository: ActivityRepository) -> Activity:
+    user = user_repository.create(
+        name="Test User",
+        enabled=True,
+        health_state=HealthState.HEALTHY,
+        mywhoosh_email_enc="test@example.com",
+        mywhoosh_password_enc="password",
+        garmin_email_enc="test@garmin.com",
+        garmin_password_enc="garmin_password",
+    )
+    activity, _ = activity_repository.get_or_create_discovered(
+        user_id=user.id,
+        mywhoosh_activity_id="mw-test-123",
+        activity_name="Test Activity",
+        activity_timestamp=None,
+    )
+    return activity

tests/db/test_sync_state.py

@@ -0,0 +1,132 @@
+from app.db.models import ActivityStatus, SyncRunStatus
+
+
+def test_failure_retains_last_completed_stage(activity_repository, seeded_activity) -> None:
+    activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
+    activity_repository.mark_failed(seeded_activity.id, "Garmin timeout", retryable=True)
+    activity = activity_repository.get(seeded_activity.id)
+
+    assert activity.status == ActivityStatus.FAILED
+    assert activity.last_completed_stage == ActivityStatus.DOWNLOADED
+    assert activity.retryable is True
+
+
+def test_converted_activity_is_pending_until_terminal(activity_repository, seeded_activity) -> None:
+    activity_repository.mark_converted(seeded_activity.id, "/data/activities/1/a/converted.fit")
+    ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
+    assert seeded_activity.id in ids
+
+
+def test_list_pending_excludes_imported(activity_repository, seeded_activity) -> None:
+    activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
+    activity_repository.mark_converted(seeded_activity.id, "/data/activities/1/a/converted.fit")
+    activity_repository.mark_imported(seeded_activity.id, "garmin-123")
+
+    ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
+    assert seeded_activity.id not in ids
+
+
+def test_list_pending_excludes_duplicate(activity_repository, seeded_activity) -> None:
+    activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
+    activity_repository.mark_duplicate(seeded_activity.id)
+
+    ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
+    assert seeded_activity.id not in ids
+
+
+def test_list_pending_excludes_non_retryable_failed(activity_repository, seeded_activity) -> None:
+    activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
+    activity_repository.mark_failed(seeded_activity.id, "Cannot retry", retryable=False)
+
+    ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
+    assert seeded_activity.id not in ids
+
+
+def test_list_pending_includes_retryable_failed(activity_repository, seeded_activity) -> None:
+    activity_repository.mark_downloaded(seeded_activity.id, "/data/activities/1/a/source.fit")
+    activity_repository.mark_failed(seeded_activity.id, "Temporary error", retryable=True)
+
+    ids = [item.id for item in activity_repository.list_pending_for_user(seeded_activity.user_id)]
+    assert seeded_activity.id in ids
+
+
+def test_sync_run_start_and_finish(sync_run_repository, user_repository) -> None:
+    user = user_repository.create(
+        name="Test User",
+        enabled=True,
+        health_state="healthy",
+        mywhoosh_email_enc="test@example.com",
+        mywhoosh_password_enc="password",
+        garmin_email_enc="test@garmin.com",
+        garmin_password_enc="garmin_password",
+    )
+
+    sync_run = sync_run_repository.start(user.id)
+    assert sync_run.status == SyncRunStatus.RUNNING
+    assert sync_run.discovered_count == 0
+    assert sync_run.imported_count == 0
+    assert sync_run.skipped_count == 0
+    assert sync_run.failed_count == 0
+
+    finished = sync_run_repository.finish(
+        sync_run.id,
+        status=SyncRunStatus.SUCCESS,
+        discovered=5,
+        imported=3,
+        skipped=1,
+        failed=1,
+    )
+
+    assert finished.status == SyncRunStatus.SUCCESS
+    assert finished.discovered_count == 5
+    assert finished.imported_count == 3
+    assert finished.skipped_count == 1
+    assert finished.failed_count == 1
+    assert finished.finished_at is not None
+
+    # Reload from DB to verify persisted
+    reloaded = sync_run_repository.get(sync_run.id)
+    assert reloaded.status == SyncRunStatus.SUCCESS
+    assert reloaded.discovered_count == 5
+    assert reloaded.imported_count == 3
+
+
+def test_sync_run_finish_with_error(sync_run_repository, user_repository) -> None:
+    user = user_repository.create(
+        name="Test User",
+        enabled=True,
+        health_state="healthy",
+        mywhoosh_email_enc="test@example.com",
+        mywhoosh_password_enc="password",
+        garmin_email_enc="test@garmin.com",
+        garmin_password_enc="garmin_password",
+    )
+
+    sync_run = sync_run_repository.start(user.id)
+    error_msg = "Connection timeout"
+
+    finished = sync_run_repository.finish(
+        sync_run.id,
+        status=SyncRunStatus.FAILED,
+        discovered=0,
+        imported=0,
+        skipped=0,
+        failed=0,
+        summary_error=error_msg,
+    )
+
+    assert finished.summary_error == error_msg
+
+    # Verify truncation works
+    long_error = "x" * 5000
+    finished_long = sync_run_repository.finish(
+        sync_run.id,
+        status=SyncRunStatus.FAILED,
+        discovered=0,
+        imported=0,
+        skipped=0,
+        failed=0,
+        summary_error=long_error,
+    )
+    assert len(finished_long.summary_error) == 2000
+    assert finished_long.summary_error == long_error[:2000]