Files
distributor/internal/app/run.go

120 lines
3.8 KiB
Go

package app
import (
"context"
"fmt"
"io"
"strings"
"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/publish"
)
type RunOptions struct {
ConfigPath string
DryRun bool
Stdout io.Writer
}
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 {
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 pipeline.Source.Backend != config.BackendLocal {
return fmt.Errorf("pipeline %s source backend %s is not implemented for execution", pipeline.ID, pipeline.Source.Backend)
}
sourceBackend, err := local.New(pipeline.Source.Path)
if err != nil {
return 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, "- %s: source=local bundles=%d destinations=%d\n", pipeline.ID, len(bundles), len(pipeline.Destinations)); err != nil {
return err
}
}
for _, sourceBundle := range bundles {
for _, destination := range pipeline.Destinations {
if destination.Backend != config.BackendLocal {
return fmt.Errorf("pipeline %s destination %s backend %s is not implemented for execution", pipeline.ID, destination.ID, destination.Backend)
}
destinationBackend, err := local.New(destination.Path)
if err != nil {
return err
}
req := publish.Request{
PipelineID: pipeline.ID,
DestinationID: destination.ID,
SourceBundle: sourceBundle,
SourceBackend: sourceBackend,
DestinationBackend: destinationBackend,
DestinationBundlePath: sourceBundle.RootRelativePath,
Publish: *destination.Publish,
Transform: destination.Transform,
Transfer: destination.Transfer,
DistributorVersion: Version,
}
plan, err := publish.Build(ctx, req)
if options.Stdout != nil {
writePlanLine(options.Stdout, plan, err)
}
if err != nil {
return fmt.Errorf("pipeline %s destination %s bundle %s: %w", pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
}
if !options.DryRun {
if err := publish.Execute(ctx, req, plan); err != nil {
return fmt.Errorf("pipeline %s destination %s bundle %s: %w", pipeline.ID, destination.ID, displayBundlePath(sourceBundle.RootRelativePath), err)
}
}
}
}
}
return nil
}
func writePlanLine(w io.Writer, plan publish.Plan, planErr error) {
if w == nil {
return
}
if planErr != nil {
fmt.Fprintf(w, " - bundle=%s destination=%s action=error reason=%q\n", displayBundlePath(plan.BundlePath), plan.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 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, ",")
}