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
}

// 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
}

// 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"`
}

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,
					}

					execErr := opts.Executor.Execute(ctx, job)

					itemRes := ItemResult{
						Schedule:  s.Name,
						Symbol:    symbol,
						Provider:  s.Provider,
						Timeframe: s.Timeframe,
					}

					if execErr != nil {
						itemRes.Status = "failed"
						itemRes.Readiness = "error"
						itemRes.Freshness = "stale"
						itemRes.Error = execErr.Error()
					} else {
						itemRes.Status = "succeeded"
						itemRes.Readiness = "ready"
						itemRes.Freshness = "fresh"
					}

					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
}