oikos/internal/actuator/actuator.go

// Package actuator executes classified actions against the fleet.
// Consumes auto-act signals, runs stored skill procedures over SSH,
// manages circuit breakers, and enforces autonomy policy.
package actuator

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/dtoro/oikos/internal/config"
	"github.com/dtoro/oikos/internal/db"
	"github.com/dtoro/oikos/internal/db/sqlcgen"
	"github.com/google/uuid"
)

// Run starts the actuator loop. Blocks until ctx is cancelled.
func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
	slog.Info("actuator: starting")
	interval := 10 * time.Second
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	circuitBreaker := newCircuitBreaker(cfg.CircuitThreshold, cfg.CircuitSeconds)

	for {
		select {
		case <-ctx.Done():
			slog.Info("actuator: shutting down")
			return
		case <-ticker.C:
			processAutoActSignals(ctx, pool, cfg, circuitBreaker)
		}
	}
}

func processAutoActSignals(ctx context.Context, pool *db.Pool, cfg config.Config, cb *circuitBreaker) {
	q := sqlcgen.New(pool)

	// Check kill-switch
	autoAct := getAutonomySetting(ctx, q, "global.auto_act")
	if autoAct == "off" || autoAct == "false" {
		slog.Debug("actuator: global auto_act disabled")
		return
	}

	signals, err := q.GetOpenSignalsForAutoAct(ctx, 5)
	if err != nil {
		slog.Error("actuator: get signals", "error", err)
		return
	}

	for _, sig := range signals {
		// Check per-target kill-switch
		slug := ""
		if sig.TargetEntityID != nil {
			var s string
			if err := pool.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", *sig.TargetEntityID).Scan(&s); err == nil {
				slug = s
			}
		}
		if slug != "" {
			ns := getAutonomySetting(ctx, q, "never_auto_act."+slug)
			if ns == "true" {
				slog.Debug("actuator: per-target auto_act disabled", "slug", slug)
				continue
			}
		}

		// Check circuit breaker
		targetKey := slug
		if targetKey == "" {
			targetKey = sig.TargetEntityID.String()
		}
		if cb.isOpen(targetKey) {
			slog.Warn("actuator: circuit open", "target", targetKey)
			continue
		}

		// Execute with advisory lock for per-target serialization
		lockKey := 0
		if sig.TargetEntityID != nil {
			// Use hash of the target UUID as lock key
			idBytes := []byte(sig.TargetEntityID.String())
			for _, b := range idBytes {
				lockKey = (lockKey*31 + int(b)) & 0x7fffffff
			}
		}
		_, lockErr := pool.Exec(ctx, "SELECT pg_advisory_xact_lock($1)", lockKey)
		if lockErr != nil {
			slog.Error("actuator: lock", "error", lockErr)
			continue
		}

		// Create execution record
		execID, _ := uuid.NewV7()
		err = q.InsertExecution(ctx, sqlcgen.InsertExecutionParams{
			EntityID:         execID,
			ClassificationID: &sig.ClassificationID,
			SignalEntityID:   &sig.EntityID,
			TargetEntityID:   sig.TargetEntityID,
			Action:           sig.Action,
			RiskClass:        sig.RiskClass,
			CorrelationID:    sig.CorrelationID,
		})
		if err != nil {
			slog.Error("actuator: insert execution", "error", err)
			continue
		}

		// Mark execution as running
		_ = q.UpdateExecutionStatus(ctx, sqlcgen.UpdateExecutionStatusParams{
			EntityID: execID,
			Status:   "running",
			Result:   []byte(`{}`),
		})

		// Execute (stub for now)
		result := map[string]any{"success": true, "message": "stub execution"}
		resultJSON, _ := json.Marshal(result)

		start := time.Now()
		duration := time.Since(start).Milliseconds()

		_ = q.UpdateExecutionStatus(ctx, sqlcgen.UpdateExecutionStatusParams{
			EntityID:   execID,
			Status:     "completed",
			Result:     resultJSON,
			DurationMs: &[]int32{int32(duration)}[0],
			Verified:   true,
		})

		// Update circuit breaker
		cb.recordSuccess(targetKey)

		slog.Info("actuator: execution complete",
			"execution", execID, "action", sig.Action, "target", targetKey)
	}
}

