orca/internal/cli/job.go

package cli

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"strings"
	"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/jobspec"
	"git.cloudinit.dev/coreci/orca/internal/model"
	"git.cloudinit.dev/coreci/orca/internal/sshpush"
	"git.cloudinit.dev/coreci/orca/internal/store"
)

var jobCmd = &cobra.Command{
	Use:   "job",
	Short: "Manage orca jobs",
	Long:  "Run, list, stop, and inspect orca jobs.",
}

func jobExecutor() (*engine.Executor, func() error, error) {
	db, closer, err := openDB()
	if err != nil {
		return nil, nil, err
	}
	jobs := store.NewJobRepo(db)
	tasks := store.NewTaskRepo(db)
	return engine.NewExecutor(jobs, tasks, newLogger()), closer, nil
}

var (
	stopID    string
	runTarget string
	runIDKey  string
	jobWatch  bool
)

var jobRunCmd = &cobra.Command{
	Use:   "run <spec.md>",
	Short: "Run a job from a markdown spec file",
	Long:  "Submit a job spec, execute its tasks, and persist the result. Use --target to pin to a specific node (overrides bin-packing); --idempotency-key for cross-node dispatch dedupe.",
	Args:  cobra.ExactArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		if strings.HasSuffix(args[0], ".hcl") {
			warnDeprecated("orca job run <spec.hcl> is deprecated: .hcl jobspec is legacy (R-013); convert to .md format (REQ-064) — see .ciagent/PRD_v0.9.md")
		}
		spec, err := jobspec.ParseFile(args[0])
		if err != nil {
			return err
		}

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

		// v0.13 phase-03 scheduler wiring (REQ-151, C-44): decide
		// whether to run locally (dev mode / no remote nodes) or
		// remotely (scheduler picks a peer, render systemd, SSH-push).
		// The deprecated mTLS Dispatcher path (--idempotency-key) is
		// retained only for the dual-write window; the new remote path
		// uses the CLI-side scheduler + sshpush.
		if runIDKey != "" {
			// Legacy --idempotency-key dispatch path (deprecated mTLS
			// Dispatcher). Retained for backward compat; routes through
			// engine.Dispatcher which is scheduled for removal in v0.10.
			exec, closer, err := jobExecutor()
			if err != nil {
				return err
			}
			defer closer()
			db, dbCloser, err := openDB()
			if err != nil {
				return err
			}
			defer dbCloser()
			peers := engine.NewPeerRegistry()
			dispatcher := engine.NewDispatcher(newLogger(), store.NewCapacityRepo(db), peers, exec)
			specBytes, _ := json.Marshal(map[string]any{
				"name":    spec.Name,
				"command": "/bin/true", // placeholder; full HCL dispatch lands in a later phase
			})
			jobID, nodeID, err := dispatcher.Submit(ctx, runTarget, specBytes, runIDKey)
			if err != nil {
				if jsonOutput {
					_ = printJSON(map[string]any{"status": "failed", "error": err.Error()})
				}
				return err
			}
			if jsonOutput {
				return printJSON(map[string]any{"id": jobID, "node_id": nodeID, "status": "dispatched"})
			}
			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job dispatched: %s to %s\n", jobID, nodeID)
			return nil
		}

		res, nodesByHost, err := dispatchDecision(ctx, spec, runTarget)
		if err != nil {
			logDispatch(nil, err)
			if jsonOutput {
				_ = printJSON(map[string]any{"status": "failed", "error": err.Error()})
			}
			return err
		}

		switch res.mode {
		case "remote":
			// Scheduler selected a node (or --target pinned one): render
			// the systemd unit / PVE container, verify it, and SSH-push.
			// C-44: a push failure is an error (no local fallback).
			unitPaths, derr := deployRemote(ctx, spec, res, nodesByHost)
			logDispatch(res, derr)
			if derr != nil {
				if jsonOutput {
					_ = printJSON(map[string]any{"status": "failed", "node": res.node, "error": derr.Error()})
				}
				return derr
			}
			res.unitPaths = unitPaths
			// REQ-166 / Phase C2: insert a Job DB record so `job list`
			// and `job stop` can find the remotely-deployed job.
			if dbErr := insertRemoteJob(spec, res.node); dbErr != nil {
				// Non-fatal: the job is deployed, just not visible to list.
				logDispatch(res, dbErr)
			}
			cacheInvalidate(cacheJobClass)
			if jsonOutput {
				return printJSON(map[string]any{
					"status":   "deployed",
					"node":     res.node,
					"alloc_id": res.allocID,
					"units":    unitPaths,
				})
			}
			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job deployed to %s: %s (%s)\n", res.node, spec.Name, strings.Join(unitPaths, ", "))
			return nil

		case "local":
			// Local exec fallback (dev mode: no remote nodes registered).
			exec, closer, err := jobExecutor()
			if err != nil {
				return err
			}
			defer closer()
			job := &model.Job{
				ID:     uuid.NewString(),
				Name:   spec.Name,
				Spec:   args[0],
				Status: model.JobStatusPending,
			}
			runErr := exec.Run(ctx, job, workloadToTaskSpecs(spec))
			logDispatch(res, runErr)
			// REQ-156 / P07 T5: invalidate the jobs cache so the next
			// `orca job list` reflects the just-run (or just-failed)
			// job instead of a stale cached list.
			cacheInvalidate(cacheJobClass)
			if runErr != nil {
				if jsonOutput {
					_ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": runErr.Error()})
					return runErr
				}
				fmt.Fprintf(cmd.ErrOrStderr(), "✗ Job %s failed: %v\n", job.ID, runErr)
				return runErr
			}
			if jsonOutput {
				return printJSON(map[string]any{"id": job.ID, "name": job.Name, "status": "complete"})
			}
			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job complete: %s (%s)\n", job.ID, job.Name)
			return nil
		}
		return fmt.Errorf("job run: unknown dispatch mode %q", res.mode)
	},
}

