// Package garmin wraps a direct garminconnect subprocess (see // wrapper/wrapper.py) as a narrow Go client interface, so the rest of // geniusrun never deals with the wire protocol directly. package garmin import ( "bufio" "context" _ "embed" "encoding/json" "errors" "fmt" "io" "log/slog" "os" "os/exec" "strconv" "sync" "time" "geniusrun/backend/internal/log" ) //go:embed wrapper/wrapper.py var wrapperScript string // maxWrapperLineBytes bounds one JSON-line response from the wrapper // subprocess -- well above the default 64KB bufio.Scanner limit, since // get_activity_details responses (per-second telemetry) can be several MB. const maxWrapperLineBytes = 16 * 1024 * 1024 // Client is the interface the rest of geniusrun depends on. The real // implementation drives an embedded Python wrapper subprocess over stdio; // internal/garmin/mock provides a fake for tests and frontend-only // development. type Client interface { // Authenticate triggers Garmin login using credentials the subprocess // was started with. Spawns the subprocess on first call. Authenticate(ctx context.Context) (AuthResult, error) // CompleteMFA submits an MFA code for a login started by Authenticate. CompleteMFA(ctx context.Context, code string) (AuthResult, error) // UpdateCredentials replaces the Garmin email/password used to spawn // the subprocess, and terminates any already-running subprocess (which // would otherwise still be authenticated under the old credentials). // The next call that needs the subprocess spawns a fresh one with the // new credentials. UpdateCredentials(email, password string) // GetActivities lists activities between start and end (YYYY-MM-DD). GetActivities(ctx context.Context, startDate, endDate string, limit int) ([]Activity, error) // GetActivitySplits fetches lap/split summaries for one activity. GetActivitySplits(ctx context.Context, activityID int64) (ActivitySplits, error) // GetActivityDetails fetches raw per-second telemetry for one activity. GetActivityDetails(ctx context.Context, activityID int64) (ActivityDetails, error) // GetWorkoutByID fetches a structured workout's step-by-step plan. GetWorkoutByID(ctx context.Context, workoutID int64) (Workout, error) // Close terminates the subprocess, if running. Close() error } // ClientConfig configures how the Garmin wrapper subprocess is spawned. type ClientConfig struct { // PythonPath is the python3 interpreter to run the embedded wrapper // script with. Empty defaults to "python3" resolved via PATH. PythonPath string GarminEmail string GarminPassword string // TokenStorePath is passed as GARMIN_TOKENSTORE so the wrapper // persists/resumes a Garmin session there. TokenStorePath string } // wireRequest is one line sent to the wrapper subprocess's stdin. type wireRequest struct { ID int `json:"id"` Cmd string `json:"cmd"` Params any `json:"params,omitempty"` } // wireResponse is one line read from the wrapper subprocess's stdout. // ErrorType/Traceback carry the Python-side failure detail in the protocol // itself (see wrapper.py's dispatch), so the Go side logs one complete // structured record per failed call instead of correlating stderr noise. type wireResponse struct { ID int `json:"id"` Result json.RawMessage `json:"result,omitempty"` Error string `json:"error,omitempty"` ErrorType string `json:"error_type,omitempty"` Traceback string `json:"traceback,omitempty"` NotFound bool `json:"not_found,omitempty"` } // ErrNotFound wraps any error a Client method returns when the wrapper // reported a definitive HTTP 404 (garminconnect's own // GarminConnectNotFoundError) -- e.g. GetWorkoutByID for a workout deleted // on Garmin's side after being linked to an activity. Callers use // errors.Is(err, ErrNotFound) to distinguish this from a transient failure // worth retrying. var ErrNotFound = errors.New("garmin: resource not found") // callParams is the Params payload for a generic "call" request: dispatches // to any garminconnect.Garmin method by name. type callParams struct { Method string `json:"method"` Args map[string]any `json:"args,omitempty"` } // authResultWire is the Result payload for authenticate/complete_mfa. type authResultWire struct { Status string `json:"status"` Message string `json:"message"` } func mapAuthStatus(s string) AuthStatus { switch s { case "success": return AuthSuccess case "mfa_required": return AuthMFARequired case "failed": return AuthFailed default: return AuthUnknown } } // subprocessClient is the real Client implementation, backed by a wrapper // subprocess spoken to over newline-delimited JSON on stdio. type subprocessClient struct { cfg ClientConfig mu sync.Mutex // serializes calls; the wrapper holds a single shared Garmin client cmd *exec.Cmd stdin io.WriteCloser enc *json.Encoder scanner *bufio.Scanner scriptPath string // temp file holding the embedded wrapper.py, written once started bool nextID int stderrDone chan struct{} // closed once the stderr-copy goroutine has finished reading } // ensureStarted spawns the wrapper subprocess if it isn't already running. // Callers must hold c.mu. func (c *subprocessClient) ensureStarted(ctx context.Context) error { if c.started { return nil } if c.scriptPath == "" { f, err := os.CreateTemp("", "geniusrun-garmin-wrapper-*.py") if err != nil { return fmt.Errorf("write embedded wrapper script: %w", err) } if _, err := f.WriteString(wrapperScript); err != nil { f.Close() return fmt.Errorf("write embedded wrapper script: %w", err) } if err := f.Close(); err != nil { return fmt.Errorf("write embedded wrapper script: %w", err) } c.scriptPath = f.Name() } pythonPath := c.cfg.PythonPath if pythonPath == "" { pythonPath = "python3" } cmd := exec.Command(pythonPath, c.scriptPath) extraEnv := []string{ "GARMIN_EMAIL=" + c.cfg.GarminEmail, "GARMIN_PASSWORD=" + c.cfg.GarminPassword, "GARMIN_TOKENSTORE=" + c.cfg.TokenStorePath, "PYTHONUNBUFFERED=1", } cmd.Env = append(os.Environ(), extraEnv...) stdin, err := cmd.StdinPipe() if err != nil { return fmt.Errorf("open wrapper stdin: %w", err) } stdout, err := cmd.StdoutPipe() if err != nil { return fmt.Errorf("open wrapper stdout: %w", err) } stderr, err := cmd.StderrPipe() if err != nil { return fmt.Errorf("open wrapper stderr: %w", err) } if err := cmd.Start(); err != nil { return fmt.Errorf("spawn garmin wrapper subprocess: %w", err) } // wrapper.py logs auth/rate-limit diagnostics to stderr as JSON lines // (see its _log helper); forwardWrapperStderr re-emits them through the // application logger so the backend's output stays one JSON stream. Per // os/exec's StderrPipe docs, it's incorrect to call Wait before all // reads from the pipe have completed, so close() waits on stderrDone // before Wait-ing. c.stderrDone = make(chan struct{}) stderrDone := c.stderrDone go func() { forwardWrapperStderr(stderr) close(stderrDone) }() c.cmd = cmd c.stdin = stdin c.enc = json.NewEncoder(stdin) scanner := bufio.NewScanner(stdout) scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) c.scanner = scanner c.started = true c.nextID = 0 return nil } // roundTrip sends one request and returns its result payload, or an error // if the wrapper reported one. Callers must hold c.mu and have already // called ensureStarted. Logs exactly one "Garmin wrapper call" line // regardless of outcome (see internal/applog) -- cmd/params are always // safe to log in full here: Garmin credentials only ever reach the // subprocess via env vars at spawn time (see ensureStarted), never through // these wire params. func (c *subprocessClient) roundTrip(ctx context.Context, cmdName string, params any) (result json.RawMessage, err error) { start := time.Now() var errorType, traceback string // Python-side failure detail, set from the error response defer func() { attrs := []slog.Attr{ slog.String("cmd", cmdName), slog.Any("params", params), slog.Int64("duration_ms", time.Since(start).Milliseconds()), } level := slog.LevelInfo if result != nil { attrs = append(attrs, slog.String("result", truncate(string(result), 500))) } if err != nil { level = slog.LevelError attrs = append(attrs, slog.String("error", err.Error())) if errorType != "" { attrs = append(attrs, slog.String("error_type", errorType)) } if traceback != "" { attrs = append(attrs, slog.String("traceback", traceback)) } } applog.FromContext(ctx).LogAttrs(context.Background(), level, "Garmin wrapper call", attrs...) }() c.nextID++ id := c.nextID if err = c.enc.Encode(wireRequest{ID: id, Cmd: cmdName, Params: params}); err != nil { err = fmt.Errorf("write %s request: %w", cmdName, err) return nil, err } if !c.scanner.Scan() { if serr := c.scanner.Err(); serr != nil { err = fmt.Errorf("read %s response: %w", cmdName, serr) } else { err = fmt.Errorf("read %s response: subprocess closed its output", cmdName) } return nil, err } var resp wireResponse if uerr := json.Unmarshal(c.scanner.Bytes(), &resp); uerr != nil { err = fmt.Errorf("parse %s response: %w", cmdName, uerr) return nil, err } if resp.ID != id { err = fmt.Errorf("%s response id mismatch: got %d, want %d", cmdName, resp.ID, id) return nil, err } if resp.Error != "" { errorType, traceback = resp.ErrorType, resp.Traceback if resp.NotFound { err = fmt.Errorf("%s: %s: %w", cmdName, resp.Error, ErrNotFound) } else { err = fmt.Errorf("%s: %s", cmdName, resp.Error) } return nil, err } result = resp.Result return result, nil } func (c *subprocessClient) Authenticate(ctx context.Context) (AuthResult, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.ensureStarted(ctx); err != nil { return AuthResult{}, err } raw, err := c.roundTrip(ctx, "authenticate", nil) if err != nil { return AuthResult{}, err } var wire authResultWire if err := json.Unmarshal(raw, &wire); err != nil { return AuthResult{}, fmt.Errorf("parse authenticate result: %w", err) } return AuthResult{Status: mapAuthStatus(wire.Status), Message: wire.Message}, nil } func (c *subprocessClient) CompleteMFA(ctx context.Context, code string) (AuthResult, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.ensureStarted(ctx); err != nil { return AuthResult{}, err } raw, err := c.roundTrip(ctx, "complete_mfa", map[string]any{"code": code}) if err != nil { return AuthResult{}, err } var wire authResultWire if err := json.Unmarshal(raw, &wire); err != nil { return AuthResult{}, fmt.Errorf("parse complete_mfa result: %w", err) } return AuthResult{Status: mapAuthStatus(wire.Status), Message: wire.Message}, nil } // UpdateCredentials implements Client. func (c *subprocessClient) UpdateCredentials(email, password string) { c.mu.Lock() defer c.mu.Unlock() c.cfg.GarminEmail = email c.cfg.GarminPassword = password c.close() } // Close implements Client. func (c *subprocessClient) Close() error { c.mu.Lock() defer c.mu.Unlock() return c.close() } // close terminates the subprocess, if running. Callers must hold c.mu. func (c *subprocessClient) close() error { if !c.started { return nil } c.started = false if c.stdin != nil { c.stdin.Close() } if c.cmd != nil && c.cmd.Process != nil { c.cmd.Process.Kill() } if c.stderrDone != nil { // Killing the process closes its end of the stderr pipe, which // unblocks the copy goroutine's Read with EOF; wait for it to finish // before Wait, per os/exec's StderrPipe doc. <-c.stderrDone } var err error if c.cmd != nil { err = c.cmd.Wait() } c.cmd, c.stdin, c.enc, c.scanner, c.stderrDone = nil, nil, nil, nil, nil return err } // forwardWrapperStderr re-emits the wrapper subprocess's stderr through // the application logger. wrapper.py writes JSON lines ({"level","msg", // ...attrs} -- see its _log helper), which are decoded and re-logged at // the corresponding level with their attrs preserved. Anything that isn't // such a line (a Python startup crash before _log exists, a chatty // third-party library printing directly) is wrapped as a warn-level // "garmin wrapper stderr" record rather than passed through raw, so the // backend's combined output is JSON no matter what the subprocess does. func forwardWrapperStderr(r io.Reader) { scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) // tracebacks can exceed the default token size for scanner.Scan() { line := scanner.Text() var entry map[string]any if err := json.Unmarshal([]byte(line), &entry); err != nil || entry["msg"] == nil { slog.Warn("garmin wrapper stderr", "line", line) continue } msg, _ := entry["msg"].(string) levelName, _ := entry["level"].(string) delete(entry, "msg") delete(entry, "level") attrs := make([]slog.Attr, 0, len(entry)+1) attrs = append(attrs, slog.String("source", "garmin-wrapper")) for k, v := range entry { attrs = append(attrs, slog.Any(k, v)) } slog.LogAttrs(context.Background(), wrapperLogLevel(levelName), msg, attrs...) } } // wrapperLogLevel maps wrapper.py's level strings onto slog levels, // defaulting to info for anything unrecognized. func wrapperLogLevel(level string) slog.Level { switch level { case "debug": return slog.LevelDebug case "warn", "warning": return slog.LevelWarn case "error": return slog.LevelError default: return slog.LevelInfo } } func truncate(s string, n int) string { if len(s) <= n { return s } return s[:n] + "...(truncated)" } var _ Client = (*subprocessClient)(nil) // NewClient builds a Client. The subprocess is not spawned until the first // call that needs it (Authenticate, or any data call once authenticated). func NewClient(cfg ClientConfig) Client { return &subprocessClient{cfg: cfg} } func (c *subprocessClient) GetActivities(ctx context.Context, startDate, endDate string, limit int) ([]Activity, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.ensureStarted(ctx); err != nil { return nil, err } raw, err := c.roundTrip(ctx, "call", callParams{ Method: "get_activities_by_date", Args: map[string]any{"startdate": startDate, "enddate": endDate}, }) if err != nil { return nil, err } var rawActivities []json.RawMessage if err := json.Unmarshal(raw, &rawActivities); err != nil { return nil, fmt.Errorf("get_activities_by_date did not return a JSON array (%q): %w", truncate(string(raw), 200), err) } if limit > 0 && limit < len(rawActivities) { rawActivities = rawActivities[:limit] } activities := make([]Activity, 0, len(rawActivities)) for _, raw := range rawActivities { var a Activity if err := json.Unmarshal(raw, &a); err != nil { return nil, fmt.Errorf("parse activity: %w", err) } a.Raw = raw activities = append(activities, a) } return activities, nil } func (c *subprocessClient) GetActivitySplits(ctx context.Context, activityID int64) (ActivitySplits, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.ensureStarted(ctx); err != nil { return ActivitySplits{}, err } raw, err := c.roundTrip(ctx, "call", callParams{ Method: "get_activity_splits", Args: map[string]any{"activity_id": strconv.FormatInt(activityID, 10)}, }) if err != nil { return ActivitySplits{}, err } var envelope struct { ActivityID int64 `json:"activityId"` Laps []json.RawMessage `json:"lapDTOs"` } if err := json.Unmarshal(raw, &envelope); err != nil { return ActivitySplits{}, fmt.Errorf("get_activity_splits did not return valid JSON (%q): %w", truncate(string(raw), 200), err) } splits := ActivitySplits{ActivityID: envelope.ActivityID, Laps: make([]Lap, 0, len(envelope.Laps))} for _, raw := range envelope.Laps { var l Lap if err := json.Unmarshal(raw, &l); err != nil { return ActivitySplits{}, fmt.Errorf("parse lap: %w", err) } l.Raw = raw splits.Laps = append(splits.Laps, l) } return splits, nil } func (c *subprocessClient) GetActivityDetails(ctx context.Context, activityID int64) (ActivityDetails, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.ensureStarted(ctx); err != nil { return ActivityDetails{}, err } raw, err := c.roundTrip(ctx, "call", callParams{ Method: "get_activity_details", Args: map[string]any{"activity_id": strconv.FormatInt(activityID, 10)}, }) if err != nil { return ActivityDetails{}, err } var details ActivityDetails if err := json.Unmarshal(raw, &details); err != nil { return ActivityDetails{}, fmt.Errorf("get_activity_details did not return valid JSON (%q): %w", truncate(string(raw), 200), err) } details.Raw = json.RawMessage(raw) return details, nil } func (c *subprocessClient) GetWorkoutByID(ctx context.Context, workoutID int64) (Workout, error) { c.mu.Lock() defer c.mu.Unlock() if err := c.ensureStarted(ctx); err != nil { return Workout{}, err } raw, err := c.roundTrip(ctx, "call", callParams{ Method: "get_workout_by_id", Args: map[string]any{"workout_id": strconv.FormatInt(workoutID, 10)}, }) if err != nil { return Workout{}, err } var workout Workout if err := json.Unmarshal(raw, &workout); err != nil { return Workout{}, fmt.Errorf("get_workout_by_id did not return valid JSON (%q): %w", truncate(string(raw), 200), err) } workout.Raw = json.RawMessage(raw) return workout, nil }