orca/internal/cli/node.go

package cli

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/google/uuid"
	"github.com/spf13/cobra"

	"git.cloudinit.dev/coreci/orca/internal/certpaths"
	"git.cloudinit.dev/coreci/orca/internal/engine"
	"git.cloudinit.dev/coreci/orca/internal/linux"
	"git.cloudinit.dev/coreci/orca/internal/model"
	"git.cloudinit.dev/coreci/orca/internal/proxmox"
	"git.cloudinit.dev/coreci/orca/internal/security"
	"git.cloudinit.dev/coreci/orca/internal/store"
)

func openDB() (*sql.DB, func() error, error) {
	db, err := store.Open(certpaths.DBPath())
	if err != nil {
		return nil, nil, err
	}
	return db, db.Close, nil
}

func newLogger() *slog.Logger {
	return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
}

func nodeRegistry() (*engine.NodeRegistry, func() error, error) {
	db, closer, err := openDB()
	if err != nil {
		return nil, nil, err
	}
	repo := store.NewNodeRepo(db)
	audit := engine.NewAudit(store.NewAuditRepo(db), newLogger())
	return engine.NewNodeRegistry(repo, audit, newLogger()), closer, nil
}

var (
	joinName        string
	joinAddr        string
	joinCAFinger    string
	joinType        string
	joinHost        string
	joinSSHUser     string
	joinSSHKey      string
	joinSSHPort     int
	joinHostKeyFP   string
	joinLXCTemplate string
	proxmoxUser     string
	proxmoxRole     string
	ingressMode     string
	floatingIP      string
	gateway         string
	macAddr         string
	netPrefix       int
	leaveID         string
	nodeWatch       bool
)

var nodeCmd = &cobra.Command{
	Use:   "node",
	Short: "Manage orca nodes",
	Long:  "Join, leave, or list orca nodes in the registry.",
}

var nodeJoinCmd = &cobra.Command{
	Use:   "join",
	Short: "Join a node to the orca registry",
	Long: `Register a node in the local orca registry. Persisted to SQLite.

Node types (via --type):
  localhost (default): register a local or Linux node (existing behavior)
  linux:               SSH-bootstrap a remote generic Linux worker
                       (Ubuntu/Debian/Alpine; deploys orca pubkey, creates orca
                       user + drift-events dir; requires --host + --ssh-key)
  proxmox:             SSH-bootstrap a remote Proxmox VE 8/9 host
                       (deploys orca pubkey, creates orca user + PVE role +
                       sudoers allowlist; requires --host + --ssh-key (R-021: no passwords))`,
	RunE: func(cmd *cobra.Command, args []string) error {
		if joinHostKeyFP != "" && joinType != "proxmox" && joinType != "linux" {
			return fmt.Errorf("--host-key-fingerprint requires --type proxmox or --type linux")
		}
		if joinType == "proxmox" {
			return joinProxmox(cmd)
		}
		if joinType == "linux" {
			return joinLinux(cmd)
		}
		return joinLocal(cmd)
	},
}

