alt/services/worker/internal/scheduler/runner.go

package scheduler

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"sort"
	"sync"
	"time"

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

// Executor represents the job execution engine.
type Executor interface {
	Execute(ctx context.Context, job jobs.Job) error
	// ExecuteWithResult runs the job and returns the import result if available.
	//
	// Three-value contract:
	//   - (res, true, nil)     → handler ran and returned result (has data)
	//   - (Result{}, false, nil) → no WithResultHandler registered (legacy fallback)
	//   - (Result{}, false, err) → handler ran but returned an error (DO NOT re-execute)
	//
	// The caller MUST distinguish the error case (false, err) from the
	// no-handler case (false, nil).  Re-executing the job on error loses
	// the first error and may cause duplicate provider/import side-effects.
	ExecuteWithResult(ctx context.Context, job jobs.Job) (jobs.Result, bool, error)
}

// TickOptions provides parameters for executing a tick.
type TickOptions struct {
	Now          time.Time
	Capabilities map[market.Provider]market.ProviderCapability
	Executor     Executor
}

// TickResult represents the outcome of a single scheduler tick.
type TickResult struct {
	Items []ItemResult
}

// RefreshEvidence carries actual import results from a single scheduler tick.
// This struct separates "we ran" from "we got data" so that status evidence
// is never fabricated from scheduler dispatch counts.
type RefreshEvidence struct {
	// HasBars is true when at least one item carried a non-empty import summary.
	HasBars bool
	// ImportedBars is the total number of bars actually imported across all
	// successful items. Can be zero if the provider returned no new data.
	ImportedBars int
	// MissingBars is the count of bars that were expected but not present
	// (e.g., provider returned fewer bars than expected).
	MissingBars int
	// GapBars is the count of bars with missing intermediate timestamps.
	GapBars int
	// DuplicateBars is the count of bars rejected as duplicates.
	DuplicateBars int
	// ProviderDelayDays is the estimated delay in days reported by the provider.
	ProviderDelayDays int
}

// ItemResult represents the outcome of a single symbol import within a schedule.
type ItemResult struct {
	Schedule  string `json:"schedule"`
	Symbol    string `json:"symbol"`
	Status    string `json:"status"` // "succeeded", "failed", "skipped"
	Provider  string `json:"provider"`
	Timeframe string `json:"timeframe"`
	Readiness string `json:"readiness"`
	Freshness string `json:"freshness"`
	Error     string `json:"error,omitempty"`
	// ImportSummary carries actual import bar count for this item when
	// status is "succeeded". Empty means we don't know the actual count.
	ImportSummary *RefreshEvidence `json:"import_summary,omitempty"`
}

var (
	activeJobs   = make(map[string]bool)
	activeJobsMu sync.Mutex
)

// resetActiveJobs clears the active running schedules (for testing).
func resetActiveJobs() {
	activeJobsMu.Lock()
	defer activeJobsMu.Unlock()
	activeJobs = make(map[string]bool)
}

func currentWindow(now time.Time, loc *time.Location, cadence time.Duration) time.Time {
	next := nextWindow(now, loc, cadence)
	return next.Add(-cadence)
}

