Add production default pipeline modules

internal/framework/pipeline/default_modules_test.go

@@ -0,0 +1,124 @@
+package pipeline_test
+
+import (
+	"context"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/config"
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/chunk/generic"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop"
+	jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/output/json"
+)
+
+func TestPipelineConfigResolvesWithProductionDefaultsRegistered(t *testing.T) {
+	cfg := config.Default()
+	cfg.Pipelines = map[string]pipeline.PipelineProfile{
+		"defaults": {
+			Input: pipeline.Binding("input"),
+			Artifacts: map[string]pipeline.ArtifactLaneProfile{
+				"events": {Extract: pipeline.Binding("extract")},
+			},
+		},
+	}
+
+	resolved, err := cfg.Resolve(config.ResolveInput{
+		PipelineID: "defaults",
+		Catalog:    defaultModuleCatalog(t),
+	})
+	if err != nil {
+		t.Fatalf("Resolve() error = %v, want nil", err)
+	}
+
+	pipeline := resolved.ResolvedPipeline
+	if pipeline.Chunk.Module != generic.Key {
+		t.Fatalf("Chunk.Module = %q, want %q", pipeline.Chunk.Module, generic.Key)
+	}
+	if pipeline.Output.Module != jsonoutput.Key {
+		t.Fatalf("Output.Module = %q, want %q", pipeline.Output.Module, jsonoutput.Key)
+	}
+	lane := pipeline.ArtifactLanes[0]
+	if lane.Merge.Module != appendorder.Key {
+		t.Fatalf("Merge.Module = %q, want %q", lane.Merge.Module, appendorder.Key)
+	}
+	if lane.Normalize.Module != noop.Key {
+		t.Fatalf("Normalize.Module = %q, want %q", lane.Normalize.Module, noop.Key)
+	}
+}
+
+func defaultModuleCatalog(t *testing.T) pipeline.ModuleCatalog {
+	t.Helper()
+
+	inputs := pipeline.NewInputAdapterRegistry()
+	chunkers := pipeline.NewChunkerRegistry()
+	extractors := pipeline.NewExtractorRegistry()
+	mergers := pipeline.NewMergerRegistry()
+	normalizers := pipeline.NewNormalizerRegistry()
+	outputs := pipeline.NewOutputEncoderRegistry()
+
+	if err := inputs.RegisterWithSpec(pipeline.ModuleSpec{
+		Key:      "input",
+		Stage:    pipeline.StageInput,
+		Provides: []string{"source"},
+	}, func() (contracts.InputAdapter, error) {
+		return defaultInput{}, nil
+	}); err != nil {
+		t.Fatalf("register input: %v", err)
+	}
+	if err := generic.Register(chunkers); err != nil {
+		t.Fatalf("register generic chunker: %v", err)
+	}
+	if err := extractors.RegisterWithSpec(pipeline.ModuleSpec{
+		Key:      "extract",
+		Stage:    pipeline.StageExtract,
+		Requires: []string{"chunks"},
+		Provides: []string{"records"},
+	}, func() (contracts.Extractor, error) {
+		return defaultExtractor{}, nil
+	}); err != nil {
+		t.Fatalf("register extractor: %v", err)
+	}
+	if err := appendorder.Register(mergers); err != nil {
+		t.Fatalf("register appendorder merger: %v", err)
+	}
+	if err := noop.Register(normalizers); err != nil {
+		t.Fatalf("register noop normalizer: %v", err)
+	}
+	if err := jsonoutput.Register(outputs); err != nil {
+		t.Fatalf("register json output: %v", err)
+	}
+
+	return pipeline.ModuleCatalog{
+		Inputs:      inputs,
+		Chunkers:    chunkers,
+		Extractors:  extractors,
+		Mergers:     mergers,
+		Normalizers: normalizers,
+		Outputs:     outputs,
+	}
+}
+
+type defaultInput struct{}
+
+func (defaultInput) Key() string { return "input" }
+
+func (defaultInput) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
+	return nil, nil
+}
+
+type defaultExtractor struct{}
+
+func (defaultExtractor) Key() string { return "extract" }
+
+func (defaultExtractor) ArtifactType() string { return "record" }
+
+func (defaultExtractor) SchemaVersion() string { return "v1" }
+
+func (defaultExtractor) Validators() []contracts.Validator { return nil }
+
+func (defaultExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) {
+	return contracts.ExtractionResult{}, nil
+}

