distributor/internal/app/run.go

package app

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"

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

type RunOptions struct {
	ConfigPath string
	DryRun     bool
	Stdout     io.Writer
	Notifier   notify.Notifier
}

func Run(ctx context.Context, options RunOptions) error {
	if err := ctx.Err(); err != nil {
		return err
	}

	configPath := options.ConfigPath
	if configPath == "" {
		configPath = config.DefaultConfigPath
	}
	cfg, err := config.LoadFile(configPath)
	if err != nil {
		return err
	}
	return runConfig(ctx, cfg, options)
}

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
	backends := newBackendFactory()
	transforms := newTransformRegistry()
	if options.Stdout != nil {
		if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil {
			return err
		}
	}
	for _, pipeline := range cfg.Pipelines {
		sourceBackend, err := backends.openSource(ctx, pipeline.Source)
		if err != nil {
			return fmt.Errorf("pipeline %s: %w", pipeline.ID, 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, "- 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 {
				destinationBackend, err := backends.openDestination(ctx, destination)
				if err != nil {
					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,
					DestinationID:         destination.ID,
					SourceBundle:          sourceBundle,
					SourceBackend:         sourceBackend,
					DestinationBackend:    destinationBackend,
					DestinationBundlePath: sourceBundle.RootRelativePath,
					Publish:               *destination.Publish,
					Transform:             destination.Transform,
					Transformers:          transforms,
					Transfer:              destination.Transfer,
					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 {
					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 {
						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
}

func writePlanLine(w io.Writer, plan publish.Plan, planErr error) {
	if w == nil {
		return
	}
	if planErr != nil {
		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"
	}
	paths := make([]string, 0, len(outputs))
	for _, output := range outputs {
		paths = append(paths, output.DestinationPath)
	}
	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...)
}