var jobListCmd = &cobra.Command{
	Use:   "list",
	Short: "List all jobs",
	Long:  "Display all jobs and their status. Use --watch to stream updates until Ctrl-C.",
	RunE: func(cmd *cobra.Command, args []string) error {
		if jobWatch {
			return watchJobs(cmd)
		}

		// Cache (R-008): read path only; --watch bypasses.
		var cachedJobs []*model.Job
		if cacheGetList(cacheJobClass, cacheListKey, &cachedJobs) {
			return renderJobs(cmd, cachedJobs)
		}

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

		db, closer, err := openDB()
		if err != nil {
			return err
		}
		defer closer()

		jobs, err := store.NewJobRepo(db).List(ctx)
		if err != nil {
			return err
		}
		cachePutList(cacheJobClass, cacheListKey, jobs, cacheJobTTL)
		return renderJobs(cmd, jobs)
	},
}

// renderJobs prints the job list in either JSON or table form.
func renderJobs(cmd *cobra.Command, jobs []*model.Job) error {
	if jsonOutput {
		return printJSON(jobs)
	}
	if len(jobs) == 0 {
		fmt.Fprintln(cmd.OutOrStdout(), "No jobs. Use 'orca job run <spec.md>' to submit one.")
		return nil
	}
	fmt.Fprintf(cmd.OutOrStdout(), "%-10s  %-20s  %-12s  %-20s  %-5s\n", "ID", "NAME", "STATUS", "NODE", "EXIT")
	for _, j := range jobs {
		shortID := j.ID
		if len(shortID) > 8 {
			shortID = shortID[:8]
		}
		exit := "-"
		if j.Status == model.JobStatusComplete || j.Status == model.JobStatusFailed || j.Status == model.JobStatusStopped {
			exit = fmt.Sprintf("%d", j.ExitCode)
		}
		fmt.Fprintf(cmd.OutOrStdout(), "%-10s  %-20s  %-12s  %-20s  %-5s\n", shortID, j.Name, j.Status, j.Node, exit)
	}
	return nil
}

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

