From e296361042f4bccd25e7432902784e6c475a5aca Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 31 May 2026 02:21:36 +0000 Subject: [PATCH] Add local source publication --- README.md | 8 +- docs/cli.md | 20 +- docs/config.md | 22 +- docs/internal/publish.md | 29 +++ docs/operations.md | 39 ++++ examples/local-publish.yml | 12 ++ internal/app/run.go | 82 ++++++-- internal/app/run_test.go | 331 +++++++++++++++++++++++++++++-- internal/cli/root_test.go | 96 ++++++--- internal/config/load_test.go | 1 + internal/publish/execute.go | 98 +++++++++ internal/publish/execute_test.go | 91 +++++++++ internal/publish/output.go | 54 +++++ internal/publish/output_test.go | 29 +++ internal/publish/plan.go | 141 +++++++++++++ internal/publish/reconcile.go | 31 +++ internal/publish/safety.go | 19 ++ 17 files changed, 1019 insertions(+), 84 deletions(-) create mode 100644 docs/internal/publish.md create mode 100644 docs/operations.md create mode 100644 examples/local-publish.yml create mode 100644 internal/publish/execute.go create mode 100644 internal/publish/execute_test.go create mode 100644 internal/publish/output.go create mode 100644 internal/publish/output_test.go create mode 100644 internal/publish/plan.go create mode 100644 internal/publish/reconcile.go create mode 100644 internal/publish/safety.go diff --git a/README.md b/README.md index 376f102..72bf993 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,11 @@ # distributor -`distributor` is a Go application shell for validating and publishing manifested report bundles. +`distributor` validates and publishes manifested report bundles. -Current implemented behavior is limited to CLI help, version output, config validation, source bundle validation, source bundle inspection, and placeholder run execution: +Run the local example pipeline: ```sh -go run ./cmd/distributor validate examples/source-bundle +go run ./cmd/distributor run --config examples/local-publish.yml ``` -See [docs/cli.md](docs/cli.md) and [docs/config.md](docs/config.md) for the implemented CLI and configuration surface. Current design and implementation planning lives under `docs/roadmap/`. +See [docs/cli.md](docs/cli.md), [docs/config.md](docs/config.md), and [docs/operations.md](docs/operations.md) for the implemented CLI, configuration, and operating notes. Current design and implementation planning lives under `docs/roadmap/`. diff --git a/docs/cli.md b/docs/cli.md index 6c75a33..ea13070 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -3,10 +3,10 @@ ## Shortest useful command ```sh -go run ./cmd/distributor validate examples/source-bundle +go run ./cmd/distributor run --config examples/local-publish.yml ``` -This validates a local source bundle fixture. +This validates and publishes the example source bundle to `workspace/published/source-bundle`. ## Command overview @@ -20,13 +20,15 @@ distributor inspect `version` prints the application name and version. The default development version is `dev`; release builds may replace it at build time. -`run --config --dry-run` loads and validates configuration, then prints a concise summary of configured pipelines and destinations. It does not discover bundles or publish files yet. +`run --config ` executes configured local-to-local pipelines that publish source files only. + +`run --config --dry-run` discovers source bundles, inspects destination state, and prints planned actions without writing files. `validate ` validates a local source bundle directory or a local tree containing source bundles. `inspect ` validates discovered local source bundles and prints a concise normalized summary. -`run` without `--dry-run` intentionally fails with a clear `not implemented` error until execution behavior exists. +Remote backends and HTML publication are not implemented yet. ## Flag reference @@ -57,8 +59,14 @@ Inspect a source bundle: go run ./cmd/distributor inspect examples/source-bundle ``` -Validate a config file without publishing: +Preview local publication without writing: ```sh -go run ./cmd/distributor run --config examples/local-to-local.yml --dry-run +go run ./cmd/distributor run --config examples/local-publish.yml --dry-run +``` + +Publish the local example: + +```sh +go run ./cmd/distributor run --config examples/local-publish.yml ``` diff --git a/docs/config.md b/docs/config.md index c24217d..0716b2b 100644 --- a/docs/config.md +++ b/docs/config.md @@ -2,15 +2,15 @@ ## Config file location -`distributor run --config --dry-run` loads the YAML config at the path provided by `--config`. +`distributor run --config ` loads the YAML config at the path provided by `--config`. -If `--config` is omitted during dry-run, the built-in default path is: +If `--config` is omitted during run, the built-in default path is: ```text /usr/local/etc/distributor/config.yml ``` -The current implementation loads and validates configuration only. Bundle discovery and publication are not implemented yet. +The current implementation supports local-to-local source-file publication. Remote backends and HTML publication are not implemented yet. ## Minimal config @@ -40,18 +40,16 @@ pipelines: on_digest_mismatch: fail destinations: - id: archive - backend: s3 - endpoint: https://s3.example.com - bucket: reports - prefix: archive - region: us-east-1 - force_path_style: true - credentials: - access_key_id_env: DISTRIBUTOR_S3_ACCESS_KEY_ID - secret_access_key_env: DISTRIBUTOR_S3_SECRET_ACCESS_KEY + backend: local + path: /srv/reports/archive publish: source: true html: false + transfer: + on_destination_same: skip + on_destination_older: replace + on_destination_newer: skip + on_conflict: fail ``` ## Reference diff --git a/docs/internal/publish.md b/docs/internal/publish.md new file mode 100644 index 0000000..baf7742 --- /dev/null +++ b/docs/internal/publish.md @@ -0,0 +1,29 @@ +# Publish + +## Purpose + +`internal/publish` plans and executes publication for one validated source bundle and one destination. + +## Inputs and outputs + +Inputs are a source bundle, source backend, destination backend, pipeline id, destination id, publish policy, transfer policy, destination bundle path, and existing destination state. + +Output is a plan with an action, reason, and selected source outputs. Execution writes selected source files and `.distributor.json` for publish or replacement actions. + +## Actions + +Supported actions are publish new, replace older destination, skip same source, skip newer destination, fail conflict, and fail unmanaged destination. + +## Boundaries + +The current implementation publishes source files only. HTML generation and remote backend execution are not implemented. + +The package uses `internal/state` for destination comparison and `internal/storage` for IO. It does not parse CLI flags or load config files. + +## Safety + +Replacement deletes only outputs recorded in existing destination state plus `.distributor.json`. Failed local writes trigger cleanup of outputs written during the failed attempt. + +## Tests + +Before changing publish behavior, inspect tests under `internal/publish` and local run tests under `internal/app`. diff --git a/docs/operations.md b/docs/operations.md new file mode 100644 index 0000000..d9ef6f7 --- /dev/null +++ b/docs/operations.md @@ -0,0 +1,39 @@ +# Distributor Operations + +## Normal workflow + +Preview a local publication: + +```sh +go run ./cmd/distributor run --config examples/local-publish.yml --dry-run +``` + +Run the local publication: + +```sh +go run ./cmd/distributor run --config examples/local-publish.yml +``` + +## Filesystem layout + +Source bundles are discovered beneath the configured local source root. Destination bundle paths preserve the source bundle path relative to that source root. + +The maintained example writes under `workspace/`, which is ignored by Git. + +## Destination state + +Each published destination bundle contains `.distributor.json`. This state file records the source manifest and copied source outputs. It is the authoritative marker that a destination path is managed by `distributor`. + +`manifest.json` from the source bundle is not copied as destination state. + +## Retry behavior + +If a destination already has matching `.distributor.json`, publication skips it as already published. + +If destination state is older than the source manifest, publication replaces only managed outputs recorded in `.distributor.json` plus the state file. + +If a write fails during local publication, `distributor` removes outputs written during that failed attempt where possible so a retry does not see an unmanaged destination. + +## Caveats + +Only local-to-local source-file publication is implemented. SSH, S3, HTML generation, notification, and force overwrite behavior are not implemented. diff --git a/examples/local-publish.yml b/examples/local-publish.yml new file mode 100644 index 0000000..de9147d --- /dev/null +++ b/examples/local-publish.yml @@ -0,0 +1,12 @@ +pipelines: + - id: example-source-bundle + source: + backend: local + path: examples/source-bundle + destinations: + - id: local-archive + backend: local + path: workspace/published/source-bundle + publish: + source: true + html: false diff --git a/internal/app/run.go b/internal/app/run.go index 56867fc..cedb9fb 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -5,7 +5,10 @@ import ( "fmt" "io" + "gitea.maximumdirect.net/eric/distributor/internal/adapters/local" + "gitea.maximumdirect.net/eric/distributor/internal/bundle" "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/publish" ) type RunOptions struct { @@ -18,9 +21,6 @@ func Run(ctx context.Context, options RunOptions) error { if err := ctx.Err(); err != nil { return err } - if !options.DryRun { - return fmt.Errorf("run command: %w", ErrNotImplemented) - } configPath := options.ConfigPath if configPath == "" { @@ -30,25 +30,77 @@ func Run(ctx context.Context, options RunOptions) error { if err != nil { return err } - return writeRunSummary(options.Stdout, cfg) + return runConfig(ctx, cfg, options) } -func writeRunSummary(w io.Writer, cfg config.Config) error { - if w == nil { - return nil - } - if _, err := fmt.Fprintf(w, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil { - return err - } - for _, pipeline := range cfg.Pipelines { - if _, err := fmt.Fprintf(w, "- %s: source=%s destinations=%d\n", pipeline.ID, pipeline.Source.Backend, len(pipeline.Destinations)); err != nil { +func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error { + if options.Stdout != nil { + if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil { return err } - for _, destination := range pipeline.Destinations { - if _, err := fmt.Fprintf(w, " - %s: backend=%s publish_source=%t publish_html=%t\n", destination.ID, destination.Backend, destination.Publish.Source, destination.Publish.HTML); err != nil { + } + for _, pipeline := range cfg.Pipelines { + if pipeline.Source.Backend != config.BackendLocal { + return fmt.Errorf("pipeline %s source backend %s is not implemented for execution", pipeline.ID, pipeline.Source.Backend) + } + sourceBackend, err := local.New(pipeline.Source.Path) + if err != nil { + return err + } + bundles, err := bundle.Discover(ctx, sourceBackend, "") + if err != nil { + return fmt.Errorf("pipeline %s discover source bundles: %w", pipeline.ID, err) + } + if options.Stdout != nil { + if _, err := fmt.Fprintf(options.Stdout, "- %s: source=local bundles=%d destinations=%d\n", pipeline.ID, len(bundles), len(pipeline.Destinations)); err != nil { return err } } + for _, sourceBundle := range bundles { + for _, destination := range pipeline.Destinations { + if destination.Backend != config.BackendLocal { + return fmt.Errorf("pipeline %s destination %s backend %s is not implemented for execution", pipeline.ID, destination.ID, destination.Backend) + } + destinationBackend, err := local.New(destination.Path) + if err != nil { + return err + } + req := publish.Request{ + PipelineID: pipeline.ID, + DestinationID: destination.ID, + SourceBundle: sourceBundle, + SourceBackend: sourceBackend, + DestinationBackend: destinationBackend, + DestinationBundlePath: sourceBundle.RootRelativePath, + Publish: *destination.Publish, + Transfer: destination.Transfer, + DistributorVersion: Version, + } + plan, err := publish.Build(ctx, req) + if options.Stdout != nil { + writePlanLine(options.Stdout, plan, err) + } + if err != nil { + return fmt.Errorf("pipeline %s destination %s bundle %s: %w", pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + } + if !options.DryRun { + if err := publish.Execute(ctx, req, plan); err != nil { + return fmt.Errorf("pipeline %s destination %s bundle %s: %w", pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + } + } + } + } } return nil } + +func writePlanLine(w io.Writer, plan publish.Plan, planErr error) { + if w == nil { + return + } + if planErr != nil { + fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, planErr.Error()) + return + } + fmt.Fprintf(w, " - bundle=%s destination=%s action=%s outputs=%d reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, plan.Action, len(plan.Outputs), plan.Reason) +} diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 88dffbe..17d970e 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -3,31 +3,25 @@ package app import ( "bytes" "context" + "encoding/json" "os" "path/filepath" "strings" "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/state" ) func TestRunDryRunPrintsConfigSummary(t *testing.T) { - configPath := filepath.Join(t.TempDir(), "config.yml") - err := os.WriteFile(configPath, []byte(` -pipelines: - - id: reports - source: - backend: local - path: /source - destinations: - - id: archive - backend: local - path: /archive -`), 0o600) - if err != nil { - t.Fatalf("write config: %v", err) - } + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + configPath := writeLocalConfig(t, sourceRoot, destinationRoot) var stdout bytes.Buffer - err = Run(context.Background(), RunOptions{ + err := Run(context.Background(), RunOptions{ ConfigPath: configPath, DryRun: true, Stdout: &stdout, @@ -39,8 +33,8 @@ pipelines: output := stdout.String() for _, want := range []string{ "Configured pipelines: 1", - "- reports: source=local destinations=1", - "archive: backend=local publish_source=true publish_html=false", + "- reports: source=local bundles=1 destinations=1", + "bundle=. destination=archive action=publish_new", } { if !strings.Contains(output, want) { t.Fatalf("Run() output = %q, want substring %q", output, want) @@ -48,9 +42,302 @@ pipelines: } } -func TestRunWithoutDryRunIsNotImplemented(t *testing.T) { - err := Run(context.Background(), RunOptions{}) - if err == nil || !strings.Contains(err.Error(), "not implemented") { - t.Fatalf("Run() error = %v, want not implemented", err) +func TestRunPublishesNewLocalBundle(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + + var stdout bytes.Buffer + err := Run(context.Background(), RunOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + Stdout: &stdout, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n") + assertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n") + if _, err := os.Stat(filepath.Join(destinationRoot, "manifest.json")); !os.IsNotExist(err) { + t.Fatalf("destination manifest stat error = %v, want not exist", err) + } + destinationState := readStateFile(t, filepath.Join(destinationRoot, ".distributor.json")) + if destinationState.PipelineID != "reports" || destinationState.DestinationID != "archive" { + t.Fatalf("state identity = %s/%s", destinationState.PipelineID, destinationState.DestinationID) + } + if destinationState.Source.Manifest.ID != manifest.ID { + t.Fatalf("state source id = %q, want %q", destinationState.Source.Manifest.ID, manifest.ID) + } + if got, want := len(destinationState.Outputs), 2; got != want { + t.Fatalf("state output count = %d, want %d", got, want) + } +} + +func TestRunSkipsWhenDestinationStateMatches(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + configPath := writeLocalConfig(t, sourceRoot, destinationRoot) + if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil { + t.Fatalf("first Run() error = %v", err) + } + + var stdout bytes.Buffer + err := Run(context.Background(), RunOptions{ConfigPath: configPath, Stdout: &stdout}) + if err != nil { + t.Fatalf("second Run() error = %v", err) + } + if !strings.Contains(stdout.String(), "action=skip_same") { + t.Fatalf("stdout = %q, want skip_same", stdout.String()) + } +} + +func TestRunReplacesOlderDestination(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + older := manifest + older.Created = older.Created.Add(-time.Hour) + writeDestinationState(t, destinationRoot, "", older) + if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("old\n"), 0o600); err != nil { + t.Fatalf("write old output: %v", err) + } + + var stdout bytes.Buffer + err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), Stdout: &stdout}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if !strings.Contains(stdout.String(), "action=replace_older") { + t.Fatalf("stdout = %q, want replace_older", stdout.String()) + } + assertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n") +} + +func TestRunSkipsNewerDestination(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + newer := manifest + newer.Created = newer.Created.Add(time.Hour) + writeDestinationState(t, destinationRoot, "", newer) + if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("newer\n"), 0o600); err != nil { + t.Fatalf("write newer output: %v", err) + } + + var stdout bytes.Buffer + err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), Stdout: &stdout}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if !strings.Contains(stdout.String(), "action=skip_destination_newer") { + t.Fatalf("stdout = %q, want skip_destination_newer", stdout.String()) + } + assertFile(t, filepath.Join(destinationRoot, "report.md"), "newer\n") +} + +func TestRunFailsOnConflict(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + manifest := writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + manifest.ID = "other.source" + writeDestinationState(t, destinationRoot, "", manifest) + + err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)}) + if err == nil || !strings.Contains(err.Error(), "fail_conflict") { + t.Fatalf("Run() error = %v, want fail_conflict", err) + } +} + +func TestRunFailsOnUnmanagedDestination(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + if err := os.WriteFile(filepath.Join(destinationRoot, "unmanaged.txt"), []byte("data"), 0o600); err != nil { + t.Fatalf("write unmanaged file: %v", err) + } + + err := Run(context.Background(), RunOptions{ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot)}) + if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") { + t.Fatalf("Run() error = %v, want fail_unmanaged", err) + } +} + +func TestRunFansOutToLocalDestinations(t *testing.T) { + sourceRoot := t.TempDir() + firstDestination := t.TempDir() + secondDestination := t.TempDir() + writeSourceBundle(t, sourceRoot, "daily", testBundleOptions{}) + + err := Run(context.Background(), RunOptions{ConfigPath: writeFanoutConfig(t, sourceRoot, firstDestination, secondDestination)}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + assertFile(t, filepath.Join(firstDestination, "daily", "report.md"), "# Report\nSunny.\n") + assertFile(t, filepath.Join(secondDestination, "daily", "summary.txt"), "Summary\n") +} + +func TestRunDryRunDoesNotWrite(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + + err := Run(context.Background(), RunOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + DryRun: true, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 { + t.Fatalf("destination entries = %v err=%v, want empty", entries, err) + } +} + +type testBundleOptions struct { + ID string + Created time.Time +} + +func writeSourceBundle(t *testing.T, root, relative string, opts testBundleOptions) bundle.Manifest { + t.Helper() + if opts.ID == "" { + opts.ID = "weather.daily.brentwood.2026-05-30" + } + if opts.Created.IsZero() { + opts.Created = time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC) + } + bundleRoot := filepath.Join(root, filepath.FromSlash(relative)) + if err := os.MkdirAll(bundleRoot, 0o755); err != nil { + t.Fatalf("mkdir bundle: %v", err) + } + files := []struct { + path string + data string + }{ + {path: "report.md", data: "# Report\nSunny.\n"}, + {path: "summary.txt", data: "Summary\n"}, + } + manifestFiles := make([]bundle.ManifestFile, 0, len(files)) + for _, file := range files { + if err := os.WriteFile(filepath.Join(bundleRoot, filepath.FromSlash(file.path)), []byte(file.data), 0o600); err != nil { + t.Fatalf("write source file: %v", err) + } + manifestFiles = append(manifestFiles, bundle.ManifestFile{ + Path: file.path, + SHA256: bundle.FileDigest([]byte(file.data)), + Size: int64(len(file.data)), + }) + } + manifest := bundle.Manifest{ + SchemaVersion: 1, + ID: opts.ID, + Created: opts.Created, + Files: manifestFiles, + } + manifest.Digest = bundle.BundleDigest(manifest.Files) + data, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + t.Fatalf("marshal manifest: %v", err) + } + data = append(data, '\n') + if err := os.WriteFile(filepath.Join(bundleRoot, "manifest.json"), data, 0o600); err != nil { + t.Fatalf("write manifest: %v", err) + } + return manifest +} + +func writeLocalConfig(t *testing.T, sourceRoot, destinationRoot string) string { + t.Helper() + return writeConfigFile(t, ` +pipelines: + - id: reports + source: + backend: local + path: `+sourceRoot+` + destinations: + - id: archive + backend: local + path: `+destinationRoot+` + publish: + source: true + html: false +`) +} + +func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestination string) string { + t.Helper() + return writeConfigFile(t, ` +pipelines: + - id: reports + source: + backend: local + path: `+sourceRoot+` + destinations: + - id: archive-one + backend: local + path: `+firstDestination+` + - id: archive-two + backend: local + path: `+secondDestination+` +`) +} + +func writeConfigFile(t *testing.T, body string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "config.yml") + if err := os.WriteFile(path, []byte(strings.TrimSpace(body)+"\n"), 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + return path +} + +func writeDestinationState(t *testing.T, root, relative string, manifest bundle.Manifest) { + t.Helper() + bundleRoot := filepath.Join(root, filepath.FromSlash(relative)) + if err := os.MkdirAll(bundleRoot, 0o755); err != nil { + t.Fatalf("mkdir destination: %v", err) + } + destinationState := state.DistributorState{ + SchemaVersion: state.SchemaVersion, + PipelineID: "reports", + DestinationID: "archive", + PublishedAt: time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC), + Source: state.SourceState{Manifest: manifest}, + Outputs: []state.OutputFile{ + {Path: "report.md", Kind: state.OutputKindSource, SourcePath: "report.md", SHA256: manifest.Files[0].SHA256, Size: manifest.Files[0].Size}, + {Path: "summary.txt", Kind: state.OutputKindSource, SourcePath: "summary.txt", SHA256: manifest.Files[1].SHA256, Size: manifest.Files[1].Size}, + }, + } + data, err := json.MarshalIndent(destinationState, "", " ") + if err != nil { + t.Fatalf("marshal state: %v", err) + } + data = append(data, '\n') + if err := os.WriteFile(filepath.Join(bundleRoot, ".distributor.json"), data, 0o600); err != nil { + t.Fatalf("write state: %v", err) + } +} + +func readStateFile(t *testing.T, path string) state.DistributorState { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read state: %v", err) + } + destinationState, err := state.Parse(data) + if err != nil { + t.Fatalf("parse state: %v", err) + } + return destinationState +} + +func assertFile(t *testing.T, path, want string) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read file %s: %v", path, err) + } + if got := string(data); got != want { + t.Fatalf("%s = %q, want %q", path, got, want) } } diff --git a/internal/cli/root_test.go b/internal/cli/root_test.go index bcad2ad..248fad3 100644 --- a/internal/cli/root_test.go +++ b/internal/cli/root_test.go @@ -41,28 +41,6 @@ func TestExecuteVersion(t *testing.T) { } } -func TestRunWithoutDryRunFailsClearly(t *testing.T) { - tests := []string{"run"} - - for _, command := range tests { - t.Run(command, func(t *testing.T) { - var stdout, stderr bytes.Buffer - - code := Execute(context.Background(), []string{command}, &stdout, &stderr) - - if code != exitError { - t.Fatalf("exit code = %d, want %d", code, exitError) - } - if stdout.Len() != 0 { - t.Fatalf("stdout = %q, want empty", stdout.String()) - } - if !strings.Contains(stderr.String(), "not implemented") { - t.Fatalf("stderr = %q, want not implemented error", stderr.String()) - } - }) - } -} - func TestExecuteValidate(t *testing.T) { var stdout, stderr bytes.Buffer @@ -90,17 +68,19 @@ func TestExecuteInspect(t *testing.T) { } func TestExecuteRunDryRun(t *testing.T) { + sourceRoot := t.TempDir() + writeCLIBundle(t, sourceRoot) configPath := filepath.Join(t.TempDir(), "config.yml") err := os.WriteFile(configPath, []byte(` pipelines: - id: reports source: backend: local - path: /source + path: `+sourceRoot+` destinations: - id: archive backend: local - path: /archive + path: `+t.TempDir()+` `), 0o600) if err != nil { t.Fatalf("write config: %v", err) @@ -113,7 +93,7 @@ pipelines: if code != exitOK { t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String()) } - if !strings.Contains(stdout.String(), "Configured pipelines: 1") { + if !strings.Contains(stdout.String(), "action=publish_new") { t.Fatalf("stdout = %q, want config summary", stdout.String()) } if stderr.Len() != 0 { @@ -121,6 +101,38 @@ pipelines: } } +func TestExecuteRunPublishes(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeCLIBundle(t, sourceRoot) + configPath := filepath.Join(t.TempDir(), "config.yml") + err := os.WriteFile(configPath, []byte(` +pipelines: + - id: reports + source: + backend: local + path: `+sourceRoot+` + destinations: + - id: archive + backend: local + path: `+destinationRoot+` +`), 0o600) + if err != nil { + t.Fatalf("write config: %v", err) + } + + var stdout, stderr bytes.Buffer + + code := Execute(context.Background(), []string{"run", "--config", configPath}, &stdout, &stderr) + + if code != exitOK { + t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String()) + } + if _, err := os.Stat(filepath.Join(destinationRoot, ".distributor.json")); err != nil { + t.Fatalf("state stat error = %v", err) + } +} + func TestUnknownCommandIsUsageError(t *testing.T) { var stdout, stderr bytes.Buffer @@ -133,3 +145,37 @@ func TestUnknownCommandIsUsageError(t *testing.T) { t.Fatalf("stderr = %q, want unknown command error", stderr.String()) } } + +func writeCLIBundle(t *testing.T, root string) { + t.Helper() + for _, file := range []struct { + path string + data string + }{ + {"manifest.json", `{ + "schema_version": 1, + "id": "weather.daily.brentwood.2026-05-30", + "digest": "sha256:099b205780d2b050024868399961b05731729a548d5d6329c7b06a6740dd75fe", + "created": "2026-05-30T11:10:00Z", + "files": [ + { + "path": "report.md", + "sha256": "sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6", + "size": 16 + }, + { + "path": "summary.txt", + "sha256": "sha256:3cbb36aca330b3bd113955dfbada0adb7a5f95ad9f678bd61f175406c6a37e95", + "size": 8 + } + ] +} +`}, + {"report.md", "# Report\nSunny.\n"}, + {"summary.txt", "Summary\n"}, + } { + if err := os.WriteFile(filepath.Join(root, file.path), []byte(file.data), 0o600); err != nil { + t.Fatalf("write bundle file: %v", err) + } + } +} diff --git a/internal/config/load_test.go b/internal/config/load_test.go index 8769b5f..940d3a5 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -264,6 +264,7 @@ pipelines: func TestExampleConfigsLoad(t *testing.T) { for _, path := range []string{ "../../examples/local-to-local.yml", + "../../examples/local-publish.yml", "../../examples/fan-out.yml", } { t.Run(path, func(t *testing.T) { diff --git a/internal/publish/execute.go b/internal/publish/execute.go new file mode 100644 index 0000000..2077c7d --- /dev/null +++ b/internal/publish/execute.go @@ -0,0 +1,98 @@ +package publish + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +func Execute(ctx context.Context, req Request, plan Plan) error { + switch plan.Action { + case ActionSkipSame, ActionSkipDestinationNewer: + return nil + case ActionPublishNew, ActionReplaceOlder: + default: + return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason) + } + + if plan.Action == ActionReplaceOlder { + if plan.ExistingState == nil { + return fmt.Errorf("replace requires existing destination state") + } + if err := req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, existingManagedOutputPaths(*plan.ExistingState), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { + return err + } + if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil { + return err + } + } + + writtenOutputs := make([]Output, 0, len(plan.Outputs)) + cleanup := func() { + _ = req.DestinationBackend.DeleteManagedBundle(ctx, req.DestinationBundlePath, managedOutputPaths(writtenOutputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}) + } + for _, output := range plan.Outputs { + sourcePath, err := storage.Join(req.SourceBundle.RootRelativePath, output.SourcePath) + if err != nil { + cleanup() + return err + } + destinationPath, err := storage.Join(req.DestinationBundlePath, output.DestinationPath) + if err != nil { + cleanup() + return err + } + data, err := req.SourceBackend.ReadFile(ctx, sourcePath) + if err != nil { + cleanup() + return err + } + if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: false, PreferAtomic: true}); err != nil { + cleanup() + return err + } + writtenOutputs = append(writtenOutputs, output) + } + + destinationState := state.DistributorState{ + SchemaVersion: state.SchemaVersion, + DistributorVersion: req.DistributorVersion, + PipelineID: req.PipelineID, + DestinationID: req.DestinationID, + PublishedAt: time.Now().UTC(), + Source: state.SourceState{Manifest: req.SourceBundle.Manifest}, + Outputs: stateOutputs(plan.Outputs), + } + if err := state.Validate(destinationState); err != nil { + cleanup() + return err + } + data, err := json.MarshalIndent(destinationState, "", " ") + if err != nil { + cleanup() + return err + } + data = append(data, '\n') + statePath, err := storage.StatePath(req.DestinationBundlePath) + if err != nil { + cleanup() + return err + } + if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: false, PreferAtomic: true}); err != nil { + cleanup() + return err + } + return nil +} + +func existingManagedOutputPaths(destinationState state.DistributorState) []string { + paths := make([]string, 0, len(destinationState.Outputs)) + for _, output := range destinationState.Outputs { + paths = append(paths, output.Path) + } + return paths +} diff --git a/internal/publish/execute_test.go b/internal/publish/execute_test.go new file mode 100644 index 0000000..cd3836b --- /dev/null +++ b/internal/publish/execute_test.go @@ -0,0 +1,91 @@ +package publish + +import ( + "context" + "fmt" + "io" + "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/storage" + "gitea.maximumdirect.net/eric/distributor/internal/storage/fake" +) + +func TestExecuteCleansUpAfterWriteFailure(t *testing.T) { + sourceBackend := fake.New() + destinationBackend := &failingBackend{Backend: fake.New(), failPath: "summary.txt"} + sourceBundle := writeFakeSourceBundle(t, sourceBackend) + req := Request{ + PipelineID: "reports", + DestinationID: "archive", + SourceBundle: sourceBundle, + SourceBackend: sourceBackend, + DestinationBackend: destinationBackend, + DestinationBundlePath: "", + Publish: config.PublishPolicy{Source: true}, + Transfer: config.TransferPolicy{OnDestinationSame: config.TransferActionSkip, OnDestinationOlder: config.TransferActionReplace, OnDestinationNewer: config.TransferActionSkip, OnConflict: config.TransferActionFail}, + DistributorVersion: "test", + } + plan, err := Build(context.Background(), req) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + err = Execute(context.Background(), req, plan) + if err == nil { + t.Fatal("Execute() error = nil, want error") + } + found, err := destinationBackend.HasAny(context.Background(), "") + if err != nil { + t.Fatalf("HasAny() error = %v", err) + } + if found { + t.Fatal("destination has content after failed execution") + } +} + +type failingBackend struct { + *fake.Backend + failPath string +} + +func (b *failingBackend) WriteFile(ctx context.Context, path string, data []byte, opts storage.WriteOptions) (storage.Entry, error) { + if path == b.failPath { + return storage.Entry{}, fmt.Errorf("injected write failure") + } + return b.Backend.WriteFile(ctx, path, data, opts) +} + +func (b *failingBackend) WriteFrom(ctx context.Context, path string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) { + if path == b.failPath { + return storage.Entry{}, fmt.Errorf("injected write failure") + } + return b.Backend.WriteFrom(ctx, path, r, opts) +} + +func writeFakeSourceBundle(t *testing.T, backend *fake.Backend) bundle.Bundle { + t.Helper() + files := []struct { + path string + data string + }{ + {path: "report.md", data: "# Report\nSunny.\n"}, + {path: "summary.txt", data: "Summary\n"}, + } + manifestFiles := make([]bundle.ManifestFile, 0, len(files)) + for _, file := range files { + if _, err := backend.WriteFile(context.Background(), file.path, []byte(file.data), storage.WriteOptions{}); err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + manifestFiles = append(manifestFiles, bundle.ManifestFile{Path: file.path, SHA256: bundle.FileDigest([]byte(file.data)), Size: int64(len(file.data))}) + } + manifest := bundle.Manifest{ + SchemaVersion: 1, + ID: "weather.daily.brentwood.2026-05-30", + Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC), + Files: manifestFiles, + } + manifest.Digest = bundle.BundleDigest(manifest.Files) + return bundle.Bundle{Manifest: manifest} +} diff --git a/internal/publish/output.go b/internal/publish/output.go new file mode 100644 index 0000000..83abcd6 --- /dev/null +++ b/internal/publish/output.go @@ -0,0 +1,54 @@ +package publish + +import ( + "fmt" + + "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +func PlanSourceOutputs(req Request) ([]Output, error) { + if !req.Publish.Source { + return nil, nil + } + outputs := make([]Output, 0, len(req.SourceBundle.Manifest.Files)) + seen := make(map[string]struct{}, len(req.SourceBundle.Manifest.Files)) + for _, file := range req.SourceBundle.Manifest.Files { + if _, exists := seen[file.Path]; exists { + return nil, fmt.Errorf("destination output path collision: %s", file.Path) + } + seen[file.Path] = struct{}{} + if err := storage.ValidatePath(file.Path); err != nil { + return nil, fmt.Errorf("destination output path %q: %w", file.Path, err) + } + outputs = append(outputs, Output{ + SourcePath: file.Path, + DestinationPath: file.Path, + SHA256: file.SHA256, + Size: file.Size, + }) + } + return outputs, nil +} + +func stateOutputs(outputs []Output) []state.OutputFile { + files := make([]state.OutputFile, 0, len(outputs)) + for _, output := range outputs { + files = append(files, state.OutputFile{ + Path: output.DestinationPath, + Kind: state.OutputKindSource, + SourcePath: output.SourcePath, + SHA256: output.SHA256, + Size: output.Size, + }) + } + return files +} + +func managedOutputPaths(outputs []Output) []string { + paths := make([]string, 0, len(outputs)) + for _, output := range outputs { + paths = append(paths, output.DestinationPath) + } + return paths +} diff --git a/internal/publish/output_test.go b/internal/publish/output_test.go new file mode 100644 index 0000000..fd4899c --- /dev/null +++ b/internal/publish/output_test.go @@ -0,0 +1,29 @@ +package publish + +import ( + "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" +) + +func TestPlanSourceOutputsRejectsCollision(t *testing.T) { + _, err := PlanSourceOutputs(Request{ + SourceBundle: bundle.Bundle{ + Manifest: bundle.Manifest{ + SchemaVersion: 1, + ID: "bundle", + Created: time.Date(2026, 5, 30, 11, 10, 0, 0, time.UTC), + Files: []bundle.ManifestFile{ + {Path: "report.md", SHA256: "sha256:1111111111111111111111111111111111111111111111111111111111111111", Size: 1}, + {Path: "report.md", SHA256: "sha256:2222222222222222222222222222222222222222222222222222222222222222", Size: 1}, + }, + }, + }, + Publish: config.PublishPolicy{Source: true}, + }) + if err == nil { + t.Fatal("PlanSourceOutputs() error = nil, want collision") + } +} diff --git a/internal/publish/plan.go b/internal/publish/plan.go new file mode 100644 index 0000000..874aed8 --- /dev/null +++ b/internal/publish/plan.go @@ -0,0 +1,141 @@ +package publish + +import ( + "context" + "fmt" + + "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +type Action string + +const ( + ActionPublishNew Action = "publish_new" + ActionReplaceOlder Action = "replace_older" + ActionSkipSame Action = "skip_same" + ActionSkipDestinationNewer Action = "skip_destination_newer" + ActionFailConflict Action = "fail_conflict" + ActionFailUnmanaged Action = "fail_unmanaged" +) + +type Request struct { + PipelineID string + DestinationID string + SourceBundle bundle.Bundle + SourceBackend storage.Backend + DestinationBackend storage.Backend + DestinationBundlePath string + Publish config.PublishPolicy + Transfer config.TransferPolicy + DistributorVersion string +} + +type Plan struct { + PipelineID string + DestinationID string + BundleID string + BundlePath string + DestinationBundlePath string + Action Action + Reason string + Outputs []Output + ExistingState *state.DistributorState +} + +type Output struct { + SourcePath string + DestinationPath string + SHA256 string + Size int64 +} + +func Build(ctx context.Context, req Request) (Plan, error) { + if err := validateRequest(req); err != nil { + return Plan{}, err + } + outputs, err := PlanSourceOutputs(req) + if err != nil { + return Plan{}, err + } + status, err := inspectDestination(ctx, req.DestinationBackend, req.DestinationBundlePath) + if err != nil { + return Plan{}, err + } + comparison := state.Compare(req.SourceBundle.Manifest, req.PipelineID, req.DestinationID, status) + action, reason := actionForComparison(comparison, req.Transfer) + plan := Plan{ + PipelineID: req.PipelineID, + DestinationID: req.DestinationID, + BundleID: req.SourceBundle.Manifest.ID, + BundlePath: req.SourceBundle.RootRelativePath, + DestinationBundlePath: req.DestinationBundlePath, + Action: action, + Reason: reason, + Outputs: outputs, + ExistingState: status.State, + } + if action == ActionFailConflict || action == ActionFailUnmanaged { + return plan, fmt.Errorf("%s: %s", action, reason) + } + return plan, nil +} + +func validateRequest(req Request) error { + if req.PipelineID == "" { + return fmt.Errorf("pipeline id is required") + } + if req.DestinationID == "" { + return fmt.Errorf("destination id is required") + } + if req.SourceBackend == nil { + return fmt.Errorf("source backend is required") + } + if req.DestinationBackend == nil { + return fmt.Errorf("destination backend is required") + } + if req.Publish.HTML { + return fmt.Errorf("publish html is not implemented") + } + if !req.Publish.Source { + return fmt.Errorf("publish source must be enabled") + } + return nil +} + +func actionForComparison(comparison state.Comparison, transfer config.TransferPolicy) (Action, string) { + switch comparison.Outcome { + case state.OutcomeDestinationAbsent: + return ActionPublishNew, comparison.Reason + case state.OutcomeDestinationUnmanaged: + return ActionFailUnmanaged, comparison.Reason + case state.OutcomeInvalidState, state.OutcomeIdentityMismatch, state.OutcomeSameCreatedConflict, state.OutcomeDifferentSourceConflict: + return ActionFailConflict, comparison.Reason + case state.OutcomeSameSource: + if transfer.OnDestinationSame == config.TransferActionFail { + return ActionFailConflict, "destination matches source and transfer policy requires failure" + } + return ActionSkipSame, comparison.Reason + case state.OutcomeDestinationOlder: + if transfer.OnDestinationOlder == config.TransferActionFail { + return ActionFailConflict, "destination is older and transfer policy requires failure" + } + return ActionReplaceOlder, comparison.Reason + case state.OutcomeDestinationNewer: + if transfer.OnDestinationNewer == config.TransferActionFail { + return ActionFailConflict, "destination is newer and transfer policy requires failure" + } + return ActionSkipDestinationNewer, comparison.Reason + default: + return ActionFailConflict, "unsupported comparison outcome" + } +} + +func displayPath(path string) string { + if path == "" { + return "." + } + return path +} diff --git a/internal/publish/reconcile.go b/internal/publish/reconcile.go new file mode 100644 index 0000000..77a92f1 --- /dev/null +++ b/internal/publish/reconcile.go @@ -0,0 +1,31 @@ +package publish + +import ( + "context" + + "gitea.maximumdirect.net/eric/distributor/internal/state" + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +func inspectDestination(ctx context.Context, backend storage.Backend, bundlePath string) (state.DestinationStatus, error) { + statePath, err := storage.StatePath(bundlePath) + if err != nil { + return state.DestinationStatus{}, err + } + data, err := backend.ReadFile(ctx, statePath) + if err == nil { + destinationState, parseErr := state.Parse(data) + if parseErr != nil { + return state.DestinationStatus{StateErr: parseErr}, nil + } + return state.DestinationStatus{State: &destinationState, HasContents: true}, nil + } + if !storage.IsNotFound(err) { + return state.DestinationStatus{}, err + } + hasContents, err := backend.HasAny(ctx, bundlePath) + if err != nil { + return state.DestinationStatus{}, err + } + return state.DestinationStatus{HasContents: hasContents}, nil +} diff --git a/internal/publish/safety.go b/internal/publish/safety.go new file mode 100644 index 0000000..a2cb9a4 --- /dev/null +++ b/internal/publish/safety.go @@ -0,0 +1,19 @@ +package publish + +import ( + "context" + "fmt" + + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +func ensureDestinationEmpty(ctx context.Context, backend storage.Backend, bundlePath string) error { + hasAny, err := backend.HasAny(ctx, bundlePath) + if err != nil { + return err + } + if hasAny { + return fmt.Errorf("destination bundle path %q is not empty after managed cleanup", displayPath(bundlePath)) + } + return nil +}