meta-core/internal/kv/kv_test.go

package kv

import (
	"context"
	"errors"
	"fmt"
	"path/filepath"
	"strings"
	"testing"
	"time"
)

func openTest(t *testing.T, path string) *Store {
	t.Helper()
	if path == "" {
		path = filepath.Join(t.TempDir(), "kv.db")
	}
	s, err := Open(path, Options{
		EventStream: "meta:events",
		EventMaxLen: 100_000,
		EventKey:    func(k string) bool { return strings.HasPrefix(k, "file:") },
	})
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	t.Cleanup(func() { _ = s.Close() })
	return s
}

func update(t *testing.T, s *Store, fn func(tx *Tx) error) {
	t.Helper()
	if err := s.Update(context.Background(), fn); err != nil {
		t.Fatalf("update: %v", err)
	}
}

func events(t *testing.T, s *Store) []Entry {
	t.Helper()
	es, err := s.ReadAfter(context.Background(), "meta:events", "0", 0)
	if err != nil {
		t.Fatalf("read events: %v", err)
	}
	return es
}

// The replacement for keyspace notifications: a change and its event commit
// together. `set` only for a real change (an unchanged value is a no-op: no
// write, no event), `del` only for a key that existed, nothing for keys outside
// the event scope.
func TestChangesPublishEventsWithRedisSemantics(t *testing.T) {
	s := openTest(t, "")
	update(t, s, func(tx *Tx) error {
		_ = tx.Put("file:A/title", "x")
		_ = tx.Put("file:A/title", "x") // unchanged: no write, no event
		_ = tx.Put("cid:bafk", "A")     // outside the event scope
		_, _ = tx.Delete("file:A/missing")
		_, _ = tx.Delete("file:A/title")
		return nil
	})
	got := events(t, s)
	want := []string{"set file:A/title", "del file:A/title"}
	if len(got) != len(want) {
		t.Fatalf("events = %+v, want %v", got, want)
	}
	for i, e := range got {
		if e.Values["type"]+" "+e.Values["key"] != want[i] {
			t.Errorf("event %d = %v, want %q", i, e.Values, want[i])
		}
	}
}

func TestRolledBackChangesPublishNothing(t *testing.T) {
	s := openTest(t, "")
	err := s.Update(context.Background(), func(tx *Tx) error {
		_ = tx.Put("file:A/title", "x")
		return errors.New("boom")
	})
	if err == nil {
		t.Fatal("want the error back")
	}
	if _, ok, _ := s.Get(context.Background(), "file:A/title"); ok {
		t.Fatal("rolled-back write is visible")
	}
	if es := events(t, s); len(es) != 0 {
		t.Fatalf("rolled-back write published %+v", es)
	}
}

func TestDeletePrefixPublishesOneDelPerKey(t *testing.T) {
	s := openTest(t, "")
	update(t, s, func(tx *Tx) error {
		for _, k := range []string{"file:A/a", "file:A/b", "file:AB/c"} {
			_ = tx.Put(k, "1")
		}
		return nil
	})
	var n int64
	update(t, s, func(tx *Tx) error {
		var err error
		n, err = tx.DeletePrefix("file:A/", 0)
		return err
	})
	if n != 2 {
		t.Fatalf("deleted %d, want 2 (file:AB/ is not under file:A/)", n)
	}
	es := events(t, s)
	if dels := es[3:]; len(dels) != 2 || dels[0].Values["type"] != "del" || dels[1].Values["type"] != "del" {
		t.Fatalf("want two del events, got %+v", es)
	}
}