func watchJobsCtx(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.NewJobRepo(db).Watch(ctx) {
			current := make(map[string]bool, len(snapshot))
			for _, j := range snapshot {
				current[j.ID] = true
				compact, _ := json.Marshal(j)
				key := string(compact)
				if prev, ok := seen[j.ID]; !ok || prev != key {
					event := "init"
					if ok {
						event = "update"
					}
					line, _ := json.Marshal(map[string]any{"event": event, "job": j})
					fmt.Fprintln(out, string(line))
					seen[j.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.NewJobRepo(db).Watch(ctx) {
		table := renderJobTable(snapshot)
		if table != prevTable {
			fmt.Fprint(out, "\033[2J\033[H")
			fmt.Fprint(out, table)
			prevTable = table
		}
	}
	return nil
}

func renderJobTable(jobs []*model.Job) string {
	if len(jobs) == 0 {
		return "No jobs.\n"
	}
	out := fmt.Sprintf("%-10s  %-20s  %-12s  %-20s  %-5s\n", "ID", "NAME", "STATUS", "NODE", "EXIT")
	for _, j := range jobs {
		shortID := j.ID
		if len(shortID) > 8 {
			shortID = shortID[:8]
		}
		exit := "-"
		if j.Status == model.JobStatusComplete || j.Status == model.JobStatusFailed || j.Status == model.JobStatusStopped {
			exit = fmt.Sprintf("%d", j.ExitCode)
		}
		out += fmt.Sprintf("%-10s  %-20s  %-12s  %-20s  %-5s\n", shortID, j.Name, j.Status, j.Node, exit)
	}
	return out
}

// jobStopTransport is the SSH command-execution seam used by
// `orca job stop`. *sshpush.Transport satisfies it via Exec; tests
// inject a mock (same pattern as driftTransport / drainExecer).
type jobStopTransport interface {
	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
}

// jobStopTransportOverride is the package-level test seam for the
// SSH transport used by `orca job stop`. When non-nil it replaces the
// production transport; tests set it and restore nil in cleanup.
var jobStopTransportOverride jobStopTransport

// jobStopTimeout is the SSH command timeout for `orca job stop`.
var jobStopTimeout time.Duration

// jobStopPeer is the optional --peer override for `orca job stop`.
// When empty, the node is looked up from the alloc_history table
// (latest entry for the job id). When set, the SSH stop targets that
// peer directly.
var jobStopPeer string

func jobStopTransportFromCtx() (jobStopTransport, error) {
	if jobStopTransportOverride != nil {
		return jobStopTransportOverride, nil
	}
	keyPath := certpaths.SSHKeyPath()
	khPath := certpaths.KnownHostsPath()
	return sshpush.NewTransport(keyPath, khPath), nil
}

// nodeForJob looks up the node that ran (or is running) a job by
// searching the alloc_history table for the latest entry for the
// given job id. Returns nil if no history entry exists (the job may
// have been run locally or pre-dates alloc_history).
func nodeForJob(ctx context.Context, db *sql.DB, jobID string) (*model.Node, error) {
	hist := store.NewAllocHistoryRepo(db)
	if err := hist.EnsureSchema(ctx); err != nil {
		return nil, fmt.Errorf("alloc history schema: %w", err)
	}
	entries, err := hist.List(ctx, store.HistoryFilter{JobID: jobID})
	if err != nil {
		return nil, fmt.Errorf("alloc history list: %w", err)
	}
	if len(entries) == 0 {
		return nil, nil
	}
	// Pick the latest entry (List returns ASC; take the last).
	latest := entries[len(entries)-1]
	if latest.NodeID == "" {
		return nil, nil
	}
	nodeRepo := store.NewNodeRepo(db)
	n, err := nodeRepo.Get(ctx, latest.NodeID)
	if err != nil {
		if errors.Is(err, store.ErrNotFound) {
			return nil, nil
		}
		return nil, fmt.Errorf("lookup node %s: %w", latest.NodeID, err)
	}
	return n, nil
}

var jobStopCmd = &cobra.Command{
	Use:   "stop [job-id]",
	Short: "Stop a running job",
	Long: `Stop a running job by sending 'systemctl stop orca-alloc-<name>-*'
to the node running the allocation via SSH, then mark the job as
stopped in the DB (REQ-158, P09 T1).

If --peer is not given, the node is looked up from the allocation
history. If no node is found, the DB status is updated anyway (soft
stop fallback for local-run jobs).`,
	Args: cobra.MaximumNArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		id := stopID
		if id == "" && len(args) > 0 {
			id = args[0]
		}
		if id == "" {
			return fmt.Errorf("job id required (--id or argument)")
		}

		db, closer, err := openDB()
		if err != nil {
			return err
		}
		defer closer()

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

		repo := store.NewJobRepo(db)
		job, err := repo.Get(ctx, id)
		if err != nil {
			if errors.Is(err, store.ErrNotFound) {
				return fmt.Errorf("job not found: %s", id)
			}
			return err
		}

		// Determine the peer to SSH to. --peer takes precedence;
		// otherwise look up the node from alloc_history.
		peer := jobStopPeer
		var node *model.Node
		if peer == "" {
			node, err = nodeForJob(ctx, db, id)
			if err != nil {
				return fmt.Errorf("lookup node for job %s: %w", id, err)
			}
			if node != nil {
				peer = peerAddrForNode(node)
			}
		}

		// Validate the job name before interpolation into the shell
		// command (same injection guard as logs --job / stopAlloc).
		jobName := job.Name
		if !validSafeName(jobName) {
			return fmt.Errorf("job stop: invalid job name %q (allowed: A-Z a-z 0-9 _ -)", jobName)
		}

		sshRan := false
		if peer != "" {
			transport, terr := jobStopTransportFromCtx()
			if terr != nil {
				return fmt.Errorf("job stop: ssh transport: %w", terr)
			}
			stopCtx, stopCancel := sshCmdCtx(ctx, jobStopTimeout)
			defer stopCancel()
			// Match the drift.go job restart unit pattern: orca-alloc-<name>.
			// Use a glob (orca-alloc-<name>-*) to stop all task units in
			// a multi-task allocation group.
			unitPattern := fmt.Sprintf("orca-alloc-%s-*", jobName)
			stopCmd := fmt.Sprintf("systemctl stop %s", shellQuote(unitPattern))
			out, sErr := transport.Exec(stopCtx, peer, stopCmd)
			if sErr != nil {
				// Non-fatal: the unit may not be running (already
				// stopped) or SSH may fail. We still update the DB
				// status so the operator's intent is recorded.
				fmt.Fprintf(cmd.ErrOrStderr(), "⚠ job stop: SSH systemctl stop failed on %s: %v (output: %s)\n", peer, sErr, strings.TrimSpace(string(out)))
			} else {
				sshRan = true
			}
		}

		if err := repo.UpdateStatus(ctx, id, model.JobStatusStopped, 130); err != nil {
			return err
		}
		// REQ-156 / P07 T5: invalidate the jobs cache so the next
		// `orca job list` reflects the just-stopped job.
		cacheInvalidate(cacheJobClass)
		if jsonOutput {
			result := map[string]any{"id": id, "status": "stopped", "previous_status": job.Status}
			if peer != "" {
				result["peer"] = peer
				result["ssh_stop"] = sshRan
			}
			return printJSON(result)
		}
		if peer != "" && sshRan {
			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (systemctl stop on %s)\n", id, peer)
		} else if peer != "" {
			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (DB only; SSH stop failed — see stderr)\n", id)
		} else {
			fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (DB only; no node found)\n", id)
		}
		return nil
	},
}

var jobLogsCmd = &cobra.Command{
	Use:   "logs [job-id]",
	Short: "Show task output for a job",
	Long:  "Display captured stdout/stderr for all tasks in a job.",
	Args:  cobra.MaximumNArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		id := stopID
		if id == "" && len(args) > 0 {
			id = args[0]
		}
		if id == "" {
			return fmt.Errorf("job id required (--id or argument)")
		}
		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
		defer cancel()

		db, closer, err := openDB()
		if err != nil {
			return err
		}
		defer closer()

		taskRepo := store.NewTaskRepo(db)
		tasks, err := taskRepo.ListByJob(ctx, id)
		if err != nil {
			return err
		}
		if jsonOutput {
			return printJSON(tasks)
		}
		if len(tasks) == 0 {
			fmt.Fprintln(cmd.OutOrStdout(), "No tasks for this job.")
			return nil
		}
		for i, t := range tasks {
			fmt.Fprintf(cmd.OutOrStdout(), "--- task[%d] %s (%s) exit=%d ---\n", i, t.Command, t.Status, t.ExitCode)
			if t.Stdout != "" {
				fmt.Fprintln(cmd.OutOrStdout(), t.Stdout)
			}
			if t.Stderr != "" {
				fmt.Fprintln(cmd.OutOrStderr(), t.Stderr)
			}
		}
		return nil
	},
}

func init() {
	jobStopCmd.Flags().StringVar(&stopID, "id", "", "job id")
	jobStopCmd.Flags().StringVar(&jobStopPeer, "peer", "", "peer address (host:port) running the allocation (auto-detected from alloc history if empty)")
	jobStopCmd.Flags().DurationVar(&jobStopTimeout, "timeout", sshCmdDefaultTimeout, "SSH command timeout")
	jobLogsCmd.Flags().StringVar(&stopID, "id", "", "job id")
	jobRunCmd.Flags().StringVar(&runTarget, "target", "", "pin job to a specific node id (overrides bin-packing)")
	jobRunCmd.Flags().StringVar(&runIDKey, "idempotency-key", "", "X-Orca-Idempotency-Key for cross-node dispatch dedupe")
	jobListCmd.Flags().BoolVar(&jobWatch, "watch", false, "stream jobs until Ctrl-C (table refresh or --json per-event)")

	jobCmd.AddCommand(jobRunCmd)
	jobCmd.AddCommand(jobListCmd)
	jobCmd.AddCommand(jobStopCmd)
	jobCmd.AddCommand(jobLogsCmd)
	rootCmd.AddCommand(jobCmd)
}

func toTaskSpecs(in []jobspec.TaskSpec) []engine.TaskSpec {
	out := make([]engine.TaskSpec, len(in))
	for i, t := range in {
		out[i] = engine.TaskSpec{
			Name:    t.Name,
			Command: t.Command,
			Args:    t.Args,
			Env:     t.Env,
		}
	}
	return out
}

// workloadToTaskSpecs converts a *WorkloadSpec into the engine.TaskSpec
// slice consumed by the executor. For the HCL adapter path the runtime
// block carries the legacy task[0].Command; for the Markdown path the
// runtime block is the canonical runtime abstraction (P07 will expand
// this). When Runtime is nil we emit a single no-op task to preserve
// the legacy "at least one task" invariant.
//
// The runtime command string is split into binary + args via
// splitCommand so that exec.Command receives the binary path and the
// args as separate elements. Without this split, a command like
// "/usr/bin/httpd -f /etc/orca/web-app/httpd.conf" is treated as a
// single file path and fork/exec fails with "no such file or directory".
func workloadToTaskSpecs(spec *jobspec.WorkloadSpec) []engine.TaskSpec {
	if spec == nil {
		return nil
	}
	if spec.Runtime == nil {
		return []engine.TaskSpec{{Name: spec.Name, Command: "/bin/true"}}
	}
	bin, args := splitCommand(spec.Runtime.Command)
	return []engine.TaskSpec{{
		Name:    spec.Name,
		Command: bin,
		Args:    args,
	}}
}

// splitCommand splits a command string into binary + args using
// strings.Fields (handles multiple spaces/tabs). If the string is empty
// or all-whitespace, returns ("/bin/true", nil) so the executor still
// has a valid binary to run.
func splitCommand(s string) (string, []string) {
	parts := strings.Fields(s)
	if len(parts) == 0 {
		return "/bin/true", nil
	}
	return parts[0], parts[1:]
}

// insertRemoteJob inserts a model.Job row for a remotely-deployed job
// (REQ-166, Phase C2). Without this, `job list` shows nothing for remote
// deployments and `job stop` can't find the node.
func insertRemoteJob(spec *jobspec.WorkloadSpec, node string) error {
	db, closer, err := openDB()
	if err != nil {
		return err
	}
	defer closer()
	repo := store.NewJobRepo(db)
	return repo.Insert(context.Background(), &model.Job{
		ID:        uuid.NewString(),
		Name:      spec.Name,
		Spec:      "",
		Status:    model.JobStatusRunning,
		CreatedAt: time.Now().UTC(),
		Node:      node,
	})
}