package scheduler import ( "encoding/json" "fmt" "io" "strings" "time" "git.toki-labs.com/toki/alt/packages/domain/market" ) // FreshnessObservation captures the current freshness state of a scheduled // market data refresh for a given schedule, provider, selector, timeframe and // backfill window. type FreshnessObservation struct { Schedule string `json:"schedule"` Provider market.Provider `json:"provider"` Selector string `json:"selector"` Timeframe string `json:"timeframe"` Timezone string `json:"timezone"` Cadence string `json:"cadence"` BackfillWindow string `json:"backfill_window"` ExpectedFrom time.Time `json:"expected_from,omitempty"` ExpectedTo time.Time `json:"expected_to,omitempty"` LatestBarDate time.Time `json:"latest_bar_date,omitempty"` MissingDates []time.Time `json:"missing_dates,omitempty"` GapBuckets []GapBucket `json:"gap_buckets,omitempty"` DuplicateCount int `json:"duplicate_count"` ProviderDelay int `json:"provider_delay_days"` LastError string `json:"last_error,omitempty"` Status string `json:"status"` // "success", "error" } // GapBucket represents a contiguous range of missing dates. type GapBucket struct { From time.Time `json:"from"` To time.Time `json:"to"` } // BackfillDecision holds the computed backfill decision for a freshness // observation. type BackfillDecision struct { Schedule string `json:"schedule"` Status string `json:"status"` // "fresh", "stale", "error" Reason string `json:"reason"` // human readable reason BackfillFrom time.Time `json:"backfill_from,omitempty"` BackfillTo time.Time `json:"backfill_to,omitempty"` HasBackfill bool `json:"has_backfill"` Retry bool `json:"retry"` MissingCount int `json:"missing_count"` GapCount int `json:"gap_count"` ProviderDelay int `json:"provider_delay_days"` LastError string `json:"last_error,omitempty"` StaleDates []time.Time `json:"stale_dates,omitempty"` } // DecideBackfill computes a backfill decision from a freshness observation and // a backfill window duration. It distinguishes the following cases: // // - fresh: no missing dates, no gaps, no error → no backfill needed. // - stale/missing: missing dates exist → backfill window is computed from // the earliest missing date. // - stale/gap: contiguous gap buckets exist → backfill window covers the // earliest gap. // - stale/provider_delay: provider reports delay > 0 with no missing dates // → marks retry=false, no backfill. // - error: Status=="error" with LastError → preserves error state, includes // the error text. func DecideBackfill(obs FreshnessObservation, window time.Duration) BackfillDecision { dec := BackfillDecision{ Schedule: obs.Schedule, ProviderDelay: obs.ProviderDelay, LastError: obs.LastError, } // Error state: preserve and report. if obs.Status == "error" && obs.LastError != "" { dec.Status = "error" dec.Reason = fmt.Sprintf("previous operation failed: %s", obs.LastError) dec.HasBackfill = false dec.Retry = true return dec } // Collect missing dates and gap buckets. missingDates := obs.MissingDates if missingDates == nil { missingDates = []time.Time{} } gapBuckets := obs.GapBuckets if gapBuckets == nil { gapBuckets = []GapBucket{} } // Fresh case: nothing missing, no gaps, no error. if len(missingDates) == 0 && len(gapBuckets) == 0 && obs.ProviderDelay == 0 { dec.Status = "fresh" dec.Reason = "all expected dates are present" dec.HasBackfill = false dec.Retry = false return dec } // Provider delay only: missing may be due to delay, not data loss. if obs.ProviderDelay > 0 && len(missingDates) == 0 && len(gapBuckets) == 0 { dec.Status = "stale" dec.Reason = fmt.Sprintf("provider delay %d day(s); no backfill needed", obs.ProviderDelay) dec.HasBackfill = false dec.Retry = false dec.MissingCount = 0 dec.GapCount = 0 return dec } // Stale with missing dates or gaps. dec.Status = "stale" dec.MissingCount = len(missingDates) dec.GapCount = len(gapBuckets) // Compute backfill window from the earliest missing/gap date. today := obs.ExpectedTo if today.IsZero() { today = time.Now() } // Find the earliest missing date. var earliestMissing time.Time hasMissing := false if len(missingDates) > 0 { earliestMissing = missingDates[0] for _, d := range missingDates { if d.Before(earliestMissing) { earliestMissing = d } } hasMissing = true } // Find the earliest gap start and latest gap end. var earliestGapStart time.Time var latestGapEnd time.Time hasGap := len(gapBuckets) > 0 if hasGap { earliestGapStart = gapBuckets[0].From latestGapEnd = gapBuckets[0].To for _, gb := range gapBuckets { if gb.From.Before(earliestGapStart) { earliestGapStart = gb.From } if gb.To.After(latestGapEnd) { latestGapEnd = gb.To } } } // Determine the overall earliest date. hasBackfillData := hasMissing || hasGap if hasBackfillData { if hasMissing && hasGap { if earliestGapStart.Before(earliestMissing) { earliestMissing = earliestGapStart } } else if hasGap { earliestMissing = earliestGapStart } // Set backfill window. dec.BackfillFrom = earliestMissing if hasGap && latestGapEnd.After(today) { dec.BackfillTo = latestGapEnd } else { dec.BackfillTo = today } // Bound by window: clamp backfill_from to window start. windowStart := today.Add(-window) if dec.BackfillFrom.Before(windowStart) { dec.BackfillFrom = windowStart } dec.HasBackfill = true dec.Retry = true // Build reason text. var parts []string if hasMissing { parts = append(parts, fmt.Sprintf("%d missing date(s)", len(missingDates))) } if hasGap { parts = append(parts, fmt.Sprintf("%d gap bucket(s)", len(gapBuckets))) } dec.Reason = fmt.Sprintf("backfill needed: %s from %s to %s", strings.Join(parts, ", "), dec.BackfillFrom.Format("2006-01-02"), dec.BackfillTo.Format("2006-01-02"), ) // Collect all stale dates (missing + gap ranges). dec.StaleDates = append(dec.StaleDates, missingDates...) for _, gb := range gapBuckets { for d := gb.From; !d.After(gb.To); d = d.Add(24 * time.Hour) { dec.StaleDates = append(dec.StaleDates, d) } } } return dec } // WriteBackfillText renders stable operator-facing dry-run lines for backfill // decisions. // // Format: // // schedule= status= reason=<...> backfill_from= backfill_to= retry= func WriteBackfillText(w io.Writer, decisions []BackfillDecision) error { for _, dec := range decisions { backfillFrom := "NONE" if dec.HasBackfill && !dec.BackfillFrom.IsZero() { backfillFrom = dec.BackfillFrom.Format("20060102") } backfillTo := "NONE" if dec.HasBackfill && !dec.BackfillTo.IsZero() { backfillTo = dec.BackfillTo.Format("20060102") } line := fmt.Sprintf( "schedule=%s status=%s reason=%q backfill_from=%s backfill_to=%s retry=%s\n", dec.Schedule, dec.Status, dec.Reason, backfillFrom, backfillTo, fmt.Sprintf("%t", dec.Retry), ) if _, err := fmt.Fprint(w, line); err != nil { return err } } return nil } // WriteBackfillJSONL renders one JSON object per backfill decision. func WriteBackfillJSONL(w io.Writer, decisions []BackfillDecision) error { enc := json.NewEncoder(w) for _, dec := range decisions { obj := map[string]interface{}{ "schedule": dec.Schedule, "status": dec.Status, "reason": dec.Reason, "has_backfill": dec.HasBackfill, "retry": dec.Retry, "missing_count": dec.MissingCount, "gap_count": dec.GapCount, "provider_delay": dec.ProviderDelay, } if !dec.BackfillFrom.IsZero() { obj["backfill_from"] = dec.BackfillFrom.Format("2006-01-02") } if !dec.BackfillTo.IsZero() { obj["backfill_to"] = dec.BackfillTo.Format("2006-01-02") } if dec.LastError != "" { obj["last_error"] = dec.LastError } if len(dec.StaleDates) > 0 { dates := make([]string, 0, len(dec.StaleDates)) for _, d := range dec.StaleDates { dates = append(dates, d.Format("2006-01-02")) } obj["stale_dates"] = dates } if err := enc.Encode(obj); err != nil { return err } } return nil } // RefreshStatus represents the operator-facing status of a scheduled refresh. type RefreshStatus struct { Schedule string `json:"schedule"` Status string `json:"status"` // "success", "stale", "error" LastSuccess time.Time `json:"last_success"` LastError string `json:"last_error"` NextRun time.Time `json:"next_run"` ImportedBarCount int `json:"imported_bar_count"` MissingCount int `json:"missing_count"` GapCount int `json:"gap_count"` DuplicateCount int `json:"duplicate_count"` ProviderDelay int `json:"provider_delay_days"` } // MakeRefreshStatus computes a RefreshStatus from a freshness observation, a backfill decision, // and external inputs (lastSuccess, nextRun, importedBarCount). func MakeRefreshStatus(obs FreshnessObservation, dec BackfillDecision, lastSuccess time.Time, nextRun time.Time, importedBarCount int) RefreshStatus { status := "success" if obs.Status == "error" || dec.Status == "error" { status = "error" } else if dec.Status == "stale" { status = "stale" } lastErr := obs.LastError if lastErr == "" { lastErr = dec.LastError } return RefreshStatus{ Schedule: obs.Schedule, Status: status, LastSuccess: lastSuccess, LastError: lastErr, NextRun: nextRun, ImportedBarCount: importedBarCount, MissingCount: dec.MissingCount, GapCount: dec.GapCount, DuplicateCount: obs.DuplicateCount, ProviderDelay: obs.ProviderDelay, } } // WriteRefreshStatusText renders stable operator-facing text lines for refresh statuses. // Format: // // schedule= status= last_success= last_error=<...> next_run= imported_bar_count= ... func WriteRefreshStatusText(w io.Writer, statuses []RefreshStatus) error { for _, s := range statuses { lastSuccess := "NONE" if !s.LastSuccess.IsZero() { lastSuccess = s.LastSuccess.Format(time.RFC3339) } nextRun := "NONE" if !s.NextRun.IsZero() { nextRun = s.NextRun.Format(time.RFC3339) } line := fmt.Sprintf( "schedule=%s status=%s last_success=%s last_error=%q next_run=%s imported_bar_count=%d missing_count=%d gap_count=%d duplicate_count=%d provider_delay=%d\n", s.Schedule, s.Status, lastSuccess, s.LastError, nextRun, s.ImportedBarCount, s.MissingCount, s.GapCount, s.DuplicateCount, s.ProviderDelay, ) if _, err := fmt.Fprint(w, line); err != nil { return err } } return nil } // WriteRefreshStatusJSONL renders one JSON object per refresh status. func WriteRefreshStatusJSONL(w io.Writer, statuses []RefreshStatus) error { enc := json.NewEncoder(w) for _, s := range statuses { obj := map[string]interface{}{ "schedule": s.Schedule, "status": s.Status, "last_error": s.LastError, // Keep even if empty "imported_bar_count": s.ImportedBarCount, // Keep even if zero "missing_count": s.MissingCount, "gap_count": s.GapCount, "duplicate_count": s.DuplicateCount, "provider_delay": s.ProviderDelay, } if !s.LastSuccess.IsZero() { obj["last_success"] = s.LastSuccess.Format(time.RFC3339) } else { obj["last_success"] = nil } if !s.NextRun.IsZero() { obj["next_run"] = s.NextRun.Format(time.RFC3339) } else { obj["next_run"] = nil } if err := enc.Encode(obj); err != nil { return err } } return nil }