// Package garmin 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 garmin import ( "context" "encoding/json" "errors" "fmt" "sync" "time" "geniusrun/backend/internal/classify" applog "geniusrun/backend/internal/log" "geniusrun/backend/internal/store" ) // Config tunes sync behavior. Zero values fall back to sensible defaults in // NewSync. 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 SyncConfig 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 SyncConfig) withDefaults() SyncConfig { 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 } // Phase values reported by Progress.Phase. const ( PhaseIdle = "idle" PhaseDiscovering = "discovering" PhaseActivities = "activities" PhaseWorkouts = "workouts" ) // Progress reports how far a currently-running (or just-finished) FullSync // has gotten, for a status modal to poll. PhaseDiscovering (backfillCore/ // incrementalSyncCore) has no meaningful Total -- discovering how many // activities exist IS the act of fetching them, so it's reported as an // indeterminate step (Done/Total both 0) rather than a fake percentage. // PhaseActivities/PhaseWorkouts do have a real Total, taken once from a // local DB count at the start of each pass (see FillPendingDetails). type Progress struct { Phase string Done int Total int } // Sync is the sync orchestrator, scoped to one user -- every store call // it makes is for userID's data only. type Sync struct { garmin Client db *store.DB userID int64 cfg SyncConfig now func() time.Time progressMu sync.Mutex progress Progress } // NewSync builds a Sync scoped to userID. now defaults to time.Now if // nil (tests can override it for deterministic date windows). func NewSync(g Client, db *store.DB, userID int64, cfg SyncConfig, now func() time.Time) *Sync { if now == nil { now = time.Now } return &Sync{garmin: g, db: db, userID: userID, cfg: cfg.withDefaults(), now: now, progress: Progress{Phase: PhaseIdle}} } // Progress returns the current sync progress (phase idle, 0/0 when nothing // is running). func (s *Sync) Progress() Progress { s.progressMu.Lock() defer s.progressMu.Unlock() return s.progress } func (s *Sync) setProgress(phase string, done, total int) { s.progressMu.Lock() s.progress = Progress{Phase: phase, Done: done, Total: total} s.progressMu.Unlock() } // backfillCore 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. Used by FullSync as one step of its single // combined SyncRun; there is no standalone entrypoint for this anymore // (the periodic background sync loop that used to call one is gone -- see // 4d2cbe4 refactor: remove automatic background incremental sync). func (s *Sync) backfillCore(ctx context.Context) (int, error) { profile, err := s.db.GetProfile(ctx, s.userID) 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, s.userID) 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, s.userID, dateStr(start), true); err != nil { return total, err } break } if err := s.db.UpdateSyncState(ctx, s.userID, 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, s.userID, dateStr(horizon), true); err != nil { return total, err } } return total, nil } // incrementalSyncCore fetches activities from just before the latest known // activity (or a short recent window if none exist yet) through today. Used // by FullSync as one step of its single combined SyncRun -- see // backfillCore's comment for why there's no standalone entrypoint. func (s *Sync) incrementalSyncCore(ctx context.Context) (int, error) { start := s.now().AddDate(0, 0, -s.cfg.IncrementalOverlapDays) if latest, ok, err := s.db.LatestActivityStartTime(ctx, s.userID); 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 -- backfillCore // (resumes from the watermark), then incrementalSyncCore (catches anything // new since the latest known activity), then FillPendingDetails -- // recorded as a single SyncRun so the reported activity count covers the // whole action instead of only whichever stage happened to finish last. func (s *Sync) FullSync(ctx context.Context, detailFillLimit int) error { runID, err := s.db.StartSyncRun(ctx, s.userID, store.SyncKindFull) if err != nil { return err } // Reset to idle on every return path, including an early error return // from backfillCore/incrementalSyncCore before FillPendingDetails (which // otherwise owns its own idle-reset) ever runs. s.setProgress(PhaseDiscovering, 0, 0) defer s.setProgress(PhaseIdle, 0, 0) backfillCount, err := s.backfillCore(ctx) if err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, s.userID, runID, backfillCount, &msg) return err } incrementalCount, err := s.incrementalSyncCore(ctx) total := backfillCount + incrementalCount if err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, s.userID, runID, total, &msg) return err } if err := s.FillPendingDetails(ctx, detailFillLimit); err != nil { msg := err.Error() s.db.FinishSyncRun(ctx, s.userID, runID, total, &msg) return err } return s.db.FinishSyncRun(ctx, s.userID, 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 *Sync) ResetAll(ctx context.Context) error { return s.db.ResetAllSyncedData(ctx, s.userID) } // 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 *Sync) 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, s.userID, a.ActivityID) if err != nil { return 0, 0, err } if _, err := s.db.UpsertActivity(ctx, s.userID, 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 activity details/splits for up to limit // activities missing them, then fetches workouts for up to limit activities // missing those (independently -- see ActivitiesMissingWorkout), then // (re)classifies every activity touched by the first pass. Each pass makes // its Garmin calls sequentially with Config.InterCallDelay between them to // avoid Garmin/Cloudflare rate limiting. func (s *Sync) FillPendingDetails(ctx context.Context, limit int) error { defer s.setProgress(PhaseIdle, 0, 0) if err := s.fillPendingActivityDetails(ctx, limit); err != nil { return err } return s.fillPendingWorkouts(ctx, limit) } func (s *Sync) fillPendingActivityDetails(ctx context.Context, limit int) error { pending, err := s.db.ActivitiesMissingDetails(ctx, s.userID, limit) if err != nil { return err } profile, err := s.db.GetProfile(ctx, s.userID) if err != nil { return fmt.Errorf("load profile: %w", err) } s.setProgress(PhaseActivities, 0, len(pending)) 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(PhaseActivities, i+1, len(pending)) } return nil } // fillPendingWorkouts fetches get_workout_by_id for up to limit activities // missing it. Unlike fillPendingActivityDetails, one activity's workout // fetch failing is non-fatal (logged, loop continues) -- workout target // bands are enrichment, not core activity data, and workout_raw_json // staying NULL means ActivitiesMissingWorkout will naturally retry it on // the next sync. func (s *Sync) fillPendingWorkouts(ctx context.Context, limit int) error { pending, err := s.db.ActivitiesMissingWorkout(ctx, s.userID, limit) if err != nil { return err } profile, err := s.db.GetProfile(ctx, s.userID) if err != nil { return fmt.Errorf("load profile: %w", err) } s.setProgress(PhaseWorkouts, 0, len(pending)) for i, a := range pending { if i > 0 { select { case <-ctx.Done(): return ctx.Err() case <-time.After(s.cfg.InterCallDelay): } } if err := s.fillActivityWorkout(ctx, a, profile); err != nil { if errors.Is(err, ErrNotFound) { // A definitive 404 (the workout was deleted on Garmin's side // after being linked to this activity) will never succeed on // retry -- mark it so ActivitiesMissingWorkout stops // surfacing it, instead of retrying forever. applog.App().Warn("workout not found on Garmin, marking as such (will not retry)", "garmin_activity_id", a.GarminActivityID, "error", err) if serr := s.db.SetActivityWorkoutNotFound(ctx, s.userID, a.ID); serr != nil { return fmt.Errorf("mark activity %d workout not found: %w", a.GarminActivityID, serr) } } else { applog.App().Warn("fill workout failed, will retry next sync", "garmin_activity_id", a.GarminActivityID, "error", err) } } s.setProgress(PhaseWorkouts, i+1, len(pending)) } return nil } func (s *Sync) 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) } samples := ExtractSamples(details) if err := s.db.ReplaceActivitySamples(ctx, s.userID, a.ID, toSampleRows(samples)); err != nil { return err } // No workout-target alignment here -- that's fillActivityWorkout's job, // run as its own later pass (see fillPendingWorkouts). targets is an // all-nil placeholder the same length as splits.Laps. targets := make([]*WorkoutStep, len(splits.Laps)) if err := s.db.ReplaceLaps(ctx, s.userID, a.ID, toLapRows(splits.Laps, samples, targets, profile)); err != nil { return err } if err := s.db.SetActivityDetails(ctx, s.userID, a.ID, string(details.Raw)); err != nil { return err } return s.db.SetActivitySplitsFetched(ctx, s.userID, a.ID) } // fillActivityWorkout fetches a's structured workout and re-derives its // laps' target pace/HR bands from it. Requires a.WorkoutID to be set -- // only ever called for activities ActivitiesMissingWorkout returned, which // already filters on that. Reads laps back from the DB (already written by // fillActivityDetails, in some earlier pass or run) rather than needing the // original garmin.Lap data again, since alignWorkoutTargets only needs a // count. func (s *Sync) fillActivityWorkout(ctx context.Context, a store.Activity, profile store.Profile) error { workout, err := s.garmin.GetWorkoutByID(ctx, *a.WorkoutID) if err != nil { return fmt.Errorf("get_workout_by_id: %w", err) } laps, err := s.db.LapsForActivity(ctx, s.userID, a.ID) if err != nil { return err } targets := alignWorkoutTargets(len(laps), workout) for i := range laps { if i < len(targets) && targets[i] != nil { laps[i].TargetPaceLowMps, laps[i].TargetPaceHighMps = targetPaceRange(*targets[i]) laps[i].TargetHRLowBpm, laps[i].TargetHRHighBpm = targetHRRange(*targets[i], profile) } } if err := s.db.ReplaceLaps(ctx, s.userID, a.ID, laps); err != nil { return err } return s.db.SetActivityWorkout(ctx, s.userID, a.ID, string(workout.Raw)) } // 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 *Sync) ClassifyActivity(ctx context.Context, activityID int64) error { activity, ok, err := s.db.GetActivity(ctx, s.userID, activityID) if err != nil { return err } if !ok { return fmt.Errorf("activity %d not found", activityID) } laps, err := s.db.LapsForActivity(ctx, s.userID, activityID) if err != nil { return err } kindRows, err := s.db.ListWorkoutKinds(ctx, s.userID, 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, s.userID) 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, s.userID, store.KindAssignment{ ActivityID: activityID, WorkoutKindID: result.WorkoutKindID, AssignmentSource: store.AssignmentSourceRuleEngine, Status: result.Status, Confidence: result.Confidence, CandidateKindsJSON: string(candidatesJSON), }) return err }