alt/services/worker/internal/jobs/backtest_jobs.go

package jobs

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"git.toki-labs.com/toki/alt/packages/domain/backtest"
	"git.toki-labs.com/toki/alt/packages/domain/market"
	"git.toki-labs.com/toki/alt/services/worker/internal/storage"
)

// BacktestExecutor defines the interface for running backtests.
type BacktestExecutor interface {
	Execute(ctx context.Context, run backtest.Run) error
}

// RunBacktestPayload is the decoded KindRunBacktest job payload.
type RunBacktestPayload struct {
	RunID      string `json:"run_id"`
	StrategyID string `json:"strategy_id"`
	Market     string `json:"market"`
	Timeframe  string `json:"timeframe"`
	From       string `json:"from"`
	To         string `json:"to"`
}

// parsePayloadDate parses standard date strings supported by the payload.
func parsePayloadDate(value string) (time.Time, error) {
	for _, layout := range []string{time.RFC3339, "2006-01-02", "20060102"} {
		if t, err := time.Parse(layout, value); err == nil {
			return t.UTC(), nil
		}
	}
	return time.Time{}, fmt.Errorf("invalid date format %q", value)
}

// DecodeRunBacktestPayload decodes and validates a KindRunBacktest payload.
func DecodeRunBacktestPayload(raw json.RawMessage) (RunBacktestPayload, error) {
	var p RunBacktestPayload
	if err := json.Unmarshal(raw, &p); err != nil {
		return RunBacktestPayload{}, fmt.Errorf("decode run backtest payload: %w", err)
	}
	if p.RunID == "" {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: run_id is required")
	}
	if p.StrategyID == "" {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: strategy_id is required")
	}
	if p.Market == "" {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: market is required")
	}
	if p.Timeframe == "" {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: timeframe is required")
	}
	if p.From == "" {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: from is required")
	}
	if p.To == "" {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: to is required")
	}

	from, err := parsePayloadDate(p.From)
	if err != nil {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: invalid from date: %w", err)
	}
	to, err := parsePayloadDate(p.To)
	if err != nil {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: invalid to date: %w", err)
	}

	if from.After(to) {
		return RunBacktestPayload{}, fmt.Errorf("run backtest payload: from date cannot be after to date")
	}

	return p, nil
}

// RegisterRunBacktestHandler registers the concrete KindRunBacktest job handler.
// It manages transition of run states: Decodes payload -> status:running -> call executor -> status:succeeded/failed/canceled.
func RegisterRunBacktestHandler(
	runner *Runner,
	store storage.BacktestRunStore,
	executor BacktestExecutor,
	now func() time.Time,
) {
	runner.Register(KindRunBacktest, func(ctx context.Context, payload json.RawMessage) error {
		p, err := DecodeRunBacktestPayload(payload)
		if err != nil {
			return err
		}

		runID := backtest.RunID(p.RunID)
		fromTime, _ := parsePayloadDate(p.From)
		toTime, _ := parsePayloadDate(p.To)

		spec := backtest.RunSpec{
			StrategyID: backtest.StrategyID(p.StrategyID),
			Market:     market.Market(p.Market),
			Timeframe:  market.Timeframe(p.Timeframe),
			From:       fromTime,
			To:         toTime,
		}

		currentTime := now().UTC()
		run, err := store.GetRun(ctx, runID)
		if err != nil {
			if errors.Is(err, storage.ErrRunNotFound) {
				// Create new run in pending status first
				run = backtest.Run{
					ID:        runID,
					Spec:      spec,
					Status:    backtest.RunStatusPending,
					CreatedAt: currentTime,
					UpdatedAt: currentTime,
				}
				if err := store.UpsertRun(ctx, run); err != nil {
					return fmt.Errorf("failed to transition run to pending: %w", err)
				}

				// Transition to running
				run.Status = backtest.RunStatusRunning
				run.UpdatedAt = currentTime
			} else {
				// Other store errors prevent execution and are returned immediately
				return fmt.Errorf("failed to get run: %w", err)
			}
		} else {
			run.Spec = spec
			run.Status = backtest.RunStatusRunning
			run.UpdatedAt = currentTime
		}

		// Record running status
		if err := store.UpsertRun(ctx, run); err != nil {
			return fmt.Errorf("failed to transition run to running: %w", err)
		}

		// Execute
		execErr := executor.Execute(ctx, run)

		// Record final status
		run.UpdatedAt = now().UTC()
		if execErr != nil {
			run.Status = backtest.RunStatusFailed
			if errors.Is(execErr, context.Canceled) {
				run.Status = backtest.RunStatusCanceled
			}
			if upsertErr := store.UpsertRun(ctx, run); upsertErr != nil {
				return fmt.Errorf("failed to transition run to %s (original error: %v): %w", run.Status, execErr, upsertErr)
			}
			return execErr
		}

		run.Status = backtest.RunStatusSucceeded
		if upsertErr := store.UpsertRun(ctx, run); upsertErr != nil {
			return fmt.Errorf("failed to transition run to succeeded: %w", upsertErr)
		}

		return nil
	})
}