orca/internal/cli/job.go

package cli

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

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

	"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/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.hcl>",
	Short: "Run a job from an HCL 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 {
		spec, err := jobspec.ParseFile(args[0])
		if err != nil {
			return err
		}

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

		exec, closer, err := jobExecutor()
		if err != nil {
			return err
		}
		defer closer()

		// If --target or --idempotency-key is set, route through the
		// dispatcher (which may land the job locally or on a peer
		// based on capacity).
		if runTarget != "" || runIDKey != "" {
			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
		}

		job := &model.Job{
			ID:     uuid.NewString(),
			Name:   spec.Name,
			Spec:   args[0],
			Status: model.JobStatusPending,
		}
		if err := exec.Run(ctx, job, workloadToTaskSpecs(spec)); err != nil {
			if jsonOutput {
				_ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": err.Error()})
				return err
			}
			fmt.Fprintf(cmd.ErrOrStderr(), "✗ Job %s failed: %v\n", job.ID, err)
			return err
		}
		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
	},
}

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)
		}
		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
		}
		if jsonOutput {
			return printJSON(jobs)
		}
		if len(jobs) == 0 {
			fmt.Fprintln(cmd.OutOrStdout(), "No jobs. Use 'orca job run <spec.hcl>' to submit one.")
			return nil
		}
		fmt.Fprintf(cmd.OutOrStdout(), "%-36s  %-20s  %-12s  %-8s\n", "ID", "NAME", "STATUS", "EXIT")
		for _, j := range jobs {
			fmt.Fprintf(cmd.OutOrStdout(), "%-36s  %-20s  %-12s  %-8d\n", j.ID, j.Name, j.Status, j.ExitCode)
		}
		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("%-36s  %-20s  %-12s  %-8s\n", "ID", "NAME", "STATUS", "EXIT")
	for _, j := range jobs {
		out += fmt.Sprintf("%-36s  %-20s  %-12s  %-8d\n", j.ID, j.Name, j.Status, j.ExitCode)
	}
	return out
}

var jobStopCmd = &cobra.Command{
	Use:   "stop [job-id]",
	Short: "Stop a running job",
	Long:  "Mark a job as stopped. Note: this is a soft stop (cancel context for the daemon).",
	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()

		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
		}
		if err := repo.UpdateStatus(ctx, id, model.JobStatusStopped, 130); err != nil {
			return err
		}
		if jsonOutput {
			return printJSON(map[string]any{"id": id, "status": "stopped", "previous_status": job.Status})
		}
		fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s\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")
	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.
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"}}
	}
	return []engine.TaskSpec{{
		Name:    spec.Name,
		Command: spec.Runtime.Command,
	}}
}