relspecgo/cmd/relspec/job.go

package main

import (
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/spf13/cobra"

	"git.warky.dev/wdevs/relspecgo/pkg/jobs"
	"git.warky.dev/wdevs/relspecgo/pkg/merge"
	"git.warky.dev/wdevs/relspecgo/pkg/models"
	"git.warky.dev/wdevs/relspecgo/pkg/readers"
	"git.warky.dev/wdevs/relspecgo/pkg/readers/sqldir"
	wpgsql "git.warky.dev/wdevs/relspecgo/pkg/writers/pgsql"
)

var (
	jobDir    string
	jobFiles  []string
	jobDryRun bool
	jobNoDeps bool
)

var jobCmd = &cobra.Command{
	Use:   "job",
	Short: "Run declarative RelSpec jobs from job files",
	Long: `Run named jobs declared in job files instead of repeating command-line arguments.

A job file is a YAML manifest (relspec.yml, or relspec.<name>.yml for extra
files) describing one or more jobs. Each job names a RelSpec command plus its
inputs, output and options:

  version: 1
  jobs:
    build-schema:
      command: convert
      description: Merge the DBML sources and emit PostgreSQL DDL
      inputs:
        - path: schema/core.dbml
          format: dbml
        - path: schema/tenant.dbml
          format: dbml
      output:
        format: pgsql
        path: build/schema.sql
        overwrite: true
      options:
        flatten_schema: false
      logfile: .relspec/log/build-schema.log

Rules and guarantees:
  - command is a closed allow-list (convert, merge, scripts-list). Arbitrary
    shell strings are never executed.
  - Every path is relative to the directory holding the job file and may not
    escape it. Absolute and home-relative paths are rejected.
  - Remote database credentials are referenced by environment-variable name
    via conn_env; connection strings are never stored in the manifest and are
    redacted from logs and diagnostics.
  - Discovery and listing are deterministic.
  - The whole plan is validated - unknown commands/formats, duplicate job
    names, missing inputs, path traversal, dependency cycles - before any job
    runs. Nothing is read, written or executed when validation fails.
  - A failed job propagates the underlying non-zero exit status and writes no
    success marker.`,
}

var jobListCmd = &cobra.Command{
	Use:   "list",
	Short: "List jobs discovered in job files (deterministic order)",
	RunE:  runJobList,
}

var jobRunCmd = &cobra.Command{
	Use:   "run <job-name>",
	Short: "Run a named job (and its dependencies) from a job file",
	Args:  cobra.ExactArgs(1),
	RunE:  runJobRun,
}

func init() {
	for _, c := range []*cobra.Command{jobListCmd, jobRunCmd} {
		c.Flags().StringVar(&jobDir, "dir", ".", "Directory to discover job files in")
		c.Flags().StringSliceVar(&jobFiles, "file", nil, "Explicit job file(s) to load (repeatable); disables discovery")
	}
	jobRunCmd.Flags().BoolVar(&jobDryRun, "dry-run", false, "Validate and print the execution plan without running anything")
	jobRunCmd.Flags().BoolVar(&jobDryRun, "plan", false, "Alias for --dry-run")
	jobRunCmd.Flags().BoolVar(&jobNoDeps, "no-deps", false, "Run only the named job, skipping its declared dependencies")

	jobCmd.AddCommand(jobListCmd)
	jobCmd.AddCommand(jobRunCmd)
}

// loadJobSet discovers or loads the requested job files and runs full
// validation. The returned Set is safe to plan and execute.
func loadJobSet() (*jobs.Set, error) {
	paths := jobFiles
	if len(paths) == 0 {
		discovered, err := jobs.Discover(jobDir)
		if err != nil {
			return nil, err
		}
		paths = discovered
	} else {
		for i, p := range paths {
			if _, err := os.Stat(p); err != nil {
				return nil, fmt.Errorf("job file %q: %w", p, err)
			}
			paths[i] = p
		}
	}
	set, err := jobs.Load(paths)
	if err != nil {
		return nil, err
	}
	if err := set.Validate(); err != nil {
		return nil, err
	}
	return set, nil
}

