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

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

type RunPipelineOptions struct {
	ConfigPath string
	PipelineID string
	DryRun     bool
	Force      bool
	Notifier   notify.Notifier
}

type RunPipelineWithLocalSourceOptions struct {
	ConfigPath string
	PipelineID string
	SourceRoot string
	DryRun     bool
	Force      bool
	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
	}

	setup, err := loadRuntimeSetup(options.ConfigPath)
	if err != nil {
		return err
	}
	return runSetup(ctx, setup, options)
}

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

	setup, err := loadRuntimeSetup(options.ConfigPath)
	if err != nil {
		return RunReport{}, err
	}
	return runPipelineSetup(ctx, setup, options)
}

func RunPipelineWithLocalSource(ctx context.Context, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
	if err := ctx.Err(); err != nil {
		return RunReport{}, err
	}
	if options.SourceRoot == "" {
		return RunReport{}, fmt.Errorf("source root is required")
	}

	setup, err := loadRuntimeSetup(options.ConfigPath)
	if err != nil {
		return RunReport{}, err
	}
	return runPipelineSetupWithLocalSource(ctx, setup, options)
}

func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return err
	}
	return runSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}

type backendFactoryProvider func(config.Environment) *backendFactory

func runPipelineConfig(ctx context.Context, cfg config.Config, options RunPipelineOptions) (RunReport, error) {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return RunReport{}, err
	}
	return runPipelineSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}

func runPipelineConfigWithLocalSource(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return RunReport{}, err
	}
	return runPipelineSetupWithLocalSourceAndBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}

func runPipelineConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return RunReport{}, err
	}
	return runPipelineSetupWithBackendFactory(ctx, setup, options, provider)
}

func runPipelineSetup(ctx context.Context, setup runtimeSetup, options RunPipelineOptions) (RunReport, error) {
	return runPipelineSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}

func runPipelineSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options RunPipelineOptions, provider backendFactoryProvider) (RunReport, error) {
	pipeline, ok := findPipeline(setup.Config, options.PipelineID)
	if !ok {
		return RunReport{}, PipelineNotFoundError{ID: options.PipelineID}
	}
	return buildRunReportWithSetup(ctx, setup.withPipelines([]config.Pipeline{pipeline}), RunOptions{
		DryRun:   options.DryRun,
		Force:    options.Force,
		Notifier: options.Notifier,
	}, provider, nil)
}

func runPipelineConfigWithLocalSourceAndBackendFactory(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions, provider backendFactoryProvider) (RunReport, error) {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return RunReport{}, err
	}
	return runPipelineSetupWithLocalSourceAndBackendFactory(ctx, setup, options, provider)
}

func runPipelineSetupWithLocalSource(ctx context.Context, setup runtimeSetup, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
	return runPipelineSetupWithLocalSourceAndBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}

func runPipelineSetupWithLocalSourceAndBackendFactory(ctx context.Context, setup runtimeSetup, options RunPipelineWithLocalSourceOptions, provider backendFactoryProvider) (RunReport, error) {
	pipeline, ok := findPipeline(setup.Config, options.PipelineID)
	if !ok {
		return RunReport{}, PipelineNotFoundError{ID: options.PipelineID}
	}
	return buildRunReportWithSetup(ctx, setup.withPipelines([]config.Pipeline{pipeline}), RunOptions{
		DryRun:   options.DryRun,
		Force:    options.Force,
		Notifier: options.Notifier,
	}, provider, &localSourceRoot{
		pipelineID: options.PipelineID,
		root:       options.SourceRoot,
	})
}

func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return err
	}
	return runSetupWithBackendFactory(ctx, setup, options, provider)
}

func runSetup(ctx context.Context, setup runtimeSetup, options RunOptions) error {
	return runSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}

func runSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options RunOptions, provider backendFactoryProvider) error {
	report, err := buildRunReportWithSetup(ctx, setup, options, provider, nil)
	if err != nil && !IsPartialResultError(err) {
		return err
	}
	if outputErr := WriteRunReport(options.Stdout, options.OutputFormat, report); outputErr != nil {
		return outputErr
	}
	return err
}

func buildRunReportWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) (RunReport, error) {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return RunReport{}, err
	}
	return buildRunReportWithSetup(ctx, setup, options, provider, nil)
}

type localSourceRoot struct {
	pipelineID string
	root       string
}

func buildRunReport(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider, sourceRoot *localSourceRoot) (RunReport, error) {
	setup, err := runtimeSetupFromConfig("", cfg)
	if err != nil {
		return RunReport{}, err
	}
	return buildRunReportWithSetup(ctx, setup, options, provider, sourceRoot)
}

func buildRunReportWithSetup(ctx context.Context, setup runtimeSetup, options RunOptions, provider backendFactoryProvider, sourceRoot *localSourceRoot) (RunReport, error) {
	notifier := options.Notifier
	if notifier == nil {
		notifier = notify.Noop{}
	}
	summary := runSummary{dryRun: options.DryRun}
	report := RunReport{
		DryRun:    options.DryRun,
		Pipelines: []RunPipelineSummary{},
		Actions:   []RunActionRecord{},
	}
	var failures runFailures
	recorder := runReportRecorder{
		report:   &report,
		summary:  &summary,
		failures: &failures,
	}
	report.PreambleWarnings = append(report.PreambleWarnings, setup.Warnings...)
	report.addWarnings(setup.Warnings)
	backends := provider(setup.Environment)
	backends.readOnlyKnownHosts = options.DryRun
	transforms := newTransformRegistry()
	for _, pipeline := range setup.Config.Pipelines {
		pipelineWarnings := sshWarnings(pipeline)
		report.addWarnings(pipelineWarnings)
		sourceBackend, bundles, sourceBackendName, err := openPipelineSource(ctx, backends, pipeline, sourceRoot)
		if err != nil {
			return report, err
		}
		report.Pipelines = append(report.Pipelines, RunPipelineSummary{
			ID:            pipeline.ID,
			SourceBackend: sourceBackendName,
			BundleCount:   len(bundles),
			Destinations:  destinationIDs(pipeline.Destinations),
			Warnings:      pipelineWarnings,
		})
		pipelineIndex := len(report.Pipelines) - 1
		for _, destination := range pipeline.Destinations {
			processDestination(ctx, runDestinationRequest{
				options:       options,
				notifier:      notifier,
				backends:      backends,
				transforms:    transforms,
				pipeline:      pipeline,
				pipelineIndex: pipelineIndex,
				sourceBackend: sourceBackend,
				bundles:       bundles,
				destination:   destination,
				recorder:      &recorder,
			})
		}
		closeBackend(sourceBackend)
	}
	report.Summary = summary.Result()
	report.OutputErrors = failures.outputErrors()
	if len(failures.items) > 0 {
		return report, failures
	}
	return report, nil
}

type runReportRecorder struct {
	report   *RunReport
	summary  *runSummary
	failures *runFailures
}

func openPipelineSource(ctx context.Context, backends *backendFactory, pipeline config.Pipeline, sourceRoot *localSourceRoot) (storage.Backend, []bundle.Bundle, string, error) {
	if sourceRoot != nil && sourceRoot.pipelineID == pipeline.ID {
		sourceBackend, err := backends.openLocalPath(ctx, sourceRoot.root)
		if err != nil {
			return nil, nil, config.BackendLocal, fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, config.BackendLocal, err)
		}
		sourceBundle, err := bundle.Validate(ctx, sourceBackend, "")
		if err != nil {
			closeBackend(sourceBackend)
			return nil, nil, config.BackendLocal, fmt.Errorf("pipeline %s source backend %s validate source bundle: %w", pipeline.ID, config.BackendLocal, err)
		}
		return sourceBackend, []bundle.Bundle{sourceBundle}, config.BackendLocal, nil
	}

	sourceBackend, err := backends.openSource(ctx, pipeline.Source)
	if err != nil {
		return nil, nil, pipeline.Source.Backend, 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 nil, nil, pipeline.Source.Backend, fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
	}
	return sourceBackend, bundles, pipeline.Source.Backend, nil
}

type closeableBackend interface {
	Close() error
}

func closeBackend(backend storage.Backend) {
	closeable, ok := backend.(closeableBackend)
	if !ok {
		return
	}
	_ = closeable.Close()
}