api: build per-user garmin.Client/sync.Service via lazy caching
Server no longer holds one fixed Garmin/Sync pair -- garminFor/syncFor build and cache one instance per user, keyed off their own profile's Garmin credentials and a per-user token-store subdirectory. The background incremental sync loop now iterates every provisioned user each tick instead of syncing one global account.
This commit is contained in:
@@ -5,8 +5,11 @@ package api
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"sync"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
@@ -17,23 +20,125 @@ import (
|
||||
appsync "geniusrun/backend/internal/sync"
|
||||
)
|
||||
|
||||
// Server wires the HTTP handlers to the app's dependencies.
|
||||
// Server wires the HTTP handlers to the app's dependencies. garmin.Client
|
||||
// and sync.Service are per-user (each user might have their own Garmin
|
||||
// account), built lazily on first use via GarminFactory and cached.
|
||||
type Server struct {
|
||||
DB *store.DB
|
||||
Garmin garmin.Client
|
||||
Sync *appsync.Service
|
||||
Auth auth.Verifier
|
||||
Session SessionConfig
|
||||
|
||||
mu sync.Mutex
|
||||
authStatus garmin.AuthStatus
|
||||
authMessage string
|
||||
syncRunning bool
|
||||
// GarminFactory builds a real (or fake, in tests) garmin.Client from a
|
||||
// fully-resolved per-user Config. Production wiring passes
|
||||
// garmin.NewClient; tests inject a factory returning a shared
|
||||
// *mock.Client (see newTestServer in api_test.go).
|
||||
GarminFactory func(garmin.Config) garmin.Client
|
||||
// GarminBase holds the plumbing shared by every user's garmin.Config
|
||||
// (subprocess paths + the token-store root directory); only
|
||||
// GarminEmail/GarminPassword/TokenStorePath vary per user, filled in by
|
||||
// garminFor.
|
||||
GarminBase garmin.Config
|
||||
SyncConfig appsync.Config
|
||||
|
||||
mu sync.Mutex
|
||||
userGarmin map[int64]garmin.Client
|
||||
userSync map[int64]*appsync.Service
|
||||
userAuthStatus map[int64]garmin.AuthStatus
|
||||
userAuthMessage map[int64]string
|
||||
userSyncRunning map[int64]bool
|
||||
}
|
||||
|
||||
// NewServer builds a Server.
|
||||
func NewServer(db *store.DB, g garmin.Client, s *appsync.Service, authVerifier auth.Verifier, session SessionConfig) *Server {
|
||||
return &Server{DB: db, Garmin: g, Sync: s, Auth: authVerifier, Session: session}
|
||||
func NewServer(db *store.DB, garminFactory func(garmin.Config) garmin.Client, garminBase garmin.Config, syncConfig appsync.Config, authVerifier auth.Verifier, session SessionConfig) *Server {
|
||||
return &Server{
|
||||
DB: db, GarminFactory: garminFactory, GarminBase: garminBase, SyncConfig: syncConfig,
|
||||
Auth: authVerifier, Session: session,
|
||||
userGarmin: map[int64]garmin.Client{},
|
||||
userSync: map[int64]*appsync.Service{},
|
||||
userAuthStatus: map[int64]garmin.AuthStatus{},
|
||||
userAuthMessage: map[int64]string{},
|
||||
userSyncRunning: map[int64]bool{},
|
||||
}
|
||||
}
|
||||
|
||||
// garminFor returns userID's garmin.Client, building and caching it (from
|
||||
// userID's own profile row) on first use.
|
||||
func (s *Server) garminFor(ctx context.Context, userID int64) (garmin.Client, error) {
|
||||
s.mu.Lock()
|
||||
if c, ok := s.userGarmin[userID]; ok {
|
||||
s.mu.Unlock()
|
||||
return c, nil
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
profile, err := s.DB.GetProfile(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load profile for garmin client (user %d): %w", userID, err)
|
||||
}
|
||||
cfg := s.GarminBase
|
||||
cfg.GarminEmail = profile.GarminEmail
|
||||
cfg.GarminPassword = profile.GarminPassword
|
||||
if cfg.TokenStorePath != "" {
|
||||
cfg.TokenStorePath = filepath.Join(cfg.TokenStorePath, strconv.FormatInt(userID, 10))
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if c, ok := s.userGarmin[userID]; ok {
|
||||
return c, nil // built concurrently by another request between our unlock and re-lock
|
||||
}
|
||||
client := s.GarminFactory(cfg)
|
||||
s.userGarmin[userID] = client
|
||||
return client, nil
|
||||
}
|
||||
|
||||
// syncFor returns userID's sync.Service, building and caching it on first use.
|
||||
func (s *Server) syncFor(ctx context.Context, userID int64) (*appsync.Service, error) {
|
||||
s.mu.Lock()
|
||||
if svc, ok := s.userSync[userID]; ok {
|
||||
s.mu.Unlock()
|
||||
return svc, nil
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
client, err := s.garminFor(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if svc, ok := s.userSync[userID]; ok {
|
||||
return svc, nil
|
||||
}
|
||||
svc := appsync.NewService(client, s.DB, userID, s.SyncConfig, nil)
|
||||
s.userSync[userID] = svc
|
||||
return svc, nil
|
||||
}
|
||||
|
||||
// RunIncrementalSyncForAllUsers is called on a timer (see main.go) to sync
|
||||
// every provisioned user in turn, replacing the old single-global-Service
|
||||
// background loop.
|
||||
func (s *Server) RunIncrementalSyncForAllUsers(ctx context.Context) {
|
||||
users, err := s.DB.ListUsers(ctx)
|
||||
if err != nil {
|
||||
log.Printf("api: list users for incremental sync: %v", err)
|
||||
return
|
||||
}
|
||||
for _, u := range users {
|
||||
svc, err := s.syncFor(ctx, u.ID)
|
||||
if err != nil {
|
||||
log.Printf("api: sync service for user %d: %v", u.ID, err)
|
||||
continue
|
||||
}
|
||||
if err := svc.IncrementalSync(ctx); err != nil {
|
||||
log.Printf("api: incremental sync for user %d: %v", u.ID, err)
|
||||
continue
|
||||
}
|
||||
if err := svc.FillPendingDetails(ctx, 50); err != nil {
|
||||
log.Printf("api: fill pending details for user %d: %v", u.ID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Router builds the HTTP routes.
|
||||
@@ -56,45 +161,49 @@ func (s *Server) Router() http.Handler {
|
||||
r.Post("/session/logout", s.handleSessionLogout)
|
||||
r.Post("/setup", s.handleSetup)
|
||||
|
||||
r.Route("/profile", func(r chi.Router) {
|
||||
r.Get("/", s.handleGetProfile)
|
||||
r.Put("/", s.handleUpdateProfile)
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Use(requireProvisionedUser)
|
||||
|
||||
r.Route("/profile", func(r chi.Router) {
|
||||
r.Get("/", s.handleGetProfile)
|
||||
r.Put("/", s.handleUpdateProfile)
|
||||
})
|
||||
|
||||
r.Route("/auth", func(r chi.Router) {
|
||||
r.Post("/login", s.handleAuthLogin)
|
||||
r.Post("/mfa", s.handleAuthMFA)
|
||||
r.Get("/status", s.handleAuthStatus)
|
||||
})
|
||||
|
||||
r.Route("/sync", func(r chi.Router) {
|
||||
r.Post("/run", s.handleSyncRun)
|
||||
r.Post("/reset", s.handleSyncReset)
|
||||
r.Get("/runs", s.handleSyncRuns)
|
||||
r.Get("/status", s.handleSyncStatus)
|
||||
})
|
||||
|
||||
r.Route("/activities", func(r chi.Router) {
|
||||
r.Get("/", s.handleListActivities)
|
||||
r.Get("/{id}", s.handleGetActivity)
|
||||
})
|
||||
|
||||
r.Route("/workout-kinds", func(r chi.Router) {
|
||||
r.Get("/", s.handleListWorkoutKinds)
|
||||
r.Get("/{id}", s.handleGetWorkoutKind)
|
||||
r.Put("/{id}", s.handleUpdateWorkoutKind)
|
||||
})
|
||||
|
||||
r.Post("/reclassify", s.handleReclassifyAll)
|
||||
|
||||
r.Route("/review-queue", func(r chi.Router) {
|
||||
r.Get("/", s.handleReviewQueue)
|
||||
r.Post("/{activityID}/resolve", s.handleResolveReview)
|
||||
r.Post("/{activityID}/unlock", s.handleUnlockReview)
|
||||
r.Post("/{activityID}/unassign", s.handleUnassignReview)
|
||||
})
|
||||
|
||||
r.Get("/progression/{kindID}", s.handleProgression)
|
||||
})
|
||||
|
||||
r.Route("/auth", func(r chi.Router) {
|
||||
r.Post("/login", s.handleAuthLogin)
|
||||
r.Post("/mfa", s.handleAuthMFA)
|
||||
r.Get("/status", s.handleAuthStatus)
|
||||
})
|
||||
|
||||
r.Route("/sync", func(r chi.Router) {
|
||||
r.Post("/run", s.handleSyncRun)
|
||||
r.Post("/reset", s.handleSyncReset)
|
||||
r.Get("/runs", s.handleSyncRuns)
|
||||
r.Get("/status", s.handleSyncStatus)
|
||||
})
|
||||
|
||||
r.Route("/activities", func(r chi.Router) {
|
||||
r.Get("/", s.handleListActivities)
|
||||
r.Get("/{id}", s.handleGetActivity)
|
||||
})
|
||||
|
||||
r.Route("/workout-kinds", func(r chi.Router) {
|
||||
r.Get("/", s.handleListWorkoutKinds)
|
||||
r.Get("/{id}", s.handleGetWorkoutKind)
|
||||
r.Put("/{id}", s.handleUpdateWorkoutKind)
|
||||
})
|
||||
|
||||
r.Post("/reclassify", s.handleReclassifyAll)
|
||||
|
||||
r.Route("/review-queue", func(r chi.Router) {
|
||||
r.Get("/", s.handleReviewQueue)
|
||||
r.Post("/{activityID}/resolve", s.handleResolveReview)
|
||||
r.Post("/{activityID}/unlock", s.handleUnlockReview)
|
||||
r.Post("/{activityID}/unassign", s.handleUnassignReview)
|
||||
})
|
||||
|
||||
r.Get("/progression/{kindID}", s.handleProgression)
|
||||
})
|
||||
})
|
||||
return r
|
||||
@@ -137,25 +246,25 @@ func writeError(w http.ResponseWriter, status int, msg string) {
|
||||
}
|
||||
|
||||
// backgroundSync runs fn in a goroutine with a fresh context, guarded so
|
||||
// only one sync operation runs at a time. Returns false if one is already
|
||||
// in progress.
|
||||
func (s *Server) backgroundSync(fn func(ctx context.Context) error) bool {
|
||||
// only one sync operation per userID runs at a time. Returns false if one
|
||||
// is already in progress for that user.
|
||||
func (s *Server) backgroundSync(userID int64, fn func(ctx context.Context) error) bool {
|
||||
s.mu.Lock()
|
||||
if s.syncRunning {
|
||||
if s.userSyncRunning[userID] {
|
||||
s.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
s.syncRunning = true
|
||||
s.userSyncRunning[userID] = true
|
||||
s.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
s.mu.Lock()
|
||||
s.syncRunning = false
|
||||
s.userSyncRunning[userID] = false
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
if err := fn(context.Background()); err != nil {
|
||||
log.Printf("api: background sync error: %v", err)
|
||||
log.Printf("api: background sync error (user %d): %v", userID, err)
|
||||
}
|
||||
}()
|
||||
return true
|
||||
|
||||
Reference in New Issue
Block a user