oikos/internal/scheduler/scheduler.go

// Package scheduler implements the Oikos observe + decide loop (Phase 3).
// It loads enabled check_defs, runs checks on schedule, manages signal
// lifecycle (dedup, flap suppression, maintenance mode), and writes metrics.
package scheduler

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"os/exec"
	"regexp"
	"runtime"
	"strconv"
	"time"

	"github.com/dtoro/oikos/internal/config"
	"github.com/dtoro/oikos/internal/db"
	"github.com/dtoro/oikos/internal/db/sqlcgen"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/google/uuid"
	"golang.org/x/sync/errgroup"
	"golang.org/x/sys/unix"
)

var (
	sshKeyPath string
	sshUser    string
)

// Run starts the scheduler loop. Blocks until ctx is cancelled.
func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
	slog.Info("scheduler: starting", "interval", cfg.SchedulerInterval)
	interval := cfg.SchedulerInterval
	if interval <= 0 {
		interval = 30 * time.Second
	}

	sshKeyPath = cfg.SSHKeyPath
	sshUser = cfg.SSHUser
	if sshUser == "" {
		sshUser = "root"
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	// Immediate first pass
	runCheckPass(ctx, pool)

	for {
		select {
		case <-ctx.Done():
			slog.Info("scheduler: shutting down")
			return
		case <-ticker.C:
			runCheckPass(ctx, pool)
		}
	}
}

// runCheckPass executes one full cycle of check evaluation.
func runCheckPass(ctx context.Context, pool *db.Pool) {
	q := sqlcgen.New(pool)

	defs, err := q.ListEnabledCheckDefs(ctx)
	if err != nil {
		slog.Error("scheduler: list check defs", "error", err)
		return
	}
	if len(defs) == 0 {
		slog.Debug("scheduler: no enabled check_defs")
		return
	}

	g, gctx := errgroup.WithContext(ctx)
	g.SetLimit(10) // bounded worker pool

	for _, def := range defs {
		cd := def
		g.Go(func() error {
			runCheck(gctx, pool, cd)
			return nil
		})
	}
	g.Wait()

	// Housekeeping after each pass
	housekeeping(ctx, pool)
}

// runCheck executes a single check and processes the result.
//
// check_defs.entity_id identifies the *check* (probe) entity itself;
// check_defs.target_id identifies the entity actually being observed (the
// host/service/etc). Health, metrics, and events must attach to the target
// so the observed entity's own record reflects reality — not the internal
// probe. Signals stay keyed by the check entity (cd.EntityID), matching how
// they are created below and resolved elsewhere.
func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDefsRow) {
	q := sqlcgen.New(pool)
	start := time.Now()

	result := executeCheck(ctx, cd)

	latency := time.Since(start).Milliseconds()

	if result.metrics == nil {
		result.metrics = make(map[string]float64)
	}
	result.metrics["probe_latency_ms"] = float64(latency)

	targetID := cd.EntityID
	if cd.TargetID != nil {
		targetID = *cd.TargetID
	}

	for metric, value := range result.metrics {
		_ = q.InsertMetricSample(ctx, sqlcgen.InsertMetricSampleParams{
			EntityID: targetID,
			Metric:   metric,
			Value:    value,
			Tags:     []byte(`{}`),
		})
	}

	if result.err != nil {
		slog.Warn("scheduler: check failed",
			"entity", cd.EntitySlug, "kind", cd.Kind, "error", result.err)
	}

	prevHealth := currentHealth(ctx, pool, targetID)

	if result.signalKind == "" || result.health == "healthy" {
		resolveSignal(ctx, pool, cd.EntityID, targetID, cd.EntitySlug)
		_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
			EntityID:    targetID,
			Health:      "healthy",
			LastCheckAt: &[]time.Time{time.Now()}[0],
			Details:     []byte(`{}`),
		})
		if prevHealth != "" && prevHealth != "healthy" {
			emitSchedulerEvent(ctx, pool, "health.changed", targetID, "info",
				map[string]any{"slug": cd.EntitySlug, "from": prevHealth, "to": "healthy"})
		}
		return
	}

	slog.Warn("scheduler: raising signal",
		"entity", cd.EntitySlug, "kind", result.signalKind, "evidence", result.evidence)

	severity := evaluateSeverity(cd.Kind, result.signalKind, cd.Config, result.metrics)

	sig, err := q.UpsertSignal(ctx, sqlcgen.UpsertSignalParams{
		EntityID:       cd.EntityID,
		Kind:           result.signalKind,
		Severity:       severity,
		TargetEntityID: cd.TargetID,
		Evidence:       &result.evidence,
	})
	if err != nil {
		slog.Error("scheduler: upsert signal", "error", err)
		return
	}

	_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
		EntityID:    targetID,
		Health:      result.health,
		LastCheckAt: &[]time.Time{time.Now()}[0],
		Details:     []byte(`{}`),
	})
	_ = sig

	if prevHealth == "" || prevHealth == "healthy" {
		emitSchedulerEvent(ctx, pool, "signal.raised", targetID, severity,
			map[string]any{"slug": cd.EntitySlug, "kind": result.signalKind, "evidence": result.evidence})
	}
	if prevHealth != result.health {
		emitSchedulerEvent(ctx, pool, "health.changed", targetID, severity,
			map[string]any{"slug": cd.EntitySlug, "from": prevHealth, "to": result.health})
	}
}