func runJobList(cmd *cobra.Command, args []string) error {
	set, err := loadJobSet()
	if err != nil {
		return err
	}
	out := cmd.OutOrStdout()

	fmt.Fprintf(os.Stderr, "\n=== RelSpec Jobs ===\n")
	fmt.Fprintf(os.Stderr, "Job files:\n")
	for _, f := range set.Files {
		fmt.Fprintf(os.Stderr, "  - %s\n", f)
	}
	fmt.Fprintln(os.Stderr)

	names := set.Names()
	if len(names) == 0 {
		fmt.Fprintln(out, "(no jobs defined)")
		return nil
	}

	nameW, cmdW, srcW := len("NAME"), len("COMMAND"), len("SOURCE")
	for _, n := range names {
		j := set.Jobs[n]
		nameW = maxInt(nameW, len(n))
		cmdW = maxInt(cmdW, len(j.Command))
		srcW = maxInt(srcW, len(j.SourceFile))
	}
	fmt.Fprintf(out, "%-*s  %-*s  %-*s  %s\n", nameW, "NAME", cmdW, "COMMAND", srcW, "SOURCE", "DESCRIPTION")
	for _, n := range names {
		j := set.Jobs[n]
		fmt.Fprintf(out, "%-*s  %-*s  %-*s  %s\n", nameW, n, cmdW, j.Command, srcW, j.SourceFile, j.Description)
	}
	return nil
}

func runJobRun(cmd *cobra.Command, args []string) error {
	set, err := loadJobSet()
	if err != nil {
		return err
	}
	return executeJobPlan(set, args[0], jobDryRun, jobNoDeps, cmd.OutOrStdout())
}

// executeJobPlan resolves the plan for name, runs pre-flight checks over
// EVERY job in the plan, and only then executes. When dryRun is set it prints
// the plan and returns without touching any input, output or database.
func executeJobPlan(set *jobs.Set, name string, dryRun, noDeps bool, out io.Writer) error {
	plan, err := set.Plan(name, !noDeps)
	if err != nil {
		return err
	}

	// Pre-flight: resolve and check paths, output policy and env vars for the
	// whole plan before anything runs. A failure here means no job executes.
	resolved := make([]*resolvedJob, len(plan))
	for i, j := range plan {
		rj, perr := preflightJob(j)
		if perr != nil {
			return fmt.Errorf("job %q: %w", j.Name, perr)
		}
		resolved[i] = rj
	}

	if dryRun {
		fmt.Fprintf(out, "RelSpec job plan for %q (dry run - nothing executed):\n\n", name)
		for i, rj := range resolved {
			printResolvedJob(out, i+1, len(resolved), rj)
		}
		return nil
	}

	for _, rj := range resolved {
		if err := executeResolvedJob(rj); err != nil {
			// Propagate the underlying failure; no success marker is written.
			return fmt.Errorf("job %q failed: %w", rj.job.Name, err)
		}
	}
	fmt.Fprintf(os.Stderr, "\n=== Job %q complete ===\n", name)
	return nil
}

// resolvedJob is a job with every manifest path turned into a checked
// absolute filesystem path and every conn_env resolved to its value.
type resolvedJob struct {
	job           *jobs.Job
	root          string
	inputs        []resolvedInput
	scriptDirs    []string
	outputPath    string // "" when the output is a database
	outputConn    string // resolved connection string (secret)
	outputConnEnv string
	logPath       string
	secrets       []string // resolved secret values to redact from logs
}

type resolvedInput struct {
	format  string
	path    string // "" when the input is a database
	conn    string // resolved connection string (secret)
	connEnv string
}

