diff --git a/catalog/consumer.go b/catalog/consumer.go index 044a92a..a3bd168 100644 --- a/catalog/consumer.go +++ b/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) +} diff --git a/catalog/consumer_test.go b/catalog/consumer_test.go index 0739c2e..1acc5fb 100644 --- a/catalog/consumer_test.go +++ b/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) + } +}