package web import ( "encoding/json" "errors" "fmt" "log/slog" "net/http" "net/url" "os" "path/filepath" "strings" "time" "github.com/martinlehoux/biking_home/ride" "github.com/martinlehoux/biking_home/strava" "github.com/martinlehoux/kagamigo/kcore" ) func (s *Server) handleSyncForm(w http.ResponseWriter, r *http.Request) { data := SyncPageData{ From: queryOrDefault(r, "from", time.Now().AddDate(0, 0, -30).Format(dateFormat)), To: queryOrDefault(r, "to", time.Now().Format(dateFormat)), Notice: syncNotice(r), HasAuth: s.hasStravaToken(), } if _, _, err := s.stravaClient(); err != nil { data.Error = err.Error() } kcore.RenderPage(r.Context(), SyncPage(data), w) } func (s *Server) handleSync(w http.ResponseWriter, r *http.Request) { if err := r.ParseMultipartForm(10 << 20); err != nil && !errors.Is(err, http.ErrNotMultipart) { http.Error(w, "invalid form", http.StatusBadRequest) return } from, to, err := parseDateRange(r.FormValue("from"), r.FormValue("to")) if err != nil { s.renderSyncError(w, r, r.FormValue("from"), r.FormValue("to"), err) return } appConfig, err := s.loadConfig() if err != nil { s.renderSyncError(w, r, r.FormValue("from"), r.FormValue("to"), err) return } client, authorized, err := newStravaClient(appConfig) if err != nil { s.renderSyncError(w, r, r.FormValue("from"), r.FormValue("to"), err) return } if !authorized { query := url.Values{} query.Set("return_to", "/sync?from="+r.FormValue("from")+"&to="+r.FormValue("to")) http.Redirect(w, r, "/strava/login?"+query.Encode(), http.StatusFound) return } if !s.beginSync() { http.Error(w, "a Strava sync is already in progress", http.StatusConflict) return } defer s.endSync() if refreshed, err := client.RefreshIfNeeded(); err != nil { s.renderSyncError(w, r, r.FormValue("from"), r.FormValue("to"), err) return } else if refreshed { if err := s.saveStravaToken(client); err != nil { s.renderSyncError(w, r, r.FormValue("from"), r.FormValue("to"), err) return } } flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "streaming responses are not supported", http.StatusInternalServerError) return } w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Content-Type", "text/event-stream") progress, err := s.syncRides(client, appConfig.Storage.RidesDir, from, to, func(progress SyncProgress) error { enrichProgressRateLimit(&progress, client) return writeSyncEvent(w, flusher, "progress", progress) }) if err != nil { if rateErr, ok := errors.AsType[*strava.RateLimitError](err); ok { slog.Warn("Strava sync rate limited", "total", progress.Total, "imported", progress.Imported, "retryAfter", rateErr.RetryAfter, "isDaily", rateErr.IsDaily, "readDaily", fmt.Sprintf("%d/%d", rateErr.State.Read.UsageDaily, rateErr.State.Read.LimitDaily), "overallDaily", fmt.Sprintf("%d/%d", rateErr.State.Overall.UsageDaily, rateErr.State.Overall.LimitDaily)) enrichProgressWithState(&progress, rateErr.State) _ = writeSyncEvent(w, flusher, "error", syncErrorEvent{ Message: rateLimitMessage(rateErr, progress), Progress: progress, RetryAfter: formatRetryAfter(rateErr.RetryAfter), IsRateLimited: true, RateLimit: progress.RateLimit, }) return } enrichProgressRateLimit(&progress, client) _ = writeSyncEvent(w, flusher, "error", syncErrorEvent{Message: err.Error(), Progress: progress, RateLimit: progress.RateLimit}) return } enrichProgressRateLimit(&progress, client) slog.Info("Completed Strava sync", "from", from.Format(dateFormat), "to", to.Add(-24*time.Hour).Format(dateFormat), "total", progress.Total, "imported", progress.Imported, "skipped", progress.Skipped, "updated", progress.Updated, "readDaily", rateLimitLogValue(progress.RateLimit, true), "overallDaily", rateLimitLogValue(progress.RateLimit, false)) _ = writeSyncEvent(w, flusher, "complete", progress) } func (s *Server) syncRides(client syncClient, ridesDir string, from, to time.Time, report func(SyncProgress) error) (SyncProgress, error) { activities, err := client.List(from, to) if err != nil { if rateErr, ok := errors.AsType[*strava.RateLimitError](err); ok { slog.Warn("Strava sync rate limited while listing activities", "retryAfter", rateErr.RetryAfter, "isDaily", rateErr.IsDaily, "error", err) } return SyncProgress{}, err } progress := SyncProgress{Total: len(activities)} if err := reportSyncProgress(report, progress); err != nil { return progress, err } slog.Info("Started Strava sync", "from", from.Format(dateFormat), "to", to.Add(-24*time.Hour).Format(dateFormat), "total", progress.Total) if err := os.MkdirAll(ridesDir, 0o755); err != nil { return progress, fmt.Errorf("create rides directory: %w", err) } for _, summary := range activities { externalID := fmt.Sprintf("strava:%d", summary.ID) existing, exists, err := ride.GetByExternalID(s.db, externalID) if err != nil { return progress, err } if exists { if existing.Name != summary.Name { existing.Name = summary.Name if err := ride.Save(s.db, existing); err != nil { if reportErr := skipSyncActivity(&progress, report, summary, fmt.Errorf("resync metadata: %w", err)); reportErr != nil { return progress, reportErr } continue } progress.Updated++ slog.Info("Resynced Strava ride metadata", "activity", summary.ID, "name", summary.Name) } else { progress.Skipped++ } progress.Completed++ if err := reportSyncProgress(report, progress); err != nil { return progress, err } continue } activity, gpxData, fitData, err := client.Get(summary.ID) if err != nil { if rateErr, ok := errors.AsType[*strava.RateLimitError](err); ok { slog.Warn("Strava sync rate limited", "activity", summary.ID, "name", summary.Name, "retryAfter", rateErr.RetryAfter, "isDaily", rateErr.IsDaily, "completed", progress.Completed, "total", progress.Total) return progress, err } if errors.Is(err, strava.ErrNoTrackPoints) { _ = os.Remove(filepath.Join(ridesDir, fmt.Sprintf("activity_%d.gpx", summary.ID))) _ = os.Remove(filepath.Join(ridesDir, fmt.Sprintf("activity_%d.fit", summary.ID))) if saveErr := ride.Save(s.db, rideFromActivity(externalID, "", "", activity)); saveErr != nil { if reportErr := skipSyncActivity(&progress, report, summary, fmt.Errorf("save metadata: %w", saveErr)); reportErr != nil { return progress, reportErr } continue } progress.Imported++ progress.Completed++ if reportErr := reportSyncProgress(report, progress); reportErr != nil { return progress, reportErr } slog.Info("Imported Strava activity without route", "activity", activity.ID, "name", activity.Name) continue } if err := skipSyncActivity(&progress, report, summary, err); err != nil { return progress, err } continue } gpxPath := filepath.Join(ridesDir, fmt.Sprintf("activity_%d.gpx", activity.ID)) fitPath := filepath.Join(ridesDir, fmt.Sprintf("activity_%d.fit", activity.ID)) if err := os.WriteFile(gpxPath, gpxData, 0o644); err != nil { _ = os.Remove(gpxPath) _ = os.Remove(fitPath) if reportErr := skipSyncActivity(&progress, report, summary, fmt.Errorf("write GPX: %w", err)); reportErr != nil { return progress, reportErr } continue } if err := os.WriteFile(fitPath, fitData, 0o644); err != nil { _ = os.Remove(gpxPath) _ = os.Remove(fitPath) if reportErr := skipSyncActivity(&progress, report, summary, fmt.Errorf("write FIT: %w", err)); reportErr != nil { return progress, reportErr } continue } if err := ride.Save(s.db, rideFromActivity(externalID, gpxPath, fitPath, activity)); err != nil { _ = os.Remove(gpxPath) _ = os.Remove(fitPath) if reportErr := skipSyncActivity(&progress, report, summary, fmt.Errorf("save activity: %w", err)); reportErr != nil { return progress, reportErr } continue } progress.Imported++ progress.Completed++ if err := reportSyncProgress(report, progress); err != nil { return progress, err } slog.Info("Imported Strava ride", "activity", activity.ID, "name", activity.Name) } return progress, nil } func rideFromActivity(externalID, gpxPath, fitPath string, activity strava.Activity) ride.StoredRide { activityType := activity.SportType if activityType == "" { activityType = activity.Type } stored := ride.StoredRide{ ExternalID: externalID, GPXPath: gpxPath, FitPath: fitPath, Name: activity.Name, ExternalSportType: activityType, StartDate: activity.StartDate, DistanceM: activity.DistanceM, ExternalMovingTimeS: activity.MovingTimeS, ElapsedTimeS: activity.ElapsedTimeS, TotalElevationGainM: activity.TotalElevationGainM, AverageSpeedMps: activity.AverageSpeedMps, } if bikeType, ok := ride.BikeTypeForExternalSportType(activityType); ok { stored.BikeType = bikeType } return stored } func skipSyncActivity(progress *SyncProgress, report func(SyncProgress) error, activity strava.Activity, err error) error { progress.Skipped++ progress.Completed++ slog.Warn("Skipped Strava activity", "activity", activity.ID, "name", activity.Name, "error", err) return reportSyncProgress(report, *progress) } func reportSyncProgress(report func(SyncProgress) error, progress SyncProgress) error { if report == nil { return nil } return report(progress) } func (s *Server) beginSync() bool { s.syncMu.Lock() defer s.syncMu.Unlock() if s.syncActive { return false } s.syncActive = true return true } func (s *Server) endSync() { s.syncMu.Lock() s.syncActive = false s.syncMu.Unlock() } type syncErrorEvent struct { Message string `json:"message"` Progress SyncProgress `json:"progress"` RetryAfter string `json:"retryAfter,omitempty"` IsRateLimited bool `json:"isRateLimited,omitempty"` RateLimit *RateLimitSnapshot `json:"rateLimit,omitempty"` } func enrichProgressRateLimit(progress *SyncProgress, client syncClient) { if provider, ok := client.(interface{ RateLimitState() strava.RateLimitState }); ok { state := provider.RateLimitState() if isZeroRateLimit(state) { return } enrichProgressWithState(progress, state) } } func enrichProgressWithState(progress *SyncProgress, state strava.RateLimitState) { progress.RateLimit = &RateLimitSnapshot{Read: state.Read, Overall: state.Overall} } func isZeroRateLimit(state strava.RateLimitState) bool { return state.Read.Limit15 == 0 && state.Read.LimitDaily == 0 && state.Overall.Limit15 == 0 && state.Overall.LimitDaily == 0 } func rateLimitLogValue(snapshot *RateLimitSnapshot, read bool) string { if snapshot == nil { return "" } if read { if snapshot.Read.LimitDaily == 0 { return "" } return fmt.Sprintf("%d/%d", snapshot.Read.UsageDaily, snapshot.Read.LimitDaily) } if snapshot.Overall.LimitDaily == 0 { return "" } return fmt.Sprintf("%d/%d", snapshot.Overall.UsageDaily, snapshot.Overall.LimitDaily) } func rateLimitMessage(rateErr *strava.RateLimitError, progress SyncProgress) string { scope := "15-minute" if rateErr.IsDaily { scope = "daily" } retryText := "" if !rateErr.RetryAfter.IsZero() { until := max(time.Until(rateErr.RetryAfter).Truncate(time.Second), 0) retryText = fmt.Sprintf(" Retry after %s (in %s).", rateErr.RetryAfter.UTC().Format(time.RFC3339), until) if rateErr.IsDaily { retryText = fmt.Sprintf(" Daily limit resets at %s (in %s).", rateErr.RetryAfter.UTC().Format(time.RFC3339), until) } } usageText := formatRateLimitUsage(rateErr.State) if usageText != "" { usageText = " " + usageText } progressText := "" if progress.Total > 0 { remaining := progress.Total - progress.Completed progressText = fmt.Sprintf(" Progress: %d/%d completed, %d imported, %d skipped, %d updated, %d remaining.", progress.Completed, progress.Total, progress.Imported, progress.Skipped, progress.Updated, remaining) } return fmt.Sprintf("Strava %s rate limit reached.%s%s%s Re-run sync after the reset time to import remaining rides.", scope, progressText, usageText, retryText) } func formatRateLimitUsage(state strava.RateLimitState) string { var parts []string if state.Read.LimitDaily > 0 { parts = append(parts, fmt.Sprintf("Read daily %d/%d", state.Read.UsageDaily, state.Read.LimitDaily)) } if state.Read.Limit15 > 0 { parts = append(parts, fmt.Sprintf("Read 15-min %d/%d", state.Read.Usage15, state.Read.Limit15)) } if state.Overall.LimitDaily > 0 && (state.Overall.LimitDaily != state.Read.LimitDaily || state.Overall.UsageDaily != state.Read.UsageDaily) { parts = append(parts, fmt.Sprintf("Overall daily %d/%d", state.Overall.UsageDaily, state.Overall.LimitDaily)) } if len(parts) == 0 { return "" } return "Usage: " + strings.Join(parts, ", ") + "." } func formatRetryAfter(t time.Time) string { if t.IsZero() { return "" } return t.UTC().Format(time.RFC3339) } func writeSyncEvent(w http.ResponseWriter, flusher http.Flusher, event string, data any) error { payload, err := json.Marshal(data) if err != nil { return err } if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, payload); err != nil { return err } flusher.Flush() return nil } func (s *Server) renderSyncError(w http.ResponseWriter, r *http.Request, from, to string, err error) { if strings.Contains(r.Header.Get("Accept"), "text/event-stream") { http.Error(w, err.Error(), http.StatusBadRequest) return } w.WriteHeader(http.StatusBadRequest) kcore.RenderPage(r.Context(), SyncPage(SyncPageData{From: from, To: to, Error: err.Error(), HasAuth: s.hasStravaToken()}), w) } func parseDateRange(from, to string) (time.Time, time.Time, error) { start, err := time.Parse(dateFormat, from) if err != nil { return time.Time{}, time.Time{}, errors.New("a valid start date is required") } end, err := time.Parse(dateFormat, to) if err != nil { return time.Time{}, time.Time{}, errors.New("a valid end date is required") } if end.Before(start) { return time.Time{}, time.Time{}, errors.New("the end date must not be before the start date") } return start.UTC(), end.AddDate(0, 0, 1).UTC(), nil }