geniusrun/backend/internal/sync/service.go

// 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
	}

	profile, err := s.db.GetProfile(ctx)
	if err != nil {
		msg := err.Error()
		s.db.FinishSyncRun(ctx, runID, 0, &msg)
		return fmt.Errorf("load profile: %w", err)
	}
	horizon := s.now().AddDate(0, 0, -profile.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)
}

// 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)
}

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 {
		// 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. len(activities)
		// below stays the unfiltered count, since it drives the backfill
		// watermark's "reached start of history" check -- a page containing
		// only non-running activities must not look like an empty page.
		if !isRunningActivityType(a.ActivityType.TypeKey) {
			continue
		}
		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
	}
	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)
		}
	}

	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
	}
	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)
	}
	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
}