geniusrun/backend/internal/garmin/client.go

// Package garmin wraps a direct garminconnect subprocess (see
// pyscript/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/applog"
)

//go:embed pyscript/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
}

// Config configures how the Garmin wrapper subprocess is spawned.
type Config 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.
type wireResponse struct {
	ID       int             `json:"id"`
	Result   json.RawMessage `json:"result,omitempty"`
	Error    string          `json:"error,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 Config

	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. 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() {
		io.Copy(os.Stderr, 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

	applog.FromContext(ctx).Info("garmin wrapper spawning",
		"python_path", pythonPath,
	)
	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()
	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_preview", truncate(string(result), 500)))
		}
		if err != nil {
			level = slog.LevelWarn
			attrs = append(attrs, slog.String("error", err.Error()))
		}
		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 != "" {
		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
}

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 Config) 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
}