// joinLocal is the existing localhost/Linux node join flow (fingerprint
// check + registry.Insert).
//
// Deprecated: v0.9 re-architecture replaces daemon-to-daemon mTLS join
// with SSH-push bootstrap (R-001). The mTLS join path is retained for
// the dual-write window and scheduled for deletion in v0.10-P14. See
// .ciagent/PRD_v0.9.md.
func joinLocal(cmd *cobra.Command) error {
	warnDeprecated("orca node join (mTLS path): v0.9 R-001 replaces daemon-to-daemon mTLS join with SSH-push bootstrap; the mTLS join path is deprecated — see .ciagent/PRD_v0.9.md")
	if joinName == "" {
		return fmt.Errorf("--name is required")
	}
	if joinAddr == "" {
		joinAddr = "localhost:8443"
	}

	// REQ-026: if --ca-fingerprint is set, verify the on-disk CA
	// matches the pinned value before we touch the registry. This
	// prevents typos in the operator-supplied fingerprint from
	// silently degrading to "no pin" and accepting any cert.
	if joinCAFinger != "" {
		fp, err := security.Fingerprint(certpaths.CACertPath())
		if err != nil {
			return fmt.Errorf("--ca-fingerprint set but local CA is missing: %w (run `orca cert ca-init` first)", err)
		}
		if fp != joinCAFinger {
			return fmt.Errorf(
				"CA fingerprint mismatch: on-disk=%s, pinned=%s — refusing to join (REQ-026)",
				fp, joinCAFinger,
			)
		}
	}

	ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
	defer cancel()

	registry, closer, err := nodeRegistry()
	if err != nil {
		return err
	}
	defer closer()

	node := &model.Node{
		ID:       uuid.NewString(),
		Name:     joinName,
		Address:  joinAddr,
		State:    model.NodeStateReady,
		JoinedAt: time.Now().UTC(),
		LastSeen: time.Now().UTC(),
	}
	if err := registry.Join(ctx, node); err != nil {
		return err
	}
	// REQ-156 / P07 T5: invalidate the nodes cache so the next
	// `orca node list` does not surface a stale list missing the
	// just-joined node.
	cacheInvalidate(cacheNodeClass)
	if jsonOutput {
		return printJSON(node)
	}
	fmt.Fprintf(cmd.OutOrStdout(), "✓ Node joined: %s (%s) at %s\n", node.ID, node.Name, node.Address)
	return nil
}

// joinProxmox bootstraps a remote Proxmox VE 8/9 host via SSH and
// registers it as an orca node (REQ-050, REQ-051). Uses SSH key auth
// (R-021: no passwords). The operator pre-stages the orca SSH public
// key on the remote host out-of-band.
func joinProxmox(cmd *cobra.Command) error {
	if joinHost == "" {
		return fmt.Errorf("--host is required for --type proxmox")
	}
	sshKeyPath := joinSSHKey
	if sshKeyPath == "" {
		sshKeyPath = certpaths.SSHKeyPath()
	}
	if sshKeyPath == "" {
		return fmt.Errorf("SSH key path is required for --type proxmox (R-021: no passwords; use --ssh-key or pre-stage the orca key)")
	}

	ctx, cancel := context.WithTimeout(cmd.Context(), 60*time.Second)
	defer cancel()

	// Default ingress mode to "native" if not specified (R-024).
	effectiveIngressMode := ingressMode
	if effectiveIngressMode == "" {
		effectiveIngressMode = "native"
	}

	result, err := proxmox.BootstrapProxmox(ctx, proxmox.Options{
		Host:               joinHost,
		SSHUser:            joinSSHUser,
		SSHKeyPath:         sshKeyPath,
		ProxmoxUser:        proxmoxUser,
		ProxmoxRole:        proxmoxRole,
		SSHPort:            joinSSHPort,
		HostKeyFingerprint: joinHostKeyFP,
		Logger:             newLogger(),
		LXCTemplate:        joinLXCTemplate,
		IngressMode:        effectiveIngressMode,
	})
	if err != nil {
		return fmt.Errorf("proxmox bootstrap: %w", err)
	}

	// Register the proxmox node in the orca registry.
	registry, closer, err := nodeRegistry()
	if err != nil {
		return err
	}
	defer closer()

	regCtx, regCancel := context.WithTimeout(ctx, 5*time.Second)
	defer regCancel()

	node := &model.Node{
		ID:          uuid.NewString(),
		Name:        result.NodeName,
		Address:     result.NodeAddress,
		State:       model.NodeStateReady,
		JoinedAt:    time.Now().UTC(),
		LastSeen:    time.Now().UTC(),
		Kind:        string(model.NodeKindProxmox),
		OS:          "pve",
		IngressMode: effectiveIngressMode,
	}
	if err := registry.Join(regCtx, node); err != nil {
		return fmt.Errorf("register proxmox node: %w", err)
	}
	// REQ-156 / P07 T5: invalidate the nodes cache.
	cacheInvalidate(cacheNodeClass)
	if jsonOutput {
		return printJSON(node)
	}
	fmt.Fprintf(cmd.OutOrStdout(), "✓ Proxmox node joined: %s (%s) at %s\n", node.ID, node.Name, node.Address)
	fmt.Fprintf(cmd.OutOrStdout(), "  role: %s, user: %s@pam\n", proxmoxRole, proxmoxUser)
	return nil
}

