feat(job): complete deferred job-file features

Implements the remaining items from issue #20: - version is now forward-permissive: any value >= 1 is accepted; a newer-than-known version loads best-effort (unknown fields ignored, warning printed) instead of hard-failing on "must be 1" - from_job input reference: `inputs: [{ from_job: <job> }]` resolves to that job's single-file output + format and implies a dependency edge; combined depends_on + from_job graph gets topological ordering and cycle detection - logfile size-rotation, on by default (5MB, keep 3), overridable per job (log_max_size / log_keep) or file-wide via a top-level defaults block - new commands: split (schema/table subsetting via select:), inspect (rule validation -> markdown/json report, fails job on enforced-rule errors), diff (compare exactly two schemas, never fails), scripts-exec (run SQL script dirs against a live PostgreSQL database) - atomic single-file output/report writes (temp file + rename) - symlink-escape hardening in SafeJoin via EvalSymlinks preflight Updates docs/JOB_FILES.md and examples/jobs/relspec.yml accordingly.
2026-09-08 14:32:07 +02:00
parent f968e3d4a6
commit 84a6b31873
7 changed files with 1434 additions and 62 deletions
@@ -1,6 +1,7 @@
 package main

 import (
+	"bytes"
 	"fmt"
 	"io"
 	"os"
@@ -11,6 +12,8 @@ import (

 	"github.com/spf13/cobra"

+	"git.warky.dev/wdevs/relspecgo/pkg/diff"
+	"git.warky.dev/wdevs/relspecgo/pkg/inspector"
 	"git.warky.dev/wdevs/relspecgo/pkg/jobs"
 	"git.warky.dev/wdevs/relspecgo/pkg/merge"
 	"git.warky.dev/wdevs/relspecgo/pkg/models"
@@ -18,6 +21,7 @@ import (
 	"git.warky.dev/wdevs/relspecgo/pkg/readers/sqldir"
 	"git.warky.dev/wdevs/relspecgo/pkg/writers"
 	wpgsql "git.warky.dev/wdevs/relspecgo/pkg/writers/pgsql"
+	"git.warky.dev/wdevs/relspecgo/pkg/writers/sqlexec"
 	wtemplate "git.warky.dev/wdevs/relspecgo/pkg/writers/template"
 )

@@ -122,6 +126,9 @@ func loadJobSet() (*jobs.Set, error) {
 	if err := set.Validate(); err != nil {
 		return nil, err
 	}
+	for _, w := range set.Warnings {
+		fmt.Fprintf(os.Stderr, "warning: %s\n", w)
+	}
 	return set, nil
 }

@@ -180,12 +187,14 @@ func executeJobPlan(set *jobs.Set, name string, dryRun, noDeps bool, out io.Writ
 	// 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))
+	byName := make(map[string]*resolvedJob, len(plan))
 	for i, j := range plan {
-		rj, perr := preflightJob(j)
+		rj, perr := preflightJob(j, byName)
 		if perr != nil {
 			return fmt.Errorf("job %q: %w", j.Name, perr)
 		}
 		resolved[i] = rj
+		byName[j.Name] = rj
 	}

 	if dryRun {
@@ -217,7 +226,12 @@ type resolvedJob struct {
 	outputConn    string // resolved connection string (secret)
 	outputConnEnv string
 	logPath       string
+	logPolicy     jobs.LogPolicy
 	templatePath  string
+	reportPath    string // "" for a diff summary written to the log
+	reportFormat  string
+	rulesPath     string // "" means inspector defaults
+	selection     *splitSelection
 	secrets       []string // resolved secret values to redact from logs
 }

@@ -226,11 +240,12 @@ type resolvedInput struct {
 	path    string // "" when the input is a database
 	conn    string // resolved connection string (secret)
 	connEnv string
+	fromJob string // producer job name when this input came from from_job
 }

-func preflightJob(j *jobs.Job) (*resolvedJob, error) {
+func preflightJob(j *jobs.Job, resolvedByName map[string]*resolvedJob) (*resolvedJob, error) {
 	root := j.Dir()
-	rj := &resolvedJob{job: j, root: root}
+	rj := &resolvedJob{job: j, root: root, logPolicy: j.ResolvedLogPolicy()}

 	if j.Logfile != "" {
 		p, err := jobs.SafeJoin(root, j.Logfile)
@@ -253,6 +268,20 @@ func preflightJob(j *jobs.Job) (*resolvedJob, error) {

 	for i, in := range j.Inputs {
 		ri := resolvedInput{format: strings.ToLower(in.Format)}
+		if in.FromJob != "" {
+			producer, ok := resolvedByName[in.FromJob]
+			if !ok {
+				return nil, fmt.Errorf("input[%d]: from_job %q is not in this plan (do not use --no-deps with from_job inputs)", i, in.FromJob)
+			}
+			if producer.outputPath == "" {
+				return nil, fmt.Errorf("input[%d]: from_job %q does not write a file output", i, in.FromJob)
+			}
+			ri.path = producer.outputPath
+			ri.format = strings.ToLower(producer.job.Output.Format)
+			ri.fromJob = in.FromJob
+			rj.inputs = append(rj.inputs, ri)
+			continue
+		}
 		if in.ConnEnv != "" {
 			v, ok := os.LookupEnv(in.ConnEnv)
 			if !ok || v == "" {
@@ -313,6 +342,43 @@ func preflightJob(j *jobs.Job) (*resolvedJob, error) {
 			rj.outputPath = p
 		}
 	}
+
+	if j.Rules != "" {
+		p, err := jobs.SafeJoin(root, j.Rules)
+		if err != nil {
+			return nil, fmt.Errorf("rules: %w", err)
+		}
+		info, err := os.Stat(p)
+		if err != nil || info.IsDir() {
+			return nil, fmt.Errorf("rules %q: not found or is a directory", j.Rules)
+		}
+		rj.rulesPath = p
+	}
+
+	if j.Report != nil {
+		rj.reportFormat = strings.ToLower(j.Report.Format)
+		if j.Report.Path != "" {
+			p, err := jobs.SafeJoin(root, j.Report.Path)
+			if err != nil {
+				return nil, fmt.Errorf("report: %w", err)
+			}
+			if _, err := os.Stat(p); err == nil && !j.Report.Overwrite {
+				return nil, fmt.Errorf("report %s already exists (set report.overwrite: true to replace it)", j.Report.Path)
+			}
+			rj.reportPath = p
+		}
+	}
+
+	if j.Select != nil {
+		rj.selection = &splitSelection{
+			Schemas:        j.Select.Schemas,
+			Tables:         j.Select.Tables,
+			ExcludeSchemas: j.Select.ExcludeSchemas,
+			ExcludeTables:  j.Select.ExcludeTables,
+			DatabaseName:   j.Select.DatabaseName,
+		}
+	}
+
 	return rj, nil
 }

@@ -325,9 +391,12 @@ func printResolvedJob(out io.Writer, n, total int, rj *resolvedJob) {
 	}
 	fmt.Fprintf(out, "      job file: %s\n", j.SourceFile)
 	for _, ri := range rj.inputs {
-		if ri.path != "" {
+		switch {
+		case ri.fromJob != "":
+			fmt.Fprintf(out, "      input: %s (%s) from job %q\n", ri.path, ri.format, ri.fromJob)
+		case ri.path != "":
 			fmt.Fprintf(out, "      input: %s (%s)\n", ri.path, ri.format)
-		} else {
+		default:
 			fmt.Fprintf(out, "      input: env:%s (%s)\n", ri.connEnv, ri.format)
 		}
 	}
@@ -339,15 +408,31 @@ func printResolvedJob(out io.Writer, n, total int, rj *resolvedJob) {
 	} else if rj.outputConnEnv != "" {
 		fmt.Fprintf(out, "      output: env:%s (%s)\n", rj.outputConnEnv, j.Output.Format)
 	}
+	if j.Report != nil {
+		format := valueOr(rj.reportFormat, "default")
+		if rj.reportPath != "" {
+			fmt.Fprintf(out, "      report: %s (%s)\n", rj.reportPath, format)
+		} else {
+			fmt.Fprintf(out, "      report: (log) (%s)\n", format)
+		}
+	}
+	if rj.rulesPath != "" {
+		fmt.Fprintf(out, "      rules: %s\n", rj.rulesPath)
+	} else if j.Command == jobs.CommandInspect {
+		fmt.Fprintf(out, "      rules: (built-in defaults)\n")
+	}
+	if rj.selection != nil {
+		fmt.Fprintf(out, "      select: %s\n", rj.selection.summary())
+	}
 	if rj.logPath != "" {
-		fmt.Fprintf(out, "      logfile: %s\n", rj.logPath)
+		fmt.Fprintf(out, "      logfile: %s (rotate >= %d bytes, keep %d)\n", rj.logPath, rj.logPolicy.MaxSizeBytes, rj.logPolicy.Keep)
 	}
 	fmt.Fprintln(out)
 }

 // executeResolvedJob runs a single already-validated job.
 func executeResolvedJob(rj *resolvedJob) (err error) {
-	lg, closeLog, lerr := newJobLogger(rj.logPath, rj.secrets)
+	lg, closeLog, lerr := newJobLogger(rj.logPath, rj.logPolicy, rj.secrets)
 	if lerr != nil {
 		return lerr
 	}
@@ -364,6 +449,14 @@ func executeResolvedJob(rj *resolvedJob) (err error) {
 		err = runScriptsListJob(rj, lg)
 	case jobs.CommandTempl:
 		err = runTemplJob(rj, lg)
+	case jobs.CommandSplit:
+		err = runSplitJob(rj, lg)
+	case jobs.CommandInspect:
+		err = runInspectJob(rj, lg)
+	case jobs.CommandDiff:
+		err = runDiffJob(rj, lg)
+	case jobs.CommandScriptsExec:
+		err = runScriptsExecJob(rj, lg)
 	default:
 		err = fmt.Errorf("unsupported command %q", rj.job.Command)
 	}
@@ -505,6 +598,168 @@ func runScriptsListJob(rj *resolvedJob, lg *jobLogger) error {
 	return nil
 }

+func runSplitJob(rj *resolvedJob, lg *jobLogger) error {
+	db, err := readJobInputs(rj, lg)
+	if err != nil {
+		return err
+	}
+	sel := splitSelection{}
+	if rj.selection != nil {
+		sel = *rj.selection
+	}
+	filtered, err := filterDatabaseSelection(db, sel)
+	if err != nil {
+		return fmt.Errorf("split selection: %w", err)
+	}
+	if sel.DatabaseName != "" {
+		filtered.Name = sel.DatabaseName
+	}
+	tables := 0
+	for _, s := range filtered.Schemas {
+		tables += len(s.Tables)
+	}
+	lg.logf("split: selected %d schema(s), %d table(s)", len(filtered.Schemas), tables)
+	return writeJobOutput(rj, filtered, lg)
+}
+
+func runInspectJob(rj *resolvedJob, lg *jobLogger) error {
+	db, err := readJobInputs(rj, lg)
+	if err != nil {
+		return err
+	}
+	config, err := inspector.LoadConfig(rj.rulesPath) // "" -> built-in defaults
+	if err != nil {
+		return fmt.Errorf("load rules: %w", err)
+	}
+	report, err := inspector.NewInspector(db, config).Inspect()
+	if err != nil {
+		return fmt.Errorf("inspection failed: %w", err)
+	}
+
+	var formatted string
+	switch valueOr(rj.reportFormat, "markdown") {
+	case "json":
+		formatted, err = inspector.NewJSONFormatter().Format(report)
+	default:
+		formatted, err = inspector.NewMarkdownFormatter(io.Discard).Format(report)
+	}
+	if err != nil {
+		return fmt.Errorf("format report: %w", err)
+	}
+	if werr := atomicWrite(rj.reportPath, func(tmp string) error {
+		return os.WriteFile(tmp, []byte(formatted), 0o644)
+	}); werr != nil {
+		return werr
+	}
+	lg.logf("inspect: %d error(s), %d warning(s) -> %s",
+		report.Summary.ErrorCount, report.Summary.WarningCount, rj.reportPath)
+	if report.HasErrors() {
+		return fmt.Errorf("inspection found %d error(s)", report.Summary.ErrorCount)
+	}
+	return nil
+}
+
+func runDiffJob(rj *resolvedJob, lg *jobLogger) error {
+	if len(rj.inputs) != 2 {
+		return fmt.Errorf("diff requires exactly 2 inputs, got %d", len(rj.inputs))
+	}
+	source, err := readOneJobInput(rj.inputs[0])
+	if err != nil {
+		return fmt.Errorf("input[0]: %w", err)
+	}
+	lg.logf("diff source: %s", inputLabel(rj.inputs[0]))
+	target, err := readOneJobInput(rj.inputs[1])
+	if err != nil {
+		return fmt.Errorf("input[1]: %w", err)
+	}
+	lg.logf("diff target: %s", inputLabel(rj.inputs[1]))
+
+	result := diff.CompareDatabases(source, target)
+	s := diff.ComputeSummary(result)
+	lg.logf("diff: schemas %d/%d/%d, tables %d/%d/%d, columns %d/%d/%d (missing/extra/modified)",
+		s.Schemas.Missing, s.Schemas.Extra, s.Schemas.Modified,
+		s.Tables.Missing, s.Tables.Extra, s.Tables.Modified,
+		s.Columns.Missing, s.Columns.Extra, s.Columns.Modified)
+
+	format := diff.FormatSummary
+	switch rj.reportFormat {
+	case "json":
+		format = diff.FormatJSON
+	case "html":
+		format = diff.FormatHTML
+	}
+
+	if rj.reportPath == "" {
+		var buf bytes.Buffer
+		if err := diff.FormatDiff(result, format, &buf); err != nil {
+			return fmt.Errorf("format diff: %w", err)
+		}
+		for _, line := range strings.Split(strings.TrimRight(buf.String(), "\n"), "\n") {
+			lg.logf("%s", line)
+		}
+		return nil
+	}
+	if werr := atomicWrite(rj.reportPath, func(tmp string) error {
+		f, err := os.Create(tmp)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		return diff.FormatDiff(result, format, f)
+	}); werr != nil {
+		return werr
+	}
+	lg.logf("diff report written: %s", rj.reportPath)
+	return nil
+}
+
+func runScriptsExecJob(rj *resolvedJob, lg *jobLogger) error {
+	schemaName := valueOr(rj.job.Options.Schema, "public")
+	combined := &models.Schema{Name: schemaName}
+	for _, dir := range rj.scriptDirs {
+		reader := sqldir.NewReader(&readers.ReaderOptions{
+			FilePath: dir,
+			Metadata: map[string]any{
+				"schema_name":   schemaName,
+				"database_name": "database",
+			},
+		})
+		db, err := reader.ReadDatabase()
+		if err != nil {
+			return fmt.Errorf("%s: %w", dir, err)
+		}
+		if len(db.Schemas) == 0 {
+			continue
+		}
+		combined.Scripts = append(combined.Scripts, db.Schemas[0].Scripts...)
+	}
+	if len(combined.Scripts) == 0 {
+		lg.logf("no scripts found; nothing to execute")
+		return nil
+	}
+	lg.logf("executing %d script(s) against database env:%s", len(combined.Scripts), rj.outputConnEnv)
+
+	writer := sqlexec.NewWriter(&writers.WriterOptions{
+		Metadata: map[string]any{
+			"connection_string": rj.outputConn,
+			"ignore_errors":     rj.job.Options.ContinueOnError,
+		},
+	})
+	if err := writer.WriteSchema(combined); err != nil {
+		return fmt.Errorf("script execution failed: %w", err)
+	}
+
+	opts := writer.Options()
+	total, _ := opts.Metadata["execution_total"].(int)
+	success, _ := opts.Metadata["execution_success"].(int)
+	failed, _ := opts.Metadata["execution_failed"].(int)
+	lg.logf("executed %d script(s): %d succeeded, %d failed", total, success, failed)
+	if failed > 0 && !rj.job.Options.ContinueOnError {
+		return fmt.Errorf("%d script(s) failed", failed)
+	}
+	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
@@ -559,7 +814,37 @@ func writeJobOutput(rj *resolvedJob, db *models.Database, lg *jobLogger) error {
 		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, "")
+
+	write := func(target string) error {
+		return writeDatabase(db, format, target, o.Package, o.Schema, o.FlattenSchema, "", "", o.ContinueOnError, "")
+	}
+	// Single-file formats are written to a temp file and renamed into place so
+	// a failure never leaves a partial or truncated output. Directory-emitting
+	// formats (gorm/bun/drizzle/typeorm/prisma) write in place.
+	if jobs.SingleFileOutputFormat(format) {
+		return atomicWrite(rj.outputPath, write)
+	}
+	return write(rj.outputPath)
+}
+
+// atomicWrite calls produce with a temp path in the same directory as
+// finalPath, then renames it over finalPath. The temp file is removed on any
+// error so the destination is only ever replaced by a complete file.
+func atomicWrite(finalPath string, produce func(tmpPath string) error) error {
+	dir := filepath.Dir(finalPath)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return fmt.Errorf("failed to create output directory: %w", err)
+	}
+	tmp := filepath.Join(dir, fmt.Sprintf(".%s.relspec-tmp-%d", filepath.Base(finalPath), os.Getpid()))
+	if err := produce(tmp); err != nil {
+		_ = os.Remove(tmp)
+		return err
+	}
+	if err := os.Rename(tmp, finalPath); err != nil {
+		_ = os.Remove(tmp)
+		return fmt.Errorf("failed to finalize %s: %w", finalPath, err)
+	}
+	return nil
 }

 // --- logging + redaction ---------------------------------------------------
@@ -572,7 +857,7 @@ type jobLogger struct {
 // 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) {
+func newJobLogger(path string, policy jobs.LogPolicy, secrets []string) (*jobLogger, func(err error), error) {
 	lg := &jobLogger{secrets: secrets}
 	if path == "" {
 		return lg, func(error) {}, nil
@@ -580,6 +865,7 @@ func newJobLogger(path string, secrets []string) (*jobLogger, func(err error), e
 	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
 		return nil, nil, fmt.Errorf("failed to create log directory: %w", err)
 	}
+	rotateLogIfNeeded(path, policy)
 	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)
@@ -593,6 +879,28 @@ func newJobLogger(path string, secrets []string) (*jobLogger, func(err error), e
 	}, nil
 }

+// rotateLogIfNeeded renames path -> path.1 -> path.2 ... up to policy.Keep
+// when path has grown to policy.MaxSizeBytes or more. The oldest file beyond
+// Keep is deleted. A zero/negative MaxSizeBytes disables rotation.
+func rotateLogIfNeeded(path string, policy jobs.LogPolicy) {
+	if policy.MaxSizeBytes <= 0 {
+		return
+	}
+	info, err := os.Stat(path)
+	if err != nil || info.Size() < policy.MaxSizeBytes {
+		return
+	}
+	if policy.Keep < 1 {
+		_ = os.Remove(path)
+		return
+	}
+	_ = os.Remove(fmt.Sprintf("%s.%d", path, policy.Keep))
+	for i := policy.Keep - 1; i >= 1; i-- {
+		_ = os.Rename(fmt.Sprintf("%s.%d", path, i), fmt.Sprintf("%s.%d", path, i+1))
+	}
+	_ = os.Rename(path, path+".1")
+}
+
 func (l *jobLogger) logf(format string, args ...interface{}) {
 	line := l.redact(fmt.Sprintf(format, args...))
 	fmt.Fprintf(os.Stderr, "      %s\n", line)