package pipeline import ( "fmt" "reflect" "strings" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) // PreparedPipeline owns the constructed, run-local implementation set for one // resolved pipeline. Its implementation values are private so execution cannot // replace or reconfigure them after preparation. type PreparedPipeline struct { Input ModuleBinding Chunk ModuleBinding ArtifactLanes []PreparedArtifactLane Output ModuleBinding resolved ResolvedPipeline dependencies ModuleDependencies input contracts.InputAdapter chunker contracts.Chunker chunkValidators preparedValidatorChain lanes []preparedLaneExecutor output contracts.OutputEncoder } type PreparedArtifactLane struct { Resolved ResolvedArtifactLane } type preparedLaneExecutor struct { resolved ResolvedArtifactLane typed *preparedTypedLane extractValidators preparedValidatorChain mergeValidators preparedValidatorChain normalizeValidators preparedValidatorChain } type preparedTypedLane struct { extractor any merger any normalizer any extract typedExtractOperation merge typedMergeOperation normalize typedNormalizeOperation codec artifactCodecEntry } type preparedValidatorChain struct { resolved ResolvedValidatorChain validators []preparedValidator } type preparedValidator struct { resolved ResolvedValidator typed any typedValidate typedValidateOperation chunk contracts.ChunkValidator serialized contracts.SerializedValidator } // Prepare validates all configured options and constructs every selected // module and validator before any operation method can run. func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDependencies) (*PreparedPipeline, error) { if err := validateResolvedPipeline(resolved); err != nil { return nil, err } if err := validateRegistrySet(resolved, registries); err != nil { return nil, err } stable := cloneResolvedPipeline(resolved) prepared := &PreparedPipeline{ Input: cloneModuleBinding(stable.Input), Chunk: cloneModuleBinding(stable.Chunk), Output: cloneModuleBinding(stable.Output), resolved: stable, dependencies: deps, } request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { return BuildRequest{Dependencies: deps, Options: cloneOptions(binding.Options), References: references} } input, err := registries.Inputs.BuildWithRequest(stable.Input.Module, request(stable.Input, contracts.ReferenceSet{})) if err != nil { return nil, constructionError(stable.ID, "", StageInput, stable.Input.Module, "", err) } prepared.input = input chunker, err := registries.Chunkers.BuildWithRequest(stable.Chunk.Module, request(stable.Chunk, stable.ChunkReferences.ReferenceSet)) if err != nil { return nil, constructionError(stable.ID, "", StageChunk, stable.Chunk.Module, "", err) } prepared.chunker = chunker prepared.chunkValidators, err = prepareValidatorChain(stable, registries, deps, StageChunk, "", stable.Chunk.Module, stable.ChunkReferences.ReferenceSet) if err != nil { return nil, err } prepared.ArtifactLanes = make([]PreparedArtifactLane, 0, len(stable.ArtifactLanes)) prepared.lanes = make([]preparedLaneExecutor, 0, len(stable.ArtifactLanes)) for _, lane := range stable.ArtifactLanes { executor, err := prepareLane(stable, lane, registries, deps) if err != nil { return nil, err } prepared.ArtifactLanes = append(prepared.ArtifactLanes, PreparedArtifactLane{Resolved: cloneResolvedArtifactLane(lane)}) prepared.lanes = append(prepared.lanes, executor) } output, err := registries.Outputs.BuildWithRequest(stable.Output.Module, request(stable.Output, contracts.ReferenceSet{})) if err != nil { return nil, constructionError(stable.ID, "", StageOutput, stable.Output.Module, "", err) } prepared.output = output return prepared, nil } func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) { executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)} request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { return BuildRequest{Dependencies: deps, Options: cloneOptions(binding.Options), References: references} } extractEntry, ok := registries.Extractors.typedEntry(lane.Extract.Module) if !ok { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", fmt.Errorf("typed construction entry is not registered")) } module, err := buildErasedModule(extractEntry.builder, request(lane.Extract, lane.ExtractReferences.ReferenceSet), lane.Extract.Module, "extractor") if err != nil { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", err) } codec, _, codecErr := registries.ArtifactCodecs.entry(lane.ArtifactKind) if codecErr != nil { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageExtract, lane.Extract.Module, "", codecErr) } executor.typed = &preparedTypedLane{extractor: module, extract: extractEntry.extract, codec: codec} executor.extractValidators, err = prepareValidatorChain(pipeline, registries, deps, StageExtract, lane.ID, lane.Extract.Module, lane.ExtractReferences.ReferenceSet) if err != nil { return preparedLaneExecutor{}, err } mergeEntry, ok := registries.Mergers.typedEntry(lane.Merge.Module, lane.ArtifactKind) if !ok { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageMerge, lane.Merge.Module, "", fmt.Errorf("typed construction entry is not registered")) } module, err = buildErasedModule(mergeEntry.builder, request(lane.Merge, lane.MergeReferences.ReferenceSet), lane.Merge.Module, "merger") if err != nil { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageMerge, lane.Merge.Module, "", err) } executor.typed.merger = module executor.typed.merge = mergeEntry.merge executor.mergeValidators, err = prepareValidatorChain(pipeline, registries, deps, StageMerge, lane.ID, lane.Merge.Module, lane.MergeReferences.ReferenceSet) if err != nil { return preparedLaneExecutor{}, err } normalizeEntry, ok := registries.Normalizers.typedEntry(lane.Normalize.Module, lane.ArtifactKind) if !ok { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageNormalize, lane.Normalize.Module, "", fmt.Errorf("typed construction entry is not registered")) } module, err = buildErasedModule(normalizeEntry.builder, request(lane.Normalize, lane.NormalizeReferences.ReferenceSet), lane.Normalize.Module, "normalizer") if err != nil { return preparedLaneExecutor{}, constructionError(pipeline.ID, lane.ID, StageNormalize, lane.Normalize.Module, "", err) } executor.typed.normalizer = module executor.typed.normalize = normalizeEntry.normalize executor.normalizeValidators, err = prepareValidatorChain(pipeline, registries, deps, StageNormalize, lane.ID, lane.Normalize.Module, lane.NormalizeReferences.ReferenceSet) if err != nil { return preparedLaneExecutor{}, err } return executor, nil } func prepareValidatorChain(pipeline ResolvedPipeline, registries Registries, deps ModuleDependencies, stage ModuleStage, laneID, moduleKey string, references contracts.ReferenceSet) (preparedValidatorChain, error) { resolved := resolvedValidatorChain(stage, laneID, moduleKey, pipeline.ValidatorChains) prepared := preparedValidatorChain{resolved: resolved} for _, validator := range resolved.Validators { request := BuildRequest{Dependencies: deps, Options: cloneOptions(validator.Binding.Options), References: references} built, err := buildPreparedValidator(registries.Validators, validator, request) if err != nil { return preparedValidatorChain{}, constructionError(pipeline.ID, laneID, stage, moduleKey, validator.Binding.Module, err) } prepared.validators = append(prepared.validators, built) } return prepared, nil } func buildPreparedValidator(registry *ValidatorRegistry, resolved ResolvedValidator, request BuildRequest) (preparedValidator, error) { prepared := preparedValidator{resolved: resolved} key := resolved.Binding.Module var implementation any var err error switch resolved.Target { case ValidatorTargetTyped: entry, ok := registry.typedEntry(key, resolved.ArtifactKind) if !ok { return preparedValidator{}, fmt.Errorf("typed construction entry is not registered") } implementation, err = entry.builder(cloneBuildRequest(request)) prepared.typed = implementation prepared.typedValidate = entry.validate case ValidatorTargetChunk: entry, ok := registry.chunkEntry(key) if !ok { return preparedValidator{}, fmt.Errorf("chunk construction entry is not registered") } prepared.chunk, err = entry.builder(cloneBuildRequest(request)) implementation = prepared.chunk case ValidatorTargetSerialized: entry, ok := registry.serializedEntry(key) if !ok { return preparedValidator{}, fmt.Errorf("serialized construction entry is not registered") } prepared.serialized, err = entry.builder(cloneBuildRequest(request)) implementation = prepared.serialized default: return preparedValidator{}, fmt.Errorf("validator construction target %q is not supported", resolved.Target) } if err != nil { return preparedValidator{}, err } if isNilImplementation(implementation) { return preparedValidator{}, fmt.Errorf("validator %q builder returned nil", key) } identity, ok := implementation.(interface { Name() string ExecutionClass() contracts.ExecutionClass }) if !ok { return preparedValidator{}, fmt.Errorf("validator %q builder returned incompatible implementation %T", key, implementation) } if identity.Name() != key { return preparedValidator{}, fmt.Errorf("validator %q returned name %q", key, identity.Name()) } if identity.ExecutionClass() != resolved.ExecutionClass { return preparedValidator{}, fmt.Errorf("validator %q returned execution class %q, want %q", key, identity.ExecutionClass(), resolved.ExecutionClass) } return prepared, nil } func buildErasedModule(builder func(BuildRequest) (any, error), request BuildRequest, key, kind string) (any, error) { implementation, err := builder(cloneBuildRequest(request)) if err != nil { return nil, err } if isNilImplementation(implementation) { return nil, fmt.Errorf("%s %q builder returned nil", kind, key) } identity, ok := implementation.(interface{ Key() string }) if !ok { return nil, fmt.Errorf("%s %q builder returned incompatible implementation %T", kind, key, implementation) } if identity.Key() != key { return nil, fmt.Errorf("%s %q returned key %q", kind, key, identity.Key()) } return implementation, nil } func isNilImplementation(value any) bool { if value == nil { return true } reflected := reflect.ValueOf(value) switch reflected.Kind() { case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: return reflected.IsNil() default: return false } } func constructionError(pipelineID, laneID string, stage ModuleStage, moduleKey, validatorKey string, cause error) error { scope := fmt.Sprintf("pipeline %q %s module %q", pipelineID, stage, moduleKey) if laneID != "" { scope = fmt.Sprintf("pipeline %q lane %q %s module %q", pipelineID, laneID, stage, moduleKey) } if strings.TrimSpace(validatorKey) != "" { scope += fmt.Sprintf(" validator %q", validatorKey) } return fmt.Errorf("prepare %s: %w", scope, cause) } func (registries Registries) catalog() ModuleCatalog { return ModuleCatalog{ Inputs: registries.Inputs, Chunkers: registries.Chunkers, ArtifactCodecs: registries.ArtifactCodecs, Extractors: registries.Extractors, Mergers: registries.Mergers, Normalizers: registries.Normalizers, Validators: registries.Validators, ValidatorChains: registries.ValidatorChains, Outputs: registries.Outputs, } } func validateRegistrySet(resolved ResolvedPipeline, registries Registries) error { if registries.Inputs == nil { return fmt.Errorf("input registry must not be nil") } if registries.Chunkers == nil { return fmt.Errorf("chunker registry must not be nil") } if registries.Extractors == nil { return fmt.Errorf("extractor registry must not be nil") } if registries.Mergers == nil { return fmt.Errorf("merger registry must not be nil") } if registries.Normalizers == nil { return fmt.Errorf("normalizer registry must not be nil") } if registries.Validators == nil { for _, chain := range resolved.ValidatorChains { if len(chain.Validators) > 0 { return fmt.Errorf("validator registry must not be nil") } } } if registries.Outputs == nil { return fmt.Errorf("output encoder registry must not be nil") } return nil } func cloneResolvedPipeline(in ResolvedPipeline) ResolvedPipeline { out := in out.Input = cloneModuleBinding(in.Input) out.Chunk = cloneModuleBinding(in.Chunk) out.Output = cloneModuleBinding(in.Output) out.ChunkReferences = CloneReferenceTarget(in.ChunkReferences) out.ValidatorChains = cloneResolvedValidatorChains(in.ValidatorChains) if len(in.ArtifactLanes) > 0 { out.ArtifactLanes = make([]ResolvedArtifactLane, len(in.ArtifactLanes)) for i, lane := range in.ArtifactLanes { out.ArtifactLanes[i] = cloneResolvedArtifactLane(lane) } } return out } func cloneResolvedArtifactLane(in ResolvedArtifactLane) ResolvedArtifactLane { out := in out.Extract = cloneModuleBinding(in.Extract) out.Merge = cloneModuleBinding(in.Merge) out.Normalize = cloneModuleBinding(in.Normalize) out.Validators = cloneModuleBindings(in.Validators) out.ExtractReferences = CloneReferenceTarget(in.ExtractReferences) out.MergeReferences = CloneReferenceTarget(in.MergeReferences) out.NormalizeReferences = CloneReferenceTarget(in.NormalizeReferences) return out }