Dash/internal/engine/engine.go

// Package engine schedules checks, runs flap detection and groups incidents.
package engine

import (
	"context"
	"log"
	"math/rand"
	"sync"
	"time"

	"dash/internal/checker"
	"dash/internal/config"
	"dash/internal/store"
)

// State is the flap-filtered status of a target.
type State string

const (
	StateUnknown  State = "unknown"
	StateUp       State = "up"
	StateDegraded State = "degraded"
	StateDown     State = "down"
)

// CheckEvent is broadcast on every completed probe.
type CheckEvent struct {
	TargetID  string  `json:"target_id"`
	TS        int64   `json:"ts"`
	OK        bool    `json:"ok"`
	LatencyMS float64 `json:"latency_ms"`
	Detail    string  `json:"detail,omitempty"`
	Err       string  `json:"error,omitempty"`
	State     State   `json:"state"`
}

// StateEvent is broadcast when flap detection confirms a transition.
type StateEvent struct {
	TargetID  string          `json:"target_id"`
	TS        int64           `json:"ts"`
	State     State           `json:"state"`
	PrevState State           `json:"prev_state"`
	Incident  *store.Incident `json:"incident,omitempty"`
}

// Event is a named SSE payload.
type Event struct {
	Name string // "check" | "state"
	Data any
}

// Hub fans events out to SSE subscribers.
type Hub struct {
	mu   sync.Mutex
	subs map[chan Event]struct{}
}

func NewHub() *Hub {
	return &Hub{subs: map[chan Event]struct{}{}}
}

func (h *Hub) Subscribe() chan Event {
	ch := make(chan Event, 64)
	h.mu.Lock()
	h.subs[ch] = struct{}{}
	h.mu.Unlock()
	return ch
}

func (h *Hub) Unsubscribe(ch chan Event) {
	h.mu.Lock()
	delete(h.subs, ch)
	h.mu.Unlock()
}

func (h *Hub) Broadcast(e Event) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for ch := range h.subs {
		select {
		case ch <- e:
		default: // slow consumer: drop rather than block the engine
		}
	}
}

// TargetStatus is the live in-memory view of one target.
type TargetStatus struct {
	Target   config.Target
	State    State
	Last     *checker.Result
	Fails    int // consecutive failures
	Incident *store.Incident
}

// Engine owns one scheduler goroutine per target.
type Engine struct {
	cfg   *config.Config
	store *store.Store
	hub   *Hub

	mu      sync.RWMutex
	runtime map[string]*targetRuntime
}

type targetRuntime struct {
	cfg config.Target

	state       State
	last        *checker.Result
	fails       int
	oks         int
	firstFailTS int64
	incident    *store.Incident
}

func New(cfg *config.Config, st *store.Store, hub *Hub) *Engine {
	e := &Engine{cfg: cfg, store: st, hub: hub, runtime: map[string]*targetRuntime{}}
	for _, t := range cfg.Targets {
		rt := &targetRuntime{cfg: t, state: StateUnknown}
		// Resume an incident that was open when the process last stopped.
		if inc, err := st.OngoingIncident(t.ID); err == nil && inc != nil {
			rt.incident = inc
		}
		e.runtime[t.ID] = rt
	}
	return e
}

// Run blocks until ctx is cancelled.
func (e *Engine) Run(ctx context.Context) {
	var wg sync.WaitGroup
	for _, t := range e.cfg.Targets {
		wg.Add(1)
		go func(t config.Target) {
			defer wg.Done()
			e.loop(ctx, t)
		}(t)
	}

	// Daily retention pruning.
	wg.Add(1)
	go func() {
		defer wg.Done()
		tick := time.NewTicker(24 * time.Hour)
		defer tick.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-tick.C:
				cutoff := time.Now().AddDate(0, 0, -e.cfg.Settings.HistoryDays)
				if err := e.store.Prune(cutoff); err != nil {
					log.Printf("prune: %v", err)
				}
			}
		}
	}()

	wg.Wait()
}

