relspecgo/cmd/relspec/job.go

package main

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

	"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"
	"git.warky.dev/wdevs/relspecgo/pkg/readers"
	"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"
)

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, templ). 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
	}
	for _, w := range set.Warnings {
		fmt.Fprintf(os.Stderr, "warning: %s\n", w)
	}
	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))
	byName := make(map[string]*resolvedJob, len(plan))
	for i, j := range plan {
		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 {
		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
	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
}

type resolvedInput struct {
	format  string
	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, resolvedByName map[string]*resolvedJob) (*resolvedJob, error) {
	root := j.Dir()
	rj := &resolvedJob{job: j, root: root, logPolicy: j.ResolvedLogPolicy()}

	if j.Logfile != "" {
		p, err := jobs.SafeJoin(root, j.Logfile)
		if err != nil {
			return nil, fmt.Errorf("logfile: %w", err)
		}
		rj.logPath = p
	}
	if j.Template != "" {
		p, err := jobs.SafeJoin(root, j.Template)
		if err != nil {
			return nil, fmt.Errorf("template: %w", err)
		}
		info, err := os.Stat(p)
		if err != nil || info.IsDir() {
			return nil, fmt.Errorf("template %q: not found or is a directory", j.Template)
		}
		rj.templatePath = p
	}

	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 == "" {
				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
		}
	}

	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
}

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 {
		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)
		default:
			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 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 (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.logPolicy, 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)
	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)
	}
	if err != nil {
		lg.logf("FAILED: %v", err)
	} else {
		lg.logf("OK")
	}
	return err
}

func runTemplJob(rj *resolvedJob, lg *jobLogger) error {
	db, err := readJobInputs(rj, lg)
	if err != nil {
		return err
	}
	if schema := rj.job.Options.Schema; schema != "" {
		found := false
		for _, s := range db.Schemas {
			if s.Name == schema {
				db.Schemas = []*models.Schema{s}
				found = true
				break
			}
		}
		if !found {
			return fmt.Errorf("schema not found: %s", schema)
		}
	}
	mode := rj.job.Mode
	if mode == "" {
		mode = "database"
	}
	pattern := rj.job.FilenamePattern
	if pattern == "" {
		pattern = "{{.Name}}.txt"
	}
	writer, err := wtemplate.NewWriter(&writers.WriterOptions{
		OutputPath: rj.outputPath,
		Metadata: map[string]interface{}{
			"template_path":    rj.templatePath,
			"mode":             mode,
			"filename_pattern": pattern,
		},
	})
	if err != nil {
		return fmt.Errorf("create template writer: %w", err)
	}
	lg.logf("applying template: %s (mode %s)", rj.templatePath, mode)
	if err := writer.WriteDatabase(db); err != nil {
		return fmt.Errorf("execute template: %w", err)
	}
	return nil
}

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
}

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

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

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, policy jobs.LogPolicy, 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)
	}
	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)
	}
	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
}

// 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)
	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
}