func getAutonomySetting(ctx context.Context, q *sqlcgen.Queries, key string) string {
	val, err := q.GetAutonomySetting(ctx, key)
	if err != nil {
		return ""
	}
	return val
}

// circuit breaker prevents repeated attempts against failing targets.
type circuitBreaker struct {
	mu         sync.Mutex
	failures   map[string]int
	cooldowns  map[string]time.Time
	threshold  int
	cooldownS  int
}

func newCircuitBreaker(threshold, cooldownSec int) *circuitBreaker {
	if threshold <= 0 { threshold = 3 }
	if cooldownSec <= 0 { cooldownSec = 300 }
	return &circuitBreaker{
		failures:  make(map[string]int),
		cooldowns: make(map[string]time.Time),
		threshold: threshold,
		cooldownS: cooldownSec,
	}
}

func (cb *circuitBreaker) isOpen(target string) bool {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	if expiry, ok := cb.cooldowns[target]; ok {
		if time.Now().Before(expiry) {
			return true
		}
		delete(cb.cooldowns, target)
		cb.failures[target] = 0
	}
	return false
}

func (cb *circuitBreaker) recordSuccess(target string) {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	cb.failures[target] = 0
}

func (cb *circuitBreaker) recordFailure(target string) {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	cb.failures[target]++
	if cb.failures[target] >= cb.threshold {
		cb.cooldowns[target] = time.Now().Add(time.Duration(cb.cooldownS) * time.Second)
		slog.Warn("actuator: circuit opened", "target", target, "cooldown_s", cb.cooldownS)
	}
}

// ─── Provisioning ─────────────────────────────────────────────────────

