diff --git a/cmd/relspec/job.go b/cmd/relspec/job.go index 0be6426..dd3d9d0 100644 --- a/cmd/relspec/job.go +++ b/cmd/relspec/job.go @@ -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) diff --git a/cmd/relspec/job_test.go b/cmd/relspec/job_test.go index a0ca9ae..fc7938d 100644 --- a/cmd/relspec/job_test.go +++ b/cmd/relspec/job_test.go @@ -376,6 +376,286 @@ jobs: } } +func TestJobRun_SplitJob(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + extract: + command: split + inputs: + - path: schema/core.dbml + format: dbml + - path: schema/tenant.dbml + format: dbml + select: + tables: [users] + output: + format: json + path: build/subset.json + overwrite: true +`) + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + if err := executeJobPlan(set, "extract", false, false, &bytes.Buffer{}); err != nil { + t.Fatalf("execute split job: %v", err) + } + out, err := os.ReadFile(filepath.Join(dir, "build", "subset.json")) + if err != nil { + t.Fatalf("read split output: %v", err) + } + s := string(out) + if !strings.Contains(s, "users") { + t.Fatalf("split output missing selected table:\n%s", s) + } + if strings.Contains(s, "posts") { + t.Fatalf("split output should have excluded posts:\n%s", s) + } +} + +func TestJobRun_InspectJob(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + lint: + command: inspect + inputs: + - path: schema/core.dbml + format: dbml + report: + format: json + path: build/report.json + overwrite: true + logfile: .relspec/lint.log +`) + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + // Default rules only warn, so the job succeeds. + if err := executeJobPlan(set, "lint", false, false, &bytes.Buffer{}); err != nil { + t.Fatalf("execute inspect job: %v", err) + } + if _, err := os.ReadFile(filepath.Join(dir, "build", "report.json")); err != nil { + t.Fatalf("expected report file: %v", err) + } + logData, _ := os.ReadFile(filepath.Join(dir, ".relspec", "lint.log")) + if !strings.Contains(string(logData), "inspect:") { + t.Fatalf("logfile missing inspect summary:\n%s", logData) + } +} + +func TestJobRun_InspectJobFailsOnRuleError(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + lint: + command: inspect + inputs: + - path: schema/core.dbml + format: dbml + rules: rules.yaml + report: + format: json + path: build/report.json + overwrite: true + logfile: .relspec/lint.log +`) + // A rule set to "error" level for a violation the fixture triggers. + writeFile(t, filepath.Join(dir, "rules.yaml"), `version: "1.0" +rules: + primary_key_naming: + enabled: enforce + function: primary_key_naming + pattern: "^id_" + message: "Primary key columns should start with id_" +`) + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + err := executeJobPlan(set, "lint", false, false, &bytes.Buffer{}) + if err == nil || !strings.Contains(err.Error(), "error(s)") { + t.Fatalf("expected inspect job to fail on rule error, got %v", err) + } + logData, _ := os.ReadFile(filepath.Join(dir, ".relspec", "lint.log")) + if !strings.Contains(string(logData), "FAILED") { + t.Fatalf("failed inspect job should log FAILED:\n%s", logData) + } +} + +func TestJobRun_DiffJob(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + compare: + command: diff + inputs: + - path: schema/core.dbml + format: dbml + - path: schema/tenant.dbml + format: dbml + report: + format: json + path: build/diff.json + overwrite: true +`) + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + if err := executeJobPlan(set, "compare", false, false, &bytes.Buffer{}); err != nil { + t.Fatalf("execute diff job: %v", err) + } + out, err := os.ReadFile(filepath.Join(dir, "build", "diff.json")) + if err != nil { + t.Fatalf("read diff report: %v", err) + } + if len(out) == 0 { + t.Fatal("diff report is empty") + } +} + +func TestJobRun_FromJobWiring(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + a: + command: convert + inputs: + - path: schema/core.dbml + format: dbml + output: + format: json + path: build/a.json + overwrite: true + b: + command: convert + inputs: + - from_job: a + output: + format: yaml + path: build/b.yaml + overwrite: true +`) + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + if err := executeJobPlan(set, "b", false, false, &bytes.Buffer{}); err != nil { + t.Fatalf("execute from_job chain: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "build", "a.json")); err != nil { + t.Fatalf("producer output missing: %v", err) + } + out, err := os.ReadFile(filepath.Join(dir, "build", "b.yaml")) + if err != nil { + t.Fatalf("consumer output missing: %v", err) + } + if !strings.Contains(string(out), "users") { + t.Fatalf("consumer did not consume producer output:\n%s", out) + } +} + +func TestJobRun_LogRotation(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + build: + command: scripts-list + script_dirs: [migrations] + log_max_size: "150B" + log_keep: 2 + logfile: .relspec/build.log +`) + writeFile(t, filepath.Join(dir, "migrations", "1_001_a.sql"), "CREATE TABLE a();\n") + logPath := filepath.Join(dir, ".relspec", "build.log") + writeFile(t, logPath, strings.Repeat("x", 300)+"\n") + + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + if err := executeJobPlan(set, "build", false, false, &bytes.Buffer{}); err != nil { + t.Fatalf("execute job: %v", err) + } + rotated, err := os.ReadFile(logPath + ".1") + if err != nil { + t.Fatalf("expected rotated logfile build.log.1: %v", err) + } + if !strings.Contains(string(rotated), strings.Repeat("x", 300)) { + t.Fatalf("rotated logfile should hold the old content") + } + fresh, err := os.ReadFile(logPath) + if err != nil { + t.Fatalf("expected fresh logfile: %v", err) + } + if strings.Contains(string(fresh), strings.Repeat("x", 300)) { + t.Fatalf("fresh logfile should not contain the rotated-out content:\n%s", fresh) + } + if !strings.Contains(string(fresh), "OK") { + t.Fatalf("fresh logfile should hold the new run:\n%s", fresh) + } +} + +func TestJobRun_AtomicOutputLeavesOriginalOnFailure(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + x: + command: convert + inputs: + - path: schema/core.dbml + format: dbml + output: + format: json + path: build/out.json + overwrite: true +`) + // Seed the destination, then make its parent directory read-only so the + // rename step fails. The seeded file must survive intact. + seeded := filepath.Join(dir, "build", "out.json") + writeFile(t, seeded, `{"seeded":true}`) + if err := os.Chmod(filepath.Join(dir, "build"), 0o500); err != nil { + t.Skipf("cannot chmod: %v", err) + } + t.Cleanup(func() { _ = os.Chmod(filepath.Join(dir, "build"), 0o755) }) + + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + if err := executeJobPlan(set, "x", false, false, &bytes.Buffer{}); err == nil { + t.Skip("write unexpectedly succeeded (running as root?)") + } + if err := os.Chmod(filepath.Join(dir, "build"), 0o755); err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(seeded) + if err != nil { + t.Fatalf("seeded file gone: %v", err) + } + if !strings.Contains(string(data), "seeded") { + t.Fatalf("seeded file was corrupted: %s", data) + } +} + +func TestJobRun_ScriptsExecMissingConnEnv(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + migrate: + command: scripts-exec + script_dirs: [migrations] + output: + conn_env: RELSPEC_TEST_EXEC_MISSING + logfile: .relspec/migrate.log +`) + writeFile(t, filepath.Join(dir, "migrations", "1_001_a.sql"), "CREATE TABLE a();\n") + os.Unsetenv("RELSPEC_TEST_EXEC_MISSING") + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + err := executeJobPlan(set, "migrate", false, false, &bytes.Buffer{}) + if err == nil || !strings.Contains(err.Error(), "conn_env") { + t.Fatalf("expected missing conn_env error, got %v", err) + } +} + +func TestJobRun_ScriptsExecDryRun(t *testing.T) { + dir := jobFixture(t, `version: 1 +jobs: + migrate: + command: scripts-exec + script_dirs: [migrations] + output: + conn_env: RELSPEC_TEST_EXEC_CONN +`) + writeFile(t, filepath.Join(dir, "migrations", "1_001_a.sql"), "CREATE TABLE a();\n") + t.Setenv("RELSPEC_TEST_EXEC_CONN", "postgres://u:secretpw@h/db") + set := mustLoadSet(t, filepath.Join(dir, "relspec.yml")) + var buf bytes.Buffer + if err := executeJobPlan(set, "migrate", true, false, &buf); err != nil { + t.Fatalf("dry run: %v", err) + } + if strings.Contains(buf.String(), "secretpw") { + t.Fatalf("plan leaked secret:\n%s", buf.String()) + } + if !strings.Contains(buf.String(), "env:RELSPEC_TEST_EXEC_CONN") { + t.Fatalf("plan should name the env var:\n%s", buf.String()) + } +} + func TestJobRun_TemplDatabaseMode(t *testing.T) { dir := jobFixture(t, `version: 1 jobs: diff --git a/cmd/relspec/split.go b/cmd/relspec/split.go index ceadd65..6d2538f 100644 --- a/cmd/relspec/split.go +++ b/cmd/relspec/split.go @@ -205,8 +205,52 @@ func runSplit(cmd *cobra.Command, args []string) error { return nil } -// filterDatabase filters the database based on provided criteria +// splitSelection is the schema/table selection for a split, independent of the +// CLI flag globals so the job runner can build one directly. +type splitSelection struct { + Schemas []string + Tables []string + ExcludeSchemas []string + ExcludeTables []string + DatabaseName string +} + +// summary renders a one-line human description of the selection. +func (s splitSelection) summary() string { + var parts []string + if len(s.Schemas) > 0 { + parts = append(parts, "schemas="+strings.Join(s.Schemas, ",")) + } + if len(s.Tables) > 0 { + parts = append(parts, "tables="+strings.Join(s.Tables, ",")) + } + if len(s.ExcludeSchemas) > 0 { + parts = append(parts, "exclude_schemas="+strings.Join(s.ExcludeSchemas, ",")) + } + if len(s.ExcludeTables) > 0 { + parts = append(parts, "exclude_tables="+strings.Join(s.ExcludeTables, ",")) + } + if s.DatabaseName != "" { + parts = append(parts, "database_name="+s.DatabaseName) + } + if len(parts) == 0 { + return "(all schemas/tables)" + } + return strings.Join(parts, " ") +} + +// filterDatabase filters the database based on the CLI split flags. func filterDatabase(db *models.Database) (*models.Database, error) { + return filterDatabaseSelection(db, splitSelection{ + Schemas: parseCommaSeparated(splitSchemas), + Tables: parseCommaSeparated(splitTables), + ExcludeSchemas: parseCommaSeparated(splitExcludeSchema), + ExcludeTables: parseCommaSeparated(splitExcludeTables), + }) +} + +// filterDatabaseSelection filters db down to the schemas/tables named by sel. +func filterDatabaseSelection(db *models.Database, sel splitSelection) (*models.Database, error) { filteredDB := &models.Database{ Name: db.Name, Description: db.Description, @@ -220,11 +264,11 @@ func filterDatabase(db *models.Database) (*models.Database, error) { Domains: db.Domains, // Keep domains for now } - // Parse filter flags - includeSchemas := parseCommaSeparated(splitSchemas) - includeTables := parseCommaSeparated(splitTables) - excludeSchemas := parseCommaSeparated(splitExcludeSchema) - excludeTables := parseCommaSeparated(splitExcludeTables) + // Selection criteria + includeSchemas := sel.Schemas + includeTables := sel.Tables + excludeSchemas := sel.ExcludeSchemas + excludeTables := sel.ExcludeTables // Convert table names to lowercase for case-insensitive matching includeTablesLower := make(map[string]bool) diff --git a/docs/JOB_FILES.md b/docs/JOB_FILES.md index e3cc9f6..f5713ba 100644 --- a/docs/JOB_FILES.md +++ b/docs/JOB_FILES.md @@ -9,10 +9,11 @@ relspec job run build-schema --plan # validate + print plan, execute nothing relspec job run build-schema # run the job (and its dependencies) ``` -## Design contract (first release) +## Design contract -This is the smallest coherent contract that is safe and useful end to end. -Anything not listed under "Supported" is intentionally deferred. +This is a deliberately small, safe contract. Every capability is offline-testable +except live database execution (`scripts-exec`), which is validated and planned +offline and only connects at run time. ### Not a shell @@ -24,14 +25,15 @@ means adding a vetted adapter in the RelSpec source. |----------------|--------------------------------------------------------------------| | `convert` | read one or more input schemas, additively merge them, write one output | | `merge` | like `convert` but requires ≥2 inputs and exposes `skip_*` merge options | +| `split` | read one or more schemas, keep the selected schemas/tables, write one output | | `scripts-list` | deterministically list SQL scripts across one or more directories | +| `scripts-exec` | execute SQL scripts across one or more directories against a live PostgreSQL database | | `templ` | apply a custom Go text template to one or more input schemas | +| `inspect` | validate one or more schemas against rules and write a report | +| `diff` | compare exactly two schemas and write a differences report | -Deferred (documented, not implemented here): `scripts` execution against a live -database, `split`, `inspect`, `diff`, job-to-job output wiring, -log rotation/retention. Live SQL execution already exists as -`relspec scripts execute`; wiring it into the job runner is a follow-up because -it needs live database credentials and cannot be covered by offline tests. +`convert`, `merge` and `split` are **producers**: their file output can be fed +directly into another job with `from_job` (see below). ### Discovery and precedence @@ -49,12 +51,16 @@ already forbid duplicate keys within a single file. ### Paths -* Every path (`inputs[].path`, `output.path`, `script_dirs[]`, `logfile`) is - **relative to the directory containing the job file that declared the job**, - not the process working directory. +* Every path (`inputs[].path`, `output.path`, `report.path`, `rules`, + `script_dirs[]`, `template`, `logfile`) is **relative to the directory + containing the job file that declared the job**, not the process working + directory. * Absolute paths, `~`-relative paths and any path that resolves outside the job file directory (`../`, `a/../../b`, …) are **rejected during validation** — before anything runs. +* At run time each path is additionally resolved through its symlinks: a symlink + inside the job-file directory that points outside it is rejected before the + path is opened. ### Credentials @@ -76,58 +82,104 @@ already forbid duplicate keys within a single file. first. Nothing is read, written, connected to, or executed if validation fails. Checks include: -* schema `version` (must be `1`), unknown YAML fields rejected +* schema `version` — **forward-permissive**: any version `>= 1` is accepted. + An omitted `version` is treated as the current one. A version newer than this + build understands loads best-effort (unknown YAML fields are ignored and a + warning is printed); at the current version unknown YAML fields are still + rejected. * duplicate job names across files * unknown / missing `command` -* per-command input/output shape (`convert`/`merge` need inputs + output; - `scripts-list` needs `script_dirs` and forbids inputs/output) +* per-command input/output shape: + * `convert` needs ≥1 input + output; `merge` needs ≥2 inputs + output + * `split` needs ≥1 input + a file output, plus an optional `select:` block + * `scripts-list` needs `script_dirs` and forbids inputs/output + * `scripts-exec` needs `script_dirs` and `output.conn_env` (pgsql only) + * `inspect` needs ≥1 input + `report:` (format `markdown`|`json`) + * `diff` needs **exactly 2** inputs + `report:` (format `summary`|`json`|`html`) * unknown input/output `format` +* `from_job` targets exist, are producers (`convert`/`merge`/`split`) and write a + single-file output * path traversal / absolute / home-relative paths -* `depends_on` targets exist -* dependency cycles (reported as `a -> b -> c -> a`) +* `depends_on` and `from_job` targets exist +* dependency cycles over the combined `depends_on` + `from_job` graph + (reported as `a -> b -> c -> a`) Then, immediately before running, per-job pre-flight resolves paths and checks: -* every input file exists and is a file +* every input file exists and is a file (a `from_job` input is exempt — its + producer runs earlier in the same plan) * every `script_dir` exists and is a directory * every `conn_env` variable is set -* `output.path` does not already exist unless `output.overwrite: true` +* `output.path` / `report.path` does not already exist unless the matching + `overwrite: true` is set +* `rules` (inspect), when given, exists and is a file +* symlinks in every resolved path stay inside the job-file directory If any pre-flight check fails for **any** job in the plan, **no** job runs. ### Execution and exit codes -* `relspec job run ` runs the job's `depends_on` closure first, in - topological order (deterministic), then the job. `--no-deps` runs only the - named job. +* `relspec job run ` runs the job's dependency closure first + (`depends_on` plus any `from_job` producers), in topological order + (deterministic), then the job. `--no-deps` runs only the named job and is + incompatible with `from_job` inputs. * `--dry-run` (alias `--plan`) prints the resolved plan and exits 0 without touching inputs, outputs or databases. * A failing job returns the underlying non-zero status (the process exits 1) and the error names the job. The logfile records `FAILED: `; a successful job records `OK`. No separate success-marker file is written, so a failure can never leave a stale "success". +* `inspect` fails the job when the report contains rule **errors** (enforced + rules); warnings do not fail it. `diff` never fails on differences. +* Single-file outputs and reports are written to a temporary file in the target + directory and atomically renamed into place, so an interrupted run never + leaves a partial file. Directory-emitting formats (`gorm`, `bun`, `drizzle`, + `typeorm`, `prisma`) are written in place. + +### Logfile rotation + +When a job has a `logfile`, it is size-rotated before each run. Defaults are +**5 MB** with **3** rotated files kept (`build.log` → `build.log.1` → …). Override +per job with `log_max_size` / `log_keep`, or for a whole file with a top-level +`defaults:` block. `log_max_size` accepts `B`/`KB`/`MB`/`GB` suffixes (e.g. +`"512KB"`, `"5MB"`). ## Schema reference ```yaml -version: 1 # required, must be 1 +version: 1 # optional; any value >= 1 is accepted +defaults: # optional, file-wide + log_max_size: 5MB # B / KB / MB / GB + log_keep: 3 jobs: : - command: convert | merge | scripts-list # required + command: convert | merge | split | scripts-list | scripts-exec | templ | inspect | diff description: "free text" # optional, shown by `job list` depends_on: [other-job, ...] # optional - inputs: # convert (≥1) / merge (≥2) + inputs: # convert (≥1) / merge (≥2) / split (≥1) / inspect (≥1) / diff (exactly 2) - path: relative/file.dbml # file inputs format: dbml - format: pgsql # live-connection inputs conn_env: SOURCE_DB_URL # env var NAME - script_dirs: # scripts-list (≥1) + - from_job: build-schema # consume another job's file output + script_dirs: # scripts-list / scripts-exec (≥1) - migrations/core - migrations/tenant template: templates/schema.tmpl # templ (required) mode: table # templ: database/schema/script/table filename_pattern: "{{.Name}}.go" # templ multi-output modes - output: # convert / merge (required) + select: # split (optional; default = keep everything) + schemas: [public] + tables: [users, orders] + exclude_schemas: [] + exclude_tables: [] + database_name: SubsetDB # optional rename of the output database + rules: .relspec-rules.yaml # inspect (optional; built-in defaults if omitted) + report: # inspect (required) / diff (required) + format: json # inspect: markdown|json ; diff: summary|json|html + path: build/report.json # required, except a diff "summary" (goes to the log) + overwrite: false + output: # convert / merge / split (required); scripts-exec (required, conn_env) format: pgsql path: build/schema.sql # file output, OR: conn_env: TARGET_DB_URL # execute against DB (pgsql only) @@ -136,13 +188,15 @@ jobs: flatten_schema: false schema: public package: models # for gorm/bun output - continue_on_error: false # pgsql output + continue_on_error: false # pgsql / scripts-exec output skip_relations: false # merge only skip_enums: false skip_views: false skip_domains: false skip_sequences: false - logfile: .relspec/log/.log # optional; appended to + logfile: .relspec/log/.log # optional; appended to, size-rotated + log_max_size: 5MB # optional per-job override + log_keep: 3 # optional per-job override ``` For `templ`, `inputs` use the same file or `pgsql`/`conn_env` source forms as @@ -150,6 +204,11 @@ schema conversion. `output` is optional (empty means stdout); when present it contains only `path` and `overwrite`, because templates do not select a schema writer format. +A `from_job` input takes no `path`, `format` or `conn_env`: it resolves to the +named job's `output.path` and inherits its format, and implies a dependency on +that job. The producer must be a `convert`, `merge` or `split` job writing a +single-file output. + ### Supported input formats `dbml`, `dctx`, `drawdb`, `graphql`, `json`, `yaml`, `gorm`, `bun`, `drizzle`, @@ -230,3 +289,77 @@ jobs: path: snapshots/prod.dbml overwrite: true ``` + +### Chain jobs with `from_job`, then lint the result + +```yaml +version: 1 +jobs: + build-json: + command: convert + inputs: + - { path: schema/core.dbml, format: dbml } + - { path: schema/tenant.dbml, format: dbml } + output: { format: json, path: build/schema.json, overwrite: true } + lint-schema: + command: inspect + inputs: + - from_job: build-json # implies depends_on: [build-json] + rules: .relspec-rules.yaml # optional; built-in rules if omitted + report: + format: markdown + path: build/lint-report.md + overwrite: true +``` + +`relspec job run lint-schema` runs `build-json` first, then inspects its output. +The job fails (exit 1) if any enforced rule is violated. + +### Split a subset out of a larger schema + +```yaml +version: 1 +jobs: + posts-only: + command: split + inputs: + - { path: schema/core.dbml, format: dbml } + - { path: schema/tenant.dbml, format: dbml } + select: + tables: [posts] + output: { format: dbml, path: build/posts.dbml, overwrite: true } +``` + +### Diff two schemas + +```yaml +version: 1 +jobs: + drift: + command: diff + inputs: # exactly two + - { path: build/schema.json, format: json } + - format: pgsql + conn_env: PROD_DB_URL + report: + format: summary # summary → logfile; json/html need a path +``` + +`diff` reports differences and always exits 0. + +### Execute migration scripts against a live database + +```yaml +version: 1 +jobs: + apply-migrations: + command: scripts-exec + script_dirs: + - migrations/core + - migrations/tenant + output: + conn_env: TARGET_DB_URL # pgsql only; no path + options: + continue_on_error: false + logfile: .relspec/log/apply-migrations.log +``` diff --git a/examples/jobs/relspec.yml b/examples/jobs/relspec.yml index 64c6452..a6adc21 100644 --- a/examples/jobs/relspec.yml +++ b/examples/jobs/relspec.yml @@ -4,7 +4,14 @@ # relspec job list # relspec job run build-schema --plan # relspec job run build-schema +# relspec job run lint-schema # inspect, consuming build-json's output version: 1 + +# File-wide defaults. Individual jobs may override log_max_size / log_keep. +defaults: + log_max_size: 2MB + log_keep: 5 + jobs: build-schema: command: convert @@ -43,3 +50,30 @@ jobs: - migrations/core - migrations/tenant logfile: .relspec/log/migration-order.log + + lint-schema: + command: inspect + description: Validate build-json's output against the built-in rules + # No depends_on needed: the from_job input implies a dependency on build-json. + inputs: + - from_job: build-json + report: + format: markdown + path: build/lint-report.md + overwrite: true + logfile: .relspec/log/lint-schema.log + + posts-only: + command: split + description: Extract just the posts table into its own DBML file + inputs: + - path: schema/core.dbml + format: dbml + - path: schema/tenant.dbml + format: dbml + select: + tables: [posts] + output: + format: dbml + path: build/posts.dbml + overwrite: true diff --git a/pkg/jobs/jobs.go b/pkg/jobs/jobs.go index ea41aaf..696d0c6 100644 --- a/pkg/jobs/jobs.go +++ b/pkg/jobs/jobs.go @@ -20,24 +20,52 @@ import ( "os" "path/filepath" "sort" + "strconv" "strings" "gopkg.in/yaml.v3" ) -// SchemaVersion is the only job-file schema version this build understands. -const SchemaVersion = 1 +// CurrentSchemaVersion is the highest job-file schema version this build was +// written for. MinSchemaVersion is the oldest it still accepts. A file that +// declares a version in between loads normally; a newer version loads +// best-effort with a warning (see Load); an older-than-minimum version is a +// hard error. +const ( + CurrentSchemaVersion = 1 + MinSchemaVersion = 1 +) + +// Built-in logfile rotation policy, used when neither the job nor its file's +// defaults block sets one. +const ( + defaultLogMaxSizeBytes int64 = 5 << 20 // 5 MiB + defaultLogKeep = 3 +) // Command names are a closed allow-list. Arbitrary strings are rejected. const ( CommandConvert = "convert" // read one or more schema files, optionally merge, write one output CommandMerge = "merge" // additive merge of two or more schema files into one output CommandScriptsList = "scripts-list" // deterministically list SQL scripts across one or more directories + CommandScriptsExec = "scripts-exec" // execute SQL scripts across one or more directories against a live database CommandTempl = "templ" // apply a custom Go text template to one or more schemas + CommandSplit = "split" // extract selected schemas/tables into a separate output + CommandInspect = "inspect" // validate one or more schemas against rules and write a report + CommandDiff = "diff" // compare exactly two schemas and write a differences report ) // SupportedCommands lists every accepted command, in help order. -var SupportedCommands = []string{CommandConvert, CommandMerge, CommandScriptsList, CommandTempl} +var SupportedCommands = []string{ + CommandConvert, CommandMerge, CommandScriptsList, CommandScriptsExec, + CommandTempl, CommandSplit, CommandInspect, CommandDiff, +} + +// producerCommands are commands whose output is a schema file that another job +// may consume via from_job. +var producerCommands = map[string]bool{ + CommandConvert: true, CommandMerge: true, CommandSplit: true, +} // readerFormats are the file-based input formats a job may declare (path). var readerFormats = map[string]bool{ @@ -61,10 +89,41 @@ var writerFormats = map[string]bool{ // live database) is supported instead of writing a file. var execOutputFormats = map[string]bool{"pgsql": true} +// singleFileFormats are output formats that emit exactly one file (as opposed +// to a directory of files). Only these are eligible for atomic temp+rename +// writes and for being consumed by another job via from_job. +var singleFileFormats = map[string]bool{ + "json": true, "yaml": true, "dbml": true, "dctx": true, "drawdb": true, + "graphql": true, "pgsql": true, "mssql": true, "sqlite": true, +} + +// SingleFileOutputFormat reports whether format writes exactly one file. +func SingleFileOutputFormat(format string) bool { + return singleFileFormats[strings.ToLower(format)] +} + +// diffReportFormats and inspectReportFormats are the report.format values +// accepted by the diff and inspect commands respectively. +var ( + diffReportFormats = map[string]bool{"summary": true, "json": true, "html": true} + inspectReportFormats = map[string]bool{"markdown": true, "json": true} +) + // File is the on-disk shape of a single job file. type File struct { - Version int `yaml:"version"` - Jobs map[string]*Job `yaml:"jobs"` + Version int `yaml:"version"` + Defaults *Defaults `yaml:"defaults"` + Jobs map[string]*Job `yaml:"jobs"` +} + +// Defaults carries file-wide settings that individual jobs may override. +type Defaults struct { + // LogMaxSize is a human-readable size ("5MB", "512KB", "1GB"). Empty + // means "use the built-in default". + LogMaxSize string `yaml:"log_max_size"` + // LogKeep is how many rotated logfiles to retain. Zero means "use the + // built-in default". + LogKeep int `yaml:"log_keep"` } // Job is one named job within a job file. @@ -72,6 +131,9 @@ type Job struct { // Name and SourceFile are populated by Load, not parsed from YAML. Name string `yaml:"-"` SourceFile string `yaml:"-"` + // fileDefaults is the Defaults block of the file that declared this job, + // captured by Load. nil when the file had none. + fileDefaults *Defaults `yaml:"-"` Command string `yaml:"command"` Description string `yaml:"description"` @@ -82,8 +144,13 @@ type Job struct { Mode string `yaml:"mode"` FilenamePattern string `yaml:"filename_pattern"` Output *Output `yaml:"output"` + Rules string `yaml:"rules"` + Report *Report `yaml:"report"` + Select *Select `yaml:"select"` Options Options `yaml:"options"` Logfile string `yaml:"logfile"` + LogMaxSize string `yaml:"log_max_size"` + LogKeep *int `yaml:"log_keep"` } // Input is one declared input schema. @@ -94,6 +161,10 @@ type Input struct { // ConnEnv is the NAME of an environment variable holding a connection // string, used with database formats. The value is never stored here. ConnEnv string `yaml:"conn_env"` + // FromJob names another job in the set whose file output is used as this + // input. It implies a dependency on that job. Path/Format/ConnEnv must be + // empty when FromJob is set; the format is inherited from the producer. + FromJob string `yaml:"from_job"` } // Output is the declared output target. @@ -104,6 +175,105 @@ type Output struct { Overwrite bool `yaml:"overwrite"` } +// Report is the output target for the inspect and diff commands. +type Report struct { + // Format is the report format: diff accepts summary|json|html, inspect + // accepts markdown|json. Empty means the command's default. + Format string `yaml:"format"` + Path string `yaml:"path"` + Overwrite bool `yaml:"overwrite"` +} + +// Select carries the schema/table selection for the split command. +type Select struct { + Schemas []string `yaml:"schemas"` + Tables []string `yaml:"tables"` + ExcludeSchemas []string `yaml:"exclude_schemas"` + ExcludeTables []string `yaml:"exclude_tables"` + DatabaseName string `yaml:"database_name"` +} + +// LogPolicy is the resolved logfile rotation policy for a job. +type LogPolicy struct { + MaxSizeBytes int64 + Keep int +} + +// ResolvedLogPolicy returns the effective rotation policy: the job's own +// overrides win, then its file's defaults block, then the built-in default. +func (j *Job) ResolvedLogPolicy() LogPolicy { + p := LogPolicy{MaxSizeBytes: defaultLogMaxSizeBytes, Keep: defaultLogKeep} + + if j.fileDefaults != nil { + if n, err := parseHumanSize(j.fileDefaults.LogMaxSize); err == nil && n > 0 { + p.MaxSizeBytes = n + } + if j.fileDefaults.LogKeep > 0 { + p.Keep = j.fileDefaults.LogKeep + } + } + if n, err := parseHumanSize(j.LogMaxSize); err == nil && n > 0 { + p.MaxSizeBytes = n + } + if j.LogKeep != nil && *j.LogKeep >= 0 { + p.Keep = *j.LogKeep + } + return p +} + +// effectiveDeps returns the union of explicit depends_on entries and the jobs +// referenced by from_job inputs, deduplicated in stable order. +func (j *Job) effectiveDeps() []string { + seen := map[string]bool{} + var deps []string + add := func(name string) { + if name == "" || name == j.Name || seen[name] { + return + } + seen[name] = true + deps = append(deps, name) + } + for _, d := range j.DependsOn { + add(d) + } + for _, in := range j.Inputs { + add(in.FromJob) + } + return deps +} + +// parseHumanSize parses a byte size such as "5MB", "512 KB", "1gb" or a bare +// byte count. An empty string returns (0, nil) so callers can fall back. +func parseHumanSize(s string) (int64, error) { + s = strings.TrimSpace(s) + if s == "" { + return 0, nil + } + upper := strings.ToUpper(s) + mult := int64(1) + // Check multi-character suffixes before the bare "B". + for _, u := range []struct { + suffix string + m int64 + }{ + {"KB", 1 << 10}, {"MB", 1 << 20}, {"GB", 1 << 30}, {"B", 1}, + } { + if strings.HasSuffix(upper, u.suffix) { + mult = u.m + upper = strings.TrimSpace(strings.TrimSuffix(upper, u.suffix)) + break + } + } + n, err := strconv.ParseFloat(upper, 64) + if err != nil { + return 0, fmt.Errorf("invalid size %q", s) + } + if n < 0 { + return 0, fmt.Errorf("negative size %q", s) + } + return int64(n * float64(mult)), nil +} + // Options carries the subset of command flags a job file may set. type Options struct { FlattenSchema bool `yaml:"flatten_schema"` @@ -127,6 +297,9 @@ type Set struct { Files []string // Jobs is keyed by job name. Jobs map[string]*Job + // Warnings holds non-fatal load-time messages (e.g. a newer-than-known + // schema version). Callers should surface these to the user. + Warnings []string } // Names returns all job names in deterministic (sorted) order. @@ -199,15 +372,35 @@ func Load(paths []string) (*Set, error) { if err != nil { return nil, fmt.Errorf("failed to read job file %q: %w", path, err) } + + // Peek at the version first so a newer file can be parsed leniently + // (unknown fields ignored) instead of failing outright. + var probe struct { + Version int `yaml:"version"` + } + if err := yaml.Unmarshal(data, &probe); err != nil { + return nil, fmt.Errorf("invalid job file %q: %w", path, err) + } + version := probe.Version + if version == 0 { + version = CurrentSchemaVersion + } + if version < MinSchemaVersion { + return nil, fmt.Errorf("job file %q: unsupported version %d (this build accepts %d or newer)", path, version, MinSchemaVersion) + } + strict := version <= CurrentSchemaVersion + if !strict { + set.Warnings = append(set.Warnings, fmt.Sprintf( + "job file %q declares version %d, newer than this build understands (%d); loading best-effort and ignoring unknown fields", + path, version, CurrentSchemaVersion)) + } + dec := yaml.NewDecoder(strings.NewReader(string(data))) - dec.KnownFields(true) + dec.KnownFields(strict) var f File if err := dec.Decode(&f); err != nil { return nil, fmt.Errorf("invalid job file %q: %w", path, err) } - if f.Version != SchemaVersion { - return nil, fmt.Errorf("job file %q: unsupported version %d (expected %d)", path, f.Version, SchemaVersion) - } if len(f.Jobs) == 0 { return nil, fmt.Errorf("job file %q: no jobs defined", path) } @@ -220,6 +413,7 @@ func Load(paths []string) (*Set, error) { } job.Name = name job.SourceFile = path + job.fileDefaults = f.Defaults origin[name] = path set.Jobs[name] = job } @@ -240,13 +434,30 @@ func (s *Set) Validate() error { errs = append(errs, fmt.Sprintf("job %q: %s", name, msg)) } } - // Dependency references + cycles. + // Dependency references + cycles + from_job wiring. for _, name := range s.Names() { - for _, dep := range s.Jobs[name].DependsOn { + j := s.Jobs[name] + for _, dep := range j.DependsOn { if _, ok := s.Jobs[dep]; !ok { errs = append(errs, fmt.Sprintf("job %q: depends_on unknown job %q", name, dep)) } } + for i, in := range j.Inputs { + if in.FromJob == "" { + continue + } + producer, ok := s.Jobs[in.FromJob] + if !ok { + errs = append(errs, fmt.Sprintf("job %q: input[%d] from_job references unknown job %q", name, i, in.FromJob)) + continue + } + if !producerCommands[producer.Command] || producer.Output == nil || + producer.Output.Path == "" || !SingleFileOutputFormat(producer.Output.Format) { + errs = append(errs, fmt.Sprintf( + "job %q: input[%d] from_job %q must name a convert/merge/split job that writes a single-file output", + name, i, in.FromJob)) + } + } } if cycle := s.findCycle(); cycle != "" { errs = append(errs, fmt.Sprintf("dependency cycle detected: %s", cycle)) @@ -262,7 +473,8 @@ func (j *Job) validate() []string { var e []string switch j.Command { - case CommandConvert, CommandMerge, CommandScriptsList, CommandTempl: + case CommandConvert, CommandMerge, CommandScriptsList, CommandScriptsExec, + CommandTempl, CommandSplit, CommandInspect, CommandDiff: case "": e = append(e, "missing command") return e @@ -282,6 +494,7 @@ func (j *Job) validate() []string { } checkPath("logfile", j.Logfile) checkPath("template", j.Template) + checkPath("rules", j.Rules) for _, in := range j.Inputs { checkPath("input path", in.Path) } @@ -291,6 +504,13 @@ func (j *Job) validate() []string { if j.Output != nil { checkPath("output path", j.Output.Path) } + if j.Report != nil { + checkPath("report path", j.Report.Path) + } + + if _, err := parseHumanSize(j.LogMaxSize); err != nil { + e = append(e, fmt.Sprintf("log_max_size: %v", err)) + } switch j.Command { case CommandConvert, CommandMerge: @@ -350,11 +570,113 @@ func (j *Job) validate() []string { if j.Output != nil && j.Output.Format != "" { e = append(e, "output.format is not valid for command \"templ\"") } + case CommandSplit: + if len(j.Inputs) < 1 { + e = append(e, "command \"split\" requires at least 1 input") + } + for i, in := range j.Inputs { + e = append(e, validateInput(i, in)...) + } + if len(j.ScriptDirs) > 0 { + e = append(e, "script_dirs is not valid for command \"split\"") + } + if j.Report != nil { + e = append(e, "report is not valid for command \"split\" (use output)") + } + if j.Output == nil { + e = append(e, "missing output") + } else { + if j.Output.ConnEnv != "" { + e = append(e, "command \"split\" writes a file; output.conn_env is not supported") + } + e = append(e, validateOutput(*j.Output)...) + } + case CommandInspect: + if len(j.Inputs) < 1 { + e = append(e, "command \"inspect\" requires at least 1 input") + } + for i, in := range j.Inputs { + e = append(e, validateInput(i, in)...) + } + if len(j.ScriptDirs) > 0 { + e = append(e, "script_dirs is not valid for command \"inspect\"") + } + if j.Output != nil { + e = append(e, "output is not valid for command \"inspect\" (use report)") + } + e = append(e, validateReport(j.Report, "inspect", inspectReportFormats, "markdown")...) + case CommandDiff: + if len(j.Inputs) != 2 { + e = append(e, "command \"diff\" requires exactly 2 inputs (source, target)") + } + for i, in := range j.Inputs { + e = append(e, validateInput(i, in)...) + } + if len(j.ScriptDirs) > 0 { + e = append(e, "script_dirs is not valid for command \"diff\"") + } + if j.Output != nil { + e = append(e, "output is not valid for command \"diff\" (use report)") + } + e = append(e, validateReport(j.Report, "diff", diffReportFormats, "summary")...) + case CommandScriptsExec: + if len(j.ScriptDirs) == 0 { + e = append(e, "command \"scripts-exec\" requires at least one script_dir") + } + if len(j.Inputs) > 0 { + e = append(e, "inputs is not valid for command \"scripts-exec\"") + } + if j.Report != nil { + e = append(e, "report is not valid for command \"scripts-exec\"") + } + if j.Output == nil || j.Output.ConnEnv == "" { + e = append(e, "command \"scripts-exec\" requires output.conn_env (an environment variable name holding a connection string)") + } else { + if j.Output.Path != "" { + e = append(e, "command \"scripts-exec\" executes against a database; output.path is not supported") + } + f := strings.ToLower(j.Output.Format) + if f != "" && f != "pgsql" { + e = append(e, fmt.Sprintf("command \"scripts-exec\" only supports pgsql databases (got %q)", j.Output.Format)) + } + if looksLikeSecret(j.Output.ConnEnv) { + e = append(e, "output: conn_env must be an environment variable name, not a connection string") + } + } + } + return e +} + +// validateReport checks a Report block for the inspect/diff commands. +func validateReport(r *Report, cmd string, allowed map[string]bool, defFmt string) []string { + if r == nil { + return []string{fmt.Sprintf("command %q requires a report block", cmd)} + } + var e []string + f := strings.ToLower(r.Format) + if f == "" { + f = defFmt + } + if !allowed[f] { + names := make([]string, 0, len(allowed)) + for k := range allowed { + names = append(names, k) + } + sort.Strings(names) + e = append(e, fmt.Sprintf("command %q report.format %q is not supported (use: %s)", cmd, r.Format, strings.Join(names, ", "))) + } + // A diff summary may be written to the log; everything else needs a path. + summaryToLog := cmd == "diff" && f == "summary" + if r.Path == "" && !summaryToLog { + e = append(e, fmt.Sprintf("command %q requires report.path", cmd)) } return e } func validateTemplInput(i int, in Input) []string { + if in.FromJob != "" { + return fromJobInputShape(i, in) + } var e []string if in.Format == "" { return []string{fmt.Sprintf("input[%d]: missing format", i)} @@ -383,7 +705,27 @@ func validateTemplInput(i int, in Input) []string { return e } +// fromJobInputShape checks the structural rules for an input that pulls its +// schema from another job's output. The referenced job's existence and kind +// are checked in Set.Validate, which can see the whole set. +func fromJobInputShape(i int, in Input) []string { + var e []string + if in.Path != "" { + e = append(e, fmt.Sprintf("input[%d]: from_job takes no path", i)) + } + if in.Format != "" { + e = append(e, fmt.Sprintf("input[%d]: from_job inherits the producer's format; drop format", i)) + } + if in.ConnEnv != "" { + e = append(e, fmt.Sprintf("input[%d]: from_job takes no conn_env", i)) + } + return e +} + func validateInput(i int, in Input) []string { + if in.FromJob != "" { + return fromJobInputShape(i, in) + } var e []string if in.Format == "" { e = append(e, fmt.Sprintf("input[%d]: missing format", i)) @@ -487,9 +829,42 @@ func SafeJoin(root, rel string) (string, error) { if rp == ".." || strings.HasPrefix(rp, ".."+string(filepath.Separator)) { return "", fmt.Errorf("path %q escapes the job file directory", rel) } + // Symlink hardening: resolve symlinks on the root and on the deepest + // existing ancestor of the target, and require the target to still live + // inside the resolved root. This catches a symlink inside the job-file + // directory that points outside it. + realRoot, err := filepath.EvalSymlinks(absRoot) + if err != nil { + return "", fmt.Errorf("cannot resolve job file directory: %w", err) + } + realAnc, err := filepath.EvalSymlinks(deepestExistingAncestor(joined)) + if err != nil { + return "", fmt.Errorf("cannot resolve path %q: %w", rel, err) + } + if realAnc != realRoot { + if r, err := filepath.Rel(realRoot, realAnc); err != nil || + r == ".." || strings.HasPrefix(r, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("path %q resolves outside the job file directory via a symlink", rel) + } + } return joined, nil } +// deepestExistingAncestor returns p itself if it exists, otherwise the nearest +// existing parent directory (falling back to the filesystem root). +func deepestExistingAncestor(p string) string { + for { + if _, err := os.Lstat(p); err == nil { + return p + } + parent := filepath.Dir(p) + if parent == p { + return p + } + p = parent + } +} + // Plan returns the jobs to execute for name in dependency order. When // includeDeps is false only the named job is returned (its declared // dependencies are still validated to exist and be acyclic by Validate). @@ -514,7 +889,7 @@ func (s *Set) Plan(name string, includeDeps bool) ([]*Job, error) { } inProgress[n] = true j := s.Jobs[n] - deps := append([]string(nil), j.DependsOn...) + deps := j.effectiveDeps() sort.Strings(deps) for _, d := range deps { if _, ok := s.Jobs[d]; !ok { @@ -543,7 +918,7 @@ func (s *Set) findCycle() string { dfs = func(n string) []string { color[n] = 1 stack = append(stack, n) - deps := append([]string(nil), s.Jobs[n].DependsOn...) + deps := s.Jobs[n].effectiveDeps() sort.Strings(deps) for _, d := range deps { if _, ok := s.Jobs[d]; !ok { diff --git a/pkg/jobs/jobs_test.go b/pkg/jobs/jobs_test.go index 6f50dfe..a757251 100644 --- a/pkg/jobs/jobs_test.go +++ b/pkg/jobs/jobs_test.go @@ -56,13 +56,82 @@ func TestLoadRejectsUnknownFields(t *testing.T) { } } -func TestLoadRejectsBadVersion(t *testing.T) { +func TestLoadWarnsOnNewerVersion(t *testing.T) { dir := t.TempDir() p := filepath.Join(dir, "relspec.yml") - write(t, p, "version: 2\njobs:\n a:\n command: convert\n") - _, err := Load([]string{p}) - if err == nil || !strings.Contains(err.Error(), "unsupported version") { - t.Fatalf("expected unsupported version error, got %v", err) + // A newer version loads best-effort with a warning, and unknown fields + // from the newer schema are ignored rather than rejected. + write(t, p, "version: 99\njobs:\n a:\n command: convert\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n"+ + " output:\n format: json\n path: out.json\n"+ + " future_field: whatever\n") + set, err := Load([]string{p}) + if err != nil { + t.Fatalf("newer version should load, got %v", err) + } + if len(set.Warnings) == 0 { + t.Fatal("expected a warning about the newer version") + } + if err := set.Validate(); err != nil { + t.Fatalf("validate: %v", err) + } +} + +func TestLoadAcceptsOmittedVersion(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "relspec.yml") + write(t, p, "jobs:\n a:\n command: convert\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n"+ + " output:\n format: json\n path: out.json\n") + set, err := Load([]string{p}) + if err != nil { + t.Fatalf("omitted version should load, got %v", err) + } + if len(set.Warnings) != 0 { + t.Fatalf("omitted version should not warn, got %v", set.Warnings) + } +} + +func TestLoadStillRejectsUnknownFieldsAtCurrentVersion(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "relspec.yml") + write(t, p, "version: 1\njobs:\n a:\n command: convert\n bogus: true\n") + if _, err := Load([]string{p}); err == nil { + t.Fatal("expected unknown-field rejection at the current version") + } +} + +func TestParseHumanSize(t *testing.T) { + cases := []struct { + in string + want int64 + bad bool + }{ + {"", 0, false}, + {"512", 512, false}, + {"512B", 512, false}, + {"1KB", 1 << 10, false}, + {"5MB", 5 << 20, false}, + {"1gb", 1 << 30, false}, + {" 2 MB ", 2 << 20, false}, + {"nonsense", 0, true}, + {"-1MB", 0, true}, + } + for _, c := range cases { + got, err := parseHumanSize(c.in) + if c.bad { + if err == nil { + t.Errorf("parseHumanSize(%q): expected error", c.in) + } + continue + } + if err != nil { + t.Errorf("parseHumanSize(%q): %v", c.in, err) + continue + } + if got != c.want { + t.Errorf("parseHumanSize(%q) = %d, want %d", c.in, got, c.want) + } } } @@ -239,6 +308,135 @@ func TestShippedExampleIsValid(t *testing.T) { } } +func TestFromJobWiring(t *testing.T) { + content := "version: 1\njobs:\n" + + " producer:\n command: convert\n" + + " inputs:\n - path: a.dbml\n format: dbml\n" + + " output:\n format: json\n path: build/schema.json\n" + + " consumer:\n command: convert\n" + + " inputs:\n - from_job: producer\n" + + " output:\n format: yaml\n path: build/schema.yaml\n" + set := loadOne(t, content) + if err := set.Validate(); err != nil { + t.Fatalf("validate: %v", err) + } + plan, err := set.Plan("consumer", true) + if err != nil { + t.Fatal(err) + } + if len(plan) != 2 || plan[0].Name != "producer" || plan[1].Name != "consumer" { + t.Fatalf("plan = %v, want [producer consumer]", plan) + } +} + +func TestFromJobRejectsNonProducer(t *testing.T) { + content := "version: 1\njobs:\n" + + " lister:\n command: scripts-list\n script_dirs: [migrations]\n" + + " consumer:\n command: convert\n" + + " inputs:\n - from_job: lister\n" + + " output:\n format: yaml\n path: out.yaml\n" + set := loadOne(t, content) + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "from_job") { + t.Fatalf("want from_job producer error, got %v", err) + } +} + +func TestFromJobRejectsUnknownJob(t *testing.T) { + content := "version: 1\njobs:\n" + + " consumer:\n command: convert\n" + + " inputs:\n - from_job: ghost\n" + + " output:\n format: yaml\n path: out.yaml\n" + set := loadOne(t, content) + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "unknown job") { + t.Fatalf("want unknown job error, got %v", err) + } +} + +func TestFromJobCycleDetected(t *testing.T) { + content := "version: 1\njobs:\n" + + " a:\n command: convert\n" + + " inputs:\n - from_job: b\n" + + " output:\n format: json\n path: a.json\n" + + " b:\n command: convert\n" + + " inputs:\n - from_job: a\n" + + " output:\n format: json\n path: b.json\n" + set := loadOne(t, content) + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "cycle") { + t.Fatalf("want cycle error, got %v", err) + } +} + +func TestSplitJobValidation(t *testing.T) { + set := loadOne(t, "version: 1\njobs:\n s:\n command: split\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n") + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "missing output") { + t.Fatalf("want missing output, got %v", err) + } + set = loadOne(t, "version: 1\njobs:\n s:\n command: split\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n"+ + " select:\n tables: [users]\n"+ + " output:\n format: json\n path: out.json\n") + if err := set.Validate(); err != nil { + t.Fatalf("expected valid split job, got %v", err) + } +} + +func TestInspectJobValidation(t *testing.T) { + set := loadOne(t, "version: 1\njobs:\n i:\n command: inspect\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n") + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "report") { + t.Fatalf("want report required, got %v", err) + } + set = loadOne(t, "version: 1\njobs:\n i:\n command: inspect\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n"+ + " report:\n format: json\n path: build/report.json\n") + if err := set.Validate(); err != nil { + t.Fatalf("expected valid inspect job, got %v", err) + } +} + +func TestDiffJobValidation(t *testing.T) { + set := loadOne(t, "version: 1\njobs:\n d:\n command: diff\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n"+ + " report:\n format: summary\n") + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "exactly 2 inputs") { + t.Fatalf("want exactly 2 inputs, got %v", err) + } + set = loadOne(t, "version: 1\njobs:\n d:\n command: diff\n"+ + " inputs:\n - path: a.dbml\n format: dbml\n"+ + " - path: b.dbml\n format: dbml\n"+ + " report:\n format: summary\n") + if err := set.Validate(); err != nil { + t.Fatalf("expected valid diff job, got %v", err) + } +} + +func TestScriptsExecValidation(t *testing.T) { + set := loadOne(t, "version: 1\njobs:\n x:\n command: scripts-exec\n"+ + " script_dirs: [migrations]\n") + if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "conn_env") { + t.Fatalf("want output.conn_env required, got %v", err) + } + set = loadOne(t, "version: 1\njobs:\n x:\n command: scripts-exec\n"+ + " script_dirs: [migrations]\n"+ + " output:\n conn_env: TARGET_DB_URL\n") + if err := set.Validate(); err != nil { + t.Fatalf("expected valid scripts-exec job, got %v", err) + } +} + +func TestSafeJoinRejectsSymlinkEscape(t *testing.T) { + root := t.TempDir() + outside := t.TempDir() + link := filepath.Join(root, "link") + if err := os.Symlink(outside, link); err != nil { + t.Skipf("symlink not supported: %v", err) + } + if _, err := SafeJoin(root, "link/x.sql"); err == nil { + t.Fatal("expected rejection of a path escaping via a symlink") + } +} + func TestScriptsListValidation(t *testing.T) { set := loadOne(t, "version: 1\njobs:\n s:\n command: scripts-list\n") if err := set.Validate(); err == nil || !strings.Contains(err.Error(), "script_dir") {