eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Add notification hook and run summary

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-31 02:33:09 +00:00
parent 5408f14195
commit 78f92154ab
8 changed files with 373 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
internal/app/run.go
View File

@@ -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...)
}

149
internal/app/run_test.go
View File

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

4
internal/cli/run.go
View File

@@ -45,7 +45,7 @@ Options:
--config <path> 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.
`)
}

9
internal/notify/noop.go Normal file
View File

@@ -0,0 +1,9 @@
package notify
import "context"
type Noop struct{}
func (Noop) Notify(ctx context.Context, event Event) error {
return ctx.Err()
}

25
internal/notify/notify.go Normal file
View File

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