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"

	"geniusrun/backend/internal/classify"
	"geniusrun/backend/internal/garmin"
	"geniusrun/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, scoped to one user -- every store call
// it makes is for userID's data only.
type Service struct {
	garmin garmin.Client
	db     *store.DB
	userID int64
	cfg    Config
	now    func() time.Time

	progressMu sync.Mutex
	progress   Progress
}

// NewService builds a Service scoped to userID. now defaults to time.Now if
// nil (tests can override it for deterministic date windows).
func NewService(g garmin.Client, db *store.DB, userID int64, cfg Config, now func() time.Time) *Service {
	if now == nil {
		now = time.Now
	}
	return &Service{garmin: g, db: db, userID: userID, 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()
}

// 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 *Service) 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 *Service) 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 *Service) FullSync(ctx context.Context, detailFillLimit int) error {
	runID, err := s.db.StartSyncRun(ctx, s.userID, store.SyncKindFull)
	if err != nil {
		return err
	}

	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 *Service) 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 *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, 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 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, 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(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, s.userID, a.ID, string(workout.Raw)); err != nil {
				return err
			}
		}
	}

	samples := garmin.ExtractSamples(details)
	if err := s.db.ReplaceActivitySamples(ctx, s.userID, a.ID, toSampleRows(samples)); err != nil {
		return err
	}
	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)
}

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