// joinLinux bootstraps a remote generic Linux worker via SSH and
// registers it as an orca node (REQ-161, P12). Uses SSH key auth
// (R-021: no passwords).
func joinLinux(cmd *cobra.Command) error {
	if joinHost == "" {
		return fmt.Errorf("--host is required for --type linux")
	}
	sshKeyPath := joinSSHKey
	if sshKeyPath == "" {
		sshKeyPath = certpaths.SSHKeyPath()
	}
	if sshKeyPath == "" {
		return fmt.Errorf("SSH key path is required for --type linux (R-021: no passwords; use --ssh-key or pre-stage the orca key)")
	}

	ctx, cancel := context.WithTimeout(cmd.Context(), 60*time.Second)
	defer cancel()

	result, err := linux.BootstrapLinux(ctx, linux.Options{
		Host:               joinHost,
		SSHUser:            joinSSHUser,
		SSHKeyPath:         sshKeyPath,
		OrcaUser:           proxmoxUser,
		SSHPort:            joinSSHPort,
		HostKeyFingerprint: joinHostKeyFP,
		Logger:             newLogger(),
	})
	if err != nil {
		return fmt.Errorf("linux bootstrap: %w", err)
	}

	registry, closer, err := nodeRegistry()
	if err != nil {
		return err
	}
	defer closer()

	regCtx, regCancel := context.WithTimeout(ctx, 5*time.Second)
	defer regCancel()

	node := &model.Node{
		ID:       uuid.NewString(),
		Name:     result.NodeName,
		Address:  result.NodeAddress,
		State:    model.NodeStateReady,
		JoinedAt: time.Now().UTC(),
		LastSeen: time.Now().UTC(),
		Kind:     string(model.NodeKindLinux),
		OS:       "linux",
	}
	if err := registry.Join(regCtx, node); err != nil {
		return fmt.Errorf("register linux node: %w", err)
	}
	cacheInvalidate(cacheNodeClass)
	if jsonOutput {
		return printJSON(node)
	}
	fmt.Fprintf(cmd.OutOrStdout(), "\xe2\x9c\x93 Linux worker joined: %s (%s) at %s\n", node.ID, node.Name, node.Address)
	if result.HostKeyFingerprint != "" {
		fmt.Fprintf(cmd.OutOrStdout(), "  host key: %s\n", result.HostKeyFingerprint)
	}
	return nil
}

var nodeLeaveCmd = &cobra.Command{
	Use:   "leave [node-id]",
	Short: "Remove a node from the orca registry",
	Long:  "Mark a node as left. Use --id to specify, or pass as argument.",
	Args:  cobra.MaximumNArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		id := leaveID
		if id == "" && len(args) > 0 {
			id = args[0]
		}
		if id == "" {
			return fmt.Errorf("node id required (use --id or pass as argument)")
		}
		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
		defer cancel()

		registry, closer, err := nodeRegistry()
		if err != nil {
			return err
		}
		defer closer()

		if err := registry.Leave(ctx, id); err != nil {
			return err
		}
		// REQ-156 / P07 T5: invalidate the nodes cache so the next
		// `orca node list` does not surface the just-left node.
		cacheInvalidate(cacheNodeClass)
		if jsonOutput {
			return printJSON(map[string]string{"id": id, "state": "left"})
		}
		fmt.Fprintf(cmd.OutOrStdout(), "✓ Node left: %s\n", id)
		return nil
	},
}

var nodeListCmd = &cobra.Command{
	Use:   "list",
	Short: "List all nodes in the orca registry",
	Long:  "Display all registered nodes and their state. Use --watch to stream updates until Ctrl-C.",
	RunE: func(cmd *cobra.Command, args []string) error {
		if nodeWatch {
			return watchNodes(cmd)
		}

		// Cache (R-008): read path only; --watch bypasses. On hit,
		// unmarshal cached JSON and render without touching the DB.
		var cachedNodes []*model.Node
		if cacheGetList(cacheNodeClass, cacheListKey, &cachedNodes) {
			return renderNodes(cmd, cachedNodes)
		}

		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
		defer cancel()

		registry, closer, err := nodeRegistry()
		if err != nil {
			return err
		}
		defer closer()

		nodes, err := registry.List(ctx)
		if err != nil {
			return err
		}
		cachePutList(cacheNodeClass, cacheListKey, nodes, cacheNodeTTL)
		return renderNodes(cmd, nodes)
	},
}

