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" "gitea.maximumdirect.net/eric/distributor/internal/storage" ) 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 secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil) if err != nil { return err } if options.Stdout != nil { if err := writeSecretConflictWarnings(options.Stdout, secretLoad.Conflicts); err != nil { return err } } backends := newBackendFactoryWithEnvironment(secretLoad.Environment) 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 { if options.Stdout != nil { if err := writeSSHWarnings(options.Stdout, pipeline); err != nil { return err } } 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 { closeBackend(sourceBackend) 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 { closeBackend(sourceBackend) 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, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() if options.Stdout != nil { writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, err) } continue } closeDestination := true deferCloseDestination := func() { if closeDestination { closeBackend(destinationBackend) closeDestination = false } } 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 { deferCloseDestination() failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } summary.recordPlan(plan.Action) if !options.DryRun { if err := publish.Execute(ctx, req, plan); err != nil { deferCloseDestination() failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } if shouldNotify(plan.Action) { if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil { deferCloseDestination() failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } } } deferCloseDestination() } } closeBackend(sourceBackend) } 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 } type closeableBackend interface { Close() error } func closeBackend(backend storage.Backend) { closeable, ok := backend.(closeableBackend) if !ok { return } _ = closeable.Close() } 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", storage.DisplayPath(plan.BundlePath), destinationID, planErr.Error()) return } fmt.Fprintf(w, " - bundle=%s destination=%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(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", storage.DisplayPath(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 writeSecretConflictWarnings(w io.Writer, conflicts []config.SecretConflict) error { for _, conflict := range conflicts { if _, err := fmt.Fprintf(w, "Warning: secret %s ignored because the real environment already has that variable\n", conflict.Name); err != nil { return err } } return nil } func writeSSHWarnings(w io.Writer, pipeline config.Pipeline) error { if pipeline.Source.Backend == config.BackendSSH && pipeline.Source.SSH.HostKeyPolicy == config.HostKeyPolicyOff { if _, err := fmt.Fprintf(w, "Warning: pipeline=%s source host_key_policy=off disables SSH host key checking\n", pipeline.ID); err != nil { return err } } for _, destination := range pipeline.Destinations { if destination.Backend == config.BackendSSH && destination.SSH.HostKeyPolicy == config.HostKeyPolicyOff { if _, err := fmt.Fprintf(w, "Warning: pipeline=%s destination=%s host_key_policy=off disables SSH host key checking\n", pipeline.ID, destination.ID); err != nil { return err } } } return nil } 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...) }