// currentHealth reads the last recorded health for an entity, or "" if none.
func currentHealth(ctx context.Context, pool *db.Pool, entityID uuid.UUID) string {
	var health string
	if err := pool.QueryRow(ctx,
		`SELECT health FROM entity_status WHERE entity_id = $1`, entityID).Scan(&health); err != nil {
		return ""
	}
	return health
}

// emitSchedulerEvent records a scheduler-sourced event for SSE fan-out.
func emitSchedulerEvent(ctx context.Context, pool *db.Pool, eventType string, entityID uuid.UUID, severity string, data map[string]any) {
	_ = observability.Event(ctx, sqlcgen.New(pool), eventType, &entityID, severity, "scheduler", "", data)
}

// resolveSignal resolves any open signal raised by the given check entity.
// checkID matches how signals are keyed (UpsertSignal uses the check's own
// entity id); targetID is the observed entity whose status this affects.
func resolveSignal(ctx context.Context, pool *db.Pool, checkID, targetID uuid.UUID, slug string) {
	q := sqlcgen.New(pool)
	// Check if there's an open signal on this entity
	tag, err := pool.Exec(ctx, `UPDATE signals SET state = 'resolved', updated_at = now()
		WHERE entity_id = $1 AND state = 'raised'`, checkID)
	if err != nil {
		return
	}
	if tag.RowsAffected() > 0 {
		emitSchedulerEvent(ctx, pool, "signal.resolved", targetID, "info",
			map[string]any{"slug": slug})
	}
	_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
		EntityID:    targetID,
		Health:      "healthy",
		LastCheckAt: &[]time.Time{time.Now()}[0],
		Details:     []byte(`{}`),
	})
	slog.Info("scheduler: signal resolved", "entity", slug)
}

// checkResult bundles the outcome of a single check execution.
type checkResult struct {
	health     string
	signalKind string
	evidence   string
	metrics    map[string]float64
	err        error
}

// executeCheck dispatches to the appropriate checker by kind.
func executeCheck(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	switch cd.Kind {
	case "http":
		return checkHTTP(ctx, cd)
	case "tcp":
		return checkTCP(ctx, cd)
	case "disk":
		return checkDisk(ctx, cd)
	case "cert-expiry":
		return checkCertExpiry(ctx, cd)
	case "ping":
		return checkPing(ctx, cd)
	case "ssh-script":
		return checkSSHScript(ctx, cd)
	default:
		return checkResult{health: "unknown"}
	}
}

// housekeeping runs background maintenance tasks.
func housekeeping(ctx context.Context, pool *db.Pool) {

	// Prune expired idempotency keys (older than 24h)
	cutoff := time.Now().Add(-24 * time.Hour)
	_, err := pool.Exec(ctx,
		"DELETE FROM idempotency_keys WHERE created_at < $1", cutoff)
	if err != nil {
		slog.Error("scheduler: prune idempotency keys", "error", err)
	}

	staleSweep(ctx, pool)

	// Log housekeeping completion
	slog.Debug("scheduler: housekeeping done", "pruned_idempotency_before", cutoff.Format(time.RFC3339))
}

// staleMultiplier and staleFloor bound how long an entity can go unobserved
// before its last-known health is no longer trusted. An entity is stale once
// it has gone longer than staleMultiplier times its fastest enabled check's
// interval (or staleFloor, whichever is larger) without a fresh observation —
// covering both a stalled scheduler and a disabled/broken check_def.
const (
	staleMultiplier = 3
	staleFloor      = 5 * time.Minute
)