func preflightJob(j *jobs.Job) (*resolvedJob, error) {
	root := j.Dir()
	rj := &resolvedJob{job: j, root: root}

	if j.Logfile != "" {
		p, err := jobs.SafeJoin(root, j.Logfile)
		if err != nil {
			return nil, fmt.Errorf("logfile: %w", err)
		}
		rj.logPath = p
	}

	for i, in := range j.Inputs {
		ri := resolvedInput{format: strings.ToLower(in.Format)}
		if in.ConnEnv != "" {
			v, ok := os.LookupEnv(in.ConnEnv)
			if !ok || v == "" {
				return nil, fmt.Errorf("input[%d]: environment variable %q (conn_env) is not set", i, in.ConnEnv)
			}
			ri.conn = v
			ri.connEnv = in.ConnEnv
			rj.secrets = append(rj.secrets, v)
		} else {
			p, err := jobs.SafeJoin(root, in.Path)
			if err != nil {
				return nil, fmt.Errorf("input[%d]: %w", i, err)
			}
			info, err := os.Stat(p)
			if err != nil {
				return nil, fmt.Errorf("input[%d]: %s: file not found", i, in.Path)
			}
			if info.IsDir() {
				return nil, fmt.Errorf("input[%d]: %s: is a directory, not a file", i, in.Path)
			}
			ri.path = p
		}
		rj.inputs = append(rj.inputs, ri)
	}

	for _, d := range j.ScriptDirs {
		p, err := jobs.SafeJoin(root, d)
		if err != nil {
			return nil, fmt.Errorf("script_dir %q: %w", d, err)
		}
		info, err := os.Stat(p)
		if err != nil {
			return nil, fmt.Errorf("script_dir %q: not found", d)
		}
		if !info.IsDir() {
			return nil, fmt.Errorf("script_dir %q: not a directory", d)
		}
		rj.scriptDirs = append(rj.scriptDirs, p)
	}

	if j.Output != nil {
		if j.Output.ConnEnv != "" {
			v, ok := os.LookupEnv(j.Output.ConnEnv)
			if !ok || v == "" {
				return nil, fmt.Errorf("output: environment variable %q (conn_env) is not set", j.Output.ConnEnv)
			}
			rj.outputConn = v
			rj.outputConnEnv = j.Output.ConnEnv
			rj.secrets = append(rj.secrets, v)
		} else {
			p, err := jobs.SafeJoin(root, j.Output.Path)
			if err != nil {
				return nil, fmt.Errorf("output: %w", err)
			}
			if _, err := os.Stat(p); err == nil && !j.Output.Overwrite {
				return nil, fmt.Errorf("output %s already exists (set output.overwrite: true to replace it)", j.Output.Path)
			}
			rj.outputPath = p
		}
	}
	return rj, nil
}

func printResolvedJob(out io.Writer, n, total int, rj *resolvedJob) {
	j := rj.job
	fmt.Fprintf(out, "[%d/%d] %s\n", n, total, j.Name)
	fmt.Fprintf(out, "      command: %s\n", j.Command)
	if j.Description != "" {
		fmt.Fprintf(out, "      description: %s\n", j.Description)
	}
	fmt.Fprintf(out, "      job file: %s\n", j.SourceFile)
	for _, ri := range rj.inputs {
		if ri.path != "" {
			fmt.Fprintf(out, "      input: %s (%s)\n", ri.path, ri.format)
		} else {
			fmt.Fprintf(out, "      input: env:%s (%s)\n", ri.connEnv, ri.format)
		}
	}
	for _, d := range rj.scriptDirs {
		fmt.Fprintf(out, "      script dir: %s\n", d)
	}
	if rj.outputPath != "" {
		fmt.Fprintf(out, "      output: %s (%s)\n", rj.outputPath, j.Output.Format)
	} else if rj.outputConnEnv != "" {
		fmt.Fprintf(out, "      output: env:%s (%s)\n", rj.outputConnEnv, j.Output.Format)
	}
	if rj.logPath != "" {
		fmt.Fprintf(out, "      logfile: %s\n", rj.logPath)
	}
	fmt.Fprintln(out)
}