// IDs are Redis-shaped and strictly increasing, within one millisecond and
// across a reopen — the SSE cursors consumers persist depend on both.
func TestStreamIDsAreMonotonicAcrossReopen(t *testing.T) {
	path := filepath.Join(t.TempDir(), "kv.db")
	s, err := Open(path, Options{})
	if err != nil {
		t.Fatal(err)
	}
	var ids []string
	if err := s.Update(context.Background(), func(tx *Tx) error {
		for i := 0; i < 50; i++ {
			id, err := tx.Append("file:events", 0, map[string]string{"n": "x"})
			if err != nil {
				return err
			}
			ids = append(ids, id)
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	_ = s.Close()

	s2 := openTest(t, path)
	var after string
	update(t, s2, func(tx *Tx) error {
		var err error
		after, err = tx.Append("file:events", 0, map[string]string{"n": "y"})
		return err
	})
	ids = append(ids, after)
	for i := 1; i < len(ids); i++ {
		if !idLess(t, ids[i-1], ids[i]) {
			t.Fatalf("id %d (%s) does not follow %s", i, ids[i], ids[i-1])
		}
	}
	if !strings.Contains(ids[0], "-") || strings.HasPrefix(ids[0], "0") {
		t.Fatalf("id %q is not the unpadded <ms>-<seq> form", ids[0])
	}
}

func idLess(t *testing.T, a, b string) bool {
	t.Helper()
	am, as, err := ParseID(a)
	if err != nil {
		t.Fatal(err)
	}
	bm, bs, err := ParseID(b)
	if err != nil {
		t.Fatal(err)
	}
	return am < bm || (am == bm && as < bs)
}

// The watcher's reset clears file:events. The next ID must still exceed every
// ID issued before the clear, or a consumer holding an old cursor would skip
// the reset event.
func TestClearStreamKeepsTheHighWaterMark(t *testing.T) {
	s := openTest(t, "")
	var before, after string
	update(t, s, func(tx *Tx) error {
		var err error
		for i := 0; i < 5; i++ {
			if before, err = tx.Append("file:events", 0, map[string]string{"type": "add"}); err != nil {
				return err
			}
		}
		return nil
	})
	update(t, s, func(tx *Tx) error { return tx.ClearStream("file:events") })
	if n, _ := s.Len(context.Background(), "file:events"); n != 0 {
		t.Fatalf("len after clear = %d", n)
	}
	update(t, s, func(tx *Tx) error {
		var err error
		after, err = tx.Append("file:events", 0, map[string]string{"type": "reset"})
		return err
	})
	if !idLess(t, before, after) {
		t.Fatalf("post-clear id %s does not follow pre-clear %s", after, before)
	}
	got, err := s.ReadAfter(context.Background(), "file:events", before, 10)
	if err != nil || len(got) != 1 || got[0].Values["type"] != "reset" {
		t.Fatalf("an old cursor must see the reset: %+v err=%v", got, err)
	}
}

// MAXLEN ~ semantics across many small transactions (the common case: one
// event per write).
func TestStreamTrimsAcrossTransactions(t *testing.T) {
	defer func(old int64) { trimEvery = old }(trimEvery)
	trimEvery = 100
	s := openTest(t, "")
	const maxLen = 50
	for i := int64(0); i < 3*trimEvery; i++ {
		update(t, s, func(tx *Tx) error {
			_, err := tx.Append("x", maxLen, map[string]string{"i": "1"})
			return err
		})
	}
	n, err := s.Len(context.Background(), "x")
	if err != nil {
		t.Fatal(err)
	}
	if n > maxLen+trimEvery {
		t.Fatalf("stream holds %d entries, want <= %d", n, maxLen+trimEvery)
	}
	oldest, ok, err := s.Oldest(context.Background(), "x")
	if err != nil || !ok {
		t.Fatal(err)
	}
	if last, _ := s.LastID(context.Background(), "x"); !idLess(t, oldest.ID, last) {
		t.Fatalf("oldest %s should precede last %s", oldest.ID, last)
	}
}

// A subscriber that took the wake channel before reading is woken by the next
// commit — the property the SSE pump relies on to not miss an append.
func TestWaitIsClosedByTheNextCommit(t *testing.T) {
	s := openTest(t, "")
	wake := s.Wait("meta:events")
	update(t, s, func(tx *Tx) error { return tx.Put("file:A/x", "1") })
	select {
	case <-wake:
	case <-time.After(2 * time.Second):
		t.Fatal("commit did not wake the waiter")
	}
}

func TestReadAfterZeroFormsReadFromTheStart(t *testing.T) {
	s := openTest(t, "")
	update(t, s, func(tx *Tx) error {
		_, err := tx.Append("x", 0, map[string]string{"n": "1"})
		return err
	})
	for _, id := range []string{"", "0", "0-0"} {
		got, err := s.ReadAfter(context.Background(), "x", id, 10)
		if err != nil || len(got) != 1 {
			t.Errorf("ReadAfter(%q) = %d entries, err %v; want 1", id, len(got), err)
		}
	}
}

func TestRangeIsExactlyThePrefix(t *testing.T) {
	s := openTest(t, "")
	update(t, s, func(tx *Tx) error {
		for _, k := range []string{"file:A/a", "file:A/b/c", "file:A0", "file:AB/x", "file:A"} {
			_ = tx.Put(k, "v")
		}
		return nil
	})
	rows, err := s.Range(context.Background(), "file:A/", "", 0)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 2 || rows[0].K != "file:A/a" || rows[1].K != "file:A/b/c" {
		t.Fatalf("range = %+v", rows)
	}
}

// Concurrent Updates share commits; a failing one must roll back alone and
// leave its neighbours' writes and events intact.
func TestGroupCommitIsolatesFailures(t *testing.T) {
	s := openTest(t, "")
	const n = 200
	errs := make(chan error, n)
	for i := 0; i < n; i++ {
		go func(i int) {
			errs <- s.Update(context.Background(), func(tx *Tx) error {
				k := fmt.Sprintf("file:R%03d/title", i)
				if err := tx.Put(k, "x"); err != nil {
					return err
				}
				if i%10 == 0 {
					return errors.New("fail")
				}
				return nil
			})
		}(i)
	}
	failed := 0
	for i := 0; i < n; i++ {
		if <-errs != nil {
			failed++
		}
	}
	if failed != n/10 {
		t.Fatalf("failed = %d, want %d", failed, n/10)
	}
	rows, _ := s.Range(context.Background(), "file:", "", 0)
	if len(rows) != n-n/10 {
		t.Fatalf("committed %d rows, want %d", len(rows), n-n/10)
	}
	if es := events(t, s); len(es) != n-n/10 {
		t.Fatalf("events = %d, want %d (one per committed write, none for failures)", len(es), n-n/10)
	}
	for _, r := range rows {
		var i int
		fmt.Sscanf(r.K, "file:R%03d/title", &i)
		if i%10 == 0 {
			t.Fatalf("failed request's write %s committed", r.K)
		}
	}
}

// The change hook sees real changes only, with old and new values, and
// counters follow the request's fate (a failed request's deltas are dropped).
func TestOnChangeAndCounters(t *testing.T) {
	type change struct{ k, old, new string }
	var seen []change
	s, err := Open(filepath.Join(t.TempDir(), "kv.db"), Options{
		OnChange: func(tx *Tx, k, old string, hadOld bool, v string, hasNew bool) {
			seen = append(seen, change{k, old, v})
			if hasNew {
				tx.AddCounter("count/"+v, 1)
			}
			if hadOld {
				tx.AddCounter("count/"+old, -1)
			}
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	update(t, s, func(tx *Tx) error {
		_ = tx.Put("a", "x")
		_ = tx.Put("b", "x")
		_ = tx.Put("a", "x") // no-op
		_ = tx.Put("a", "y")
		return nil
	})
	_ = s.Update(context.Background(), func(tx *Tx) error {
		_ = tx.Put("c", "x")
		return errors.New("fail")
	})
	update(t, s, func(tx *Tx) error { _, err := tx.DeletePrefix("b", 0); return err })

	if len(seen) != 5 { // a=x, b=x, a x→y, c=x (rolled back, hook still ran), del b; stats count attempts
		t.Fatalf("changes = %+v", seen)
	}
	for k, want := range map[string]int64{"count/x": 0, "count/y": 1} {
		got, _, err := s.Counter(context.Background(), k)
		if err != nil || got != want {
			t.Errorf("%s = %d (err %v), want %d", k, got, err, want)
		}
	}
	if changed, unchanged := s.WriteStats(); changed != 4 || unchanged != 1 {
		t.Errorf("write stats = %d changed / %d unchanged", changed, unchanged)
	}
}