meta-core/internal/storage/claims_test.go

package storage

import (
	"context"
	"database/sql"
	"os"
	"path/filepath"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"testing"

	"github.com/metazla/meta-core/internal/claim"
	"github.com/metazla/meta-core/internal/kv"
)

func TestWriteClaimStatuses(t *testing.T) {
	c := newTestClient(t)
	a := signer(t, 1)
	good := vote(t, a, "bafyA", "title", "The Matrix", 100)
	tampered := vote(t, a, "bafyA", "year", "1999", 100)
	tampered.Value = claim.Text("2000")
	highS := vote(t, a, "bafyA", "studio", "WB", 100)
	highS.Sig = flipS(highS.Sig)

	got := write(t, c,
		good,
		tampered,
		highS,
		claim.Claim{CID: "bafyA", UID: a.UID, TS: 1}, // no key
		vote(t, a, "bafyA", "cid_sha256", "x", 100),  // removed CID shape
	)
	want := []ClaimStatus{StatusAccepted, StatusBadSignature, StatusBadSignature, StatusBadRequest, StatusBadRequest}
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("statuses %v, want %v", got, want)
	}
	if st := write(t, c, good); st[0] != StatusUnchanged {
		t.Fatalf("identical re-send: %s, want unchanged", st[0])
	}
	if st := write(t, c, vote(t, a, "bafyA", "title", "Older", 50)); st[0] != StatusStale {
		t.Fatalf("older ts: %s, want stale", st[0])
	}
	if st := write(t, c, vote(t, a, "bafyA", "title", "Same ts other value", 100)); st[0] != StatusStale {
		t.Fatalf("equal ts, different claim: %s, want stale", st[0])
	}
	if got := record(t, c, "bafyA")["title"]; got != "The Matrix" {
		t.Fatalf("title = %q", got)
	}
	// A re-persist of the same value with a newer ts keeps the stored row (and
	// its older ts): it is "unchanged", not a new version.
	if st := write(t, c, vote(t, a, "bafyA", "title", "The Matrix", 200)); st[0] != StatusUnchanged {
		t.Fatalf("same value, newer ts: %s, want unchanged", st[0])
	}
	// Proof the stored ts stayed 100: a new value at 150 (< 200) is accepted.
	if st := write(t, c, vote(t, a, "bafyA", "title", "Updated", 150)); st[0] != StatusAccepted {
		t.Fatalf("new value at ts 150 after a same-value re-send at 200: %s, want accepted", st[0])
	}
}

func flipS(sig []byte) []byte {
	// n - s for secp256k1, computed via the vectors' helper in internal/claim
	// would be cleaner; here big-endian subtraction on bytes is enough.
	n := []byte{
		0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xfe,
		0xba, 0xae, 0xdc, 0xe6, 0xaf, 0x48, 0xa0, 0x3b, 0xbf, 0xd2, 0x5e, 0x8c, 0xd0, 0x36, 0x41, 0x41,
	}
	out := append([]byte{}, sig...)
	borrow := 0
	for i := 31; i >= 0; i-- {
		d := int(n[i]) - int(sig[32+i]) - borrow
		borrow = 0
		if d < 0 {
			d += 256
			borrow = 1
		}
		out[32+i] = byte(d)
	}
	return out
}

// A retraction is kept as a tombstone, so the vote it retracted cannot be
// replayed — by its signer's old copy or by a peer that still holds it.
func TestRetractionKeepsTombstoneAndBlocksReplay(t *testing.T) {
	c := newTestClient(t)
	a := signer(t, 1)
	old := vote(t, a, "bafyA", "genres/Drama", "true", 10)
	mustAccept(t, c, old, vote(t, a, "bafyA", "title", "X", 10))
	mustAccept(t, c, vote(t, a, "bafyA", "genres/Drama", nil, 20))
	if _, ok := record(t, c, "bafyA")["genres/Drama"]; ok {
		t.Fatal("retracted member still merged")
	}
	if st := write(t, c, old); st[0] != StatusStale {
		t.Fatalf("replay of the retracted vote: %s, want stale", st[0])
	}
	if n, _ := c.CountClaims(); n != 2 {
		t.Fatalf("claims = %d, want 2 (tombstone kept)", n)
	}
	live, _ := c.ClaimsForCIDs([]string{"bafyA"}, "")
	if len(live) != 1 || live[0].Key != "title" {
		t.Fatalf("live claims = %+v, want only title", live)
	}
}

