Initial commit: smartrun MVP
Garmin run classification and progression tracker. Go backend (MCP client to mcp-garmin, SQLite store, deterministic rule engine, REST API) and React/TS frontend (Dashboard, Review Queue, Workout Kinds). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
317
backend/internal/sync/service.go
Normal file
317
backend/internal/sync/service.go
Normal file
@@ -0,0 +1,317 @@
|
||||
// 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"
|
||||
"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.
|
||||
type Config struct {
|
||||
// BackfillHorizonDays bounds how far back a full backfill reaches.
|
||||
BackfillHorizonDays int
|
||||
// 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
|
||||
// MaxHR is used only to derive avg_hr_pct_max for the rule engine.
|
||||
MaxHR float64
|
||||
}
|
||||
|
||||
func (c Config) withDefaults() Config {
|
||||
if c.BackfillHorizonDays == 0 {
|
||||
c.BackfillHorizonDays = 3 * 365
|
||||
}
|
||||
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
|
||||
// Config.BackfillHorizonDays is reached or Garmin returns an empty page.
|
||||
// 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.
|
||||
func (s *Service) Backfill(ctx context.Context) error {
|
||||
runID, err := s.db.StartSyncRun(ctx, store.SyncKindBackfill)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
horizon := s.now().AddDate(0, 0, -s.cfg.BackfillHorizonDays)
|
||||
|
||||
state, err := s.db.GetSyncState(ctx)
|
||||
if err != nil {
|
||||
msg := err.Error()
|
||||
s.db.FinishSyncRun(ctx, runID, 0, &msg)
|
||||
return 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 s.db.FinishSyncRun(ctx, runID, 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
|
||||
}
|
||||
|
||||
n, err := s.fetchAndStoreWindow(ctx, dateStr(start), dateStr(end))
|
||||
if err != nil {
|
||||
msg := err.Error()
|
||||
s.db.FinishSyncRun(ctx, runID, total, &msg)
|
||||
return fmt.Errorf("backfill window %s..%s: %w", dateStr(start), dateStr(end), err)
|
||||
}
|
||||
total += n
|
||||
|
||||
if n == 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 {
|
||||
msg := err.Error()
|
||||
s.db.FinishSyncRun(ctx, runID, total, &msg)
|
||||
return err
|
||||
}
|
||||
break
|
||||
}
|
||||
if err := s.db.UpdateSyncState(ctx, dateStr(start), false); err != nil {
|
||||
msg := err.Error()
|
||||
s.db.FinishSyncRun(ctx, runID, total, &msg)
|
||||
return 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 {
|
||||
msg := err.Error()
|
||||
s.db.FinishSyncRun(ctx, runID, total, &msg)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return s.db.FinishSyncRun(ctx, runID, 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
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
n, err := s.fetchAndStoreWindow(ctx, dateStr(start), dateStr(s.now()))
|
||||
if err != nil {
|
||||
msg := err.Error()
|
||||
s.db.FinishSyncRun(ctx, runID, n, &msg)
|
||||
return err
|
||||
}
|
||||
return s.db.FinishSyncRun(ctx, runID, n, nil)
|
||||
}
|
||||
|
||||
func (s *Service) fetchAndStoreWindow(ctx context.Context, startDate, endDate string) (int, error) {
|
||||
activities, err := s.garmin.GetActivities(ctx, startDate, endDate, 500)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get_activities(%s, %s): %w", startDate, endDate, err)
|
||||
}
|
||||
for _, a := range activities {
|
||||
if _, err := s.db.UpsertActivity(ctx, toActivityRow(a)); err != nil {
|
||||
return 0, fmt.Errorf("store activity %d: %w", a.ActivityID, err)
|
||||
}
|
||||
}
|
||||
return len(activities), 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
|
||||
}
|
||||
|
||||
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); 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) 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 := 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)); err != nil {
|
||||
return err
|
||||
}
|
||||
detailsRaw, _ := json.Marshal(details)
|
||||
if err := s.db.SetActivityDetails(ctx, a.ID, string(detailsRaw)); 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)
|
||||
}
|
||||
|
||||
ctxMetrics := buildMetricContext(activity, laps, s.cfg.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
|
||||
}
|
||||
Reference in New Issue
Block a user