382 lines
12 KiB
Go
382 lines
12 KiB
Go
package publish
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/config"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/state"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/storage"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/transform"
|
|
)
|
|
|
|
type Action string
|
|
|
|
const (
|
|
ActionPublishNew Action = "publish_new"
|
|
ActionSkipSame Action = "skip_same"
|
|
ActionFailConflict Action = "fail_conflict"
|
|
ActionFailUnmanaged Action = "fail_unmanaged"
|
|
ActionForceReplace Action = "force_replace"
|
|
ActionUpsertAdditive Action = "upsert_additive"
|
|
ActionReplaceCatalog Action = "replace_catalog"
|
|
)
|
|
|
|
type Request struct {
|
|
PipelineID string
|
|
DestinationID string
|
|
SourceBundle bundle.Bundle
|
|
SourceBackend storage.Backend
|
|
DestinationBackend storage.Backend
|
|
DestinationBundlePath string
|
|
PathMapping string
|
|
Publish config.PublishPolicy
|
|
Transform config.Transform
|
|
Links *config.Links
|
|
Workflow string
|
|
Transformers TransformerResolver
|
|
DistributorVersion string
|
|
Force bool
|
|
Now time.Time
|
|
}
|
|
|
|
type TransformerResolver interface {
|
|
Get(name string) (transform.Transformer, bool)
|
|
}
|
|
|
|
type Output struct {
|
|
SourcePath string
|
|
DestinationPath string
|
|
Kind string
|
|
Transform string
|
|
URL string
|
|
Data []byte
|
|
SHA256 string
|
|
Size int64
|
|
}
|
|
|
|
type Plan struct {
|
|
PipelineID string
|
|
DestinationID string
|
|
BundleID string
|
|
BundlePath string
|
|
DestinationBundlePath string
|
|
PathMapping string
|
|
Action Action
|
|
Reason string
|
|
Force bool
|
|
PrimaryURL string
|
|
Workflow string
|
|
OwnerScope state.OwnerScope
|
|
Outputs []Output
|
|
ExistingCatalog *state.CatalogState
|
|
SupersededLegacy *state.SupersededLegacyState
|
|
CatalogOutputsToWrite []state.CatalogOutputFile
|
|
CatalogOutputsToRetain []state.CatalogOutputFile
|
|
CatalogOutputsToDelete []state.CatalogOutputFile
|
|
ClearDestinationRoot bool
|
|
}
|
|
|
|
type catalogPlanDetails struct {
|
|
Action Action
|
|
Reason string
|
|
CatalogOutputsToWrite []state.CatalogOutputFile
|
|
CatalogOutputsToRetain []state.CatalogOutputFile
|
|
CatalogOutputsToDelete []state.CatalogOutputFile
|
|
ClearDestinationRoot bool
|
|
}
|
|
|
|
func Build(ctx context.Context, req Request) (Plan, error) {
|
|
if err := validateRequest(req); err != nil {
|
|
return Plan{}, err
|
|
}
|
|
outputs, err := PlanOutputs(ctx, req)
|
|
if err != nil {
|
|
return Plan{}, err
|
|
}
|
|
outputs, primaryURL, err := PlanLinks(req, outputs)
|
|
if err != nil {
|
|
return Plan{}, err
|
|
}
|
|
status, err := inspectDestination(ctx, req.DestinationBackend, req.DestinationBundlePath)
|
|
if err != nil {
|
|
return Plan{}, err
|
|
}
|
|
workflow := normalizeWorkflow(req.Workflow)
|
|
scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
|
|
now := requestTime(req)
|
|
plan := Plan{
|
|
PipelineID: req.PipelineID,
|
|
DestinationID: req.DestinationID,
|
|
BundleID: req.SourceBundle.Manifest.ID,
|
|
BundlePath: req.SourceBundle.RootRelativePath,
|
|
DestinationBundlePath: req.DestinationBundlePath,
|
|
PathMapping: req.PathMapping,
|
|
PrimaryURL: primaryURL,
|
|
Workflow: workflow,
|
|
OwnerScope: scope,
|
|
Outputs: outputs,
|
|
ExistingCatalog: status.Catalog,
|
|
SupersededLegacy: status.SupersededLegacy,
|
|
}
|
|
if status.StateErr != nil {
|
|
plan.Reason = status.StateErr.Error()
|
|
if req.Force {
|
|
plan.Action = ActionForceReplace
|
|
plan.Force = true
|
|
plan.ClearDestinationRoot = true
|
|
plan.CatalogOutputsToWrite = catalogOutputsForPlan(req, outputs, nil, scope, now)
|
|
return plan, nil
|
|
}
|
|
plan.Action = ActionFailConflict
|
|
return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason)
|
|
}
|
|
|
|
var details catalogPlanDetails
|
|
switch {
|
|
case status.Catalog != nil:
|
|
details, err = planExistingCatalog(ctx, req, *status.Catalog, outputs, workflow, scope, now)
|
|
case status.SupersededLegacy != nil:
|
|
details = planSupersededLegacy(req, outputs, workflow, scope, now)
|
|
default:
|
|
details, err = planWithoutCatalog(ctx, req, outputs, workflow, scope, now, status.HasContents)
|
|
}
|
|
plan.Action = details.Action
|
|
plan.Reason = details.Reason
|
|
plan.CatalogOutputsToWrite = details.CatalogOutputsToWrite
|
|
plan.CatalogOutputsToRetain = details.CatalogOutputsToRetain
|
|
plan.CatalogOutputsToDelete = details.CatalogOutputsToDelete
|
|
plan.ClearDestinationRoot = details.ClearDestinationRoot
|
|
if err != nil && req.Force && forceCanReplace(details.Action) {
|
|
plan.Action = ActionForceReplace
|
|
plan.Force = true
|
|
plan.ClearDestinationRoot = true
|
|
plan.CatalogOutputsToWrite = catalogOutputsForPlan(req, outputs, nil, scope, now)
|
|
plan.CatalogOutputsToRetain = nil
|
|
plan.CatalogOutputsToDelete = nil
|
|
return plan, nil
|
|
}
|
|
if err != nil {
|
|
return plan, err
|
|
}
|
|
return plan, nil
|
|
}
|
|
|
|
func validateRequest(req Request) error {
|
|
if req.PipelineID == "" {
|
|
return fmt.Errorf("pipeline id is required")
|
|
}
|
|
if req.DestinationID == "" {
|
|
return fmt.Errorf("destination id is required")
|
|
}
|
|
if req.SourceBackend == nil {
|
|
return fmt.Errorf("source backend is required")
|
|
}
|
|
if req.DestinationBackend == nil {
|
|
return fmt.Errorf("destination backend is required")
|
|
}
|
|
if err := config.ValidatePublishTransformPolicy(req.Publish, req.Transform); err != nil {
|
|
return fmt.Errorf("publish/transform policy: %w", err)
|
|
}
|
|
switch normalizeWorkflow(req.Workflow) {
|
|
case config.WorkflowAdditive, config.WorkflowReplacement:
|
|
default:
|
|
return fmt.Errorf("destination.workflow must be %s or %s", config.WorkflowAdditive, config.WorkflowReplacement)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func normalizeWorkflow(workflow string) string {
|
|
if workflow == "" {
|
|
return config.WorkflowAdditive
|
|
}
|
|
return workflow
|
|
}
|
|
|
|
func requestTime(req Request) time.Time {
|
|
if req.Now.IsZero() {
|
|
return time.Now().UTC()
|
|
}
|
|
return req.Now.UTC()
|
|
}
|
|
|
|
func planExistingCatalog(ctx context.Context, req Request, catalog state.CatalogState, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) {
|
|
if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, catalog.Outputs, outputs); err != nil {
|
|
return catalogPlanDetails{
|
|
Action: ActionFailUnmanaged,
|
|
Reason: err.Error(),
|
|
}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
|
|
}
|
|
planned := outputPathSet(outputs)
|
|
details := catalogPlanDetails{
|
|
Action: actionForWorkflow(workflow),
|
|
CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, catalog.Outputs, scope, now),
|
|
}
|
|
allPlannedOutputsMatch := catalogContainsMatchingOutputs(req, catalog.Outputs, outputs, scope)
|
|
for _, output := range catalog.Outputs {
|
|
if _, exists := planned[output.Path]; exists {
|
|
continue
|
|
}
|
|
if workflow == config.WorkflowReplacement && output.PipelineID == scope.PipelineID && output.DestinationID == scope.DestinationID {
|
|
details.CatalogOutputsToDelete = append(details.CatalogOutputsToDelete, output)
|
|
continue
|
|
}
|
|
details.CatalogOutputsToRetain = append(details.CatalogOutputsToRetain, output)
|
|
}
|
|
if allPlannedOutputsMatch && (workflow == config.WorkflowAdditive || len(details.CatalogOutputsToDelete) == 0) {
|
|
details.Action = ActionSkipSame
|
|
details.CatalogOutputsToWrite = nil
|
|
details.CatalogOutputsToDelete = nil
|
|
}
|
|
return details, nil
|
|
}
|
|
|
|
func catalogContainsMatchingOutputs(req Request, existing []state.CatalogOutputFile, outputs []Output, scope state.OwnerScope) bool {
|
|
for _, output := range outputs {
|
|
catalogOutput, ok := state.FindCatalogOutputByPath(existing, output.DestinationPath)
|
|
if !ok || !catalogOutputMatchesPlan(req, catalogOutput, output, scope) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func catalogOutputMatchesPlan(req Request, catalogOutput state.CatalogOutputFile, output Output, scope state.OwnerScope) bool {
|
|
if catalogOutput.PipelineID != scope.PipelineID ||
|
|
catalogOutput.DestinationID != scope.DestinationID ||
|
|
catalogOutput.Source.ID != req.SourceBundle.Manifest.ID ||
|
|
catalogOutput.Source.Digest != req.SourceBundle.Manifest.Digest ||
|
|
!catalogOutput.Source.Created.Equal(req.SourceBundle.Manifest.Created) ||
|
|
catalogOutput.Path != output.DestinationPath ||
|
|
catalogOutput.Kind != output.Kind ||
|
|
catalogOutput.URL != output.URL ||
|
|
catalogOutput.SHA256 != output.SHA256 ||
|
|
catalogOutput.Size != output.Size {
|
|
return false
|
|
}
|
|
if output.Kind == state.OutputKindGenerated {
|
|
return catalogOutput.SourcePath == output.SourcePath && catalogOutput.Transform == output.Transform
|
|
}
|
|
return catalogOutput.SourcePath == "" && catalogOutput.Transform == ""
|
|
}
|
|
|
|
func planSupersededLegacy(req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) catalogPlanDetails {
|
|
details := catalogPlanDetails{
|
|
Action: actionForWorkflow(workflow),
|
|
CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now),
|
|
}
|
|
if workflow == config.WorkflowReplacement {
|
|
details.ClearDestinationRoot = true
|
|
}
|
|
return details
|
|
}
|
|
|
|
func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time, hasContents bool) (catalogPlanDetails, error) {
|
|
if hasContents {
|
|
err := fmt.Errorf("destination has content but no distributor state")
|
|
return catalogPlanDetails{
|
|
Action: ActionFailUnmanaged,
|
|
Reason: err.Error(),
|
|
}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
|
|
}
|
|
if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, nil, outputs); err != nil {
|
|
return catalogPlanDetails{
|
|
Action: ActionFailUnmanaged,
|
|
Reason: err.Error(),
|
|
}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
|
|
}
|
|
return catalogPlanDetails{
|
|
Action: ActionPublishNew,
|
|
CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now),
|
|
}, nil
|
|
}
|
|
|
|
func forceCanReplace(action Action) bool {
|
|
return action == ActionFailUnmanaged || action == ActionFailConflict
|
|
}
|
|
|
|
func actionForWorkflow(workflow string) Action {
|
|
if workflow == config.WorkflowReplacement {
|
|
return ActionReplaceCatalog
|
|
}
|
|
return ActionUpsertAdditive
|
|
}
|
|
|
|
func catalogOutputsForPlan(req Request, outputs []Output, existing []state.CatalogOutputFile, scope state.OwnerScope, now time.Time) []state.CatalogOutputFile {
|
|
files := make([]state.CatalogOutputFile, 0, len(outputs))
|
|
source := state.CatalogSourceIdentity{
|
|
ID: req.SourceBundle.Manifest.ID,
|
|
Digest: req.SourceBundle.Manifest.Digest,
|
|
Created: req.SourceBundle.Manifest.Created,
|
|
}
|
|
for _, output := range outputs {
|
|
createdAt := now
|
|
if existingOutput, ok := state.FindCatalogOutputByPath(existing, output.DestinationPath); ok {
|
|
createdAt = existingOutput.CreatedAt
|
|
}
|
|
file := state.CatalogOutputFile{
|
|
Path: output.DestinationPath,
|
|
PipelineID: scope.PipelineID,
|
|
DestinationID: scope.DestinationID,
|
|
Source: source,
|
|
Kind: output.Kind,
|
|
URL: output.URL,
|
|
SHA256: output.SHA256,
|
|
Size: output.Size,
|
|
CreatedAt: createdAt,
|
|
UpdatedAt: now,
|
|
}
|
|
if output.Kind == state.OutputKindGenerated {
|
|
file.SourcePath = output.SourcePath
|
|
file.Transform = output.Transform
|
|
}
|
|
files = append(files, file)
|
|
}
|
|
return files
|
|
}
|
|
|
|
func rejectCatalogUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, existing []state.CatalogOutputFile, outputs []Output) error {
|
|
managed := catalogOutputPathSet(existing)
|
|
for _, output := range outputs {
|
|
if _, exists := managed[output.DestinationPath]; exists {
|
|
continue
|
|
}
|
|
destinationPath, err := storage.Join(bundlePath, output.DestinationPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := backend.Stat(ctx, destinationPath); err == nil {
|
|
return fmt.Errorf("destination output path %s exists but is not managed by catalog state", storage.DisplayPath(output.DestinationPath))
|
|
} else if !storage.IsNotFound(err) {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func outputPaths(outputs []Output) []string {
|
|
paths := make([]string, 0, len(outputs))
|
|
for _, output := range outputs {
|
|
paths = append(paths, output.DestinationPath)
|
|
}
|
|
return paths
|
|
}
|
|
|
|
func outputPathSet(outputs []Output) map[string]struct{} {
|
|
paths := make(map[string]struct{}, len(outputs))
|
|
for _, output := range outputs {
|
|
paths[output.DestinationPath] = struct{}{}
|
|
}
|
|
return paths
|
|
}
|
|
|
|
func catalogOutputPathSet(outputs []state.CatalogOutputFile) map[string]struct{} {
|
|
paths := make(map[string]struct{}, len(outputs))
|
|
for _, output := range outputs {
|
|
paths[output.Path] = struct{}{}
|
|
}
|
|
return paths
|
|
}
|