test

catalog/consumer.go

@@ -12,9 +12,13 @@ package catalog
 // docs/plan-projectionstore-generalization.md for why the split exists.
 
 import (
+	"encoding/json"
+	"fmt"
 	"time"
 
+	"git.k6n.net/mats/platform/event"
 	"git.k6n.net/mats/platform/projection"
+	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 // Generic machinery, re-exported under their historical catalog names. These
@@ -97,3 +101,28 @@ func decodeProjectionUpdates(payload []byte) ([]projection.Update[Projection], e
 	}
 	return out, nil
 }
+
+// MakeAmqpHandler returns a RabbitMQ delivery handler that wraps a ProjectionStore[T].
+// It decodes the event.Event envelope, recovers from panics, and applies the frame to the store.
+func MakeAmqpHandler[T any](store *ProjectionStore[T]) func(amqp.Delivery) error {
+	return func(d amqp.Delivery) (err error) {
+		defer func() {
+			if r := recover(); r != nil {
+				err = fmt.Errorf("projection handler panic recovered: %v", r)
+			}
+		}()
+		var ev event.Event
+		if err := json.Unmarshal(d.Body, &ev); err != nil {
+			return fmt.Errorf("decode event envelope: %w", err)
+		}
+		if err := store.HandleFrame(ev.Meta, ev.Payload); err != nil {
+			return fmt.Errorf("apply projection frame: %w", err)
+		}
+		return nil
+	}
+}
+
+// HandleEvent feeds an already-decoded event.Event directly into the projection store.
+func HandleEvent[T any](store *ProjectionStore[T], ev event.Event) error {
+	return store.HandleFrame(ev.Meta, ev.Payload)
+}

catalog/consumer_test.go

@@ -1,10 +1,13 @@
 package catalog
 
 import (
+	"encoding/json"
 	"strconv"
 	"testing"
 
+	"git.k6n.net/mats/platform/event"
 	"git.k6n.net/mats/platform/money"
+	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 func snapMeta(phase string, epoch int64, seq, count int) map[string]string {
@@ -223,3 +226,53 @@ func TestProjectionStore_BadPayloadErrors(t *testing.T) {
 		t.Fatal("delta with bad payload should error")
 	}
 }
+
+func TestMakeAmqpHandler(t *testing.T) {
+	s := newStore(t, nil)
+	handler := MakeAmqpHandler(s)
+
+	// Build a mock AMQP delivery containing an event envelope.
+	payload, err := EncodeUpdates([]ProjectionUpdate{{Projection: proj("A", 123_45)}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ev := event.Event{
+		Type:    event.CatalogProjectionPublished,
+		Payload: payload,
+		Meta:    snapMeta(SnapshotBegin, 1, 0, 1),
+	}
+	envelope, err := json.Marshal(ev)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	d := amqp.Delivery{
+		Body: envelope,
+	}
+
+	// 1. Process Begin frame
+	if err := handler(d); err != nil {
+		t.Fatalf("handler failed on Begin: %v", err)
+	}
+
+	// 2. Process Chunk frame
+	ev.Meta = snapMeta(SnapshotChunk, 1, 0, 0)
+	envelope, _ = json.Marshal(ev)
+	if err := handler(amqp.Delivery{Body: envelope}); err != nil {
+		t.Fatalf("handler failed on Chunk: %v", err)
+	}
+
+	// 3. Process End frame
+	ev.Meta = snapMeta(SnapshotEnd, 1, 1, 1)
+	ev.Payload = nil
+	envelope, _ = json.Marshal(ev)
+	if err := handler(amqp.Delivery{Body: envelope}); err != nil {
+		t.Fatalf("handler failed on End: %v", err)
+	}
+
+	// Validate results in store
+	p, ok := s.Get("A")
+	if !ok || p.PriceIncVat != 123_45 {
+		t.Fatalf("A = %+v ok=%v, want price 12345", p, ok)
+	}
+}