orca/internal/cli/drift.go

// Package cli: drift.go implements the `orca drift` subcommand family
// (P10b, v0.11; R-018/R-019/R-020, REQ-104). Subcommands:
//
//	orca drift                       show current drift state (table)
//	orca drift watch [--interval=2s] [--paths=...] [--json]
//	                                 stream drift events (iter.Seq2, D-017)
//	                                 ctrl-c cancels via signal.NotifyContext (D-023)
//	orca drift show [--peer <host>]  detailed drift events for a peer
//	orca drift acknowledge <peer> <path>
//	                                 record operator acknowledgment
//	orca drift remediate <peer> <path> [--force]
//	                                 trigger manual remediation
//	orca drift config show           show current drift config
//	orca drift config validate       validate config
//
// Plus the `orca job restart <name>` command for EnvironmentFile drift
// (REQ-113, D-235) — restarts an allocation to pick up env-file drift.
package cli

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"syscall"
	"time"

	"github.com/spf13/cobra"

	"git.cloudinit.dev/coreci/orca/internal/certpaths"
	"git.cloudinit.dev/coreci/orca/internal/drift"
	"git.cloudinit.dev/coreci/orca/internal/sshpush"
)

var (
	driftWatchInterval     time.Duration
	driftWatchPaths        []string
	driftShowPeer          string
	driftConfigPath        string
	driftRemediateForce    bool
	driftAckPeer           string
	driftAckPath           string
	driftRemediatePeer     string
	driftRemediatePath     string
	driftWatchPollOverride time.Duration
)

// driftTransport is the SSH-push surface the drift CLI needs. It
// mirrors drift.Transport; tests substitute a mock.
type driftTransport interface {
	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
	WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error)
	ReadFile(ctx context.Context, peer string, path string) ([]byte, error)
}

// driftTransportOverride is the package-level test seam.
var driftTransportOverride driftTransport

func driftTransportFromCtx() (driftTransport, error) {
	if driftTransportOverride != nil {
		return driftTransportOverride, nil
	}
	keyPath := certpaths.SSHKeyPath()
	khPath := certpaths.KnownHostsPath()
	return sshpush.NewTransport(keyPath, khPath), nil
}

// driftDetectorOverride is the package-level test seam for the
// Detector itself. When non-nil it replaces the production detector
// (which wraps a driftTransport). Tests set it and restore nil.
var driftDetectorOverride drift.Detector

func driftDetector() (drift.Detector, error) {
	if driftDetectorOverride != nil {
		return driftDetectorOverride, nil
	}
	t, err := driftTransportFromCtx()
	if err != nil {
		return nil, err
	}
	return drift.NewDefaultDetector(t), nil
}

var driftCmd = &cobra.Command{
	Use:   "drift",
	Short: "Detect and remediate control-plane drift (P10b)",
	Long: `Orca's drift detector is a BACKSTOP (R-019): the primary
consistency mechanism is systemd / Traefik / step-ca / Syncthing
themselves. The detector polls the lead's aggregated drift state
(drift-events-aggregated.json) and can trigger orca-remediate.sh for
auto-remediable paths. Pre-flight drift blocks txn apply (R-020).`,
}

var driftShowCmd = &cobra.Command{
	Use:   "show",
	Short: "Show current drift state (table format)",
	Long:  `Show the aggregated drift events from the lead. Use --peer to filter to a single peer.`,
	Args:  cobra.NoArgs,
	RunE: func(cmd *cobra.Command, args []string) error {
		d, err := driftDetector()
		if err != nil {
			return fmt.Errorf("drift detector: %w", err)
		}
		events, err := d.Aggregate(cmd.Context(), driftShowPeer)
		if err != nil {
			return fmt.Errorf("aggregate: %w", err)
		}
		if driftShowPeer != "" {
			var filtered []drift.Event
			for _, e := range events {
				if e.Host == driftShowPeer {
					filtered = append(filtered, e)
				}
			}
			events = filtered
		}
		if jsonOutput {
			return printJSON(events)
		}
		out := cmd.OutOrStdout()
		if len(events) == 0 {
			fmt.Fprintln(out, "No drift events.")
			return nil
		}
		fmt.Fprintf(out, "%-20s  %-20s  %-40s  %-10s  %-10s\n", "EVENT-ID", "HOST", "PATH", "STATUS", "CONFIRMED")
		for _, e := range events {
			path := e.Path
			if len(path) > 40 {
				path = "..." + path[len(path)-37:]
			}
			confirmed := "no"
			if e.DriftConfirmed {
				confirmed = "yes"
			}
			fmt.Fprintf(out, "%-20s  %-20s  %-40s  %-10s  %-10s\n", e.EventID, e.Host, path, e.Status, confirmed)
		}
		return nil
	},
}

