path: root/web/sync.go

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.GPXDir, 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, gpxDir 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(gpxDir, 0o755); err != nil {
		return progress, fmt.Errorf("create GPX 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, 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(gpxDir, fmt.Sprintf("activity_%d.gpx", 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(gpxDir, fmt.Sprintf("activity_%d.gpx", activity.ID))
		if err := os.WriteFile(gpxPath, gpxData, 0o644); err != nil {
			_ = os.Remove(gpxPath)
			if reportErr := skipSyncActivity(&progress, report, summary, fmt.Errorf("write GPX: %w", err)); reportErr != nil {
				return progress, reportErr
			}
			continue
		}
		if err := ride.Save(s.db, rideFromActivity(externalID, gpxPath, activity)); err != nil {
			_ = os.Remove(gpxPath)
			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 string, activity strava.Activity) ride.StoredRide {
	activityType := activity.SportType
	if activityType == "" {
		activityType = activity.Type
	}
	return ride.StoredRide{
		ExternalID:          externalID,
		GPXPath:             gpxPath,
		Name:                activity.Name,
		Type:                activityType,
		StartDate:           activity.StartDate,
		DistanceM:           activity.DistanceM,
		ExternalMovingTimeS: activity.MovingTimeS,
		ElapsedTimeS:        activity.ElapsedTimeS,
		TotalElevationGainM: activity.TotalElevationGainM,
		AverageSpeedMps:     activity.AverageSpeedMps,
	}
}

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
}