func TestMergeRule(t *testing.T) {
	tx := func(s string) claim.Value { return claim.Text(s) }
	cases := []struct {
		name   string
		votes  []Vote
		viewer string
		want   string
		nsrc   int
		ok     bool
	}{
		{"support wins", []Vote{{Uid: "a", Value: tx("X"), TS: 9}, {Uid: "b", Value: tx("x "), TS: 9}, {Uid: "c", Value: tx("Y"), TS: 1}}, "", "X", 2, true},
		{"tie: oldest wins", []Vote{{Uid: "a", Value: tx("New"), TS: 9}, {Uid: "b", Value: tx("Old"), TS: 1}}, "", "Old", 1, true},
		{"tie: smallest wins", []Vote{{Uid: "a", Value: tx("b"), TS: 5}, {Uid: "b", Value: tx("a"), TS: 5}}, "", "a", 1, true},
		{"signer counts once, newest", []Vote{{Uid: "a", Value: tx("X"), TS: 1}, {Uid: "a", Value: tx("Y"), TS: 2}, {Uid: "b", Value: tx("Y"), TS: 3}}, "", "Y", 2, true},
		{"retraction drops", []Vote{{Uid: "a", Value: tx("X"), TS: 1}, {Uid: "a", Value: claim.Null(), TS: 2}}, "", "", 0, false},
		{"ciphertext never merges", []Vote{{Uid: "a", Value: claim.Bytes([]byte{1}), TS: 1}}, "", "", 0, false},
		{"viewer wins", []Vote{{Uid: "a", Value: tx("X"), TS: 1}, {Uid: "b", Value: tx("X"), TS: 1}, {Uid: "v", Value: tx("Mine"), TS: 9}}, "v", "Mine", 1, true},
		{"viewer retraction does not hide", []Vote{{Uid: "a", Value: tx("X"), TS: 1}, {Uid: "v", Value: claim.Null(), TS: 9}}, "v", "X", 1, true},
	}
	for _, tc := range cases {
		got := Merge(tc.votes, tc.viewer)
		if got.OK != tc.ok || got.Value != tc.want || (tc.ok && got.Nsrc != tc.nsrc) {
			t.Errorf("%s: got %+v, want %q nsrc %d ok %v", tc.name, got, tc.want, tc.nsrc, tc.ok)
		}
	}
}

func TestMergedViewAndClaims(t *testing.T) {
	c := newTestClient(t)
	gwA, gwB, gwC := signer(t, 1), signer(t, 2), signer(t, 3)
	mustAccept(t, c,
		vote(t, gwA, "bafyA", "title", "Matrix", 10),
		vote(t, gwB, "bafyA", "title", "Matrix", 20),
		vote(t, gwC, "bafyA", "title", "The Matrix (1999)", 5),
	)
	rec := record(t, c, "bafyA")
	if rec["title"] != "Matrix" || rec["cids/bafyA"] != "true" {
		t.Fatalf("merged %v", rec)
	}
	claims, err := c.ClaimsForRID(rid(t, c, "bafyA"), "title")
	if err != nil || len(claims) != 3 {
		t.Fatalf("claims %d (err %v), want 3", len(claims), err)
	}
	for _, cl := range claims {
		if err := cl.Verify(); err != nil {
			t.Fatalf("returned claim does not verify: %v", err)
		}
	}
}

