package publish import ( "context" "encoding/json" "fmt" "sort" "gitea.maximumdirect.net/eric/distributor/internal/state" "gitea.maximumdirect.net/eric/distributor/internal/storage" ) func Execute(ctx context.Context, req Request, plan Plan) error { switch plan.Action { case ActionSkipSame: return nil case ActionPublishNew, ActionUpsertAdditive, ActionReplaceCatalog, ActionForceReplace: return executeCatalog(ctx, req, plan) default: return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason) } } func executeCatalog(ctx context.Context, req Request, plan Plan) error { if plan.Action == ActionForceReplace { if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { return err } if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil { return err } } else if plan.Action == ActionReplaceCatalog { if plan.ClearDestinationRoot { if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { return err } if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil { return err } } else if len(plan.CatalogOutputsToDelete) > 0 { if err := req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, catalogOutputPaths(plan.CatalogOutputsToDelete), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil { return err } } } writtenOutputs := make([]Output, 0, len(plan.Outputs)) newOutputs := make([]Output, 0, len(plan.Outputs)) cleanup := func() { outputs := writtenOutputs if plan.Action == ActionUpsertAdditive { outputs = newOutputs } _ = req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, ManagedOutputPaths(outputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}) } for _, output := range plan.Outputs { destinationPath, err := storage.Join(req.DestinationBundlePath, output.DestinationPath) if err != nil { cleanup() return err } created, err := catalogWriteCreatesOutput(ctx, req.DestinationBackend, destinationPath) if err != nil { cleanup() return err } data := output.Data if output.Kind == state.OutputKindSource { sourcePath, err := storage.Join(req.SourceBundle.RootRelativePath, output.SourcePath) if err != nil { cleanup() return err } data, err = req.SourceBackend.ReadFile(ctx, sourcePath) if err != nil { cleanup() return err } } if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: catalogOutputOverwriteAllowed(plan, output), PreferAtomic: true}); err != nil { cleanup() return err } writtenOutputs = append(writtenOutputs, output) if created { newOutputs = append(newOutputs, output) } } catalogState := catalogStateForPlan(req, plan) if err := state.ValidateCatalog(catalogState); err != nil { cleanup() return err } data, err := json.MarshalIndent(catalogState, "", " ") if err != nil { cleanup() return err } data = append(data, '\n') statePath, err := storage.StatePath(req.DestinationBundlePath) if err != nil { cleanup() return err } if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: catalogStateWriteOverwrites(plan), PreferAtomic: true}); err != nil { cleanup() return err } return nil } func catalogWriteCreatesOutput(ctx context.Context, backend storage.Backend, destinationPath string) (bool, error) { if _, err := backend.Stat(ctx, destinationPath); err == nil { return false, nil } else if storage.IsNotFound(err) { return true, nil } else { return false, err } } func catalogOutputOverwriteAllowed(plan Plan, output Output) bool { if plan.ClearDestinationRoot { return false } if plan.SupersededLegacy != nil { return true } if plan.ExistingCatalog == nil { return false } _, ok := state.FindCatalogOutputByPath(plan.ExistingCatalog.Outputs, output.DestinationPath) return ok } func catalogStateForPlan(req Request, plan Plan) state.CatalogState { now := requestTime(req) createdAt := now if plan.ExistingCatalog != nil { createdAt = plan.ExistingCatalog.CreatedAt } outputs := make([]state.CatalogOutputFile, 0, len(plan.CatalogOutputsToRetain)+len(plan.CatalogOutputsToWrite)) outputs = append(outputs, plan.CatalogOutputsToRetain...) outputs = append(outputs, plan.CatalogOutputsToWrite...) sort.SliceStable(outputs, func(i, j int) bool { if outputs[i].Path != outputs[j].Path { return outputs[i].Path < outputs[j].Path } if outputs[i].PipelineID != outputs[j].PipelineID { return outputs[i].PipelineID < outputs[j].PipelineID } return outputs[i].DestinationID < outputs[j].DestinationID }) return state.CatalogState{ SchemaVersion: state.CatalogSchemaVersion, DistributorVersion: req.DistributorVersion, CreatedAt: createdAt, UpdatedAt: now, State: state.StatePolicy{Mode: state.StateModeCatalog}, Outputs: outputs, } } func catalogStateWriteOverwrites(plan Plan) bool { return plan.ExistingCatalog != nil || plan.SupersededLegacy != nil } func catalogOutputPaths(outputs []state.CatalogOutputFile) []string { paths := make([]string, 0, len(outputs)) for _, output := range outputs { paths = append(paths, output.Path) } return paths }