from __future__ import annotations import asyncio import logging from pathlib import Path from typing import Any, Callable from app.db.models import ActivityStatus, HealthState, SyncRunStatus from app.db.repositories import ActivityRepository, SyncRunRepository, UserRepository from app.fit.rewriter import FitFormatError from app.garmin.uploader import GarminAuthError, GarminTransientError, GarminUploadBlocked from app.mywhoosh.client import MyWhooshAuthError, MyWhooshIntegrationError, MyWhooshTransientError from app.mywhoosh.tokenstore import MyWhooshTokenStore from app.security.credentials import CredentialCipher from app.sync.states import SyncOutcome logger = logging.getLogger(__name__) class SyncAlreadyRunning(RuntimeError): pass class SyncManager: """Resumable single-user MyWhoosh -> Garmin sync pipeline. `_sync_user_locked` implements the state machine for one user's sync run. `sync_user` wraps it with a per-user `asyncio.Lock` so only one sync can run for a given user at a time (raising `SyncAlreadyRunning` on overlap), and `sync_all_enabled` fans out across all enabled users, isolating each user's failure to its own result rather than cancelling siblings. """ def __init__( self, *, session_factory: Callable[[], Any], credential_cipher: CredentialCipher, settings: Any, mywhoosh_factory: Callable[[MyWhooshTokenStore], Any], garmin_factory: Callable[[str, str, Path], Any], fit_converter: Callable[[Path, Path], Any], ) -> None: self.session_factory = session_factory self.credential_cipher = credential_cipher self.settings = settings self.mywhoosh_factory = mywhoosh_factory self.garmin_factory = garmin_factory self.fit_converter = fit_converter self._locks: dict[int, asyncio.Lock] = {} self._locks_guard = asyncio.Lock() async def _lock_for(self, user_id: int) -> asyncio.Lock: async with self._locks_guard: return self._locks.setdefault(user_id, asyncio.Lock()) async def sync_user(self, user_id: int, mfa_code: str | None = None) -> SyncOutcome: lock = await self._lock_for(user_id) if lock.locked(): raise SyncAlreadyRunning(f"sync already running for user {user_id}") async with lock: return await self._sync_user_locked(user_id, mfa_code) def _load_enabled_user_ids(self) -> list[int]: with self.session_factory() as session: return [user.id for user in UserRepository(session).list_enabled()] async def sync_all_enabled(self) -> list[SyncOutcome | Exception]: user_ids = self._load_enabled_user_ids() return await asyncio.gather( *(self.sync_user(user_id) for user_id in user_ids), return_exceptions=True, ) async def _sync_user_locked(self, user_id: int, mfa_code: str | None = None) -> SyncOutcome: with self.session_factory() as session: user = UserRepository(session).get(user_id) if user is None: raise ValueError(f"user {user_id} not found") sync_run_repo = SyncRunRepository(session) run = sync_run_repo.start(user_id) mw_email = self.credential_cipher.decrypt(user.mywhoosh_email_enc) mw_password = self.credential_cipher.decrypt(user.mywhoosh_password_enc) garmin_email = self.credential_cipher.decrypt(user.garmin_email_enc) garmin_password = self.credential_cipher.decrypt(user.garmin_password_enc) token_dir = self.settings.tokens_dir / str(user.id) mywhoosh = self.mywhoosh_factory(MyWhooshTokenStore(token_dir / "mywhoosh.json")) garmin = self.garmin_factory(garmin_email, garmin_password, token_dir / "garmin") activity_repo = ActivityRepository(session) imported_count = 0 skipped_count = 0 failed_count = 0 stop_user_run = False summary_error: str | None = None try: remote_activities = await mywhoosh.list_activities(mw_email, mw_password) except MyWhooshTransientError as exc: user.health_state = HealthState.DEGRADED user.mywhoosh_state = "error" session.commit() remote_activities = [] stop_user_run = True summary_error = str(exc) except MyWhooshAuthError as exc: user.health_state = HealthState.ACTION_REQUIRED user.mywhoosh_state = "auth_required" user.action_reason = "mywhoosh_auth_required" session.commit() remote_activities = [] stop_user_run = True summary_error = str(exc) except MyWhooshIntegrationError as exc: user.health_state = HealthState.ACTION_REQUIRED user.mywhoosh_state = "integration_error" user.action_reason = "mywhoosh_integration_changed" session.commit() remote_activities = [] stop_user_run = True summary_error = str(exc) except Exception as exc: user.health_state = HealthState.DEGRADED user.mywhoosh_state = "error" session.commit() remote_activities = [] stop_user_run = True summary_error = f"{type(exc).__name__}: {str(exc)[:200]}" logger.warning( "sync_user: unexpected error listing activities for user %s: %s", user.id, exc.__class__.__name__, ) else: user.mywhoosh_state = "connected" session.commit() discovered = len(remote_activities) for remote in remote_activities: if stop_user_run: break activity, _created = activity_repo.get_or_create_discovered( user_id=user.id, mywhoosh_activity_id=remote.id, activity_name=remote.title, activity_timestamp=remote.started_at, ) # Non-retryable failures (e.g. corrupt/unsupported FIT files) are # terminal: never re-attempt them, and don't count them in any # counter for this run. if activity.status == ActivityStatus.FAILED and not activity.retryable: continue stage = ( activity.last_completed_stage if activity.status == ActivityStatus.FAILED else activity.status ) initial_stage = stage activity_dir = self.settings.activities_dir / str(user.id) / activity.mywhoosh_activity_id source_path = activity_dir / "source.fit" converted_path = activity_dir / "edge-1030-plus.fit" try: if stage == ActivityStatus.DISCOVERED: fit_bytes = await mywhoosh.download_fit(remote.activity_file_id, mw_email, mw_password) activity_dir.mkdir(parents=True, exist_ok=True) source_path.write_bytes(fit_bytes) activity = activity_repo.mark_downloaded(activity.id, str(source_path)) stage = activity.status if stage in {ActivityStatus.DOWNLOADED}: self.fit_converter(source_path, converted_path) activity = activity_repo.mark_converted(activity.id, str(converted_path)) stage = activity.status if stage in {ActivityStatus.CONVERTED}: upload = await asyncio.to_thread(garmin.import_fit, converted_path, mfa_code) if upload.duplicate: activity = activity_repo.mark_duplicate(activity.id) else: activity = activity_repo.mark_imported(activity.id, upload.garmin_activity_id) stage = activity.status user.garmin_state = "connected" # Only count this activity's outcome toward this run's totals # if the state machine actually did work this call. An # activity that was already terminal (IMPORTED/DUPLICATE) # before this call is resume history, not this run's work. if initial_stage not in (ActivityStatus.IMPORTED, ActivityStatus.DUPLICATE): if activity.status == ActivityStatus.IMPORTED: imported_count += 1 elif activity.status == ActivityStatus.DUPLICATE: skipped_count += 1 except MyWhooshTransientError as exc: user.health_state = HealthState.DEGRADED user.mywhoosh_state = "error" activity_repo.mark_failed(activity.id, str(exc), retryable=True) failed_count += 1 except MyWhooshAuthError as exc: user.health_state = HealthState.ACTION_REQUIRED user.mywhoosh_state = "auth_required" user.action_reason = "mywhoosh_auth_required" stop_user_run = True summary_error = str(exc) except MyWhooshIntegrationError as exc: user.health_state = HealthState.ACTION_REQUIRED user.mywhoosh_state = "integration_error" user.action_reason = "mywhoosh_integration_changed" stop_user_run = True summary_error = str(exc) except GarminUploadBlocked as exc: user.health_state = HealthState.ACTION_REQUIRED user.garmin_state = "mfa_required" user.action_reason = "garmin_mfa_required" stop_user_run = True summary_error = str(exc) except GarminAuthError as exc: user.health_state = HealthState.ACTION_REQUIRED user.garmin_state = "auth_required" user.action_reason = "garmin_auth_required" stop_user_run = True summary_error = str(exc) except GarminTransientError as exc: user.health_state = HealthState.DEGRADED user.garmin_state = "error" activity_repo.mark_failed(activity.id, str(exc), retryable=True) failed_count += 1 except FitFormatError as exc: activity_repo.mark_failed(activity.id, str(exc), retryable=False) failed_count += 1 except Exception as exc: user.health_state = HealthState.DEGRADED activity_repo.mark_failed( activity.id, f"{type(exc).__name__}: {str(exc)[:200]}", retryable=True, ) failed_count += 1 logger.warning( "sync_user: unexpected error for user %s activity %s: %s", user.id, activity.id, exc.__class__.__name__, ) session.commit() if not stop_user_run: user.action_reason = None user.health_state = HealthState.DEGRADED if failed_count > 0 else HealthState.HEALTHY session.commit() status = ( SyncRunStatus.SUCCESS if failed_count == 0 and not stop_user_run else SyncRunStatus.PARTIAL if (imported_count + skipped_count) > 0 else SyncRunStatus.FAILED ) sync_run_repo.finish( run.id, status=status, discovered=discovered, imported=imported_count, skipped=skipped_count, failed=failed_count, summary_error=summary_error, ) session.commit() return SyncOutcome( user_id=user.id, status=status.value, discovered=discovered, imported=imported_count, skipped=skipped_count, failed=failed_count, message=summary_error, )