// A cids/* link merges two groups: the older rid survives, the loser's rows
// go with `del` events, and the survivor carries both groups' votes.
func TestGroupLinkMergesRecords(t *testing.T) {
	c := newTestClient(t)
	a, b := signer(t, 1), signer(t, 2)
	mustAccept(t, c, vote(t, a, "bafyM", "title", "Local copy", 10))
	older := rid(t, c, "bafyM")
	mustAccept(t, c, vote(t, b, "bafyQ", "year", "1999", 10))
	newer := rid(t, c, "bafyQ")
	if older == newer || older > newer {
		t.Fatalf("expected two groups, older first: %s %s", older, newer)
	}
	before, _ := c.Store().LastID(context.Background(), MetaEventsStream)

	mustAccept(t, c, vote(t, a, "bafyM", "cids/bafyQ", "true", 11))
	if got := rid(t, c, "bafyQ"); got != older {
		t.Fatalf("bafyQ → %s, want the older rid %s", got, older)
	}
	rec := record(t, c, "bafyM")
	if rec["title"] != "Local copy" || rec["year"] != "1999" || rec["cids/bafyQ"] != "true" || rec["cids/bafyM"] != "true" {
		t.Fatalf("survivor %v", rec)
	}
	if f, _ := c.GetMetadataFlat(newer); f != nil {
		t.Fatalf("loser still has rows: %v", f)
	}
	entries, _ := c.Store().ReadAfter(context.Background(), MetaEventsStream, before, 1000)
	sawDel := false
	for _, e := range entries {
		if e.Values["type"] == "del" && strings.HasPrefix(e.Values["key"], "file:"+newer+"/") {
			sawDel = true
		}
	}
	if !sawDel {
		t.Fatal("no del event for the losing rid")
	}
	if n, _ := c.CountRoots(); n != 1 {
		t.Fatalf("roots = %d, want 1", n)
	}
}

func TestViewerOverlay(t *testing.T) {
	c := newTestClient(t)
	a, b, me := signer(t, 1), signer(t, 2), signer(t, 9)
	mustAccept(t, c,
		vote(t, a, "bafyA", "title", "Shared", 1),
		vote(t, b, "bafyA", "title", "Shared", 1),
		vote(t, me, "bafyA", "title", "Mine", 2),
		vote(t, me, "bafyA", "seek", []byte{1, 2, 3}, 2), // ciphertext: never shown
	)
	r := rid(t, c, "bafyA")
	shared, _ := c.GetMetadataFlat(r)
	mine, _ := c.GetMetadataFlatAs(r, me.UID)
	if shared["title"] != "Shared" || mine["title"] != "Mine" {
		t.Fatalf("shared %q, mine %q", shared["title"], mine["title"])
	}
	if _, ok := mine["seek"]; ok {
		t.Fatal("ciphertext leaked into the view")
	}
	match := func(_ string, f map[string]string) bool { return strings.Contains(strings.ToLower(f["title"]), "mine") }
	if hits, _ := c.Search("", nil, 10, match); len(hits) != 0 {
		t.Fatalf("shared search found the viewer's value: %v", hits)
	}
	hits, err := c.SearchAs("", nil, 10, me.UID, match)
	if err != nil || len(hits) != 1 {
		t.Fatalf("viewer search hits %v (err %v), want 1", hits, err)
	}
	merged, _, dropped, err := c.MergeClaims([]claim.Claim{
		vote(t, a, "bafyA", "title", "Shared", 1),
		vote(t, b, "bafyZ", "title", "Other", 1),
	}, me.UID)
	if err != nil || dropped != 0 || merged["title"] != "Mine" {
		t.Fatalf("stateless merge as viewer = %v dropped %d err %v", merged, dropped, err)
	}
}