// executeResolvedJob runs a single already-validated job.
func executeResolvedJob(rj *resolvedJob) (err error) {
	lg, closeLog, lerr := newJobLogger(rj.logPath, rj.secrets)
	if lerr != nil {
		return lerr
	}
	defer func() { closeLog(err) }()

	lg.logf("=== job %q (%s) started at %s ===", rj.job.Name, rj.job.Command, time.Now().Format(time.RFC3339))

	switch rj.job.Command {
	case jobs.CommandConvert:
		err = runConvertJob(rj, lg)
	case jobs.CommandMerge:
		err = runMergeJob(rj, lg)
	case jobs.CommandScriptsList:
		err = runScriptsListJob(rj, lg)
	default:
		err = fmt.Errorf("unsupported command %q", rj.job.Command)
	}
	if err != nil {
		lg.logf("FAILED: %v", err)
	} else {
		lg.logf("OK")
	}
	return err
}

func runConvertJob(rj *resolvedJob, lg *jobLogger) error {
	db, err := readJobInputs(rj, lg)
	if err != nil {
		return err
	}
	return writeJobOutput(rj, db, lg)
}

func runMergeJob(rj *resolvedJob, lg *jobLogger) error {
	opts := &merge.MergeOptions{
		SkipDomains:   rj.job.Options.SkipDomains,
		SkipRelations: rj.job.Options.SkipRelations,
		SkipEnums:     rj.job.Options.SkipEnums,
		SkipViews:     rj.job.Options.SkipViews,
		SkipSequences: rj.job.Options.SkipSequences,
	}
	var base *models.Database
	for i, ri := range rj.inputs {
		db, err := readOneJobInput(ri)
		if err != nil {
			return fmt.Errorf("input[%d]: %w", i, err)
		}
		if base == nil {
			base = db
			lg.logf("merge target: %s", inputLabel(ri))
			continue
		}
		lg.logf("merging: %s", inputLabel(ri))
		merge.MergeDatabases(base, db, opts)
	}
	base.UpdateDate()
	return writeJobOutput(rj, base, lg)
}

func runScriptsListJob(rj *resolvedJob, lg *jobLogger) error {
	type row struct {
		priority int
		sequence uint
		name     string
		dir      string
		lines    int
	}
	var rows []row
	for _, dir := range rj.scriptDirs {
		reader := sqldir.NewReader(&readers.ReaderOptions{
			FilePath: dir,
			Metadata: map[string]any{
				"schema_name":   valueOr(rj.job.Options.Schema, "public"),
				"database_name": "database",
			},
		})
		db, err := reader.ReadDatabase()
		if err != nil {
			return fmt.Errorf("%s: %w", dir, err)
		}
		if len(db.Schemas) == 0 {
			continue
		}
		for _, s := range db.Schemas[0].Scripts {
			lines := strings.Count(s.SQL, "\n")
			if len(s.SQL) > 0 && !strings.HasSuffix(s.SQL, "\n") {
				lines++
			}
			rows = append(rows, row{s.Priority, s.Sequence, s.Name, dir, lines})
		}
	}
	sort.Slice(rows, func(i, j int) bool {
		if rows[i].priority != rows[j].priority {
			return rows[i].priority < rows[j].priority
		}
		if rows[i].sequence != rows[j].sequence {
			return rows[i].sequence < rows[j].sequence
		}
		if rows[i].name != rows[j].name {
			return rows[i].name < rows[j].name
		}
		return rows[i].dir < rows[j].dir
	})
	lg.logf("found %d script(s) across %d director(y/ies):", len(rows), len(rj.scriptDirs))
	lg.logf("%-4s %-9s %-9s %-30s %-6s %s", "No.", "Priority", "Sequence", "Name", "Lines", "Directory")
	for i, r := range rows {
		lg.logf("%-4d %-9d %-9d %-30s %-6d %s", i+1, r.priority, r.sequence, r.name, r.lines, r.dir)
	}
	return nil
}

