package app import ( "context" "fmt" "io" "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 Force bool Stdout io.Writer OutputFormat OutputFormat Notifier notify.Notifier } func Run(ctx context.Context, options RunOptions) error { if err := ValidateOutputFormat(options.OutputFormat); err != nil { return err } 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 { return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment) } type backendFactoryProvider func(config.Environment) *backendFactory func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error { notifier := options.Notifier if notifier == nil { notifier = notify.Noop{} } jsonOutput := IsJSONOutput(options.OutputFormat) summary := runSummary{dryRun: options.DryRun} result := runResult{ DryRun: options.DryRun, Pipelines: []runPipelineResult{}, Actions: []runActionResult{}, } var warnings []OutputWarning var failures runFailures secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil) if err != nil { return err } secretWarnings := secretConflictWarnings(secretLoad.Conflicts) if jsonOutput { warnings = append(warnings, secretWarnings...) } else if options.Stdout != nil { if err := writeWarnings(options.Stdout, secretWarnings); err != nil { return err } } backends := provider(secretLoad.Environment) backends.readOnlyKnownHosts = options.DryRun transforms := newTransformRegistry() if options.Stdout != nil && !jsonOutput { if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil { return err } } for _, pipeline := range cfg.Pipelines { pipelineWarnings := sshWarnings(pipeline) if jsonOutput { warnings = append(warnings, pipelineWarnings...) } else if options.Stdout != nil { if err := writeWarnings(options.Stdout, pipelineWarnings); err != nil { return err } } sourceBackend, err := backends.openSource(ctx, pipeline.Source) if err != nil { return fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err) } bundles, err := bundle.Discover(ctx, sourceBackend, "") if err != nil { closeBackend(sourceBackend) return fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err) } result.Pipelines = append(result.Pipelines, runPipelineResult{ ID: pipeline.ID, SourceBackend: pipeline.Source.Backend, BundleCount: len(bundles), Destinations: destinationIDs(pipeline.Destinations), }) if options.Stdout != nil && !jsonOutput { 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 _, destination := range pipeline.Destinations { selections := selectDestinationBundles(destination, bundles) if isFixedPathDestination(destination) { summary.recordFixedPath() if options.DryRun { warning := fixedPathSelectionWarning(pipeline.ID, destination.ID, selections, len(bundles)) if jsonOutput { warnings = append(warnings, warning) } else if options.Stdout != nil { if err := writeWarnings(options.Stdout, []OutputWarning{warning}); err != nil { closeBackend(sourceBackend) return err } } } } if len(selections) == 0 { continue } destinationBackend, err := backends.openDestination(ctx, destination) if err != nil { for _, selection := range selections { failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(selection.SourceBundle.RootRelativePath), err) summary.recordFailure() if jsonOutput { result.Actions = append(result.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, selection.SourceBundle.RootRelativePath, err)) } else if options.Stdout != nil { writeErrorLine(options.Stdout, selection.SourceBundle.RootRelativePath, destination.ID, destination.Backend, err) } } continue } closeDestination := true deferCloseDestination := func() { if closeDestination { closeBackend(destinationBackend) closeDestination = false } } for _, selection := range selections { sourceBundle := selection.SourceBundle req := publish.Request{ PipelineID: pipeline.ID, DestinationID: destination.ID, SourceBundle: sourceBundle, SourceBackend: sourceBackend, DestinationBackend: destinationBackend, DestinationBundlePath: selection.DestinationBundlePath, PathMapping: destination.PathMap.Mode, Publish: *destination.Publish, Transform: destination.Transform, Links: destination.Links, Transformers: transforms, Transfer: destination.Transfer, DistributorVersion: Version, Force: options.Force, } plan, err := publish.Build(ctx, req) if err != nil { if plan.PipelineID == "" { plan.PipelineID = pipeline.ID } if plan.DestinationID == "" { plan.DestinationID = destination.ID } if plan.BundleID == "" { plan.BundleID = sourceBundle.Manifest.ID } if plan.BundlePath == "" { plan.BundlePath = sourceBundle.RootRelativePath } if plan.DestinationBundlePath == "" { plan.DestinationBundlePath = selection.DestinationBundlePath } } if isFixedPathDestination(destination) { plan.PathMapping = config.PathMappingFixed if options.DryRun && isDestructiveFixedPathAction(plan.Action) { warning := fixedPathReplacementWarning(plan) if jsonOutput { warnings = append(warnings, warning) } else if options.Stdout != nil { if err := writeWarnings(options.Stdout, []OutputWarning{warning}); err != nil { deferCloseDestination() closeBackend(sourceBackend) return err } } } } if jsonOutput { result.Actions = append(result.Actions, runActionFromPlan(destination.Backend, plan, err)) } else if options.Stdout != nil { writePlanLine(options.Stdout, destination.Backend, plan, err) } if err != nil { failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(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, destination.Backend, storage.DisplayPath(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, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err) summary.recordFailure() continue } } } } deferCloseDestination() } closeBackend(sourceBackend) } result.Summary = summary.Result() if jsonOutput { if err := WriteJSONEnvelope(options.Stdout, "run", len(failures.items) == 0, warnings, result, failures.outputErrors()); err != nil { return err } } else 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() }