// A crash between the truth commit and the cache apply leaves the cache
// behind; the next boot replays the missed changes from truth.
func TestCrashBetweenCommitsIsCaughtUp(t *testing.T) {
	dir := t.TempDir()
	c := openClient(t, dir)
	a := signer(t, 1)
	mustAccept(t, c, vote(t, a, "bafyA", "title", "Before", 1))

	// Commit straight to truth, as WriteClaims does, but never apply.
	lost := vote(t, a, "bafyA", "title", "After", 2)
	if err := c.truth.Update(context.Background(), func(tx *kv.Tx) error {
		seq, err := readLastSeq(tx)
		if err != nil {
			return err
		}
		pid, err := ensurePrincipal(tx, lost.UID)
		if err != nil {
			return err
		}
		seq++
		if _, err := tx.Exec(`UPDATE claim SET value = ?, ts = ?, sig = ?, seq = ? WHERE cid = ? AND key = ? AND pid = ?`,
			"After", int64(lost.TS), lost.Sig, seq, lost.CID, lost.Key, pid); err != nil {
			return err
		}
		return writeLastSeq(tx, seq)
	}); err != nil {
		t.Fatal(err)
	}
	if got := record(t, c, "bafyA")["title"]; got != "Before" {
		t.Fatalf("cache applied a change it was never given: %q", got)
	}
	c.Close()

	c = openClient(t, dir)
	defer c.Close()
	if got := record(t, c, "bafyA")["title"]; got != "After" {
		t.Fatalf("after catch-up title = %q, want After", got)
	}
}

// The cache is a pure function of truth: deleting it and rebuilding yields
// exactly what incremental application produced.
func TestRebuildEqualsIncremental(t *testing.T) {
	dir := t.TempDir()
	c := openClient(t, dir)
	a, b, u := signer(t, 1), signer(t, 2), signer(t, 3)
	mustAccept(t, c,
		vote(t, a, "bafy1", "title", "One", 1), vote(t, a, "bafy1", "genres/Action", "true", 1),
		vote(t, b, "bafy1", "title", "Uno", 2), vote(t, b, "bafy2", "title", "Two", 2),
		vote(t, a, "bafy2", "sizeByte", "123", 1), vote(t, u, "bafy2", "like", "true", 3),
		vote(t, u, "bafy2", "seek", []byte{9}, 3),
	)
	mustAccept(t, c,
		vote(t, a, "bafy1", "genres/Action", nil, 4), // retraction
		vote(t, b, "bafy1", "cids/bafy2", "true", 5), // group merge
		vote(t, a, "bafy3", "title", "Three", 5),
	)
	incremental := cacheState(t, c)
	c.Close()
	for _, f := range []string{"cache.db", "cache.db-wal", "cache.db-shm"} {
		_ = os.Remove(filepath.Join(dir, f))
	}
	c = openClient(t, dir)
	defer c.Close()
	if rebuilt := cacheState(t, c); !reflect.DeepEqual(rebuilt, incremental) {
		t.Fatalf("rebuild differs from incremental:\n incremental %v\n rebuilt     %v", incremental, rebuilt)
	}
	if c.RebuiltAt() == "" {
		t.Fatal("rebuild did not record rebuilt_at (SSE consumers would not get a gap)")
	}
}