internal/framework/pipeline/generic_stages.go

@@ -1,70 +0,0 @@
-package pipeline
-
-import (
-	"context"
-	"encoding/json"
-
-	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
-	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
-	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
-)
-
-type AppendOrderMerger struct{}
-
-func (m AppendOrderMerger) Key() string {
-	return DefaultMergeModule
-}
-
-func (m AppendOrderMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
-	var candidates []artifacts.ArtifactCandidate
-	for _, chunkArtifacts := range req.ChunkArtifacts {
-		candidates = append(candidates, copyArtifactCandidates(chunkArtifacts.Candidates)...)
-	}
-	return contracts.MergeResult{Candidates: candidates}, nil
-}
-
-type NoopNormalizer struct{}
-
-func (n NoopNormalizer) Key() string {
-	return DefaultNormalizeModule
-}
-
-func (n NoopNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
-	return contracts.NormalizeResult{Candidates: copyArtifactCandidates(req.Candidates)}, nil
-}
-
-func copyArtifactCandidates(candidates []artifacts.ArtifactCandidate) []artifacts.ArtifactCandidate {
-	if len(candidates) == 0 {
-		return nil
-	}
-
-	copied := make([]artifacts.ArtifactCandidate, 0, len(candidates))
-	for _, candidate := range candidates {
-		copied = append(copied, copyArtifactCandidate(candidate))
-	}
-	return copied
-}
-
-func copyArtifactCandidate(candidate artifacts.ArtifactCandidate) artifacts.ArtifactCandidate {
-	return artifacts.ArtifactCandidate{
-		Index:         candidate.Index,
-		ExtractorKey:  candidate.ExtractorKey,
-		ArtifactType:  candidate.ArtifactType,
-		SchemaVersion: candidate.SchemaVersion,
-		Payload:       append(json.RawMessage(nil), candidate.Payload...),
-		SourceRefs:    append([]source.SourceRef(nil), candidate.SourceRefs...),
-		Metadata:      copyArtifactMetadata(candidate.Metadata),
-	}
-}
-
-func copyArtifactMetadata(metadata map[string]any) map[string]any {
-	if len(metadata) == 0 {
-		return nil
-	}
-
-	copied := make(map[string]any, len(metadata))
-	for key, value := range metadata {
-		copied[key] = value
-	}
-	return copied
-}

internal/framework/pipeline/generic_stages_test.go