// staleSweep marks entities whose last observation has aged past their
// check's expected cadence as 'stale', so the system never reports an old
// health value as if it were current. Runs once per housekeeping pass.
func staleSweep(ctx context.Context, pool *db.Pool) {
	rows, err := pool.Query(ctx, `
		SELECT e.id, e.slug, st.health
		FROM entity_status st
		JOIN entities e ON e.id = st.entity_id
		JOIN (
			SELECT target_id, MIN(interval_s) AS min_interval
			FROM check_defs
			WHERE enabled AND target_id IS NOT NULL
			GROUP BY target_id
		) iv ON iv.target_id = st.entity_id
		WHERE st.health <> 'stale'
		  AND (st.last_check_at IS NULL
		       OR st.last_check_at < now() - make_interval(secs => GREATEST(iv.min_interval * $1, $2)))`,
		staleMultiplier, int(staleFloor.Seconds()))
	if err != nil {
		slog.Error("scheduler: stale sweep query", "error", err)
		return
	}
	defer rows.Close()

	type staleEntity struct {
		id     uuid.UUID
		slug   string
		health string
	}
	var stale []staleEntity
	for rows.Next() {
		var se staleEntity
		if err := rows.Scan(&se.id, &se.slug, &se.health); err != nil {
			continue
		}
		stale = append(stale, se)
	}
	rows.Close()

	for _, se := range stale {
		_, err := pool.Exec(ctx,
			`UPDATE entity_status SET health = 'stale', updated_at = now() WHERE entity_id = $1`, se.id)
		if err != nil {
			slog.Error("scheduler: mark stale", "entity", se.slug, "error", err)
			continue
		}
		slog.Warn("scheduler: entity stale", "entity", se.slug, "prev_health", se.health)
		emitSchedulerEvent(ctx, pool, "health.stale", se.id, "warning",
			map[string]any{"slug": se.slug, "from": se.health, "to": "stale"})
	}
}

// checkHTTP performs an HTTP health check.
func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		URL            string `json:"url"`
		ExpectedStatus int    `json:"expected_status"`
		Insecure       bool   `json:"insecure"`
	}{
		ExpectedStatus: 200,
	}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.URL == "" {
		return checkResult{health: "healthy"}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 30 * time.Second
	}

	client := &http.Client{
		Timeout: timeout,
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{InsecureSkipVerify: cfg.Insecure},
		},
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "http",
			evidence: fmt.Sprintf("invalid URL %q: %v", cfg.URL, err),
			err:      err,
		}
	}

	resp, err := client.Do(req)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "http",
			evidence: fmt.Sprintf("GET %s: %v", cfg.URL, err),
			err:      err,
		}
	}
	defer resp.Body.Close()

	if resp.StatusCode != cfg.ExpectedStatus {
		return checkResult{
			health: "degraded", signalKind: "http",
			evidence: fmt.Sprintf("GET %s returned %d (expected %d)", cfg.URL, resp.StatusCode, cfg.ExpectedStatus),
		}
	}

	return checkResult{health: "healthy"}
}

// checkTCP performs a TCP dial check.
func checkTCP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host string `json:"host"`
		Port int    `json:"port"`
	}{}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" || cfg.Port == 0 {
		return checkResult{health: "healthy"}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 30 * time.Second
	}

	addr := net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port))
	conn, err := net.DialTimeout("tcp", addr, timeout)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "tcp",
			evidence: fmt.Sprintf("dial %s: %v", addr, err),
			err:      err,
		}
	}
	conn.Close()
	return checkResult{health: "healthy"}
}