// readJobInputs reads every input and additively merges them into one model.
func readJobInputs(rj *resolvedJob, lg *jobLogger) (*models.Database, error) {
	var base *models.Database
	for i, ri := range rj.inputs {
		db, err := readOneJobInput(ri)
		if err != nil {
			return nil, fmt.Errorf("input[%d]: %w", i, err)
		}
		lg.logf("read input: %s", inputLabel(ri))
		if base == nil {
			base = db
		} else {
			merge.MergeDatabases(base, db, &merge.MergeOptions{})
		}
	}
	if base == nil {
		return nil, fmt.Errorf("no inputs produced a database")
	}
	return base, nil
}

func readOneJobInput(ri resolvedInput) (*models.Database, error) {
	if ri.conn != "" {
		return readDatabaseForConvert(ri.format, "", ri.conn)
	}
	return readDatabaseForConvert(ri.format, ri.path, "")
}

func inputLabel(ri resolvedInput) string {
	if ri.path != "" {
		return fmt.Sprintf("%s (%s)", ri.path, ri.format)
	}
	return fmt.Sprintf("env:%s (%s)", ri.connEnv, ri.format)
}

// writeJobOutput writes db to the job's output target (file or database).
func writeJobOutput(rj *resolvedJob, db *models.Database, lg *jobLogger) error {
	o := rj.job.Options
	format := strings.ToLower(rj.job.Output.Format)

	if rj.outputConn != "" {
		if format != "pgsql" {
			return fmt.Errorf("database output is only supported for pgsql (got %q)", rj.job.Output.Format)
		}
		lg.logf("writing output to database env:%s", rj.outputConnEnv)
		writerOpts := newWriterOptions("", o.Package, o.FlattenSchema, "", "", o.ContinueOnError)
		writerOpts.Metadata = map[string]interface{}{"connection_string": rj.outputConn}
		return wpgsql.NewWriter(writerOpts).WriteDatabase(db)
	}

	if err := os.MkdirAll(filepath.Dir(rj.outputPath), 0o755); err != nil {
		return fmt.Errorf("failed to create output directory: %w", err)
	}
	lg.logf("writing output: %s (%s)", rj.outputPath, format)
	return writeDatabase(db, format, rj.outputPath, o.Package, o.Schema, o.FlattenSchema, "", "", o.ContinueOnError, "")
}

// --- logging + redaction ---------------------------------------------------

type jobLogger struct {
	file    io.Writer
	secrets []string
}

// newJobLogger returns a logger that mirrors to stderr and, when path is set,
// to a job logfile. Connection strings and known secret values are redacted
// from everything it writes.
func newJobLogger(path string, secrets []string) (*jobLogger, func(err error), error) {
	lg := &jobLogger{secrets: secrets}
	if path == "" {
		return lg, func(error) {}, nil
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return nil, nil, fmt.Errorf("failed to create log directory: %w", err)
	}
	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to open logfile %q: %w", path, err)
	}
	lg.file = f
	return lg, func(runErr error) {
		if runErr != nil {
			fmt.Fprintf(f, "%s job ended with error\n", time.Now().Format(time.RFC3339))
		}
		_ = f.Close()
	}, nil
}

func (l *jobLogger) logf(format string, args ...interface{}) {
	line := l.redact(fmt.Sprintf(format, args...))
	fmt.Fprintf(os.Stderr, "      %s\n", line)
	if l.file != nil {
		fmt.Fprintf(l.file, "%s %s\n", time.Now().Format(time.RFC3339), line)
	}
}

func (l *jobLogger) redact(s string) string {
	for _, sec := range l.secrets {
		if sec != "" {
			s = strings.ReplaceAll(s, sec, "***")
		}
	}
	return maskPassword(s)
}

// --- small helpers -------------------------------------------------------

func maxInt(a, b int) int {
	if a > b {
		return a
	}
	return b
}

func valueOr(v, def string) string {
	if v == "" {
		return def
	}
	return v
}