diff --git a/docs/cli.md b/docs/cli.md index e96702d..71c9743 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -22,7 +22,7 @@ distributor inspect `run --config ` executes configured local-to-local pipelines that publish source files, generated HTML files, or both. -`run --config --dry-run` discovers source bundles, inspects destination state, and prints planned actions without writing files. +`run --config --dry-run` discovers source bundles, inspects destination state, and prints planned actions plus a final status summary without writing files. `validate ` validates a local source bundle directory or a local tree containing source bundles. @@ -43,7 +43,7 @@ Each subcommand supports: `run` supports: - `--config `: config file to load. -- `--dry-run`: validate config and print the resolved summary without publishing. +- `--dry-run`: validate config, print planned actions and final status, and do not publish. ## Common workflows diff --git a/docs/internal/notify.md b/docs/internal/notify.md new file mode 100644 index 0000000..66465ab --- /dev/null +++ b/docs/internal/notify.md @@ -0,0 +1,13 @@ +# Notify + +## Purpose + +`internal/notify` defines the internal notification interface used by the application runner. + +## Current behavior + +The implemented notifier is a no-op. It is invoked only after a successful publish or replacement. Dry-run, skipped destinations, and failed destinations do not invoke it. + +## Boundaries + +No external notification adapters are implemented. Notification configuration is not part of the current user-facing config schema. diff --git a/docs/operations.md b/docs/operations.md index 7654967..94d8690 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -40,6 +40,10 @@ If destination state is older than the source manifest, publication replaces onl 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. +If one destination fails, later destinations in the same fan-out are still planned and run where they are independent. The command exits non-zero after printing the final status when any destination fails. + +After a successful publish or replacement, the internal notifier hook runs as a no-op. Skipped destinations do not invoke it. + ## Caveats -Only local-to-local publication is implemented. SSH, S3, notification, and force overwrite behavior are not implemented. +Only local-to-local publication is implemented. SSH, S3, external notification adapters, and force overwrite behavior are not implemented. diff --git a/internal/app/run.go b/internal/app/run.go index 27f20ba..5c3fac5 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "fmt" "io" "strings" @@ -9,6 +10,7 @@ import ( "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/notify" "gitea.maximumdirect.net/eric/distributor/internal/publish" ) @@ -16,6 +18,7 @@ type RunOptions struct { ConfigPath string DryRun bool Stdout io.Writer + Notifier notify.Notifier } func Run(ctx context.Context, options RunOptions) error { @@ -35,6 +38,12 @@ func Run(ctx context.Context, options RunOptions) error { } func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error { + notifier := options.Notifier + if notifier == nil { + notifier = notify.Noop{} + } + summary := runSummary{dryRun: options.DryRun} + var failures runFailures if options.Stdout != nil { if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil { return err @@ -53,18 +62,29 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error 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 { + if _, err := fmt.Fprintf(options.Stdout, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.Source.Backend, len(bundles), destinationSummary(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) + err := fmt.Errorf("backend %s is not implemented for execution", destination.Backend) + failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + summary.recordFailure() + if options.Stdout != nil { + writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, err) + } + continue } destinationBackend, err := local.New(destination.Path) if err != nil { - return err + failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + summary.recordFailure() + if options.Stdout != nil { + writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, err) + } + continue } req := publish.Request{ PipelineID: pipeline.ID, @@ -79,20 +99,43 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error DistributorVersion: Version, } plan, err := publish.Build(ctx, req) + if err != nil && plan.DestinationID == "" { + plan = publish.Plan{DestinationID: destination.ID, BundlePath: sourceBundle.RootRelativePath} + } 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) + failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + summary.recordFailure() + continue } + summary.recordPlan(plan.Action) 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) + failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + summary.recordFailure() + continue + } + if shouldNotify(plan.Action) { + if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil { + failures.add(pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err) + summary.recordFailure() + continue + } } } } } } + if options.Stdout != nil { + if _, err := fmt.Fprintln(options.Stdout, summary.Line()); err != nil { + return err + } + } + if len(failures.items) > 0 { + return failures + } return nil } @@ -101,12 +144,23 @@ func writePlanLine(w io.Writer, plan publish.Plan, planErr error) { return } if planErr != nil { - fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, planErr.Error()) + destinationID := plan.DestinationID + if destinationID == "" { + destinationID = "unknown" + } + fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(plan.BundlePath), destinationID, planErr.Error()) return } fmt.Fprintf(w, " - bundle=%s destination=%s action=%s outputs=%s reason=%q\n", displayBundlePath(plan.BundlePath), plan.DestinationID, plan.Action, outputSummary(plan.Outputs), plan.Reason) } +func writeErrorLine(w io.Writer, bundlePath, destinationID string, err error) { + if w == nil { + return + } + fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(bundlePath), destinationID, err.Error()) +} + func outputSummary(outputs []publish.Output) string { if len(outputs) == 0 { return "none" @@ -117,3 +171,112 @@ func outputSummary(outputs []publish.Output) string { } return strings.Join(paths, ",") } + +func destinationSummary(destinations []config.Destination) string { + if len(destinations) == 0 { + return "none" + } + ids := make([]string, 0, len(destinations)) + for _, destination := range destinations { + ids = append(ids, destination.ID) + } + return strings.Join(ids, ",") +} + +func shouldNotify(action publish.Action) bool { + return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder +} + +func notifyEvent(plan publish.Plan) notify.Event { + outputs := make([]notify.Output, 0, len(plan.Outputs)) + for _, output := range plan.Outputs { + outputs = append(outputs, notify.Output{ + Path: output.DestinationPath, + Kind: output.Kind, + SourcePath: output.SourcePath, + Transform: output.Transform, + SHA256: output.SHA256, + Size: output.Size, + }) + } + return notify.Event{ + PipelineID: plan.PipelineID, + DestinationID: plan.DestinationID, + BundleID: plan.BundleID, + BundlePath: plan.BundlePath, + Action: string(plan.Action), + Outputs: outputs, + } +} + +type runSummary struct { + dryRun bool + planned int + publishNew int + replaceOlder int + skipped int + failures int +} + +func (s *runSummary) recordPlan(action publish.Action) { + s.planned++ + switch action { + case publish.ActionPublishNew: + s.publishNew++ + case publish.ActionReplaceOlder: + s.replaceOlder++ + case publish.ActionSkipSame, publish.ActionSkipDestinationNewer: + s.skipped++ + } +} + +func (s *runSummary) recordFailure() { + s.failures++ +} + +func (s runSummary) Line() string { + status := "ok" + if s.failures > 0 { + status = "failed" + } + return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d skipped=%d failed=%d dry_run=%t", status, s.planned, s.publishNew, s.replaceOlder, s.skipped, s.failures, s.dryRun) +} + +type runFailure struct { + pipelineID string + destinationID string + bundlePath string + err error +} + +type runFailures struct { + items []runFailure +} + +func (f *runFailures) add(pipelineID, destinationID, bundlePath string, err error) { + f.items = append(f.items, runFailure{ + pipelineID: pipelineID, + destinationID: destinationID, + bundlePath: bundlePath, + err: err, + }) +} + +func (f runFailures) Error() string { + if len(f.items) == 0 { + return "" + } + parts := make([]string, 0, len(f.items)) + for _, item := range f.items { + parts = append(parts, fmt.Sprintf("pipeline %s destination %s bundle %s: %v", item.pipelineID, item.destinationID, item.bundlePath, item.err)) + } + return "run failed: " + strings.Join(parts, "; ") +} + +func (f runFailures) Unwrap() error { + errs := make([]error, 0, len(f.items)) + for _, item := range f.items { + errs = append(errs, item.err) + } + return errors.Join(errs...) +} diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 4be6910..0f4e900 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -12,6 +12,7 @@ import ( "time" "gitea.maximumdirect.net/eric/distributor/internal/bundle" + "gitea.maximumdirect.net/eric/distributor/internal/notify" "gitea.maximumdirect.net/eric/distributor/internal/state" ) @@ -34,8 +35,9 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) { output := stdout.String() for _, want := range []string{ "Configured pipelines: 1", - "- reports: source=local bundles=1 destinations=1", + "- pipeline=reports source=local bundles=1 destinations=archive", "bundle=. destination=archive action=publish_new outputs=report.md,summary.txt", + "Final status: ok planned=1 publish_new=1 replace_older=0 skipped=0 failed=0 dry_run=true", } { if !strings.Contains(output, want) { t.Fatalf("Run() output = %q, want substring %q", output, want) @@ -73,6 +75,135 @@ func TestRunPublishesNewLocalBundle(t *testing.T) { } } +func TestRunNotifiesAfterPublication(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + notifier := &recordingNotifier{ + check: func() { + if _, err := os.Stat(filepath.Join(destinationRoot, ".distributor.json")); err != nil { + t.Fatalf("state stat during notify: %v", err) + } + }, + } + + err := Run(context.Background(), RunOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + Notifier: notifier, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if got, want := len(notifier.events), 1; got != want { + t.Fatalf("notification count = %d, want %d", got, want) + } + event := notifier.events[0] + if event.PipelineID != "reports" || event.DestinationID != "archive" || event.BundleID == "" || event.Action != "publish_new" { + t.Fatalf("notification event = %#v", event) + } + if got, want := len(event.Outputs), 2; got != want { + t.Fatalf("notification output count = %d, want %d", got, want) + } +} + +func TestRunNotifiesAfterReplacement(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) + } + notifier := &recordingNotifier{} + + err := Run(context.Background(), RunOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + Notifier: notifier, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if got, want := len(notifier.events), 1; got != want { + t.Fatalf("notification count = %d, want %d", got, want) + } + if notifier.events[0].Action != "replace_older" { + t.Fatalf("notification action = %q, want replace_older", notifier.events[0].Action) + } +} + +func TestRunDoesNotNotifyForSkippedDestination(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) + } + notifier := &recordingNotifier{} + + err := Run(context.Background(), RunOptions{ConfigPath: configPath, Notifier: notifier}) + if err != nil { + t.Fatalf("second Run() error = %v", err) + } + if len(notifier.events) != 0 { + t.Fatalf("notifications = %#v, want none", notifier.events) + } +} + +func TestRunDoesNotNotifyDuringDryRun(t *testing.T) { + sourceRoot := t.TempDir() + destinationRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + notifier := &recordingNotifier{} + + err := Run(context.Background(), RunOptions{ + ConfigPath: writeLocalConfig(t, sourceRoot, destinationRoot), + DryRun: true, + Notifier: notifier, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(notifier.events) != 0 { + t.Fatalf("notifications = %#v, want none", notifier.events) + } + if entries, err := os.ReadDir(destinationRoot); err != nil || len(entries) != 0 { + t.Fatalf("destination entries = %v err=%v, want empty", entries, err) + } +} + +func TestRunContinuesAfterDestinationFailure(t *testing.T) { + sourceRoot := t.TempDir() + firstDestination := t.TempDir() + secondDestination := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + if err := os.WriteFile(filepath.Join(firstDestination, "unmanaged.txt"), []byte("data"), 0o600); err != nil { + t.Fatalf("write unmanaged file: %v", err) + } + + var stdout bytes.Buffer + err := Run(context.Background(), RunOptions{ + ConfigPath: writeFanoutConfig(t, sourceRoot, firstDestination, secondDestination), + Stdout: &stdout, + }) + if err == nil || !strings.Contains(err.Error(), "fail_unmanaged") { + t.Fatalf("Run() error = %v, want unmanaged failure", err) + } + output := stdout.String() + for _, want := range []string{ + "destination=archive-one action=error", + "destination=archive-two action=publish_new", + "Final status: failed planned=1 publish_new=1 replace_older=0 skipped=0 failed=1 dry_run=false", + } { + if !strings.Contains(output, want) { + t.Fatalf("stdout = %q, want substring %q", output, want) + } + } + assertFile(t, filepath.Join(secondDestination, "report.md"), "# Report\nSunny.\n") +} + func TestRunPublishesHTMLOnly(t *testing.T) { sourceRoot := t.TempDir() destinationRoot := t.TempDir() @@ -475,3 +606,19 @@ func assertFileContains(t *testing.T, path, want string) { t.Fatalf("%s = %q, want substring %q", path, data, want) } } + +type recordingNotifier struct { + events []notify.Event + check func() +} + +func (n *recordingNotifier) Notify(ctx context.Context, event notify.Event) error { + if err := ctx.Err(); err != nil { + return err + } + if n.check != nil { + n.check() + } + n.events = append(n.events, event) + return nil +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 940d9c6..d74e5f0 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -45,7 +45,7 @@ Options: --config Path to config file --dry-run Load and validate config without publishing -Execution behavior is not implemented yet. Dry-run currently prints a resolved -configuration summary only. +Run discovers local source bundles, plans each configured destination, publishes +selected outputs unless --dry-run is set, and prints a final status summary. `) } diff --git a/internal/notify/noop.go b/internal/notify/noop.go new file mode 100644 index 0000000..58b2897 --- /dev/null +++ b/internal/notify/noop.go @@ -0,0 +1,9 @@ +package notify + +import "context" + +type Noop struct{} + +func (Noop) Notify(ctx context.Context, event Event) error { + return ctx.Err() +} diff --git a/internal/notify/notify.go b/internal/notify/notify.go new file mode 100644 index 0000000..a78f7dd --- /dev/null +++ b/internal/notify/notify.go @@ -0,0 +1,25 @@ +package notify + +import "context" + +type Event struct { + PipelineID string + DestinationID string + BundleID string + BundlePath string + Action string + Outputs []Output +} + +type Output struct { + Path string + Kind string + SourcePath string + Transform string + SHA256 string + Size int64 +} + +type Notifier interface { + Notify(ctx context.Context, event Event) error +}