// Package sync orchestrates fetching activities from Garmin (via // internal/garmin), persisting them (via internal/store), and classifying // them (via internal/classify). It's the only package that depends on all // three, keeping garmin/store/classify decoupled from each other. package sync import ( "context" "encoding/json" "fmt" "log" "sync" "time" "smartrun/backend/internal/classify" "smartrun/backend/internal/garmin" "smartrun/backend/internal/store" ) // Config tunes sync behavior. Zero values fall back to sensible defaults in // NewService. How far back Backfill reaches is not here -- it's // Profile.BackfillHorizonDays, read fresh on every call so a user-edited // value takes effect on the next sync without a server restart. type Config struct { // BackfillWindowDays is the page size for each get_activities call // during backfill. BackfillWindowDays int // IncrementalOverlapDays re-fetches a small trailing window on every // incremental sync, guarding against activities that were still // uploading/processing at the time of the previous sync. IncrementalOverlapDays int // InterCallDelay is a pause between sequential Garmin calls during // detail-fill, to avoid tripping Garmin/Cloudflare's rate limiting // (observed firsthand during development). InterCallDelay time.Duration // MinConfidence is the classify.Classify threshold below which even a // single matching kind is sent to manual review. MinConfidence float64 } func (c Config) withDefaults() Config { if c.BackfillWindowDays == 0 { c.BackfillWindowDays = 90 } if c.IncrementalOverlapDays == 0 { c.IncrementalOverlapDays = 2 } if c.InterCallDelay == 0 { c.InterCallDelay = time.Second } if c.MinConfidence == 0 { c.MinConfidence = classify.DefaultMinConfidence } return c } // Progress reports how far a currently-running (or just-finished) // FillPendingDetails pass has gotten, for a status banner to poll. type Progress struct { Done int Total int } // Service is the sync orchestrator. type Service struct { garmin garmin.Client db *store.DB cfg Config now func() time.Time progressMu sync.Mutex progress Progress } // NewService builds a Service. now defaults to time.Now if nil (tests can // override it for deterministic date windows). func NewService(g garmin.Client, db *store.DB, cfg Config, now func() time.Time) *Service { if now == nil { now = time.Now } return &Service{garmin: g, db: db, cfg: cfg.withDefaults(), now: now} } // Progress returns the current detail-fill progress (0/0 when idle). func (s *Service) Progress() Progress { s.progressMu.Lock() defer s.progressMu.Unlock() return s.progress } func (s *Service) setProgress(done, total int) { s.progressMu.Lock() s.progress = Progress{Done: done, Total: total} s.progressMu.Unlock() } // Backfill pages backward in Config.BackfillWindowDays windows until // Profile.BackfillHorizonDays is reached or Garmin returns an empty page. // The horizon is read fresh from the profile on every call (not fixed at // server startup), so a user-edited value takes effect on the very next // sync. Safe to re-run: activities are upserted by garmin_activity_id, and // thanks to the sync_state watermark (Garmin history is immutable once // recorded) a repeat call only fetches whatever's newer than the last // completed backfill, or is a fast no-op if the configured horizon is // already fully covered -- it does not re-walk years of already-known // history. Widening the horizon between calls resumes further back instead // of re-fetching everything. func (s *Service) Backfill(ctx context.Context) error { runID, err := s.db.StartSyncRun(ctx, store.SyncKindBackfill) if err != nil { return err } total, err := s.backfillCore(ctx) if err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, runID, total, &msg) return err } return s.db.FinishSyncRun(ctx, runID, total, nil) } // backfillCore holds Backfill's actual fetch logic, without the SyncRun // bookkeeping, so FullSync can run it as one step of a single combined run // instead of its own separately-recorded one. The returned count reflects // whatever was fetched even when an error is also returned, matching // Backfill's own partial-progress-on-error behavior. func (s *Service) backfillCore(ctx context.Context) (int, error) { profile, err := s.db.GetProfile(ctx) if err != nil { return 0, fmt.Errorf("load profile: %w", err) } horizon := s.now().AddDate(0, 0, -profile.BackfillHorizonDays) state, err := s.db.GetSyncState(ctx) if err != nil { return 0, err } end := s.now() if state.EarliestSyncedDate != nil { if watermark, err := time.Parse("2006-01-02", *state.EarliestSyncedDate); err == nil { if state.BackfillComplete && !watermark.After(horizon) { // Already backfilled at least as far back as the configured // horizon -- nothing new to fetch from Garmin at all. return 0, nil } end = watermark.AddDate(0, 0, -1) } } total := 0 reachedStartOfHistory := false for end.After(horizon) { start := end.AddDate(0, 0, -s.cfg.BackfillWindowDays) if start.Before(horizon) { start = horizon } rawCount, newCount, err := s.fetchAndStoreWindow(ctx, dateStr(start), dateStr(end)) if err != nil { return total, fmt.Errorf("backfill window %s..%s: %w", dateStr(start), dateStr(end), err) } total += newCount if rawCount == 0 { // Empty page: reached the start of this account's history, // regardless of the configured horizon. reachedStartOfHistory = true if err := s.db.UpdateSyncState(ctx, dateStr(start), true); err != nil { return total, err } break } if err := s.db.UpdateSyncState(ctx, dateStr(start), false); err != nil { return total, err } end = start.AddDate(0, 0, -1) } if !reachedStartOfHistory { // Reached the configured horizon (not Garmin's actual history // start) -- mark complete relative to that horizon. if err := s.db.UpdateSyncState(ctx, dateStr(horizon), true); err != nil { return total, err } } return total, nil } // IncrementalSync fetches activities from just before the latest known // activity (or a short recent window if none exist yet) through today. func (s *Service) IncrementalSync(ctx context.Context) error { runID, err := s.db.StartSyncRun(ctx, store.SyncKindIncremental) if err != nil { return err } n, err := s.incrementalSyncCore(ctx) if err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, runID, n, &msg) return err } return s.db.FinishSyncRun(ctx, runID, n, nil) } // incrementalSyncCore holds IncrementalSync's actual fetch logic, without // the SyncRun bookkeeping -- see backfillCore. func (s *Service) incrementalSyncCore(ctx context.Context) (int, error) { start := s.now().AddDate(0, 0, -s.cfg.IncrementalOverlapDays) if latest, ok, err := s.db.LatestActivityStartTime(ctx); err == nil && ok { if t, err := time.Parse("2006-01-02 15:04:05", latest); err == nil { start = t.AddDate(0, 0, -s.cfg.IncrementalOverlapDays) } } _, newCount, err := s.fetchAndStoreWindow(ctx, dateStr(start), dateStr(s.now())) return newCount, err } // FullSync performs a complete manual "Sync now" pass -- Backfill (resumes // from the watermark), then IncrementalSync (catches anything new since the // latest known activity), then FillPendingDetails -- recorded as a single // SyncRun. Backfill and IncrementalSync each record their own SyncRun when // called on their own (used by the periodic background loop), but a manual // sync runs both back to back, and FillPendingDetails records no run at all; // showing the user only the most recently *recorded* run (IncrementalSync's) // would silently hide however many activities Backfill fetched. Recording // one combined run makes the reported count match the whole action. func (s *Service) FullSync(ctx context.Context, detailFillLimit int) error { runID, err := s.db.StartSyncRun(ctx, store.SyncKindFull) if err != nil { return err } backfillCount, err := s.backfillCore(ctx) if err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, runID, backfillCount, &msg) return err } incrementalCount, err := s.incrementalSyncCore(ctx) total := backfillCount + incrementalCount if err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, runID, total, &msg) return err } if err := s.FillPendingDetails(ctx, detailFillLimit); err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, runID, total, &msg) return err } return s.db.FinishSyncRun(ctx, runID, total, nil) } // ResetAll deletes every synced activity (and its laps/samples/kind // assignments) and rewinds the backfill watermark, so the next Backfill // call performs a genuinely fresh pull from Garmin instead of resuming from // wherever the previous one left off. Workout kinds are left untouched. func (s *Service) ResetAll(ctx context.Context) error { return s.db.ResetAllSyncedData(ctx) } // fetchAndStoreWindow returns two counts: rawCount is every activity Garmin's // API returned for this date range, regardless of sport or whether it was // already known -- Backfill's "reached start of history" check needs this // exact unfiltered count, since a page containing only non-running // activities must not look like an empty page. newCount is how many running // activities were genuinely new (not already stored), which is what actually // belongs in the user-facing "activities fetched" report: the incremental // overlap window and backfill's already-covered history mean Garmin almost // always re-returns activities we already have, and reporting rawCount there // produced a confusing, meaningless number (e.g. "2 activities" on a sync // that found nothing new, just because 2 already-known activities happened // to fall inside the queried window). func (s *Service) fetchAndStoreWindow(ctx context.Context, startDate, endDate string) (rawCount, newCount int, err error) { activities, err := s.garmin.GetActivities(ctx, startDate, endDate, 500) if err != nil { return 0, 0, fmt.Errorf("get_activities(%s, %s): %w", startDate, endDate, err) } for _, a := range activities { // Only running activities are of interest here; other sports (padel, // cycling, strength training, ...) also come back from // get_activities() but are dropped rather than stored. if !isRunningActivityType(a.ActivityType.TypeKey) { continue } exists, err := s.db.ActivityExists(ctx, a.ActivityID) if err != nil { return 0, 0, err } if _, err := s.db.UpsertActivity(ctx, toActivityRow(a)); err != nil { return 0, 0, fmt.Errorf("store activity %d: %w", a.ActivityID, err) } if !exists { newCount++ } } return len(activities), newCount, nil } // FillPendingDetails fetches get_activity_splits/get_activity_details for up // to limit activities that don't have them yet, then (re)classifies each one. // Calls are made sequentially with Config.InterCallDelay between them to // avoid Garmin/Cloudflare rate limiting. func (s *Service) FillPendingDetails(ctx context.Context, limit int) error { pending, err := s.db.ActivitiesMissingDetails(ctx, limit) if err != nil { return err } profile, err := s.db.GetProfile(ctx) if err != nil { return fmt.Errorf("load profile: %w", err) } s.setProgress(0, len(pending)) defer s.setProgress(0, 0) for i, a := range pending { if i > 0 { select { case <-ctx.Done(): return ctx.Err() case <-time.After(s.cfg.InterCallDelay): } } if err := s.fillActivityDetails(ctx, a, profile); err != nil { return fmt.Errorf("fill details for activity %d: %w", a.GarminActivityID, err) } if err := s.ClassifyActivity(ctx, a.ID); err != nil { return fmt.Errorf("classify activity %d: %w", a.GarminActivityID, err) } s.setProgress(i+1, len(pending)) } return nil } func (s *Service) fillActivityDetails(ctx context.Context, a store.Activity, profile store.Profile) error { splits, err := s.garmin.GetActivitySplits(ctx, a.GarminActivityID) if err != nil { return fmt.Errorf("get_activity_splits: %w", err) } details, err := s.garmin.GetActivityDetails(ctx, a.GarminActivityID) if err != nil { return fmt.Errorf("get_activity_details: %w", err) } targets := make([]*garmin.WorkoutStep, len(splits.Laps)) if a.WorkoutID != nil { workout, err := s.garmin.GetWorkoutByID(ctx, *a.WorkoutID) if err != nil { log.Printf("sync: get_workout_by_id(%d) for activity %d failed, continuing without target zones: %v", *a.WorkoutID, a.GarminActivityID, err) } else { targets = alignWorkoutTargets(splits.Laps, workout) if err := s.db.SetActivityWorkout(ctx, a.ID, string(workout.Raw)); err != nil { return err } } } samples := garmin.ExtractSamples(details) if err := s.db.ReplaceActivitySamples(ctx, a.ID, toSampleRows(samples)); err != nil { return err } if err := s.db.ReplaceLaps(ctx, a.ID, toLapRows(splits.Laps, samples, targets, profile)); err != nil { return err } if err := s.db.SetActivityDetails(ctx, a.ID, string(details.Raw)); err != nil { return err } return s.db.SetActivitySplitsFetched(ctx, a.ID) } // ClassifyActivity (re)runs the rule engine for one activity against the // currently active workout kinds and appends a new kind_assignments row. // Safe to call repeatedly (e.g. after editing a workout kind's rule). func (s *Service) ClassifyActivity(ctx context.Context, activityID int64) error { activity, ok, err := s.db.GetActivity(ctx, activityID) if err != nil { return err } if !ok { return fmt.Errorf("activity %d not found", activityID) } laps, err := s.db.LapsForActivity(ctx, activityID) if err != nil { return err } kindRows, err := s.db.ListWorkoutKinds(ctx, true) if err != nil { return err } rules, err := loadRuleKinds(kindRows) if err != nil { return fmt.Errorf("parse workout kind rules: %w", err) } profile, err := s.db.GetProfile(ctx) if err != nil { return fmt.Errorf("load profile: %w", err) } var maxHR float64 if profile.MaxHeartRate != nil { maxHR = *profile.MaxHeartRate } ctxMetrics := buildMetricContext(activity, laps, maxHR) result := classify.Classify(ctxMetrics, rules, s.cfg.MinConfidence) candidatesJSON, err := json.Marshal(result.Candidates) if err != nil { return err } _, err = s.db.InsertKindAssignment(ctx, store.KindAssignment{ ActivityID: activityID, WorkoutKindID: result.WorkoutKindID, AssignmentSource: store.AssignmentSourceRuleEngine, Status: result.Status, Confidence: result.Confidence, CandidateKindsJSON: string(candidatesJSON), }) return err }