279 lines
9.1 KiB
Go
279 lines
9.1 KiB
Go
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 {
|
|
m := map[string]string{MetaEpoch: strconv.FormatInt(epoch, 10), MetaTenant: "SE"}
|
|
if phase != "" {
|
|
m[MetaSnapshotPhase] = phase
|
|
m[MetaSeq] = strconv.Itoa(seq)
|
|
}
|
|
if phase == SnapshotBegin || phase == SnapshotEnd {
|
|
m[MetaCount] = strconv.Itoa(count)
|
|
}
|
|
return m
|
|
}
|
|
|
|
func mustEnc(t *testing.T, ups []ProjectionUpdate) []byte {
|
|
t.Helper()
|
|
b, err := EncodeUpdates(ups)
|
|
if err != nil {
|
|
t.Fatalf("encode: %v", err)
|
|
}
|
|
return b
|
|
}
|
|
|
|
// identity is the transform used by tests: store the full Projection (T = Projection).
|
|
func identity(p Projection) Projection { return p }
|
|
|
|
// emitSnapshot drives a full begin→chunk→end stream of projs at epoch.
|
|
func emitSnapshot(t *testing.T, s *ProjectionStore[Projection], epoch int64, projs []Projection) {
|
|
t.Helper()
|
|
if err := s.HandleFrame(snapMeta(SnapshotBegin, epoch, 0, len(projs)), mustEnc(t, nil)); err != nil {
|
|
t.Fatalf("begin: %v", err)
|
|
}
|
|
ups := make([]ProjectionUpdate, len(projs))
|
|
for i, p := range projs {
|
|
ups[i] = ProjectionUpdate{Projection: p}
|
|
}
|
|
if err := s.HandleFrame(snapMeta(SnapshotChunk, epoch, 0, 0), mustEnc(t, ups)); err != nil {
|
|
t.Fatalf("chunk: %v", err)
|
|
}
|
|
if err := s.HandleFrame(snapMeta(SnapshotEnd, epoch, 1, len(projs)), mustEnc(t, nil)); err != nil {
|
|
t.Fatalf("end: %v", err)
|
|
}
|
|
}
|
|
|
|
func proj(sku string, price money.Cents) Projection {
|
|
return Projection{ID: "sku:" + sku, SKU: sku, PriceIncVat: price}
|
|
}
|
|
|
|
func newStore(t *testing.T, fb ProjectionFallback) *ProjectionStore[Projection] {
|
|
t.Helper()
|
|
s, err := NewProjectionStore(t.TempDir(), "consumer.bin", identity, fb)
|
|
if err != nil {
|
|
t.Fatalf("new store: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = s.Close() })
|
|
return s
|
|
}
|
|
|
|
func TestProjectionStore_SnapshotThenGet(t *testing.T) {
|
|
s := newStore(t, nil)
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 100_00), proj("B", 150_00)})
|
|
|
|
if got, ok := s.Get("A"); !ok || got.PriceIncVat != 100_00 {
|
|
t.Fatalf("A = %+v ok=%v, want 10000", got, ok)
|
|
}
|
|
if e, base, ov := s.Stats(); e != 100 || base != 2 || ov != 0 {
|
|
t.Fatalf("stats = epoch %d base %d overlay %d, want 100/2/0", e, base, ov)
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_DeltaOverlayWinsOverBase(t *testing.T) {
|
|
s := newStore(t, nil)
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)})
|
|
|
|
// Same-epoch delta with a fresher price overrides the base.
|
|
if err := s.HandleFrame(snapMeta("", 100, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 80_00)}})); err != nil {
|
|
t.Fatalf("delta: %v", err)
|
|
}
|
|
if got, _ := s.Get("A"); got.PriceIncVat != 80_00 {
|
|
t.Fatalf("A after delta = %d, want 8000", got.PriceIncVat)
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_DeltaTombstoneShadowsBaseAndFallback(t *testing.T) {
|
|
fbHits := 0
|
|
s := newStore(t, func(string) (Projection, bool) { fbHits++; return proj("A", 1), true })
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)})
|
|
|
|
if err := s.HandleFrame(snapMeta("", 100, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: Projection{SKU: "A"}, Deleted: true}})); err != nil {
|
|
t.Fatalf("delete: %v", err)
|
|
}
|
|
if got, ok := s.Get("A"); ok {
|
|
t.Fatalf("A after tombstone = %+v, want miss (no resurrect, no fallback)", got)
|
|
}
|
|
if fbHits != 0 {
|
|
t.Fatalf("fallback hit %d times for a tombstoned SKU, want 0", fbHits)
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_FallbackOnMiss(t *testing.T) {
|
|
s := newStore(t, func(sku string) (Projection, bool) {
|
|
if sku == "Z" {
|
|
return proj("Z", 42), true
|
|
}
|
|
return Projection{}, false
|
|
})
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)})
|
|
|
|
if got, ok := s.Get("Z"); !ok || got.PriceIncVat != 42 {
|
|
t.Fatalf("Z = %+v ok=%v, want fallback 42", got, ok)
|
|
}
|
|
if _, ok := s.Get("nope"); ok {
|
|
t.Fatalf("unknown SKU should miss")
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_StaleEpochIgnored(t *testing.T) {
|
|
s := newStore(t, nil)
|
|
emitSnapshot(t, s, 200, []Projection{proj("A", 100_00)})
|
|
|
|
// An older snapshot must not replace the newer base.
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 999_00), proj("OLD", 1)})
|
|
if got, _ := s.Get("A"); got.PriceIncVat != 100_00 {
|
|
t.Fatalf("A = %d, want base epoch 200 price 10000 (stale snapshot ignored)", got.PriceIncVat)
|
|
}
|
|
if _, ok := s.Get("OLD"); ok {
|
|
t.Fatalf("OLD from a stale snapshot should not be present")
|
|
}
|
|
|
|
// An older-epoch delta is ignored too.
|
|
_ = s.HandleFrame(snapMeta("", 100, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 1)}}))
|
|
if got, _ := s.Get("A"); got.PriceIncVat != 100_00 {
|
|
t.Fatalf("A after stale delta = %d, want 10000", got.PriceIncVat)
|
|
}
|
|
}
|
|
|
|
// A future-epoch delta is DROPPED, by contract. Under FIFO per-queue delivery a
|
|
// delta never legitimately precedes its snapshot (the producer emits begin(N)
|
|
// before any delta(N)); the only way to see one is a reconnect that missed the
|
|
// whole snapshot, which a buffer couldn't reassemble anyway. Recovery is the next
|
|
// snapshot of that epoch (and the consumer's fallback in the meantime). See
|
|
// docs/catalog-projection.md.
|
|
func TestProjectionStore_FutureEpochDeltaDropped(t *testing.T) {
|
|
s := newStore(t, nil)
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)})
|
|
|
|
// A delta ahead of our base (gap) is dropped; only a matching snapshot applies it.
|
|
_ = s.HandleFrame(snapMeta("", 200, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 7)}}))
|
|
if got, _ := s.Get("A"); got.PriceIncVat != 100_00 {
|
|
t.Fatalf("A after future delta = %d, want 10000 (gap delta dropped)", got.PriceIncVat)
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_IncompleteSnapshotDiscarded(t *testing.T) {
|
|
s := newStore(t, nil)
|
|
emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)})
|
|
|
|
// begin says 3, but only 1 chunked → end must discard and keep the old base.
|
|
if err := s.HandleFrame(snapMeta(SnapshotBegin, 200, 0, 3), mustEnc(t, nil)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.HandleFrame(snapMeta(SnapshotChunk, 200, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("B", 5)}})); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.HandleFrame(snapMeta(SnapshotEnd, 200, 1, 3), mustEnc(t, nil)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if e := s.Epoch(); e != 100 {
|
|
t.Fatalf("epoch = %d, want 100 (incomplete snapshot discarded)", e)
|
|
}
|
|
if _, ok := s.Get("B"); ok {
|
|
t.Fatalf("B from an incomplete snapshot must not be visible")
|
|
}
|
|
if got, _ := s.Get("A"); got.PriceIncVat != 100_00 {
|
|
t.Fatalf("A = %d, want old base intact", got.PriceIncVat)
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_ReopenFromDiskColdStart(t *testing.T) {
|
|
dir := t.TempDir()
|
|
s1, err := NewProjectionStore(dir, "consumer.bin", identity, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
emitSnapshot(t, s1, 500, []Projection{proj("A", 100_00), proj("B", 150_00)})
|
|
_ = s1.Close()
|
|
|
|
// A fresh store on the same dir must serve immediately from the persisted base
|
|
// and resume same-epoch deltas (epoch read from the sidecar).
|
|
s2, err := NewProjectionStore(dir, "consumer.bin", identity, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = s2.Close() })
|
|
|
|
if e := s2.Epoch(); e != 500 {
|
|
t.Fatalf("reopened epoch = %d, want 500 (from sidecar)", e)
|
|
}
|
|
if got, ok := s2.Get("A"); !ok || got.PriceIncVat != 100_00 {
|
|
t.Fatalf("A after reopen = %+v ok=%v, want 10000", got, ok)
|
|
}
|
|
if err := s2.HandleFrame(snapMeta("", 500, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 90_00)}})); err != nil {
|
|
t.Fatalf("post-reopen delta: %v", err)
|
|
}
|
|
if got, _ := s2.Get("A"); got.PriceIncVat != 90_00 {
|
|
t.Fatalf("A after post-reopen delta = %d, want 9000", got.PriceIncVat)
|
|
}
|
|
}
|
|
|
|
func TestProjectionStore_BadPayloadErrors(t *testing.T) {
|
|
s := newStore(t, nil)
|
|
if err := s.HandleFrame(snapMeta(SnapshotChunk, 1, 0, 0), []byte("{not json")); err == nil {
|
|
t.Fatal("chunk with bad payload should error (caller may NACK)")
|
|
}
|
|
if err := s.HandleFrame(snapMeta("", 1, 0, 0), []byte("{not json")); err == nil {
|
|
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)
|
|
}
|
|
}
|