distributor/internal/app/run.go

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()
}