geniusrun/backend/internal/garmin/client.go

// 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"
)

//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 wrapper script: %w", err)
		}
		if _, err := f.WriteString(wrapperScript); err != nil {
			f.Close()
			return fmt.Errorf("write wrapper script: %w", err)
		}
		if err := f.Close(); err != nil {
			return fmt.Errorf("write 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 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
}

// execute 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 type=wrapper line regardless of
// outcome (the record's meaning is carried by its fields, no msg) --
// 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) execute(ctx context.Context, cmd 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("type", "wrapper"),
			slog.String("class", "garmin.subprocessClient"),
			slog.String("method", "execute"),
			slog.String("cmd", cmd),
			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))
			}
		}
		slog.Default().LogAttrs(ctx, level, "", attrs...)
	}()

	c.nextID++
	id := c.nextID

	if err = c.enc.Encode(wireRequest{ID: id, Cmd: cmd, Params: params}); err != nil {
		err = fmt.Errorf("write %s request: %w", cmd, err)
		return nil, err
	}

	if !c.scanner.Scan() {
		if serr := c.scanner.Err(); serr != nil {
			err = fmt.Errorf("read %s response: %w", cmd, serr)
		} else {
			err = fmt.Errorf("read %s response: subprocess closed its output", cmd)
		}
		return nil, err
	}

	var resp wireResponse
	if uerr := json.Unmarshal(c.scanner.Bytes(), &resp); uerr != nil {
		err = fmt.Errorf("parse %s response: %w", cmd, uerr)
		return nil, err
	}
	if resp.ID != id {
		err = fmt.Errorf("%s response id mismatch: got %d, want %d", cmd, resp.ID, id)
		return nil, err
	}
	if resp.Error != "" {
		errorType, traceback = resp.ErrorType, resp.Traceback
		if resp.NotFound {
			err = fmt.Errorf("%s: %s: %w", cmd, resp.Error, ErrNotFound)
		} else {
			err = fmt.Errorf("%s: %s", cmd, 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.execute(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.execute(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
// type=wrapper record carrying the raw "line" rather than passed through, 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("", "type", "wrapper", "class", "garmin", "method", "forwardWrapperStderr", "line", line)
			continue
		}
		msg, _ := entry["msg"].(string)
		levelName, _ := entry["level"].(string)
		delete(entry, "msg")
		delete(entry, "level")
		// class is the Python module; method arrives in the entry itself
		// (wrapper.py's _log stamps the emitting function).
		attrs := make([]slog.Attr, 0, len(entry)+2)
		attrs = append(attrs, slog.String("type", "wrapper"), slog.String("class", "wrapper"))
		for k, v := range entry {
			attrs = append(attrs, slog.Any(k, v))
		}
		slog.Default().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.execute(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.execute(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.execute(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.execute(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
}