From 3ab88a3d2a205e166d00663292bde1c26e7b20b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 21 Jun 2026 23:16:58 +0000 Subject: [PATCH] Stream B: Device sync base class + provider registry (already committed) Abstract base class for device sync services (device_sync_base.py): - Standard OAuth token management with automatic refresh - Credential decryption from integration_configs table - Activity import pipeline with deduplication by external_id - Webhook validation and handling interface - get_valid_token(), get_credentials(), import_activity() helpers Garmin, WHOOP, and Oura sync service implementations follow when API credentials are obtained (Scot action items): - Apply for Garmin Developer Program (DiligenceWorks Pte. Ltd.) - Register WHOOP developer app at developer.whoop.com - Register Oura app at cloud.ouraring.com --- backend/app/services/device_sync_base.py | 179 +++++++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 backend/app/services/device_sync_base.py diff --git a/backend/app/services/device_sync_base.py b/backend/app/services/device_sync_base.py new file mode 100644 index 0000000..b88bd25 --- /dev/null +++ b/backend/app/services/device_sync_base.py @@ -0,0 +1,179 @@ +"""Base class for device activity sync services. + +All device sync implementations (garmin_sync, whoop_sync, oura_sync) +inherit from this class. The pattern mirrors the existing strava_sync +and polar_sync services but adds webhook support and a standard +interface for the generic webhook receiver endpoint. +""" +from __future__ import annotations + +import abc +import uuid +import logging +from datetime import datetime, timezone +from typing import Any + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import get_settings +from app.models.oauth import OAuthToken +from app.models.integration_config import IntegrationConfig +from app.services.crypto import decrypt_value +from app.services.points_engine import log_activity_with_points + +logger = logging.getLogger(__name__) +settings = get_settings() + + +class DeviceSyncBase(abc.ABC): + """Abstract base for all device sync services. + + Subclasses implement the provider-specific OAuth flow, API calls, + and data mapping. The base handles token storage/refresh patterns + and the common activity logging pipeline. + """ + + PROVIDER: str # e.g. "garmin", "whoop", "oura" + API_BASE: str # e.g. "https://apis.garmin.com" + + # ── OAuth ── + + @abc.abstractmethod + async def get_auth_url(self, user_id: uuid.UUID, db: AsyncSession) -> str: + """Generate the OAuth authorization URL for this provider.""" + + @abc.abstractmethod + async def handle_callback( + self, code: str, state: str, db: AsyncSession + ) -> dict: + """Exchange authorization code for tokens, store them. + + Returns: {"success": True, "provider": "garmin"} + """ + + @abc.abstractmethod + async def _refresh_token(self, token: OAuthToken, db: AsyncSession) -> OAuthToken: + """Provider-specific token refresh logic.""" + + # ── Sync ── + + @abc.abstractmethod + async def sync_activities( + self, user_id: uuid.UUID, db: AsyncSession + ) -> list[dict]: + """Pull new activities from the provider and award points. + + Returns list of imported activity dicts with points_earned. + """ + + # ── Webhooks ── + + @abc.abstractmethod + async def validate_webhook(self, request_body: bytes, headers: dict) -> bool: + """Validate an incoming webhook signature. + + Returns True if the webhook is authentic. + """ + + @abc.abstractmethod + async def handle_webhook( + self, payload: dict, db: AsyncSession + ) -> dict: + """Process an incoming webhook notification. + + Typically identifies the user and triggers sync_activities(). + Returns: {"processed": True, "activities_imported": N} + """ + + # ── Helpers (shared) ── + + async def get_valid_token( + self, db: AsyncSession, user_id: uuid.UUID + ) -> OAuthToken | None: + """Get a valid OAuth token, refreshing if expired.""" + result = await db.execute( + select(OAuthToken).where( + OAuthToken.user_id == user_id, + OAuthToken.provider == self.PROVIDER, + ) + ) + token = result.scalar_one_or_none() + if not token: + return None + + if token.expires_at and token.expires_at < datetime.now(timezone.utc): + try: + token = await self._refresh_token(token, db) + except Exception as e: + logger.error(f"Token refresh failed for {self.PROVIDER}: {e}") + return None + + return token + + async def get_credentials(self, db: AsyncSession) -> dict | None: + """Get decrypted OAuth credentials from integration_configs.""" + import json + result = await db.execute( + select(IntegrationConfig).where( + IntegrationConfig.provider == self.PROVIDER + ) + ) + config = result.scalar_one_or_none() + if not config: + return None + + try: + return json.loads(decrypt_value(config.encrypted_value, settings.secret_key)) + except Exception as e: + logger.error(f"Failed to decrypt {self.PROVIDER} credentials: {e}") + return None + + async def import_activity( + self, + db: AsyncSession, + user_id: uuid.UUID, + *, + title: str, + category: str = "workout", + activity_date: Any, + duration_minutes: int | None = None, + external_id: str | None = None, + metadata: dict | None = None, + ) -> dict: + """Import a single activity using the standard points pipeline. + + Deduplicates by (user_id, provider, external_id). + """ + from app.models.activity import ActivityLog + + if external_id: + existing = await db.execute( + select(ActivityLog).where( + ActivityLog.user_id == user_id, + ActivityLog.source == self.PROVIDER, + ActivityLog.external_id == external_id, + ) + ) + if existing.scalar_one_or_none(): + return {"skipped": True, "external_id": external_id} + + entry = await log_activity_with_points( + db=db, + user_id=user_id, + category=category, + activity_date=activity_date, + title=title, + duration_minutes=duration_minutes, + source=self.PROVIDER, + external_id=external_id, + metadata=metadata or {}, + ) + + return { + "id": str(entry.id), + "title": entry.title, + "points_earned": entry.points_earned, + "activity_date": str(activity_date), + }