func (e *Engine) loop(ctx context.Context, t config.Target) {
	// Jitter the first run so all targets don't fire at once.
	jitter := time.Duration(rand.Int63n(int64(2 * time.Second)))
	select {
	case <-ctx.Done():
		return
	case <-time.After(jitter):
	}

	tick := time.NewTicker(t.Interval.Duration)
	defer tick.Stop()
	for {
		e.runOnce(ctx, t)
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		}
	}
}

func (e *Engine) runOnce(ctx context.Context, t config.Target) {
	res := checker.Run(ctx, t)
	ts := res.TS.UnixMilli()

	if err := e.store.InsertCheck(t.ID, store.Check{
		TS: ts, OK: res.OK, LatencyMS: res.LatencyMS, Detail: res.Detail, Err: res.Err,
	}); err != nil {
		log.Printf("store check %s: %v", t.ID, err)
	}

	e.mu.Lock()
	rt := e.runtime[t.ID]
	prev := rt.state
	rt.last = &res

	if res.OK {
		rt.oks++
		rt.fails = 0
	} else {
		if rt.fails == 0 {
			rt.firstFailTS = ts
		}
		rt.fails++
		rt.oks = 0
	}

	var incidentEvent *store.Incident
	switch {
	case rt.state == StateDown:
		// Recovery needs RecoverThreshold consecutive successes.
		if res.OK && rt.oks >= t.RecoverThreshold {
			rt.state = stateForOK(res)
			if rt.incident != nil {
				if err := e.store.CloseIncident(rt.incident.ID, ts); err != nil {
					log.Printf("close incident %s: %v", t.ID, err)
				}
				closed := *rt.incident
				closed.EndedAt = &ts
				incidentEvent = &closed
				rt.incident = nil
			}
		}
	case !res.OK:
		if rt.fails >= t.FailThreshold || (prev == StateUnknown && rt.fails >= t.FailThreshold) {
			rt.state = StateDown
			if rt.incident == nil {
				// The incident spans the whole losing streak, starting
				// at the first failed check, not the confirming one.
				id, err := e.store.OpenIncident(t.ID, rt.firstFailTS, res.Err)
				if err != nil {
					log.Printf("open incident %s: %v", t.ID, err)
				} else {
					rt.incident = &store.Incident{
						ID: id, TargetID: t.ID, StartedAt: rt.firstFailTS, Err: res.Err,
					}
				}
			}
			incidentEvent = rt.incident
		} else {
			// Failing but not yet confirmed down: flapping.
			rt.state = StateDegraded
		}
	default: // res.OK while up/degraded/unknown
		rt.state = stateForOK(res)
		// A stale incident from a previous process run: close it.
		if rt.incident != nil && prev == StateUnknown {
			if err := e.store.CloseIncident(rt.incident.ID, ts); err != nil {
				log.Printf("close stale incident %s: %v", t.ID, err)
			}
			rt.incident = nil
		}
	}
	state := rt.state
	e.mu.Unlock()

	e.hub.Broadcast(Event{Name: "check", Data: CheckEvent{
		TargetID: t.ID, TS: ts, OK: res.OK, LatencyMS: res.LatencyMS,
		Detail: res.Detail, Err: res.Err, State: state,
	}})
	if state != prev {
		e.hub.Broadcast(Event{Name: "state", Data: StateEvent{
			TargetID: t.ID, TS: ts, State: state, PrevState: prev, Incident: incidentEvent,
		}})
	}
}

func stateForOK(res checker.Result) State {
	if res.Degraded {
		return StateDegraded
	}
	return StateUp
}

// Snapshot returns the live status of every target, in config order.
func (e *Engine) Snapshot() []TargetStatus {
	e.mu.RLock()
	defer e.mu.RUnlock()
	out := make([]TargetStatus, 0, len(e.cfg.Targets))
	for _, t := range e.cfg.Targets {
		rt := e.runtime[t.ID]
		out = append(out, TargetStatus{
			Target: rt.cfg, State: rt.state, Last: rt.last,
			Fails: rt.fails, Incident: rt.incident,
		})
	}
	return out
}