// renderNodes prints the node list in either JSON or table form.
func renderNodes(cmd *cobra.Command, nodes []*model.Node) error {
	if jsonOutput {
		return printJSON(nodes)
	}
	if len(nodes) == 0 {
		fmt.Fprintln(cmd.OutOrStdout(), "No nodes registered. Use 'orca node join' to add one.")
		return nil
	}
	fmt.Fprintf(cmd.OutOrStdout(), "%-36s  %-20s  %-22s  %-10s\n", "ID", "NAME", "ADDRESS", "STATE")
	for _, n := range nodes {
		fmt.Fprintf(cmd.OutOrStdout(), "%-36s  %-20s  %-22s  %-10s\n", n.ID, n.Name, n.Address, n.State)
	}
	return nil
}

func watchNodes(cmd *cobra.Command) error {
	ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
	defer cancel()
	return watchNodesCtx(cmd, ctx)
}

func watchNodesCtx(cmd *cobra.Command, ctx context.Context) error {
	db, closer, err := openDB()
	if err != nil {
		return err
	}
	defer closer()

	out := cmd.OutOrStdout()

	if jsonOutput {
		seen := make(map[string]string)
		for snapshot := range store.NewNodeRepo(db).Watch(ctx) {
			current := make(map[string]bool, len(snapshot))
			for _, n := range snapshot {
				current[n.ID] = true
				compact, _ := json.Marshal(n)
				key := string(compact)
				if prev, ok := seen[n.ID]; !ok || prev != key {
					event := "init"
					if ok {
						event = "update"
					}
					line, _ := json.Marshal(map[string]any{"event": event, "node": n})
					fmt.Fprintln(out, string(line))
					seen[n.ID] = key
				}
			}
			for id := range seen {
				if !current[id] {
					line, _ := json.Marshal(map[string]any{"event": "delete", "id": id})
					fmt.Fprintln(out, string(line))
					delete(seen, id)
				}
			}
		}
		return nil
	}

	prevTable := ""
	for snapshot := range store.NewNodeRepo(db).Watch(ctx) {
		table := renderNodeTable(snapshot)
		if table != prevTable {
			fmt.Fprint(out, "\033[2J\033[H")
			fmt.Fprint(out, table)
			prevTable = table
		}
	}
	return nil
}

func renderNodeTable(nodes []*model.Node) string {
	if len(nodes) == 0 {
		return "No nodes registered.\n"
	}
	out := fmt.Sprintf("%-36s  %-20s  %-22s  %-10s\n", "ID", "NAME", "ADDRESS", "STATE")
	for _, n := range nodes {
		out += fmt.Sprintf("%-36s  %-20s  %-22s  %-10s\n", n.ID, n.Name, n.Address, n.State)
	}
	return out
}

var nodeKeyResetCmd = &cobra.Command{
	Use:   "key-reset <node>",
	Short: "Reset the SSH known_hosts entry for a node",
	Long: `Remove the pinned SSH host key for <node> from the local known_hosts
file. The next connect re-pins the key via TOFU or --host-key-fingerprint.

LOCAL ONLY (D-046): does not touch the remote host's authorized_keys.

<node> is the node name (for proxmox nodes, this is the host address).`,
	Args: cobra.ExactArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		nodeArg := args[0]

		registry, closer, err := nodeRegistry()
		if err != nil {
			return err
		}
		defer closer()

		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
		defer cancel()

		nodes, err := registry.List(ctx)
		if err != nil {
			return fmt.Errorf("list nodes: %w", err)
		}
		var node *model.Node
		for _, n := range nodes {
			if n.Name == nodeArg || n.ID == nodeArg {
				node = n
				break
			}
		}
		if node == nil {
			return fmt.Errorf("node %q not found in the registry", nodeArg)
		}
		host := node.Name

		if err := proxmox.ResetHostKey(host); err != nil {
			return fmt.Errorf("reset host key: %w", err)
		}

		// Audit-log the reset (REQ-059): actor=cli, action=node.key_reset.
		db, dbCloser, dbErr := openDB()
		if dbErr == nil {
			defer dbCloser()
			audit := engine.NewAudit(store.NewAuditRepo(db), newLogger())
			audit.Record(ctx, actorFromCtx(ctx), "node.key_reset", node.ID, "success", nil, map[string]any{
				"node": node.Name,
				"host": host,
			})
		}

		fmt.Fprintf(cmd.OutOrStdout(), "✓ Host key reset for %s (next connect will re-pin via TOFU or --host-key-fingerprint)\n", node.Name)
		return nil
	},
}

