Updated processor/normalizer wiring to track Feedkit v0.7.0

cmd/weatherfeeder/main.go

@@ -17,6 +17,7 @@ import (
 	fkevent "gitea.maximumdirect.net/ejr/feedkit/event"
 	fknormalize "gitea.maximumdirect.net/ejr/feedkit/normalize"
 	fkpipeline "gitea.maximumdirect.net/ejr/feedkit/pipeline"
+	fkprocessors "gitea.maximumdirect.net/ejr/feedkit/processors"
 	fkscheduler "gitea.maximumdirect.net/ejr/feedkit/scheduler"
 	fksinks "gitea.maximumdirect.net/ejr/feedkit/sinks"
 	fksources "gitea.maximumdirect.net/ejr/feedkit/sources"
@@ -101,17 +102,24 @@ func main() {
 	// --- Normalization (optional) ---
 	//
 	// We install feedkit's normalize.Processor even before any normalizers exist.
-	// With an empty registry and RequireMatch=false, this is a no-op passthrough.
+	// With an empty normalizer list and RequireMatch=false, this is a no-op passthrough.
 	// It will begin transforming events as soon as:
 	//   1) sources emit raw schemas (raw.*), and
 	//   2) matching normalizers are registered.
-	normReg := &fknormalize.Registry{}
-	wfnormalizers.RegisterBuiltins(normReg)
+	normalizers := wfnormalizers.RegisterBuiltins(nil)
+
+	procReg := fkprocessors.NewRegistry()
+	procReg.Register("normalize", func() (fkprocessors.Processor, error) {
+		return fknormalize.NewProcessor(normalizers, false), nil
+	})
+
+	chain, err := procReg.BuildChain([]string{"normalize"})
+	if err != nil {
+		log.Fatalf("build processor chain failed: %v", err)
+	}
 
 	pl := &fkpipeline.Pipeline{
-		Processors: []fkpipeline.Processor{
-			fknormalize.Processor{Registry: normReg},
-		},
+		Processors: chain,
 	}
 
 	s := &fkscheduler.Scheduler{

cmd/weatherfeeder/main_test.go

@@ -1,11 +1,19 @@
 package main
 
 import (
+	"context"
+	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"gitea.maximumdirect.net/ejr/feedkit/config"
 	fkevent "gitea.maximumdirect.net/ejr/feedkit/event"
+	fknormalize "gitea.maximumdirect.net/ejr/feedkit/normalize"
+	fkpipeline "gitea.maximumdirect.net/ejr/feedkit/pipeline"
+	fkprocessors "gitea.maximumdirect.net/ejr/feedkit/processors"
+
+	wfnormalizers "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers"
 )
 
 type testInput struct {
@@ -88,3 +96,64 @@ func TestExampleConfigLoads(t *testing.T) {
 		t.Fatalf("config.Load(config.yml) unexpected error: %v", err)
 	}
 }
+
+func TestProcessorRegistryBuildsNormalizeChain(t *testing.T) {
+	normalizers := wfnormalizers.RegisterBuiltins(nil)
+	if len(normalizers) == 0 {
+		t.Fatalf("RegisterBuiltins() returned no normalizers")
+	}
+
+	procReg := fkprocessors.NewRegistry()
+	procReg.Register("normalize", func() (fkprocessors.Processor, error) {
+		return fknormalize.NewProcessor(normalizers, false), nil
+	})
+
+	chain, err := procReg.BuildChain([]string{"normalize"})
+	if err != nil {
+		t.Fatalf("BuildChain() unexpected error: %v", err)
+	}
+	if len(chain) != 1 {
+		t.Fatalf("BuildChain() expected 1 processor, got %d", len(chain))
+	}
+
+	pl := &fkpipeline.Pipeline{Processors: chain}
+	if len(pl.Processors) != 1 {
+		t.Fatalf("pipeline expected 1 processor, got %d", len(pl.Processors))
+	}
+}
+
+func TestNormalizeNoMatchPassThrough(t *testing.T) {
+	normalizers := wfnormalizers.RegisterBuiltins(nil)
+	procReg := fkprocessors.NewRegistry()
+	procReg.Register("normalize", func() (fkprocessors.Processor, error) {
+		return fknormalize.NewProcessor(normalizers, false), nil
+	})
+
+	chain, err := procReg.BuildChain([]string{"normalize"})
+	if err != nil {
+		t.Fatalf("BuildChain() unexpected error: %v", err)
+	}
+
+	pl := &fkpipeline.Pipeline{Processors: chain}
+	in := fkevent.Event{
+		ID:        "evt-no-match",
+		Kind:      fkevent.Kind("observation"),
+		Source:    "test",
+		EmittedAt: time.Now().UTC(),
+		Schema:    "raw.weatherfeeder.unknown.v1",
+		Payload: map[string]any{
+			"ok": true,
+		},
+	}
+
+	out, err := pl.Process(context.Background(), in)
+	if err != nil {
+		t.Fatalf("Pipeline.Process() unexpected error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("Pipeline.Process() returned nil output")
+	}
+	if !reflect.DeepEqual(*out, in) {
+		t.Fatalf("Pipeline.Process() expected passthrough output, got %#v", *out)
+	}
+}