platform/projection/emitter_test.go

package projection

import (
	"errors"
	"fmt"
	"sort"
	"testing"
)

// busToStore wires an Emitter's frames straight into a Store via HandleFrame —
// the in-process equivalent of publishing to the bus and a receiver consuming.
// Returns a PublishFunc the emitter calls per frame.
func busToStore[W, T any](t *testing.T, s *Store[W, T]) PublishFunc {
	t.Helper()
	return func(meta map[string]string, payload []byte) error {
		return s.HandleFrame(meta, payload)
	}
}

// newIdentityStore builds a Store where wire == stored (W == T == tProj), using
// the generic JSON codec. opts forwards e.g. WithKeyHashing.
func newIdentityStore(t *testing.T, opts ...StoreOption) *Store[tProj, tProj] {
	t.Helper()
	s, err := NewStore[tProj, tProj](t.TempDir(), "emit.bin",
		JSONDecode[tProj], func(p tProj) tProj { return p }, nil, opts...)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = s.Close() })
	return s
}

func ups(items ...tProj) []Update[tProj] {
	out := make([]Update[tProj], len(items))
	for i, it := range items {
		out[i] = Update[tProj]{Key: it.SKU, Value: it}
	}
	return out
}

// TestEmitter_SnapshotRoundTripsIntoStore — a full snapshot emitted frame-by-frame
// reassembles in the receiver's Store, served by Get.
func TestEmitter_SnapshotRoundTripsIntoStore(t *testing.T) {
	s := newIdentityStore(t)
	e := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s), WithTenant("se"), WithChunkSize(2))

	items := ups(
		makeProjection("A", 100), makeProjection("B", 200),
		makeProjection("C", 300), makeProjection("D", 400), makeProjection("E", 500),
	)
	chunks, err := e.Snapshot(1000, items)
	if err != nil {
		t.Fatalf("Snapshot: %v", err)
	}
	if chunks != 3 { // 5 items / chunkSize 2 = ceil = 3
		t.Errorf("chunks = %d, want 3", chunks)
	}
	if ep, base, _ := s.Stats(); ep != 1000 || base != 5 {
		t.Fatalf("stats = epoch %d base %d, want 1000/5", ep, base)
	}
	for _, k := range []string{"A", "C", "E"} {
		if _, ok := s.Get(k); !ok {
			t.Errorf("Get(%q) miss after snapshot", k)
		}
	}
}

// TestEmitter_DeltaAfterSnapshot — a same-epoch delta lands in the overlay and
// wins over the base; a tombstone hides the key.
func TestEmitter_DeltaAfterSnapshot(t *testing.T) {
	s := newIdentityStore(t)
	e := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s))

	if _, err := e.Snapshot(10, ups(makeProjection("A", 100), makeProjection("B", 200))); err != nil {
		t.Fatal(err)
	}
	// upsert A to a new price + delete B, same epoch.
	if err := e.Delta(10, []Update[tProj]{
		{Key: "A", Value: makeProjection("A", 999)},
		{Key: "B", Deleted: true},
	}); err != nil {
		t.Fatal(err)
	}
	if got, _ := s.Get("A"); got.PriceIncVat != 999 {
		t.Errorf("A = %d, want overlay 999", got.PriceIncVat)
	}
	if _, ok := s.Get("B"); ok {
		t.Error("B should be tombstoned")
	}
	if !s.IsDeleted("B") {
		t.Error("IsDeleted(B) should be true")
	}
}

// TestEmitter_EmptySnapshotRefused — emitting nothing must not wipe receivers.
func TestEmitter_EmptySnapshotRefused(t *testing.T) {
	var published int
	e := NewEmitter[tProj](JSONEncode[tProj], func(map[string]string, []byte) error {
		published++
		return nil
	})
	if _, err := e.Snapshot(1, nil); !errors.Is(err, ErrEmptySnapshot) {
		t.Fatalf("empty Snapshot err = %v, want ErrEmptySnapshot", err)
	}
	if published != 0 {
		t.Errorf("published %d frames for an empty snapshot, want 0", published)
	}
}

// TestEmitter_PublishErrorAbortsStream — a publish failure mid-stream stops the
// emitter and the receiver keeps its previous base.
func TestEmitter_PublishErrorAbortsStream(t *testing.T) {
	s := newIdentityStore(t)
	// Seed a good base at epoch 5.
	good := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s))
	if _, err := good.Snapshot(5, ups(makeProjection("OLD", 1))); err != nil {
		t.Fatal(err)
	}

	// New emitter whose publish fails on the first chunk of the next snapshot.
	calls := 0
	flaky := NewEmitter[tProj](JSONEncode[tProj], func(meta map[string]string, payload []byte) error {
		calls++
		if meta[MetaSnapshotPhase] == SnapshotChunk {
			return fmt.Errorf("boom")
		}
		return s.HandleFrame(meta, payload)
	})
	if _, err := flaky.Snapshot(9, ups(makeProjection("NEW", 2))); err == nil {
		t.Fatal("expected Snapshot to surface the publish error")
	}
	// Base stays at the old epoch; OLD still served, NEW never applied.
	if ep := s.Epoch(); ep != 5 {
		t.Errorf("epoch = %d, want 5 (aborted snapshot didn't swap)", ep)
	}
	if _, ok := s.Get("OLD"); !ok {
		t.Error("OLD should survive the aborted rebuild")
	}
	if _, ok := s.Get("NEW"); ok {
		t.Error("NEW must not be visible from an aborted snapshot")
	}
}

// TestEmitter_HashedKeysEndToEnd — long keys flow producer→consumer in hashed
// mode: Get by original key works, and Scan yields digests with the original
// recoverable from the value. Ranked top-N (the popularity use case) works.
func TestEmitter_HashedKeysEndToEnd(t *testing.T) {
	s := newIdentityStore(t, WithKeyHashing())
	e := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s))

	// Keys that blow past 32 bytes (query strings), values carry the original +
	// a score in PriceIncVat to rank by.
	mk := func(q string, score int64) tProj { return tProj{SKU: q, PriceIncVat: score} }
	q1 := "the quick brown fox jumps over the lazy dog (query one)"
	q2 := "another unbounded length search query that a user typed"
	q3 := "short-ish but still over thirty-two bytes for sure yes"
	if _, err := e.Snapshot(1, ups(mk(q1, 30), mk(q2, 10), mk(q3, 20))); err != nil {
		t.Fatal(err)
	}

	// Point lookup by the ORIGINAL key.
	if got, ok := s.Get(q1); !ok || got.PriceIncVat != 30 {
		t.Fatalf("Get(q1) = %+v ok=%v, want score 30", got, ok)
	}

	// Ranked enumeration: collect via Scan (keys are digests), sort by score,
	// recover the original query from the value. Top-N popularity in a nutshell.
	type row struct {
		query string
		score int64
	}
	var rows []row
	s.Scan(func(key string, v tProj) bool {
		if len(key) != 32 {
			t.Errorf("hashed Scan key len = %d, want 32", len(key))
		}
		rows = append(rows, row{v.SKU, v.PriceIncVat})
		return true
	})
	if len(rows) != 3 {
		t.Fatalf("scanned %d rows, want 3", len(rows))
	}
	sort.Slice(rows, func(i, j int) bool { return rows[i].score > rows[j].score })
	if rows[0].query != q1 || rows[1].query != q3 || rows[2].query != q2 {
		t.Fatalf("ranked order wrong: %+v", rows)
	}
}