// ProvisionLXC creates and configures an LXC container on a Proxmox host.
// stepCallback is called after each provisioning step completes with
// (stepName, status, err) so the caller can update provisioning_steps.
func ProvisionLXC(ctx context.Context, pool *db.Pool, entityID uuid.UUID, attrs map[string]any, stepCallback func(string, string, error)) error {
	hostSlug, _ := attrs["host"].(string)
	if hostSlug == "" {
		return fmt.Errorf("missing host attribute")
	}

	host, user, err := resolveHost(ctx, pool, hostSlug)
	if err != nil {
		return fmt.Errorf("resolve host %q: %w", hostSlug, err)
	}

	vmid, _ := attrs["vmid"].(float64)
	if vmid == 0 {
		return fmt.Errorf("missing vmid attribute")
	}
	vmIDInt := int(vmid)

	cores, _ := attrs["cores"].(float64)
	ramMB, _ := attrs["ram_mb"].(float64)
	diskGB, _ := attrs["disk_gb"].(float64)
	ip, _ := attrs["ip"].(string)
	template, _ := attrs["template"].(string)
	privileged, _ := attrs["privileged"].(bool)
	if template == "" {
		template = "debian-12-standard"
	}
	if cores == 0 {
		cores = 1
	}
	if ramMB == 0 {
		ramMB = 512
	}
	if diskGB == 0 {
		diskGB = 8
	}

	privFlag := "--unprivileged 1"
	if privileged {
		privFlag = "--unprivileged 0"
	}

	// Step 1: Validate constraints.
	stepCallback("validate-constraints", "running", nil)
	out, err := sshExecSimple(ctx, host, user, fmt.Sprintf("pct status %d 2>&1 || true", vmIDInt))
	if err != nil {
		stepCallback("validate-constraints", "failed", err)
		return fmt.Errorf("check VMID: %w", err)
	}
	if !strings.Contains(out, "does not exist") && !strings.Contains(out, "not found") {
		err := fmt.Errorf("VMID %d already in use on %s", vmIDInt, hostSlug)
		stepCallback("validate-constraints", "failed", err)
		return err
	}
	stepCallback("validate-constraints", "ok", nil)

	// Step 2: Create container.
	stepCallback("create-container", "running", nil)
	templatePath := fmt.Sprintf("/var/lib/vz/template/cache/%s.tar.zst", template)
	createCmd := fmt.Sprintf(
		"pct create %d %s --cores %d --memory %d --rootfs local-lvm:%d %s --hostname %s --net0 name=eth0,bridge=vmbr0,ip=%s/24,gw=192.168.8.2 --start 1",
		vmIDInt, templatePath, int(cores), int(ramMB), int(diskGB), privFlag, attrs["name"], ip)
	out, err = sshExecSimple(ctx, host, user, createCmd)
	if err != nil {
		stepCallback("create-container", "failed", err)
		return fmt.Errorf("pct create: %w", err)
	}
	stepCallback("create-container", "ok", nil)

	// Step 3: Configure network.
	stepCallback("configure-network", "running", nil)
	_ = out
	stepCallback("configure-network", "ok", nil)

	// Step 4: Install services.
	stepCallback("install-services", "running", nil)
	services, _ := attrs["services"].([]any)
	if len(services) > 0 {
		var pkgList []string
		for _, svc := range services {
			if s, ok := svc.(string); ok {
				pkgList = append(pkgList, s)
			}
		}
		if len(pkgList) > 0 {
			installCmd := fmt.Sprintf("pct exec %d -- bash -c 'apt update -qq && apt install -y -qq %s'", vmIDInt, strings.Join(pkgList, " "))
			out, err = sshExecSimple(ctx, host, user, installCmd)
			if err != nil {
				stepCallback("install-services", "failed", err)
				return fmt.Errorf("install services: %w", err)
			}
			_ = out
		}
	}
	stepCallback("install-services", "ok", nil)

	// Step 5: Configure mounts.
	stepCallback("configure-mounts", "running", nil)
	mounts, _ := attrs["mounts"].([]any)
	for _, m := range mounts {
		if mount, ok := m.(map[string]any); ok {
			source, _ := mount["source"].(string)
			target, _ := mount["target"].(string)
			if source != "" && target != "" {
				mountCmd := fmt.Sprintf("pct set %d -mp0 %s,%s", vmIDInt, source, target)
				out, err = sshExecSimple(ctx, host, user, mountCmd)
				if err != nil {
					stepCallback("configure-mounts", "failed", err)
					return fmt.Errorf("mount %s: %w", source, err)
				}
				_ = out
			}
		}
	}
	stepCallback("configure-mounts", "ok", nil)

	// Step 6: Health check.
	stepCallback("health-check", "running", nil)
	if ip != "" {
		checkCmd := fmt.Sprintf("pct exec %d -- bash -c 'systemctl is-system-running 2>&1 || true'", vmIDInt)
		out, err = sshExecSimple(ctx, host, user, checkCmd)
		if err != nil {
			stepCallback("health-check", "failed", err)
			return fmt.Errorf("health check: %w", err)
		}
		if strings.Contains(out, "degraded") || strings.Contains(out, "running") {
			stepCallback("health-check", "ok", nil)
		} else {
			err := fmt.Errorf("health check returned: %s", strings.TrimSpace(out))
			stepCallback("health-check", "failed", err)
			return err
		}
	} else {
		stepCallback("health-check", "skipped", nil)
	}

	return nil
}

