feat: add resumable per-user sync pipeline
This commit is contained in:
Bastian Wagner
252
app/sync/manager.py
Normal file
252
app/sync/manager.py
Normal file
@@ -0,0 +1,252 @@
|
||||
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 SyncManager:
|
||||
"""Resumable single-user MyWhoosh -> Garmin sync pipeline.
|
||||
|
||||
Locking/scheduling across multiple users is layered on top of `sync_user`
|
||||
by a later task; this class only implements the state machine for one
|
||||
user's sync run.
|
||||
"""
|
||||
|
||||
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
|
||||
|
||||
async def sync_user(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()
|
||||
|
||||
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,
|
||||
)
|
||||
Reference in New Issue
Block a user