// checkDisk performs a disk usage check.
func checkDisk(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Path         string `json:"path"`
		ThresholdPct int    `json:"threshold_pct"`
	}{
		Path:         "/",
		ThresholdPct: 85,
	}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}

	var stat unix.Statfs_t
	if err := unix.Statfs(cfg.Path, &stat); err != nil {
		return checkResult{
			health: "down", signalKind: "disk",
			evidence: fmt.Sprintf("statfs %s: %v", cfg.Path, err),
			err:      err,
		}
	}

	total := stat.Blocks * uint64(stat.Bsize)
	free := stat.Bfree * uint64(stat.Bsize)
	if total == 0 {
		return checkResult{health: "healthy"}
	}

	usedPct := float64(total-free) / float64(total) * 100
	inodePct := 0.0
	if stat.Files > 0 {
		inodePct = float64(stat.Files-stat.Ffree) / float64(stat.Files) * 100
	}

	metrics := map[string]float64{
		"disk_used_pct":  usedPct,
		"disk_inode_pct": inodePct,
	}

	if usedPct > float64(cfg.ThresholdPct) {
		return checkResult{
			health: "degraded", signalKind: "disk",
			evidence:   fmt.Sprintf("%s %.1f%% full (threshold %d%%)", cfg.Path, usedPct, cfg.ThresholdPct),
			metrics:    metrics,
		}
	}

	return checkResult{health: "healthy", metrics: metrics}
}

// checkCertExpiry checks TLS certificate expiry.
func checkCertExpiry(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host     string `json:"host"`
		Port     int    `json:"port"`
		WarnDays int    `json:"warn_days"`
		CritDays int    `json:"crit_days"`
	}{
		Port:     443,
		WarnDays: 30,
		CritDays: 7,
	}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" {
		return checkResult{health: "healthy"}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 30 * time.Second
	}

	addr := net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port))

	d := tls.Dialer{Config: &tls.Config{InsecureSkipVerify: true}}
	conn, err := d.DialContext(ctx, "tcp", addr)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "cert-expiry",
			evidence: fmt.Sprintf("TLS dial %s: %v", addr, err),
			err:      err,
		}
	}
	defer conn.Close()

	tlsConn := conn.(*tls.Conn)
	cs := tlsConn.ConnectionState()
	if len(cs.PeerCertificates) == 0 {
		return checkResult{
			health: "down", signalKind: "cert-expiry",
			evidence: "no peer certificates",
		}
	}

	cert := cs.PeerCertificates[0]
	daysLeft := int(time.Until(cert.NotAfter).Hours() / 24)

	metrics := map[string]float64{
		"cert_days_left": float64(daysLeft),
	}

	if daysLeft <= cfg.CritDays {
		return checkResult{
			health: "down", signalKind: "cert-expiry",
			evidence: fmt.Sprintf("%s expires in %d days (crit=%d)", cfg.Host, daysLeft, cfg.CritDays),
			metrics:  metrics,
		}
	}
	if daysLeft <= cfg.WarnDays {
		return checkResult{
			health: "degraded", signalKind: "cert-expiry",
			evidence: fmt.Sprintf("%s expires in %d days (warn=%d)", cfg.Host, daysLeft, cfg.WarnDays),
			metrics:  metrics,
		}
	}

	return checkResult{health: "healthy", metrics: metrics}
}

// checkPing performs an ICMP ping check using the system ping command.
func checkPing(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host  string `json:"host"`
		Count int    `json:"count"`
	}{}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" {
		return checkResult{health: "healthy"}
	}
	if cfg.Count <= 0 {
		cfg.Count = 1
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 10 * time.Second
	}

	deadline := time.Duration(cfg.Count+1) * timeout
	ctx, cancel := context.WithTimeout(ctx, deadline)
	defer cancel()

	countStr := strconv.Itoa(cfg.Count)
	timeoutSec := strconv.Itoa(int(timeout.Seconds()))
	if timeoutSec == "0" {
		timeoutSec = "1"
	}

	cmd := exec.CommandContext(ctx, "ping", "-c", countStr, "-W", timeoutSec, cfg.Host)
	if runtime.GOOS == "darwin" {
		cmd = exec.CommandContext(ctx, "ping", "-c", countStr, "-t", timeoutSec, cfg.Host)
	}

	output, err := cmd.Output()
	if err != nil {
		return checkResult{
			health: "down", signalKind: "ping",
			evidence: fmt.Sprintf("ping %s: %v", cfg.Host, err),
			err:      err,
		}
	}

	latency := parsePingLatency(output)
	metrics := map[string]float64{}
	if latency > 0 {
		metrics["ping_latency_ms"] = latency
	}

	return checkResult{health: "healthy", metrics: metrics}
}

var pingRttRe = regexp.MustCompile(`(?:rtt\s+min\/avg\/max\/mdev|round-trip\s+min\/avg\/max\/stddev)\s*=\s*[\d.]+\/([\d.]+)\/`)