// ProvisionVM creates and configures a VM on a Proxmox host.
func ProvisionVM(ctx context.Context, pool *db.Pool, entityID uuid.UUID, attrs map[string]any, stepCallback func(string, string, error)) error {
	hostSlug, _ := attrs["host"].(string)
	if hostSlug == "" {
		return fmt.Errorf("missing host attribute")
	}

	host, user, err := resolveHost(ctx, pool, hostSlug)
	if err != nil {
		return fmt.Errorf("resolve host %q: %w", hostSlug, err)
	}

	vmid, _ := attrs["vmid"].(float64)
	if vmid == 0 {
		return fmt.Errorf("missing vmid attribute")
	}
	vmIDInt := int(vmid)
	cores, _ := attrs["cores"].(float64)
	ramMB, _ := attrs["ram_mb"].(float64)
	diskGB, _ := attrs["disk_gb"].(float64)
	if cores == 0 {
		cores = 1
	}
	if ramMB == 0 {
		ramMB = 1024
	}
	if diskGB == 0 {
		diskGB = 32
	}

	stepCallback("create-vm", "running", nil)
	createCmd := fmt.Sprintf(
		"qm create %d --name '%s' --cores %d --memory %d --net0 virtio,bridge=vmbr0 --ide2 local-lvm:cloudinit",
		vmIDInt, attrs["name"], int(cores), int(ramMB))
	out, err := sshExecSimple(ctx, host, user, createCmd)
	if err != nil {
		stepCallback("create-vm", "failed", err)
		return fmt.Errorf("qm create: %w", err)
	}
	_ = out

	stepCallback("health-check", "running", nil)
	statusCmd := fmt.Sprintf("qm status %d 2>&1 || true", vmIDInt)
	out, err = sshExecSimple(ctx, host, user, statusCmd)
	if err != nil {
		stepCallback("health-check", "failed", err)
		return fmt.Errorf("qm status: %w", err)
	}
	if strings.Contains(out, "running") || strings.Contains(out, "stopped") {
		stepCallback("health-check", "ok", nil)
	} else {
		err := fmt.Errorf("health check returned: %s", strings.TrimSpace(out))
		stepCallback("health-check", "failed", err)
		return err
	}

	return nil
}

// sshExecSimple runs a command over SSH using the shared dial/run primitives.
// Uses the default SSH key from SSH_KEY_PATH or ~/.ssh/id_rsa.
func sshExecSimple(ctx context.Context, host, user, command string) (string, error) {
	keyPath := os.Getenv("SSH_KEY_PATH")
	if keyPath == "" {
		keyPath = os.Getenv("HOME") + "/.ssh/id_rsa"
	}

	signer, err := LoadSigner(keyPath)
	if err != nil {
		return "", err
	}

	client, err := Dial(ctx, DialOptions{Host: host, User: user, Signer: signer})
	if err != nil {
		return "", err
	}
	defer client.Close()

	out, err := RunCombinedOutput(ctx, client, command)
	return string(out), err
}

// resolveHost resolves a host entity slug to (address, user) for SSH.
func resolveHost(ctx context.Context, pool *db.Pool, slug string) (string, string, error) {
	var attrsJSON []byte
	err := pool.QueryRow(ctx,
		"SELECT attributes FROM entities WHERE slug = $1", slug).Scan(&attrsJSON)
	if err != nil {
		return "", "", fmt.Errorf("entity %s not found: %w", slug, err)
	}

	var attrs map[string]any
	json.Unmarshal(attrsJSON, &attrs)

	addr := ""
	mesh, ok := attrs["mesh"].(map[string]any)
	if ok {
		if nb, ok := mesh["netbird"].(map[string]any); ok {
			if ip, ok := nb["ip"].(string); ok && ip != "" {
				addr = ip
			} else if fqdn, ok := nb["fqdn"].(string); ok && fqdn != "" {
				addr = fqdn
			}
		}
	}
	if addr == "" {
		if lanIP, ok := attrs["lan_ip"].(string); ok && lanIP != "" {
			addr = lanIP
		}
	}
	if addr == "" {
		return "", "", fmt.Errorf("no reachable address for %s", slug)
	}

	user := "root"
	if u, ok := attrs["ssh"].(map[string]any); ok {
		if su, ok := u["user"].(string); ok && su != "" {
			user = su
		}
	}

	return addr, user, nil
}