// cacheState is every derived row that must be a function of truth.
func cacheState(t *testing.T, c *Client) []string {
	t.Helper()
	var out []string
	for _, k := range allKeys(t, c) {
		if strings.HasPrefix(k, "ev:") || strings.HasPrefix(k, "evhwm:") || strings.HasPrefix(k, "meta:") ||
			strings.HasPrefix(k, ftsNS) || k == countersReady {
			continue
		}
		v, _, _ := c.Store().Get(context.Background(), k)
		out = append(out, k+"="+v)
	}
	rows, err := c.Store().DB().Query(`SELECT pid, cid, key FROM claim_by_pid`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	for rows.Next() {
		var pid int64
		var cid, key string
		_ = rows.Scan(&pid, &cid, &key)
		out = append(out, "byp:"+strconv.FormatInt(pid, 10)+"/"+cid+"/"+key)
	}
	var hays []string
	hr, _ := c.Store().DB().Query(`SELECT rid, hay FROM fts`)
	for hr.Next() {
		var r, h string
		_ = hr.Scan(&r, &h)
		hays = append(hays, "fts:"+r+"="+h)
	}
	hr.Close()
	out = append(out, hays...)
	sort.Strings(out)
	return out
}

func TestLocalFileFactsAndEvents(t *testing.T) {
	c := newTestClient(t)
	before, _ := c.Store().LastID(context.Background(), MetaEventsStream)
	r, created, err := c.RegisterFileTuple("/files/watch/a.mkv", 1000, 42, "bafkmid")
	if err != nil || !created {
		t.Fatalf("register: %v created=%v", err, created)
	}
	rec, _ := c.GetMetadataFlat(r)
	if rec["filePath"] != "/files/watch/a.mkv" || rec["sizeByte"] != "1000" || rec["mtimeNano"] != "42" || rec["cids/bafkmid"] != "true" {
		t.Fatalf("record %v", rec)
	}
	entries, _ := c.Store().ReadAfter(context.Background(), MetaEventsStream, before, 100)
	sawPath := false
	for _, e := range entries {
		if e.Values["type"] == "set" && e.Values["key"] == "file:"+r+"/filePath" {
			sawPath = true
		}
	}
	if !sawPath {
		t.Fatal("no set event for file:<rid>/filePath (meta-fuse builds its tree from it)")
	}
	// A signed sizeByte (content fact) wins over the on-disk size.
	mustAccept(t, c, vote(t, signer(t, 1), "bafkmid", "sizeByte", "999", 1))
	if got, _ := c.GetProperty(r, "sizeByte"); got != "999" {
		t.Fatalf("sizeByte = %q, want the signed 999", got)
	}
	// A second path of the same content is a duplicate.
	if _, _, err := c.RegisterFileTuple("/files/watch/copy.mkv", 1000, 43, "bafkmid"); err != nil {
		t.Fatal(err)
	}
	doc, _ := c.GetMetadataDocument(r)
	if len(doc.Duplicates) != 1 || doc.Duplicates[0] != "/files/watch/copy.mkv" {
		t.Fatalf("duplicates %v", doc.Duplicates)
	}
	tuples, _ := c.GetTuplesForAllFiles()
	if len(tuples) != 2 || tuples[0].HashID != r {
		t.Fatalf("tuples %+v", tuples)
	}
	// Deleting the files drops the local facts, keeps the claims.
	mark, _ := c.Store().LastID(context.Background(), MetaEventsStream)
	for _, p := range []string{"/files/watch/a.mkv", "/files/watch/copy.mkv"} {
		if err := c.RemoveFileTuple(p); err != nil {
			t.Fatal(err)
		}
	}
	rec, _ = c.GetMetadataFlat(r)
	if _, ok := rec["filePath"]; ok || rec["sizeByte"] != "999" {
		t.Fatalf("after delete %v", rec)
	}
	entries, _ = c.Store().ReadAfter(context.Background(), MetaEventsStream, mark, 100)
	sawDel := false
	for _, e := range entries {
		if e.Values["type"] == "del" && e.Values["key"] == "file:"+r+"/filePath" {
			sawDel = true
		}
	}
	if !sawDel {
		t.Fatal("no del event for filePath")
	}
}

func TestSchemaGateRefusesV1Store(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "meta-core.db")
	db, err := sql.Open("sqlite", "file:"+path)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`CREATE TABLE kv(k TEXT PRIMARY KEY, v TEXT NOT NULL) WITHOUT ROWID;
		INSERT INTO kv VALUES('meta-core:schema-version','2')`); err != nil {
		t.Fatal(err)
	}
	db.Close()
	c := NewClient()
	if err := c.Connect(path, filepath.Join(dir, "cache.db")); err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	if err := c.EnsureSchemaVersion(SchemaVersion); err == nil || !strings.Contains(err.Error(), "v1.x") {
		t.Fatalf("v1 store accepted: %v", err)
	}
}