orca/internal/cli/job.go

package cli

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"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
)

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.Job.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.Job.Name,
			Spec:   args[0],
			Status: model.JobStatusPending,
		}
		if err := exec.Run(ctx, job, toTaskSpecs(spec.Tasks)); 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.",
	RunE: func(cmd *cobra.Command, args []string) error {
		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
	},
}

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")

	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
}