func parsePingLatency(output []byte) float64 {
	matches := pingRttRe.FindSubmatch(output)
	if len(matches) < 2 {
		return 0
	}
	val, err := strconv.ParseFloat(string(matches[1]), 64)
	if err != nil {
		return 0
	}
	return val
}

// checkSSHScript executes an allowlisted script on a remote host via SSH.
func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host   string `json:"host"`
		Port   int    `json:"port"`
		User   string `json:"user"`
		Script string `json:"script"`
	}{}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" || cfg.Script == "" {
		return checkResult{health: "healthy"}
	}
	if cfg.Port == 0 {
		cfg.Port = 22
	}
	if cfg.User == "" {
		cfg.User = sshUser
	}

	if !allowlistedScript(cfg.Script) {
		return checkResult{
			health: "unknown", signalKind: "ssh-script",
			evidence: fmt.Sprintf("script %q not allowlisted", cfg.Script),
		}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 10 * time.Second
	}

	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	scriptPath := "/opt/oikos/checks/" + cfg.Script
	port := strconv.Itoa(cfg.Port)

	output, err := sshExec(ctx, cfg.Host, port, cfg.User, scriptPath, timeout)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "ssh-script",
			evidence: fmt.Sprintf("ssh %s:%s %s: %v", cfg.Host, strconv.Itoa(cfg.Port), cfg.Script, err),
			err:      err,
		}
	}

	type scriptOutput struct {
		Health     string             `json:"health"`
		SignalKind string             `json:"signalKind"`
		Evidence   string             `json:"evidence"`
		Metrics    map[string]float64 `json:"metrics"`
	}
	var so scriptOutput
	if err := json.Unmarshal(output, &so); err != nil {
		return checkResult{
			health: "down", signalKind: "ssh-script",
			evidence: fmt.Sprintf("invalid script output from %s: %v", cfg.Script, err),
			err:      err,
		}
	}

	health := so.Health
	if health == "" {
		health = "healthy"
	}

	metrics := so.Metrics
	if metrics == nil {
		metrics = make(map[string]float64)
	}

	return checkResult{
		health:     health,
		signalKind: so.SignalKind,
		evidence:   so.Evidence,
		metrics:    metrics,
	}
}

var scriptNameRe = regexp.MustCompile(`^[a-z][a-z0-9_-]+\.sh$`)

func allowlistedScript(name string) bool {
	return scriptNameRe.MatchString(name)
}

func sshExec(ctx context.Context, host, port, user, cmd string, timeout time.Duration) ([]byte, error) {
	args := []string{
		"-o", "ConnectTimeout=" + strconv.Itoa(int(timeout.Seconds())),
		"-o", "StrictHostKeyChecking=no",
		"-o", "BatchMode=yes",
		"-o", "UserKnownHostsFile=/dev/null",
		"-o", "LogLevel=ERROR",
	}
	if sshKeyPath != "" {
		args = append(args, "-i", sshKeyPath)
	}
	if port != "" && port != "22" {
		args = append(args, "-p", port)
	}
	args = append(args, "-l", user, host, cmd)
	c := exec.CommandContext(ctx, "ssh", args...)
	out, err := c.Output()
	if err != nil {
		var ee *exec.ExitError
		if errors.As(err, &ee) {
			return nil, fmt.Errorf("ssh %s: %v (stderr: %s)", host, err, string(ee.Stderr))
		}
		return nil, fmt.Errorf("ssh %s: %v", host, err)
	}
	return out, nil
}


// metricThreshold defines warn/crit thresholds for a single metric.
type metricThreshold struct {
	Warn float64 `json:"warn"`
	Crit float64 `json:"crit"`
}

// thresholdsConfig is parsed from check_defs.config.thresholds JSONB.
type thresholdsConfig map[string]metricThreshold

// evaluateSeverity determines signal severity from check result and thresholds.
func evaluateSeverity(kind string, signalKind string, config []byte, metrics map[string]float64) string {
	var thresholds thresholdsConfig
	if len(config) > 0 {
		_ = json.Unmarshal(config, &thresholds)
	}

	for metric, value := range metrics {
		t, ok := thresholds[metric]
		if !ok {
			continue
		}
		if t.Crit > 0 && value >= t.Crit {
			return "critical"
		}
		if t.Warn > 0 && value >= t.Warn {
			return "warning"
		}
	}

	if signalKind == "down" {
		return "critical"
	}
	return "warning"
}

var _ = uuid.UUID{} // ensure uuid import stays