refactor: merge internal/sync into internal/garmin, regroup api files and routes
Garmin auth/sync routes move under /api/garmin/*; sync.Service becomes garmin.Sync with garmin.SyncConfig/ClientConfig; applog becomes internal/log; the test mock moves into the garmin package as MockClient (breaking the test-only import cycle the merge created); stale test URLs and type names updated to match. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
489
backend/internal/garmin/sync.go
Normal file
489
backend/internal/garmin/sync.go
Normal file
@@ -0,0 +1,489 @@
|
||||
// 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"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"geniusrun/backend/internal/classify"
|
||||
"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.
|
||||
log.Printf("sync: workout for activity %d not found on Garmin, marking as such (will not retry): %v", a.GarminActivityID, 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 {
|
||||
log.Printf("sync: fill workout for activity %d failed, will retry next sync: %v", a.GarminActivityID, 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
|
||||
}
|
||||
Reference in New Issue
Block a user