@@ -1,221 +0,0 @@
-package pipeline
-
-import (
-	"context"
-	"encoding/json"
-	"reflect"
-	"testing"
-
-	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
-	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
-	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
-)
-
-func TestGenericMergeAndNormalizeKeys(t *testing.T) {
-	merger := AppendOrderMerger{}
-	normalizer := NoopNormalizer{}
-
-	if merger.Key() != DefaultMergeModule {
-		t.Fatalf("AppendOrderMerger.Key() = %q, want %q", merger.Key(), DefaultMergeModule)
-	}
-	if normalizer.Key() != DefaultNormalizeModule {
-		t.Fatalf("NoopNormalizer.Key() = %q, want %q", normalizer.Key(), DefaultNormalizeModule)
-	}
-}
-
-func TestAppendOrderMergerConcatenatesByChunkAndCandidateOrder(t *testing.T) {
-	merger := AppendOrderMerger{}
-	chunks := []contracts.ChunkArtifacts{
-		{
-			Chunk: sourceChunk(0),
-			Candidates: []artifacts.ArtifactCandidate{
-				candidate(2, "first-b"),
-				candidate(1, "first-a"),
-			},
-		},
-		{
-			Chunk: sourceChunk(1),
-			Candidates: []artifacts.ArtifactCandidate{
-				candidate(4, "second-b"),
-				candidate(3, "second-a"),
-			},
-		},
-	}
-
-	result, err := merger.Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: chunks})
-	if err != nil {
-		t.Fatalf("Merge() error = %v, want nil", err)
-	}
-	if len(result.Warnings) != 0 {
-		t.Fatalf("Warnings = %#v, want none", result.Warnings)
-	}
-
-	got := candidateNames(result.Candidates)
-	want := []string{"first-b", "first-a", "second-b", "second-a"}
-	if !reflect.DeepEqual(got, want) {
-		t.Fatalf("candidate order = %#v, want %#v", got, want)
-	}
-}
-
-func TestAppendOrderMergerReturnsMutationSafeCandidates(t *testing.T) {
-	merger := AppendOrderMerger{}
-	input := []contracts.ChunkArtifacts{
-		{
-			Chunk: sourceChunk(0),
-			Candidates: []artifacts.ArtifactCandidate{
-				candidate(1, "original"),
-			},
-		},
-	}
-
-	result, err := merger.Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: input})
-	if err != nil {
-		t.Fatalf("Merge() error = %v, want nil", err)
-	}
-	if len(result.Candidates) != 1 {
-		t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates))
-	}
-
-	input[0].Candidates[0].Index = 99
-	input[0].Candidates[0].Payload[0] = '['
-	input[0].Candidates[0].SourceRefs[0].StartUnitID = "changed"
-	input[0].Candidates[0].Metadata["name"] = "changed"
-
-	got := result.Candidates[0]
-	if got.Index != 1 {
-		t.Fatalf("Index = %d, want 1", got.Index)
-	}
-	if string(got.Payload) != `{"name":"original"}` {
-		t.Fatalf("Payload = %s, want original payload", got.Payload)
-	}
-	if got.SourceRefs[0].StartUnitID != "u1" {
-		t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs)
-	}
-	if got.Metadata["name"] != "original" {
-		t.Fatalf("Metadata = %#v, want original metadata", got.Metadata)
-	}
-}
-
-func TestNoopNormalizerPreservesOrderAndValues(t *testing.T) {
-	normalizer := NoopNormalizer{}
-	input := []artifacts.ArtifactCandidate{
-		candidate(3, "third"),
-		candidate(1, "first"),
-		candidate(2, "second"),
-	}
-
-	result, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input})
-	if err != nil {
-		t.Fatalf("Normalize() error = %v, want nil", err)
-	}
-	if len(result.Warnings) != 0 {
-		t.Fatalf("Warnings = %#v, want none", result.Warnings)
-	}
-
-	got := candidateNames(result.Candidates)
-	want := []string{"third", "first", "second"}
-	if !reflect.DeepEqual(got, want) {
-		t.Fatalf("candidate order = %#v, want %#v", got, want)
-	}
-
-	if !reflect.DeepEqual(result.Candidates[0].SourceRefs, input[0].SourceRefs) {
-		t.Fatalf("SourceRefs = %#v, want %#v", result.Candidates[0].SourceRefs, input[0].SourceRefs)
-	}
-	if !reflect.DeepEqual(result.Candidates[0].Metadata, input[0].Metadata) {
-		t.Fatalf("Metadata = %#v, want %#v", result.Candidates[0].Metadata, input[0].Metadata)
-	}
-}
-
-func TestNoopNormalizerReturnsMutationSafeCandidates(t *testing.T) {
-	normalizer := NoopNormalizer{}
-	input := []artifacts.ArtifactCandidate{candidate(1, "original")}
-
-	result, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input})
-	if err != nil {
-		t.Fatalf("Normalize() error = %v, want nil", err)
-	}
-	if len(result.Candidates) != 1 {
-		t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates))
-	}
-
-	input[0].Index = 99
-	input[0].Payload[0] = '['
-	input[0].SourceRefs[0].EndUnitID = "changed"
-	input[0].Metadata["name"] = "changed"
-
-	got := result.Candidates[0]
-	if got.Index != 1 {
-		t.Fatalf("Index = %d, want 1", got.Index)
-	}
-	if string(got.Payload) != `{"name":"original"}` {
-		t.Fatalf("Payload = %s, want original payload", got.Payload)
-	}
-	if got.SourceRefs[0].EndUnitID != "u1" {
-		t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs)
-	}
-	if got.Metadata["name"] != "original" {
-		t.Fatalf("Metadata = %#v, want original metadata", got.Metadata)
-	}
-}
-
-func TestGenericMergeAndNormalizeHandleEmptyInput(t *testing.T) {
-	merger := AppendOrderMerger{}
-	normalizer := NoopNormalizer{}
-
-	mergeResult, err := merger.Merge(context.Background(), contracts.MergeRequest{})
-	if err != nil {
-		t.Fatalf("Merge() error = %v, want nil", err)
-	}
-	if len(mergeResult.Candidates) != 0 {
-		t.Fatalf("len(mergeResult.Candidates) = %d, want 0", len(mergeResult.Candidates))
-	}
-	if len(mergeResult.Warnings) != 0 {
-		t.Fatalf("merge warnings = %#v, want none", mergeResult.Warnings)
-	}
-
-	normalizeResult, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{})
-	if err != nil {
-		t.Fatalf("Normalize() error = %v, want nil", err)
-	}
-	if len(normalizeResult.Candidates) != 0 {
-		t.Fatalf("len(normalizeResult.Candidates) = %d, want 0", len(normalizeResult.Candidates))
-	}
-	if len(normalizeResult.Warnings) != 0 {
-		t.Fatalf("normalize warnings = %#v, want none", normalizeResult.Warnings)
-	}
-}
-
-func candidate(index int, name string) artifacts.ArtifactCandidate {
-	return artifacts.ArtifactCandidate{
-		Index:         index,
-		ExtractorKey:  "generic-extractor",
-		ArtifactType:  "generic-artifact",
-		SchemaVersion: "v1",
-		Payload:       json.RawMessage(`{"name":"` + name + `"}`),
-		SourceRefs: []source.SourceRef{
-			{SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"},
-		},
-		Metadata: map[string]any{
-			"name": name,
-		},
-	}
-}
-
-func candidateNames(candidates []artifacts.ArtifactCandidate) []string {
-	names := make([]string, 0, len(candidates))
-	for _, candidate := range candidates {
-		names = append(names, candidate.Metadata["name"].(string))
-	}
-	return names
-}
-
-func sourceChunk(index int) contracts.SourceChunk {
-	return contracts.SourceChunk{
-		ID:       "chunk",
-		SourceID: "source-1",
-		Index:    index,
-		Units: []source.SourceUnit{
-			{ID: "u1", Kind: "unit", Text: "Source unit."},
-		},
-	}
-}

internal/framework/pipeline/walking_skeleton_test.go

@@ -135,7 +135,7 @@ func walkingSkeletonCatalog(t *testing.T) ModuleCatalog {
 		Stage:    StageMerge,
 		Requires: []string{"fake_artifacts"},
 	}, func() (contracts.Merger, error) {
-		return AppendOrderMerger{}, nil
+		return walkingSkeletonMerger{}, nil
 	}); err != nil {
 		t.Fatalf("register append-order merger: %v", err)
 	}
