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, GarminImportRejected, GarminTransientError, GarminUploadBlocked, ) from app.mywhoosh.client import ( MyWhooshAuthError, MyWhooshDeviceConflictError, 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 # action_reason values that are resolved purely by this run's MyWhoosh listing # succeeding (reaching the end of the activity loop with stop_user_run still # False already proves MyWhoosh is working again -- every MyWhoosh-related # exception branch that could fire also sets stop_user_run=True). _MYWHOOSH_ACTION_REASONS = frozenset( {"mywhoosh_auth_required", "mywhoosh_integration_changed", "mywhoosh_device_conflict"} ) # action_reason values that can only be resolved by actual, this-run evidence # of a successful Garmin import -- the mere absence of a Garmin exception does # NOT prove anything, since no Garmin work may have been attempted this run. _GARMIN_ACTION_REASONS = frozenset({"garmin_mfa_required", "garmin_auth_required"}) 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) # Best-available counters, tracked outside the inner try so that if # something raises before/while they'd normally be populated, the # outer except below can still report whatever we do know. discovered = 0 imported_count = 0 skipped_count = 0 failed_count = 0 mywhoosh = None try: try: 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) stop_user_run = False summary_error: str | None = None # True only once an actual, this-run Garmin import/duplicate # check has succeeded for a real activity -- the sole # evidence that can justify clearing a Garmin-class # action_reason (see _GARMIN_ACTION_REASONS below). garmin_succeeded_this_run = False 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 MyWhooshDeviceConflictError as exc: user.health_state = HealthState.ACTION_REQUIRED user.mywhoosh_state = "device_conflict" user.action_reason = "mywhoosh_device_conflict" 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 # Use the DB's own numeric primary key rather than the # upstream-supplied mywhoosh_activity_id as a directory # component: activity.id is always a safe integer, so # this avoids any path-traversal risk from an # unsanitized remote id (e.g. "../../etc") while # remaining just as stable across resumed syncs. activity_dir = self.settings.activities_dir / str(user.id) / str(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" garmin_succeeded_this_run = True # 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 MyWhooshDeviceConflictError as exc: user.health_state = HealthState.ACTION_REQUIRED user.mywhoosh_state = "device_conflict" user.action_reason = "mywhoosh_device_conflict" 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" 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 GarminImportRejected as exc: # Garmin-side permanent content rejection (real # failures, no successes in detailedImportResult): # same category as a corrupt/unsupported FIT file -- # a non-retryable per-activity failure, no user # health/state change. activity_repo.mark_failed(activity.id, str(exc), retryable=False) 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: # Reaching here with stop_user_run still False proves # this run's MyWhoosh listing succeeded (every # MyWhoosh-exception branch above also sets # stop_user_run=True) -- so a MyWhoosh-class # action_reason is always safe to clear here. It does # NOT prove any Garmin problem was fixed: a Garmin-class # action_reason may only be cleared when this run # actually succeeded at a real Garmin import # (garmin_succeeded_this_run). Otherwise the # action-required condition is still live and # unverified, so leave both action_reason and # health_state untouched. reason = user.action_reason can_clear = ( reason is None or reason in _MYWHOOSH_ACTION_REASONS or (reason in _GARMIN_ACTION_REASONS and garmin_succeeded_this_run) ) if can_clear: 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, ) except Exception as exc: # Any unhandled exception escaping the block above (e.g. a # credential-decrypt failure after key rotation, a factory # constructor raising, an unexpected DB error) would # otherwise leave this SyncRun stuck at # status=RUNNING/finished_at=NULL forever. Finish it as # FAILED with whatever counters we do have, then # re-propagate so callers (e.g. sync_all_enabled's # asyncio.gather(return_exceptions=True)) still see it. session.rollback() sync_run_repo.finish( run.id, status=SyncRunStatus.FAILED, discovered=discovered, imported=imported_count, skipped=skipped_count, failed=failed_count, summary_error=f"{type(exc).__name__}: {str(exc)[:200]}", ) session.commit() raise finally: if mywhoosh is not None: aclose = getattr(mywhoosh, "aclose", None) if aclose is not None: await aclose()