core/server/health/accumulator.go

// Package health provides in-memory accumulation of provider health
// metrics with periodic flush to the database.  The accumulator is
// goroutine-safe and designed for high-throughput recording from
// concurrent completion handlers.
package health

import (
	"context"
	"log"
	"sync"
	"time"

	"git.gobha.me/xcaliber/chat-switchboard/models"
)

// ── Thresholds ──────────────────────────────

const (
	// DegradedThreshold — error rate above this = degraded.
	DegradedThreshold = 0.05 // 5%
	// DownThreshold — error rate above this = down.
	DownThreshold = 0.25 // 25%
	// FlushInterval — how often in-memory counters are flushed to DB.
	FlushInterval = 60 * time.Second
	// PruneAge — health windows older than this are deleted.
	PruneAge = 7 * 24 * time.Hour
)

// ── Store Interface ─────────────────────────

// Store is the persistence layer for health data.
type Store interface {
	// UpsertWindow merges in-memory counters into the hourly bucket row.
	UpsertWindow(ctx context.Context, w *models.ProviderHealthWindow) error

	// GetCurrentWindow returns the current hourly bucket for a provider.
	GetCurrentWindow(ctx context.Context, providerConfigID string) (*models.ProviderHealthWindow, error)

	// ListWindows returns the last N hourly buckets for a provider, newest first.
	ListWindows(ctx context.Context, providerConfigID string, hours int) ([]models.ProviderHealthWindow, error)

	// ListAllCurrent returns the current-hour bucket for every provider.
	ListAllCurrent(ctx context.Context) ([]models.ProviderHealthWindow, error)

	// Prune deletes rows older than the given time.
	Prune(ctx context.Context, before time.Time) (int64, error)
}

// ── In-Memory Bucket ────────────────────────

// bucket holds counters for one provider within one flush interval.
type bucket struct {
	providerConfigID string
	requestCount     int
	errorCount       int
	timeoutCount     int
	totalLatencyMs   int64
	maxLatencyMs     int
	lastError        string
	lastErrorAt      time.Time
}

// ── Accumulator ─────────────────────────────

// Accumulator collects health metrics in memory and periodically
// flushes them to the database.  One instance per process.
type Accumulator struct {
	mu      sync.Mutex
	buckets map[string]*bucket // keyed by provider_config_id
	store   Store
	stopCh  chan struct{}
	wg      sync.WaitGroup
}

// NewAccumulator creates and starts the background flush loop.
func NewAccumulator(store Store) *Accumulator {
	a := &Accumulator{
		buckets: make(map[string]*bucket),
		store:   store,
		stopCh:  make(chan struct{}),
	}
	a.wg.Add(1)
	go a.flushLoop()
	return a
}

// RecordSuccess records a successful provider call.
func (a *Accumulator) RecordSuccess(providerConfigID string, latencyMs int) {
	a.mu.Lock()
	defer a.mu.Unlock()
	b := a.getBucket(providerConfigID)
	b.requestCount++
	b.totalLatencyMs += int64(latencyMs)
	if latencyMs > b.maxLatencyMs {
		b.maxLatencyMs = latencyMs
	}
}

// RecordError records a failed provider call.
func (a *Accumulator) RecordError(providerConfigID string, latencyMs int, errMsg string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	b := a.getBucket(providerConfigID)
	b.requestCount++
	b.errorCount++
	b.totalLatencyMs += int64(latencyMs)
	if latencyMs > b.maxLatencyMs {
		b.maxLatencyMs = latencyMs
	}
	b.lastError = errMsg
	b.lastErrorAt = time.Now().UTC()
}

// RecordTimeout records a timed-out provider call (counted as both error and timeout).
func (a *Accumulator) RecordTimeout(providerConfigID string, latencyMs int, errMsg string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	b := a.getBucket(providerConfigID)
	b.requestCount++
	b.errorCount++
	b.timeoutCount++
	b.totalLatencyMs += int64(latencyMs)
	if latencyMs > b.maxLatencyMs {
		b.maxLatencyMs = latencyMs
	}
	b.lastError = errMsg
	b.lastErrorAt = time.Now().UTC()
}

// Stop halts the flush loop and performs a final flush.
func (a *Accumulator) Stop() {
	close(a.stopCh)
	a.wg.Wait()
}

// DeriveStatus computes the provider status from an error rate.
func DeriveStatus(errorRate float64, requestCount int) models.ProviderStatus {
	if requestCount == 0 {
		return models.StatusUnknown
	}
	if errorRate >= DownThreshold {
		return models.StatusDown
	}
	if errorRate >= DegradedThreshold {
		return models.StatusDegraded
	}
	return models.StatusHealthy
}

// ── Internal ────────────────────────────────

func (a *Accumulator) getBucket(providerConfigID string) *bucket {
	b, ok := a.buckets[providerConfigID]
	if !ok {
		b = &bucket{providerConfigID: providerConfigID}
		a.buckets[providerConfigID] = b
	}
	return b
}

func (a *Accumulator) flushLoop() {
	defer a.wg.Done()
	ticker := time.NewTicker(FlushInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			a.flush()
		case <-a.stopCh:
			a.flush() // final flush
			return
		}
	}
}

func (a *Accumulator) flush() {
	a.mu.Lock()
	if len(a.buckets) == 0 {
		a.mu.Unlock()
		return
	}
	// Swap out current buckets so we don't hold the lock during DB writes.
	snap := a.buckets
	a.buckets = make(map[string]*bucket, len(snap))
	a.mu.Unlock()

	windowStart := hourFloor(time.Now().UTC())
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	for _, b := range snap {
		if b.requestCount == 0 {
			continue
		}
		w := &models.ProviderHealthWindow{
			ProviderConfigID: b.providerConfigID,
			WindowStart:      windowStart,
			RequestCount:     b.requestCount,
			ErrorCount:       b.errorCount,
			TimeoutCount:     b.timeoutCount,
			TotalLatencyMs:   b.totalLatencyMs,
			MaxLatencyMs:     b.maxLatencyMs,
		}
		if b.lastError != "" {
			w.LastError = &b.lastError
			ts := b.lastErrorAt.Format(time.RFC3339)
			w.LastErrorAt = &ts
		}
		if err := a.store.UpsertWindow(ctx, w); err != nil {
			log.Printf("⚠ health: flush failed for provider %s: %v", b.providerConfigID, err)
		}
	}
}

// hourFloor truncates a time to the start of its hour.
func hourFloor(t time.Time) time.Time {
	return t.Truncate(time.Hour)
}