kriss/geniusrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: scope Service to one user per instance

Browse Source 
NewService now takes a userID, baked into the instance rather than passed
per-call -- matches internal/api's one-Service-per-logged-in-user model
(Task 13), so ClassifyActivity/Backfill/etc. keep their existing call
signatures unchanged everywhere they're already used.
This commit is contained in:
Kriss Vila
2026-07-25 17:45:41 +02:00
parent 3a1333017d
commit 69103b7a4b
2 changed files with 138 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
backend/internal/sync/service.go
View File

@@ -61,10 +61,12 @@ type Progress struct {
Total int
}
// Service is the sync orchestrator.
// 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
@@ -72,13 +74,13 @@ type Service struct {
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 {
// 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, cfg: cfg.withDefaults(), now: 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).
@@ -106,7 +108,7 @@ func (s *Service) setProgress(done, total int) {
// 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)
runID, err := s.db.StartSyncRun(ctx, s.userID, store.SyncKindBackfill)
if err != nil {
return err
}
@@ -114,10 +116,10 @@ func (s *Service) Backfill(ctx context.Context) error {
total, err := s.backfillCore(ctx)
if err != nil {
msg := err.Error()
s.db.FinishSyncRun(ctx, runID, total, &msg)
s.db.FinishSyncRun(ctx, s.userID, runID, total, &msg)
return err
}
return s.db.FinishSyncRun(ctx, runID, total, nil)
return s.db.FinishSyncRun(ctx, s.userID, runID, total, nil)
}
// backfillCore holds Backfill's actual fetch logic, without the SyncRun
@@ -126,13 +128,13 @@ func (s *Service) Backfill(ctx context.Context) error {
// whatever was fetched even when an error is also returned, matching
// Backfill's own partial-progress-on-error behavior.
func (s *Service) backfillCore(ctx context.Context) (int, error) {
profile, err := s.db.GetProfile(ctx)
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)
state, err := s.db.GetSyncState(ctx, s.userID)
if err != nil {
return 0, err
}
@@ -167,12 +169,12 @@ func (s *Service) backfillCore(ctx context.Context) (int, error) {
// 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 {
if err := s.db.UpdateSyncState(ctx, s.userID, dateStr(start), true); err != nil {
return total, err
}
break
}
if err := s.db.UpdateSyncState(ctx, dateStr(start), false); err != nil {
if err := s.db.UpdateSyncState(ctx, s.userID, dateStr(start), false); err != nil {
return total, err
}
end = start.AddDate(0, 0, -1)
@@ -181,7 +183,7 @@ func (s *Service) backfillCore(ctx context.Context) (int, error) {
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 {
if err := s.db.UpdateSyncState(ctx, s.userID, dateStr(horizon), true); err != nil {
return total, err
}
}
@@ -192,7 +194,7 @@ func (s *Service) backfillCore(ctx context.Context) (int, error) {
// 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)
runID, err := s.db.StartSyncRun(ctx, s.userID, store.SyncKindIncremental)
if err != nil {
return err
}
@@ -200,17 +202,17 @@ func (s *Service) IncrementalSync(ctx context.Context) error {
n, err := s.incrementalSyncCore(ctx)
if err != nil {
msg := err.Error()
s.db.FinishSyncRun(ctx, runID, n, &msg)
s.db.FinishSyncRun(ctx, s.userID, runID, n, &msg)
return err
}
return s.db.FinishSyncRun(ctx, runID, n, nil)
return s.db.FinishSyncRun(ctx, s.userID, runID, n, nil)
}
// incrementalSyncCore holds IncrementalSync's actual fetch logic, without
// the SyncRun bookkeeping -- see backfillCore.
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); err == nil && ok {
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)
}
@@ -229,7 +231,7 @@ func (s *Service) incrementalSyncCore(ctx context.Context) (int, error) {
// would silently hide however many activities Backfill fetched. Recording
// one combined run makes the reported count match the whole action.
func (s *Service) FullSync(ctx context.Context, detailFillLimit int) error {
runID, err := s.db.StartSyncRun(ctx, store.SyncKindFull)
runID, err := s.db.StartSyncRun(ctx, s.userID, store.SyncKindFull)
if err != nil {
return err
}
@@ -237,7 +239,7 @@ func (s *Service) FullSync(ctx context.Context, detailFillLimit int) error {
backfillCount, err := s.backfillCore(ctx)
if err != nil {
msg := err.Error()
s.db.FinishSyncRun(ctx, runID, backfillCount, &msg)
s.db.FinishSyncRun(ctx, s.userID, runID, backfillCount, &msg)
return err
}
@@ -245,17 +247,17 @@ func (s *Service) FullSync(ctx context.Context, detailFillLimit int) error {
total := backfillCount + incrementalCount
if err != nil {
msg := err.Error()
s.db.FinishSyncRun(ctx, runID, total, &msg)
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, runID, total, &msg)
s.db.FinishSyncRun(ctx, s.userID, runID, total, &msg)
return err
}
return s.db.FinishSyncRun(ctx, runID, total, nil)
return s.db.FinishSyncRun(ctx, s.userID, runID, total, nil)
}
// ResetAll deletes every synced activity (and its laps/samples/kind
@@ -263,7 +265,7 @@ func (s *Service) FullSync(ctx context.Context, detailFillLimit int) error {
// 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)
return s.db.ResetAllSyncedData(ctx, s.userID)
}
// fetchAndStoreWindow returns two counts: rawCount is every activity Garmin's
@@ -290,11 +292,11 @@ func (s *Service) fetchAndStoreWindow(ctx context.Context, startDate, endDate st
if !isRunningActivityType(a.ActivityType.TypeKey) {
continue
}
exists, err := s.db.ActivityExists(ctx, a.ActivityID)
exists, err := s.db.ActivityExists(ctx, s.userID, a.ActivityID)
if err != nil {
return 0, 0, err
}
if _, err := s.db.UpsertActivity(ctx, toActivityRow(a)); err != nil {
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 {
@@ -309,11 +311,11 @@ func (s *Service) fetchAndStoreWindow(ctx context.Context, startDate, endDate st
// 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)
pending, err := s.db.ActivitiesMissingDetails(ctx, s.userID, limit)
if err != nil {
return err
}
profile, err := s.db.GetProfile(ctx)
profile, err := s.db.GetProfile(ctx, s.userID)
if err != nil {
return fmt.Errorf("load profile: %w", err)
}
@@ -357,41 +359,41 @@ func (s *Service) fillActivityDetails(ctx context.Context, a store.Activity, pro
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, a.ID, string(workout.Raw)); err != nil {
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, a.ID, toSampleRows(samples)); err != nil {
if err := s.db.ReplaceActivitySamples(ctx, s.userID, a.ID, toSampleRows(samples)); err != nil {
return err
}
if err := s.db.ReplaceLaps(ctx, a.ID, toLapRows(splits.Laps, samples, targets, profile)); err != nil {
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, a.ID, string(details.Raw)); err != nil {
if err := s.db.SetActivityDetails(ctx, s.userID, a.ID, string(details.Raw)); err != nil {
return err
}
return s.db.SetActivitySplitsFetched(ctx, a.ID)
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, activityID)
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, activityID)
laps, err := s.db.LapsForActivity(ctx, s.userID, activityID)
if err != nil {
return err
}
kindRows, err := s.db.ListWorkoutKinds(ctx, true)
kindRows, err := s.db.ListWorkoutKinds(ctx, s.userID, true)
if err != nil {
return err
}
@@ -399,7 +401,7 @@ func (s *Service) ClassifyActivity(ctx context.Context, activityID int64) error
if err != nil {
return fmt.Errorf("parse workout kind rules: %w", err)
}
profile, err := s.db.GetProfile(ctx)
profile, err := s.db.GetProfile(ctx, s.userID)
if err != nil {
return fmt.Errorf("load profile: %w", err)
}
@@ -416,7 +418,7 @@ func (s *Service) ClassifyActivity(ctx context.Context, activityID int64) error
return err
}
_, err = s.db.InsertKindAssignment(ctx, store.KindAssignment{
_, err = s.db.InsertKindAssignment(ctx, s.userID, store.KindAssignment{
ActivityID: activityID,
WorkoutKindID: result.WorkoutKindID,
AssignmentSource: store.AssignmentSourceRuleEngine,