var driftWatchCmd = &cobra.Command{
	Use:   "watch",
	Short: "Stream drift events (ctrl-c to cancel)",
	Long: `Stream drift events from the lead's aggregated state. Default
poll is 2s; override with --interval. Use --paths=<glob1>,<glob2> to
filter. Uses iter.Seq2 (D-017) and signal.NotifyContext for ctrl-c
(D-023).`,
	Args: cobra.NoArgs,
	RunE: func(cmd *cobra.Command, args []string) error {
		d, err := driftDetector()
		if err != nil {
			return fmt.Errorf("drift detector: %w", err)
		}
		ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
		defer cancel()
		interval := driftWatchInterval
		if interval <= 0 {
			interval = 2 * time.Second
		}
		var specs []drift.PathSpec
		for _, p := range driftWatchPaths {
			if p == "" {
				continue
			}
			specs = append(specs, drift.PathSpec{Pattern: p, Interval: interval})
		}
		out := cmd.OutOrStdout()
		for e, err := range d.Watch(ctx, specs) {
			if err != nil {
				if errors.Is(err, context.Canceled) {
					return nil
				}
				fmt.Fprintf(out, "watch error: %v\n", err)
				continue
			}
			if jsonOutput {
				line, _ := json.Marshal(e)
				fmt.Fprintln(out, string(line))
			} else {
				fmt.Fprintf(out, "%s  [%s]  %s  %s  %s  confirmed=%t\n", e.TS.Format(time.RFC3339), e.EventID, e.Host, e.Path, e.Status, e.DriftConfirmed)
			}
		}
		return nil
	},
}

var driftAckCmd = &cobra.Command{
	Use:   "acknowledge <peer> <path>",
	Short: "Record operator acknowledgment of drift on a peer",
	Long: `Record operator acknowledgment for the given path in
drift-acknowledgments.json on the lead. Acknowledged drift no longer
blocks txn apply for that namespace (R-020).`,
	Args: cobra.ExactArgs(2),
	RunE: func(cmd *cobra.Command, args []string) error {
		peer := args[0]
		path := args[1]
		d, err := driftDetector()
		if err != nil {
			return fmt.Errorf("drift detector: %w", err)
		}
		if err := d.Acknowledge(cmd.Context(), peer, path); err != nil {
			return fmt.Errorf("acknowledge: %w", err)
		}
		printResult(fmt.Sprintf("✓ Acknowledged drift on %s for %s", peer, path), map[string]any{
			"peer": peer, "path": path, "status": "acknowledged",
		})
		return nil
	},
}

var driftRemediateCmd = &cobra.Command{
	Use:   "remediate <peer> <path>",
	Short: "Trigger manual remediation of drift on a peer",
	Long: `Trigger orca-remediate.sh on the lead for the given path.
--force bypasses the cooldown window (C4).`,
	Args: cobra.ExactArgs(2),
	RunE: func(cmd *cobra.Command, args []string) error {
		peer := args[0]
		path := args[1]
		d, err := driftDetector()
		if err != nil {
			return fmt.Errorf("drift detector: %w", err)
		}
		if err := d.Remediate(cmd.Context(), peer, path, driftRemediateForce); err != nil {
			if errors.Is(err, drift.ErrCooldown) {
				printResult(fmt.Sprintf("✗ Remediation in cooldown for %s on %s (use --force to bypass)", path, peer), map[string]any{
					"peer": peer, "path": path, "status": "cooldown",
				})
				return nil
			}
			return fmt.Errorf("remediate: %w", err)
		}
		printResult(fmt.Sprintf("✓ Remediated drift on %s for %s", peer, path), map[string]any{
			"peer": peer, "path": path, "status": "remediated", "force": driftRemediateForce,
		})
		return nil
	},
}

var driftConfigCmd = &cobra.Command{
	Use:   "config",
	Short: "Show or validate the drift config",
}

var driftConfigShowCmd = &cobra.Command{
	Use:   "show",
	Short: "Show the current drift config",
	Args:  cobra.NoArgs,
	RunE: func(cmd *cobra.Command, args []string) error {
		cfg, err := drift.LoadConfig(driftConfigPath)
		if err != nil {
			return fmt.Errorf("load config: %w", err)
		}
		if jsonOutput {
			return printJSON(cfg)
		}
		out := cmd.OutOrStdout()
		fmt.Fprintf(out, "Polling: enabled=%t default=%s max_peers=%d\n", cfg.Polling.Enabled, cfg.Polling.DefaultInterval, cfg.Polling.MaxConcurrentPeers)
		fmt.Fprintln(out, "Critical paths:")
		for _, p := range cfg.Paths.Critical {
			fmt.Fprintf(out, "  [%s] %s (interval=%s, path_unit=%t)\n", p.Tier, p.Pattern, p.Interval, p.SystemdPathUnit)
		}
		fmt.Fprintln(out, "Standard paths:")
		for _, p := range cfg.Paths.Standard {
			fmt.Fprintf(out, "  [%s] %s (interval=%s)\n", p.Tier, p.Pattern, p.Interval)
		}
		fmt.Fprintln(out, "Excluded paths:")
		for _, p := range cfg.Paths.Excluded {
			fmt.Fprintf(out, "  %s\n", p)
		}
		fmt.Fprintf(out, "Remediation: auto=%t notify=%t\n", cfg.Remediate.Auto, cfg.Remediate.NotifyOnRemediation)
		if len(cfg.Remediate.AutoPaths) > 0 {
			fmt.Fprintln(out, "  auto_paths:")
			for _, p := range cfg.Remediate.AutoPaths {
				fmt.Fprintf(out, "    %s\n", p)
			}
		}
		if len(cfg.Remediate.RequireApproval) > 0 {
			fmt.Fprintln(out, "  require_approval:")
			for _, p := range cfg.Remediate.RequireApproval {
				fmt.Fprintf(out, "    %s\n", p)
			}
		}
		return nil
	},
}