func init() {
	nodeJoinCmd.Flags().StringVar(&joinName, "name", "", "node name (required for --type localhost)")
	nodeJoinCmd.Flags().StringVar(&joinAddr, "addr", "", "node address (default localhost:8443)")
	nodeJoinCmd.Flags().StringVar(&joinCAFinger, "ca-fingerprint", "", "pin CA cert SHA-256 (REQ-026); fails if on-disk CA doesn't match")
	nodeJoinCmd.Flags().StringVar(&joinType, "type", "localhost", "node type: localhost (default), proxmox, or linux (SSH bootstrap)")
	nodeJoinCmd.Flags().StringVar(&joinHost, "host", "", "proxmox host address (IP/hostname, no port; required for --type proxmox)")
	nodeJoinCmd.Flags().StringVar(&joinSSHUser, "ssh-user", "root", "SSH username for proxmox bootstrap (default root)")
	nodeJoinCmd.Flags().StringVar(&joinSSHKey, "ssh-key", "", "SSH private key path for proxmox bootstrap (R-021: no passwords; default: orca key)")
	nodeJoinCmd.Flags().IntVar(&joinSSHPort, "ssh-port", 22, "SSH port for proxmox bootstrap (default 22)")
	nodeJoinCmd.Flags().StringVar(&proxmoxUser, "proxmox-user", "orca", "Linux system user to create on the proxmox host (config-overridable)")
	nodeJoinCmd.Flags().StringVar(&proxmoxRole, "proxmox-role", "OrcaOperator", "PVE custom role to create (config-overridable)")
	nodeJoinCmd.Flags().StringVar(&joinHostKeyFP, "host-key-fingerprint", "", "SSH host key SHA256:base64 fingerprint (pre-pin; supersedes TOFU for --type proxmox or --type linux)")
	nodeJoinCmd.Flags().StringVar(&joinLXCTemplate, "lxc-template", "ubuntu-24.04", "LXC template for Proxmox (default ubuntu-24.04; alternatives: alpine-3.20, debian-12)")
	nodeJoinCmd.Flags().StringVar(&ingressMode, "ingress-mode", "", "proxmox ingress mode: native (default, traefik in LXC) or floating-ip (ingress LXC owns floating IP)")
	nodeJoinCmd.Flags().StringVar(&floatingIP, "floating-ip", "", "floating public IP for the ingress LXC (required for --ingress-mode floating-ip)")
	nodeJoinCmd.Flags().StringVar(&gateway, "gateway", "", "gateway for the ingress LXC (required for --ingress-mode floating-ip)")
	nodeJoinCmd.Flags().StringVar(&macAddr, "mac", "", "MAC address for the ingress LXC net0 (required for --ingress-mode floating-ip in --json mode; auto-generated in interactive mode)")
	nodeJoinCmd.Flags().IntVar(&netPrefix, "net-prefix", 24, "network prefix (CIDR) for the ingress LXC IP (default 24; valid 8-32)")
	nodeLeaveCmd.Flags().StringVar(&leaveID, "id", "", "node id")
	nodeListCmd.Flags().BoolVar(&nodeWatch, "watch", false, "stream nodes until Ctrl-C (table refresh or --json per-event)")

	nodeCmd.AddCommand(nodeJoinCmd)
	nodeCmd.AddCommand(nodeLeaveCmd)
	nodeCmd.AddCommand(nodeListCmd)
	nodeCmd.AddCommand(nodeKeyResetCmd)
	rootCmd.AddCommand(nodeCmd)
}