// RunTick triggers the periodic import task scheduler.
func RunTick(ctx context.Context, cfg Config, opts TickOptions) (TickResult, error) {
	now := opts.Now
	if now.IsZero() {
		now = time.Now()
	}

	validation := ValidateConfig(cfg, ValidationOptions{
		Now:          now,
		Capabilities: opts.Capabilities,
	})

	var results []ItemResult
	var resultsMu sync.Mutex
	var wg sync.WaitGroup

	for i, sched := range cfg.Schedules {
		valItem := validation.Items[i]
		if valItem.Status != "valid" {
			continue
		}

		wg.Add(1)
		go func(s ScheduleConfig) {
			defer wg.Done()

			loc, err := time.LoadLocation(s.Timezone)
			if err != nil {
				return
			}
			cadence, _ := parseWindow(s.Cadence)
			backfill, _ := parseWindow(s.BackfillWindow)

			windowTime := currentWindow(now, loc, cadence)
			lockKey := fmt.Sprintf("%s:%s", s.Name, windowTime.Format(time.RFC3339))

			activeJobsMu.Lock()
			if activeJobs[lockKey] {
				activeJobsMu.Unlock()
				resultsMu.Lock()
				for _, sym := range s.Selector.Symbols {
					results = append(results, ItemResult{
						Schedule:  s.Name,
						Symbol:    sym,
						Status:    "skipped",
						Provider:  s.Provider,
						Timeframe: s.Timeframe,
						Readiness: "stale",
						Freshness: "stale",
						Error:     "duplicate window running",
					})
				}
				resultsMu.Unlock()
				return
			}
			activeJobs[lockKey] = true
			activeJobsMu.Unlock()

			defer func() {
				activeJobsMu.Lock()
				delete(activeJobs, lockKey)
				activeJobsMu.Unlock()
			}()

			sem := make(chan struct{}, s.ParallelismLimit)
			var schedWg sync.WaitGroup

			for _, sym := range s.Selector.Symbols {
				schedWg.Add(1)
				sem <- struct{}{}

				go func(symbol string) {
					defer func() {
						<-sem
						schedWg.Done()
					}()

					fromTime := windowTime.Add(-backfill)
					payload := jobs.DailyBarImportPayload{
						Provider:     s.Provider,
						SelectorKind: s.Selector.Kind,
						Market:       s.Selector.Market,
						Venue:        s.Selector.Venue,
						Name:         s.Selector.Name,
						Symbols:      []string{symbol},
						From:         fromTime.Format("20060102"),
						To:           windowTime.Format("20060102"),
						Timeframe:    s.Timeframe,
					}

					payloadJSON, err := json.Marshal(payload)
					if err != nil {
						resultsMu.Lock()
						results = append(results, ItemResult{
							Schedule:  s.Name,
							Symbol:    symbol,
							Status:    "failed",
							Provider:  s.Provider,
							Timeframe: s.Timeframe,
							Readiness: "error",
							Freshness: "stale",
							Error:     fmt.Sprintf("marshal payload: %v", err),
						})
						resultsMu.Unlock()
						return
					}

					jobID := fmt.Sprintf("sched-%s-%s-%d", s.Name, symbol, windowTime.Unix())
					job := jobs.Job{
						ID:      jobID,
						Kind:    jobs.KindImportDailyBars,
						Payload: payloadJSON,
					}

					// Call ExecuteWithResult. The three-value contract:
					//   (res, true, nil)       → handler ran with result
					//   (Result{}, false, nil)  → no WithResultHandler (legacy)
					//   (Result{}, false, err)  → handler error (DO NOT re-execute)
					itemRes := ItemResult{
						Schedule:  s.Name,
						Symbol:    symbol,
						Provider:  s.Provider,
						Timeframe: s.Timeframe,
					}

					res, hasResult, execErr := opts.Executor.ExecuteWithResult(ctx, job)
					if hasResult {
						// Executor returned evidence: success with data.
						itemRes.Status = "succeeded"
						itemRes.Readiness = "ready"
						itemRes.Freshness = "fresh"
						itemRes.ImportSummary = &RefreshEvidence{
							HasBars:           res.Bars > 0,
							ImportedBars:      res.Bars,
							MissingBars:       res.MissingBars,
							GapBars:           res.GapBars,
							DuplicateBars:     res.DuplicateBars,
							ProviderDelayDays: res.ProviderDelay,
						}
					} else if execErr != nil {
						// Handler ran but returned an error.
						// DO NOT re-execute via Execute() — this preserves the
						// original error and avoids duplicate provider/import side effects.
						itemRes.Status = "failed"
						itemRes.Readiness = "error"
						itemRes.Freshness = "stale"
						itemRes.Error = execErr.Error()
					} else {
						// No WithResultHandler registered: fall back to Execute.
						execErr2 := opts.Executor.Execute(ctx, job)
						if execErr2 != nil {
							itemRes.Status = "failed"
							itemRes.Readiness = "error"
							itemRes.Freshness = "stale"
							itemRes.Error = execErr2.Error()
						} else {
							itemRes.Status = "succeeded"
							itemRes.Readiness = "ready"
							itemRes.Freshness = "fresh"
							// No ImportSummary — ran but no evidence.
						}
					}

					resultsMu.Lock()
					results = append(results, itemRes)
					resultsMu.Unlock()
				}(sym)
			}
			schedWg.Wait()
		}(sched)
	}

	wg.Wait()

	sort.Slice(results, func(i, j int) bool {
		if results[i].Schedule != results[j].Schedule {
			return results[i].Schedule < results[j].Schedule
		}
		return results[i].Symbol < results[j].Symbol
	})

	return TickResult{Items: results}, nil
}

// WriteTickText renders stable tick result lines.
func WriteTickText(w io.Writer, result TickResult) error {
	for _, item := range result.Items {
		if item.Error != "" {
			if _, err := fmt.Fprintf(w, "schedule=%s symbol=%s status=%s provider=%s timeframe=%s readiness=%s freshness=%s error=%q\n",
				item.Schedule, item.Symbol, item.Status, item.Provider, item.Timeframe, item.Readiness, item.Freshness, item.Error); err != nil {
				return err
			}
		} else {
			if _, err := fmt.Fprintf(w, "schedule=%s symbol=%s status=%s provider=%s timeframe=%s readiness=%s freshness=%s\n",
				item.Schedule, item.Symbol, item.Status, item.Provider, item.Timeframe, item.Readiness, item.Freshness); err != nil {
				return err
			}
		}
	}
	return nil
}

// WriteTickJSONL renders one JSON object per tick item.
func WriteTickJSONL(w io.Writer, result TickResult) error {
	enc := json.NewEncoder(w)
	for _, item := range result.Items {
		if err := enc.Encode(item); err != nil {
			return err
		}
	}
	return nil
}