package app import ( "context" "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 runDestinationRequest struct { options RunOptions notifier notify.Notifier backends *backendFactory transforms publish.TransformerResolver pipeline config.Pipeline pipelineIndex int sourceBackend storage.Backend bundles []bundle.Bundle destination config.Destination recorder *runReportRecorder } func processDestination(ctx context.Context, request runDestinationRequest) { selections := selectDestinationBundles(request.destination, request.bundles) if isFixedPathDestination(request.destination) { request.recorder.summary.recordFixedPath() if request.options.DryRun { warning := fixedPathSelectionWarning(request.pipeline.ID, request.destination.ID, selections, len(request.bundles)) request.recorder.addPipelineWarning(request.pipelineIndex, warning) } } if len(selections) == 0 { return } destinationBackend, err := request.backends.openDestination(ctx, request.destination) if err != nil { for _, selection := range selections { sourceBundle := selection.SourceBundle request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{ pipelineID: request.pipeline.ID, destinationID: request.destination.ID, backend: request.destination.Backend, bundlePath: sourceBundle.RootRelativePath, err: err, }, errorAction(request.pipeline.ID, request.destination.ID, request.destination.Backend, sourceBundle.RootRelativePath, err), true) } return } defer closeBackend(destinationBackend) for _, selection := range selections { processDestinationSelection(ctx, request, destinationBackend, selection) } } func processDestinationSelection(ctx context.Context, request runDestinationRequest, destinationBackend storage.Backend, selection destinationBundleSelection) { sourceBundle := selection.SourceBundle publishRequest := publish.Request{ PipelineID: request.pipeline.ID, DestinationID: request.destination.ID, SourceBundle: sourceBundle, SourceBackend: request.sourceBackend, DestinationBackend: destinationBackend, DestinationBundlePath: selection.DestinationBundlePath, PathMapping: request.destination.PathMap.Mode, Publish: *request.destination.Publish, Transform: request.destination.Transform, Links: request.destination.Links, Transformers: request.transforms, Transfer: request.destination.Transfer, DistributorVersion: Version, Force: request.options.Force, } plan, err := publish.Build(ctx, publishRequest) if err != nil { plan = completePlanIdentity(plan, request.pipeline, request.destination, selection) } if isFixedPathDestination(request.destination) { plan.PathMapping = config.PathMappingFixed if request.options.DryRun && isDestructiveFixedPathAction(plan.Action) { warning := fixedPathReplacementWarning(plan) request.recorder.addPipelineWarning(request.pipelineIndex, warning) } } action := runActionFromPlan(request.destination.Backend, plan, err) if err != nil { request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{ pipelineID: request.pipeline.ID, destinationID: request.destination.ID, backend: request.destination.Backend, bundlePath: sourceBundle.RootRelativePath, err: err, }, action, true) return } request.recorder.addPipelineAction(request.pipelineIndex, action) request.recorder.summary.recordPlan(plan.Action) if request.options.DryRun { return } if err := publish.Execute(ctx, publishRequest, plan); err != nil { request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{ pipelineID: request.pipeline.ID, destinationID: request.destination.ID, backend: request.destination.Backend, bundlePath: sourceBundle.RootRelativePath, err: err, }, RunActionRecord{}, false) return } if shouldNotify(plan.Action) { if err := request.notifier.Notify(ctx, notifyEvent(plan)); err != nil { request.recorder.recordDestinationFailure(request.pipelineIndex, runFailure{ pipelineID: request.pipeline.ID, destinationID: request.destination.ID, backend: request.destination.Backend, bundlePath: sourceBundle.RootRelativePath, err: err, }, RunActionRecord{}, false) return } } } func (recorder *runReportRecorder) addPipelineWarning(pipelineIndex int, warning OutputWarning) { recorder.report.addWarning(warning) recorder.report.Pipelines[pipelineIndex].events = append(recorder.report.Pipelines[pipelineIndex].events, warningEvent(warning)) } func (recorder *runReportRecorder) addPipelineAction(pipelineIndex int, action RunActionRecord) { recorder.report.Actions = append(recorder.report.Actions, action) recorder.report.Pipelines[pipelineIndex].events = append(recorder.report.Pipelines[pipelineIndex].events, actionEvent(len(recorder.report.Actions)-1)) } func (recorder *runReportRecorder) recordDestinationFailure(pipelineIndex int, failure runFailure, action RunActionRecord, includeAction bool) { recorder.failures.add(failure.pipelineID, failure.destinationID, failure.backend, storage.DisplayPath(failure.bundlePath), failure.err) recorder.summary.recordFailure() if includeAction { recorder.addPipelineAction(pipelineIndex, action) } } func completePlanIdentity(plan publish.Plan, pipeline config.Pipeline, destination config.Destination, selection destinationBundleSelection) publish.Plan { if plan.PipelineID == "" { plan.PipelineID = pipeline.ID } if plan.DestinationID == "" { plan.DestinationID = destination.ID } if plan.BundleID == "" { plan.BundleID = selection.SourceBundle.Manifest.ID } if plan.BundlePath == "" { plan.BundlePath = selection.SourceBundle.RootRelativePath } if plan.DestinationBundlePath == "" { plan.DestinationBundlePath = selection.DestinationBundlePath } return plan }