@@ -143,7 +143,7 @@ func walkingSkeletonCatalog(t *testing.T) ModuleCatalog {
 		Key:   DefaultNormalizeModule,
 		Stage: StageNormalize,
 	}, func() (contracts.Normalizer, error) {
-		return NoopNormalizer{}, nil
+		return walkingSkeletonNormalizer{}, nil
 	}); err != nil {
 		t.Fatalf("register no-op normalizer: %v", err)
 	}
@@ -309,6 +309,30 @@ func (client *walkingSkeletonLLMClient) CompleteStructured(ctx context.Context,
 	}, nil
 }
 
+type walkingSkeletonMerger struct{}
+
+func (merger walkingSkeletonMerger) Key() string {
+	return DefaultMergeModule
+}
+
+func (merger walkingSkeletonMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
+	var candidates []artifacts.ArtifactCandidate
+	for _, chunkArtifacts := range req.ChunkArtifacts {
+		candidates = append(candidates, chunkArtifacts.Candidates...)
+	}
+	return contracts.MergeResult{Candidates: candidates}, nil
+}
+
+type walkingSkeletonNormalizer struct{}
+
+func (normalizer walkingSkeletonNormalizer) Key() string {
+	return DefaultNormalizeModule
+}
+
+func (normalizer walkingSkeletonNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
+	return contracts.NormalizeResult{Candidates: req.Candidates}, nil
+}
+
 type walkingSkeletonOutput struct{}
 
 func (output walkingSkeletonOutput) Key() string {