var driftConfigValidateCmd = &cobra.Command{
	Use:   "validate",
	Short: "Validate the drift config",
	Args:  cobra.NoArgs,
	RunE: func(cmd *cobra.Command, args []string) error {
		cfg, err := drift.LoadConfig(driftConfigPath)
		if err != nil {
			return fmt.Errorf("load config: %w", err)
		}
		if err := drift.ValidateConfig(cfg); err != nil {
			printResult(fmt.Sprintf("✗ Config invalid: %v", err), map[string]any{"valid": false, "error": err.Error()})
			return err
		}
		printResult("✓ Config valid", map[string]any{"valid": true})
		return nil
	},
}

// jobRestartCmd implements `orca job restart <name>` (REQ-113, D-235):
// restart an allocation on its peer to pick up EnvironmentFile drift.
var jobRestartCmd = &cobra.Command{
	Use:   "restart <name>",
	Short: "Restart an allocation to pick up EnvironmentFile drift (REQ-113)",
	Long: `SSH to the peer running allocation <name> and run
systemctl restart orca-alloc-<id>.service. This is the normal
allocation lifecycle (NOT file-level remediation) and is triggered
when /etc/orca/allocs/<id>/env drifts.`,
	Args: cobra.ExactArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		name := args[0]
		peer := jobRestartPeer
		if peer == "" {
			return fmt.Errorf("--peer is required for job restart")
		}
		transport, err := driftTransportFromCtx()
		if err != nil {
			return fmt.Errorf("ssh transport: %w", err)
		}
		unit := fmt.Sprintf("orca-alloc-%s.service", name)
		restartCmd := fmt.Sprintf("systemctl restart %s", shellQuoteDrift(unit))
		out, err := transport.Exec(cmd.Context(), peer, restartCmd)
		if err != nil {
			return fmt.Errorf("restart %s on %s: %w (output: %s)", unit, peer, err, string(out))
		}
		printResult(fmt.Sprintf("✓ Restarted %s on %s", unit, peer), map[string]any{
			"unit": unit, "peer": peer, "status": "restarted", "output": string(out),
		})
		return nil
	},
}

var jobRestartPeer string

func shellQuoteDrift(s string) string {
	return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
}

func init() {
	driftWatchCmd.Flags().DurationVar(&driftWatchInterval, "interval", 2*time.Second, "poll interval (default 2s)")
	driftWatchCmd.Flags().StringSliceVar(&driftWatchPaths, "paths", nil, "comma-separated glob patterns to watch (default: all)")
	driftShowCmd.Flags().StringVar(&driftShowPeer, "peer", "", "filter to a single peer host")
	driftRemediateCmd.Flags().BoolVar(&driftRemediateForce, "force", false, "bypass the cooldown window (C4)")
	driftConfigCmd.PersistentFlags().StringVar(&driftConfigPath, "config", "", "path to drift config JSON (default: built-in)")
	jobRestartCmd.Flags().StringVar(&jobRestartPeer, "peer", "", "peer address (host:port) running the allocation")

	driftCmd.AddCommand(driftShowCmd)
	driftCmd.AddCommand(driftWatchCmd)
	driftCmd.AddCommand(driftAckCmd)
	driftCmd.AddCommand(driftRemediateCmd)
	driftCmd.AddCommand(driftConfigCmd)
	driftConfigCmd.AddCommand(driftConfigShowCmd)
	driftConfigCmd.AddCommand(driftConfigValidateCmd)
	rootCmd.AddCommand(driftCmd)

	jobCmd.AddCommand(jobRestartCmd)
}

// writeClusterDefaultDriftConfig writes the canonical default drift
// config to the cluster state dir so the lead has a reference copy.
// Best-effort; caller logs failures.
func writeClusterDefaultDriftConfig() error {
	dir := filepath.Join(os.Getenv("ORCA_HOME"), "cluster", "state")
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return fmt.Errorf("mkdir cluster state: %w", err)
	}
	path := filepath.Join(dir, "drift.json")
	cfg := drift.DefaultConfig()
	raw, err := json.MarshalIndent(cfg, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal drift config: %w", err)
	}
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, raw, 0o644); err != nil {
		return fmt.Errorf("write drift config: %w", err)
	}
	if err := os.Rename(tmp, path); err != nil {
		return fmt.Errorf("rename drift config: %w", err)
	}
	return nil
}