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"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"time"

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

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

	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.
func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDefsRow) {
	q := sqlcgen.New(pool)
	start := time.Now()

	health, signalKind, evidence, checkErr := executeCheck(ctx, cd)

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

	// Write metric
	_ = q.InsertMetricSample(ctx, sqlcgen.InsertMetricSampleParams{
		EntityID: cd.EntityID,
		Metric:   "probe_latency_ms",
		Value:    float64(latency),
		Tags:     []byte(`{}`),
	})

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

	if signalKind == "" || health == "healthy" {
		// Recovery: resolve any open signal for this check
		resolveSignal(ctx, pool, cd.EntityID, cd.EntitySlug)
		// Update entity_status to healthy
		_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
			EntityID:    cd.EntityID,
			Health:      "healthy",
			LastCheckAt: &[]time.Time{time.Now()}[0],
			Details:     []byte(`{}`),
		})
		return
	}

	// Failure: upsert signal (dedup via partial unique index)
	slog.Warn("scheduler: raising signal",
		"entity", cd.EntitySlug, "kind", signalKind, "evidence", evidence)

	severity := "warning"
	if signalKind == "down" {
		severity = "critical"
	}

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

	// Update entity_status
	_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
		EntityID:    cd.EntityID,
		Health:      health,
		LastCheckAt: &[]time.Time{time.Now()}[0],
		Details:     []byte(`{}`),
	})
	_ = sig // used for flap detection below
}

// resolveSignal resolves any open signal for the given check entity.
func resolveSignal(ctx context.Context, pool *db.Pool, entityID uuid.UUID, slug string) {
	q := sqlcgen.New(pool)
	// Check if there's an open signal on this entity
	_, err := pool.Exec(ctx, `UPDATE signals SET state = 'resolved', updated_at = now()
		WHERE entity_id = $1 AND state = 'raised'`, entityID)
	if err != nil {
		return
	}
	_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
		EntityID:    entityID,
		Health:      "healthy",
		LastCheckAt: &[]time.Time{time.Now()}[0],
		Details:     []byte(`{}`),
	})
	slog.Info("scheduler: signal resolved", "entity", slug)
}

// executeCheck dispatches to the appropriate checker by kind.
func executeCheck(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (health string, signalKind string, evidence string, err error) {
	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)
	default:
		return "unknown", "", "", nil
	}
}

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

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

// checkHTTP performs an HTTP health check.
func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
	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 "healthy", "", "", nil
	}

	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 "down", "http", fmt.Sprintf("invalid URL %q: %v", cfg.URL, err), err
	}

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

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

	return "healthy", "", "", nil
}

// checkTCP performs a TCP dial check.
func checkTCP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
	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 "healthy", "", "", nil
	}

	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 "down", "tcp", fmt.Sprintf("dial %s: %v", addr, err), err
	}
	conn.Close()
	return "healthy", "", "", nil
}

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

	// Use unix.Statfs for disk usage.
	var stat unix.Statfs_t
	if err := unix.Statfs(cfg.Path, &stat); err != nil {
		return "down", "disk", fmt.Sprintf("statfs %s: %v", cfg.Path, err), err
	}

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

	usedPct := float64(total-free) / float64(total) * 100
	if usedPct > float64(cfg.ThresholdPct) {
		return "degraded", "disk",
			fmt.Sprintf("%s %.1f%% full (threshold %d%%)", cfg.Path, usedPct, cfg.ThresholdPct), nil
	}

	return "healthy", "", "", nil
}

// checkCertExpiry checks TLS certificate expiry.
func checkCertExpiry(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) (string, string, string, error) {
	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 "healthy", "", "", nil
	}

	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 "down", "cert-expiry", fmt.Sprintf("TLS dial %s: %v", addr, err), err
	}
	defer conn.Close()

	tlsConn := conn.(*tls.Conn)
	// Use crypto/tls ConnectionState to get verified chains
	cs := tlsConn.ConnectionState()
	if len(cs.PeerCertificates) == 0 {
		return "down", "cert-expiry", "no peer certificates", nil
	}

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

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

	return "healthy", "", "", nil
}


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