package storage // The truth database: the only data meta-core cannot rebuild. // // principal(pid, uid) who signed (pid = local short form) // claim(cid, key, pid) → value, ts, sig, seq every vote, signed (MMUR2) // cid_group(cid) → rid which record a cid belongs to // meta(name) → value schema version, last_seq // // Primary keys only — every other index lives in the cache, which is rebuilt // from these four tables. A claim never mentions a rid: the rid is a local // name for a group of cids, minted here (UUIDv7) and kept because consumers // hold rids. import ( "context" "database/sql" "errors" "fmt" "strconv" "strings" "github.com/metazla/meta-core/internal/claim" "github.com/metazla/meta-core/internal/kv" ) const truthSchema = ` CREATE TABLE IF NOT EXISTS principal( pid INTEGER PRIMARY KEY, uid TEXT NOT NULL UNIQUE ) STRICT; CREATE TABLE IF NOT EXISTS claim( cid TEXT NOT NULL, key TEXT NOT NULL, pid INTEGER NOT NULL, value ANY, ts INTEGER NOT NULL, sig BLOB NOT NULL, seq INTEGER NOT NULL, PRIMARY KEY (cid, key, pid) ) STRICT, WITHOUT ROWID; CREATE TABLE IF NOT EXISTS cid_group(cid TEXT PRIMARY KEY, rid TEXT NOT NULL) STRICT, WITHOUT ROWID; CREATE TABLE IF NOT EXISTS meta(name TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT, WITHOUT ROWID; ` const lastSeqName = "last_seq" // ClaimStatus is the per-claim outcome of POST /api/claims. type ClaimStatus string const ( StatusAccepted ClaimStatus = "accepted" StatusUnchanged ClaimStatus = "unchanged" StatusStale ClaimStatus = "stale" StatusBadSignature ClaimStatus = "bad_signature" StatusBadRequest ClaimStatus = "bad_request" ) // ClaimResult is one entry of the POST /api/claims response. type ClaimResult struct { Status ClaimStatus `json:"status"` Reason string `json:"reason,omitempty"` } // MaxClaimsPerBatch bounds one POST /api/claims. const MaxClaimsPerBatch = 10_000 // change is one committed truth change, queued for the apply loop in commit // order. key == "" means "membership only" (a cid joined a group). type change struct { seq int64 cid string key string pid int64 live bool // the claim now holds a value (false: tombstone) regroup *[2]string // {loser, survivor}: two groups merged } func valueArg(v claim.Value) any { switch v.Kind { case claim.KindText: return v.Text case claim.KindBytes: if v.Bytes == nil { return []byte{} } return v.Bytes } return nil } // scanValue turns a claim.value column (TEXT / BLOB / NULL) back into a Value. func scanValue(raw any) claim.Value { switch x := raw.(type) { case string: return claim.Text(x) case []byte: return claim.Bytes(x) } return claim.Null() } type storedClaim struct { ts int64 sig []byte value claim.Value } func readStored(ctx context.Context, q interface { QueryRowContext(context.Context, string, ...any) *sql.Row }, cid, key, uid string) (storedClaim, bool, error) { var sc storedClaim var raw any err := q.QueryRowContext(ctx, `SELECT c.ts, c.sig, c.value FROM claim c JOIN principal p ON p.pid = c.pid WHERE c.cid = ? AND c.key = ? AND p.uid = ?`, cid, key, uid).Scan(&sc.ts, &sc.sig, &raw) sc.value = scanValue(raw) if errors.Is(err, sql.ErrNoRows) { return sc, false, nil } return sc, err == nil, err } // precheck classifies a claim against the committed state without the writer: // skip-identical happens BEFORE verification (most writes are re-persists of // the same signed claim), stale is refused, everything else is verified. func (c *Client) precheck(ctx context.Context, cl *claim.Claim) (ClaimStatus, string) { if cl.CID == "" || cl.Key == "" || cl.UID == "" { return StatusBadRequest, "cid, uid and key are required" } if !claimKeyAllowed(cl.Key) { return StatusBadRequest, "key is reserved or malformed" } if len(cl.Key) > 4096 || len(cl.CID) > 512 { return StatusBadRequest, "cid or key too long" } if cl.TS == 0 || cl.TS > 1<<62 { return StatusBadRequest, "ts out of range" } sc, ok, err := readStored(ctx, c.truth.DB(), cl.CID, cl.Key, cl.UID) if err != nil { return StatusBadRequest, err.Error() } if ok { if sc.ts == int64(cl.TS) && string(sc.sig) == string(cl.Sig) { return StatusUnchanged, "" } // Same signer re-asserting the value it already holds (the gateway // re-persists every search hit with a fresh ts): keep the stored row and // its older ts, so oldest-ts tie-breaking stays anchored and truth takes // no write. Nothing is stored, so the new signature needs no check. if sc.value.Equal(cl.Value) { return StatusUnchanged, "" } if sc.ts >= int64(cl.TS) { return StatusStale, "a newer or equal ts is stored for this signer, cid and key" } } if err := cl.Verify(); err != nil { return StatusBadSignature, err.Error() } return StatusAccepted, "" } // WriteClaims is POST /api/claims: verify, store, regroup, and wait until the // cache has applied the batch (read-your-writes for the caller). func (c *Client) WriteClaims(ctx context.Context, claims []claim.Claim) ([]ClaimResult, error) { truth, err := c.truthStore() if err != nil { return nil, err } results := make([]ClaimResult, len(claims)) var todo []int for i := range claims { st, reason := c.precheck(ctx, &claims[i]) results[i] = ClaimResult{Status: st, Reason: reason} if st == StatusAccepted { todo = append(todo, i) } } if len(todo) == 0 { return results, nil } var maxSeq int64 err = truth.Update(ctx, func(tx *kv.Tx) error { seq, err := readLastSeq(tx) if err != nil { return err } var changes []change pids := map[string]int64{} for _, i := range todo { cl := &claims[i] // Re-check under the writer: a concurrent batch may have stored a // newer vote since the precheck. sc, ok, err := readStored(tx.Context(), txQuerier{tx}, cl.CID, cl.Key, cl.UID) if err != nil { return err } if ok && (sc.ts >= int64(cl.TS) || sc.value.Equal(cl.Value)) { if sc.value.Equal(cl.Value) { results[i] = ClaimResult{Status: StatusUnchanged} } else { results[i] = ClaimResult{Status: StatusStale, Reason: "a newer or equal ts is stored for this signer, cid and key"} } continue } pid, ok := pids[cl.UID] if !ok { if pid, err = ensurePrincipal(tx, cl.UID); err != nil { return err } pids[cl.UID] = pid } seq++ if _, err := tx.Exec( `INSERT INTO claim(cid, key, pid, value, ts, sig, seq) VALUES(?, ?, ?, ?, ?, ?, ?) ON CONFLICT(cid, key, pid) DO UPDATE SET value = excluded.value, ts = excluded.ts, sig = excluded.sig, seq = excluded.seq`, cl.CID, cl.Key, pid, valueArg(cl.Value), int64(cl.TS), cl.Sig, seq); err != nil { return fmt.Errorf("store claim: %w", err) } if _, _, err := ensureGroup(tx, cl.CID); err != nil { return err } changes = append(changes, change{seq: seq, cid: cl.CID, key: cl.Key, pid: pid, live: !cl.Value.IsNull()}) if other := cidFromKeysetField(cl.Key); other != "" && cl.Value.Kind == claim.KindText && other != cl.CID { more, err := linkGroups(tx, cl.CID, other, seq) if err != nil { return err } changes = append(changes, more...) } } if len(changes) == 0 { return nil } if err := writeLastSeq(tx, seq); err != nil { return err } maxSeq = seq tx.OnCommit(func() { c.applier.enqueue(changes) }) return nil }) if err != nil { return nil, err } if maxSeq > 0 { if err := c.applier.wait(ctx, maxSeq); err != nil { return results, err } } return results, nil } // txQuerier adapts a kv.Tx to the QueryRowContext shape readStored takes. type txQuerier struct{ tx *kv.Tx } func (q txQuerier) QueryRowContext(_ context.Context, query string, args ...any) *sql.Row { return q.tx.QueryRow(query, args...) } func readLastSeq(tx *kv.Tx) (int64, error) { var v string err := tx.QueryRow(`SELECT value FROM meta WHERE name = ?`, lastSeqName).Scan(&v) if errors.Is(err, sql.ErrNoRows) { return 0, nil } if err != nil { return 0, err } return strconv.ParseInt(v, 10, 64) } func writeLastSeq(tx *kv.Tx, seq int64) error { _, err := tx.Exec(`INSERT INTO meta(name, value) VALUES(?, ?) ON CONFLICT(name) DO UPDATE SET value = excluded.value`, lastSeqName, strconv.FormatInt(seq, 10)) return err } func ensurePrincipal(tx *kv.Tx, uid string) (int64, error) { var pid int64 err := tx.QueryRow(`SELECT pid FROM principal WHERE uid = ?`, uid).Scan(&pid) if err == nil { return pid, nil } if !errors.Is(err, sql.ErrNoRows) { return 0, err } if err := tx.QueryRow(`INSERT INTO principal(uid) VALUES(?) RETURNING pid`, uid).Scan(&pid); err != nil { return 0, fmt.Errorf("new principal: %w", err) } return pid, nil } // ensureGroup returns cid's rid, minting a new group for an unknown cid. func ensureGroup(tx *kv.Tx, cid string) (string, bool, error) { var rid string err := tx.QueryRow(`SELECT rid FROM cid_group WHERE cid = ?`, cid).Scan(&rid) if err == nil { return rid, false, nil } if !errors.Is(err, sql.ErrNoRows) { return "", false, err } if rid, err = NewUUID(); err != nil { return "", false, err } if _, err := tx.Exec(`INSERT INTO cid_group(cid, rid) VALUES(?, ?)`, cid, rid); err != nil { return "", false, fmt.Errorf("new group: %w", err) } return rid, true, nil } // linkGroups records that a and b are the same content. An unknown b joins a's // group; two known groups merge, the older rid surviving (UUIDv7 sorts by // time). A retraction of the link does not split the group (phase 1). func linkGroups(tx *kv.Tx, a, b string, seq int64) ([]change, error) { ra, _, err := ensureGroup(tx, a) if err != nil { return nil, err } var rb string err = tx.QueryRow(`SELECT rid FROM cid_group WHERE cid = ?`, b).Scan(&rb) if errors.Is(err, sql.ErrNoRows) { if _, err := tx.Exec(`INSERT INTO cid_group(cid, rid) VALUES(?, ?)`, b, ra); err != nil { return nil, err } return []change{{seq: seq, cid: b}}, nil } if err != nil { return nil, err } if ra == rb { return nil, nil } survivor, loser := ra, rb if rb < ra { survivor, loser = rb, ra } // cid_group has no rid index (truth keeps primary keys only); a merge is // rare, so the scan is acceptable. if _, err := tx.Exec(`UPDATE cid_group SET rid = ? WHERE rid = ?`, survivor, loser); err != nil { return nil, fmt.Errorf("merge groups: %w", err) } return []change{{seq: seq, regroup: &[2]string{loser, survivor}}}, nil } // truthRid reads cid's committed group. func (c *Client) truthRid(ctx context.Context, cid string) (string, error) { var rid string err := c.truth.DB().QueryRowContext(ctx, `SELECT rid FROM cid_group WHERE cid = ?`, cid).Scan(&rid) if errors.Is(err, sql.ErrNoRows) { return "", nil } return rid, err } // votesFor reads every committed vote on key across cids, with signer uids. func (c *Client) votesFor(ctx context.Context, cids []string, key string) ([]Vote, error) { var out []Vote for _, cid := range cids { rows, err := c.truth.DB().QueryContext(ctx, `SELECT c.pid, p.uid, c.value, c.ts FROM claim c JOIN principal p ON p.pid = c.pid WHERE c.cid = ? AND c.key = ?`, cid, key) if err != nil { return nil, err } for rows.Next() { var v Vote var raw any var ts int64 if err := rows.Scan(&v.Pid, &v.Uid, &raw, &ts); err != nil { rows.Close() return nil, err } v.Value, v.TS = scanValue(raw), uint64(ts) out = append(out, v) } if err := rows.Close(); err != nil { return nil, err } } return out, nil } // claimKeys lists the distinct keys claimed on any of cids. func (c *Client) claimKeys(ctx context.Context, cids []string) ([]string, error) { seen := map[string]bool{} var out []string for _, cid := range cids { rows, err := c.truth.DB().QueryContext(ctx, `SELECT DISTINCT key FROM claim WHERE cid = ?`, cid) if err != nil { return nil, err } for rows.Next() { var k string if err := rows.Scan(&k); err != nil { rows.Close() return nil, err } if !seen[k] { seen[k] = true out = append(out, k) } } if err := rows.Close(); err != nil { return nil, err } } return out, nil } // ClaimsForCIDs returns every live signed claim on cids (tombstones excluded), // optionally only for one key. func (c *Client) ClaimsForCIDs(cids []string, key string) ([]claim.Claim, error) { if _, err := c.truthStore(); err != nil { return nil, err } ctx, cancel := opCtx() defer cancel() var out []claim.Claim for _, cid := range cids { q := `SELECT c.cid, p.uid, c.key, c.value, c.ts, c.sig FROM claim c JOIN principal p ON p.pid = c.pid WHERE c.cid = ? AND c.value IS NOT NULL` args := []any{cid} if key != "" { q += ` AND c.key = ?` args = append(args, key) } q += ` ORDER BY c.key, p.uid` rows, err := c.truth.DB().QueryContext(ctx, q, args...) if err != nil { return nil, err } for rows.Next() { var cl claim.Claim var raw any var ts int64 if err := rows.Scan(&cl.CID, &cl.UID, &cl.Key, &raw, &ts, &cl.Sig); err != nil { rows.Close() return nil, err } cl.Value, cl.TS = scanValue(raw), uint64(ts) out = append(out, cl) } if err := rows.Close(); err != nil { return nil, err } } return out, nil } // ClaimsForRID returns the live claims of every cid in rid's group. func (c *Client) ClaimsForRID(rid, key string) ([]claim.Claim, error) { _, claims, err := c.RecordClaims(rid, key) return claims, err } // RecordClaims returns the cids of rid's group (empty: no such record) and // their live claims, optionally for one key. func (c *Client) RecordClaims(rid, key string) ([]string, []claim.Claim, error) { s, err := c.kv() if err != nil { return nil, nil, err } ctx, cancel := opCtx() defer cancel() cids, err := cidsForRoot(ctx, s, rid) if err != nil || len(cids) == 0 { return cids, nil, err } claims, err := c.ClaimsForCIDs(cids, key) return cids, claims, err } // LiveClaimsBySigner lists (cid, key) of every live claim uid has signed — the // set DELETE /api/identity retracts. func (c *Client) LiveClaimsBySigner(uid string) ([][2]string, error) { s, err := c.kv() if err != nil { return nil, err } pid, ok, err := c.pidOf(uid) if err != nil || !ok { return nil, err } ctx, cancel := opCtx() defer cancel() rows, err := s.DB().QueryContext(ctx, `SELECT cid, key FROM claim_by_pid WHERE pid = ? ORDER BY cid, key`, pid) if err != nil { return nil, err } defer rows.Close() var out [][2]string for rows.Next() { var p [2]string if err := rows.Scan(&p[0], &p[1]); err != nil { return nil, err } out = append(out, p) } return out, rows.Err() } func (c *Client) pidOf(uid string) (int64, bool, error) { if _, err := c.truthStore(); err != nil { return 0, false, err } ctx, cancel := opCtx() defer cancel() var pid int64 err := c.truth.DB().QueryRowContext(ctx, `SELECT pid FROM principal WHERE uid = ?`, uid).Scan(&pid) if errors.Is(err, sql.ErrNoRows) { return 0, false, nil } return pid, err == nil, err } // CountClaims counts claim rows (tombstones included) — /api/stats' redisKeys. func (c *Client) CountClaims() (int64, error) { if _, err := c.truthStore(); err != nil { return 0, err } ctx, cancel := opCtx() defer cancel() var n int64 err := c.truth.DB().QueryRowContext(ctx, `SELECT count(*) FROM claim`).Scan(&n) return n, err } func isLocalField(key string) bool { return key == "filePath" || key == "mtimeNano" || key == "sizeByte" } // claimKeyAllowed rejects keys that would corrupt the merged view's key // layout, and the removed per-algorithm CID shapes (cid_*, midhash256, // canonical_cid — METADATA_KEYS.md §14.13: write the cids/ key-set). func claimKeyAllowed(key string) bool { if strings.HasPrefix(key, "cid_") || key == "midhash256" || key == "canonical_cid" { return false } return !strings.HasPrefix(key, "/") && !strings.Contains(key, "\x00") && !isReservedFieldName(key) } // GetMetadataFlatAs is GetMetadataFlat seen by viewer: the merged record with // the viewer's own live text claims overlaid (their value wins in their view). func (c *Client) GetMetadataFlatAs(rid, viewer string) (map[string]string, error) { s, err := c.kv() if err != nil { return nil, err } ctx, cancel := opCtx() defer cancel() return c.readRecordAs(ctx, s, rid, viewer) } func (c *Client) readRecordAs(ctx context.Context, s *kv.Store, rid, viewer string) (map[string]string, error) { fields, err := readRecord(ctx, s, rid) if err != nil || viewer == "" { return fields, err } over, err := c.viewerOverrides(ctx, s, rid, viewer) if err != nil { return nil, err } if len(over) == 0 { return fields, nil } if fields == nil { fields = make(map[string]string, len(over)) } for k, v := range over { fields[k] = v } return fields, nil } // viewerOverrides returns the viewer's newest live text vote per key on the // cids of rid's group. func (c *Client) viewerOverrides(ctx context.Context, s *kv.Store, rid, viewer string) (map[string]string, error) { pid, ok, err := c.pidOf(viewer) if err != nil || !ok { return nil, err } cids, err := cidsForRoot(ctx, s, rid) if err != nil || len(cids) == 0 { return nil, err } out := map[string]string{} newest := map[string]int64{} for _, cid := range cids { rows, err := s.DB().QueryContext(ctx, `SELECT key FROM claim_by_pid WHERE pid = ? AND cid = ?`, pid, cid) if err != nil { return nil, err } var keys []string for rows.Next() { var k string if err := rows.Scan(&k); err != nil { rows.Close() return nil, err } keys = append(keys, k) } if err := rows.Close(); err != nil { return nil, err } for _, k := range keys { var raw any var ts int64 if err := c.truth.DB().QueryRowContext(ctx, `SELECT value, ts FROM claim WHERE cid = ? AND key = ? AND pid = ?`, cid, k, pid).Scan(&raw, &ts); err != nil { continue } if v := scanValue(raw); v.Kind == claim.KindText && ts > newest[k] { out[k], newest[k] = v.Text, ts } } } return out, nil } // viewerRIDs lists the records the viewer holds live claims on. func (c *Client) viewerRIDs(ctx context.Context, viewer string) ([]string, error) { pid, ok, err := c.pidOf(viewer) if err != nil || !ok { return nil, err } s := c.store rows, err := s.DB().QueryContext(ctx, `SELECT DISTINCT k.v FROM claim_by_pid b JOIN kv k ON k.k = 'cid:' || b.cid WHERE b.pid = ? LIMIT 10000`, pid) if err != nil { return nil, err } defer rows.Close() var out []string for rows.Next() { var rid string if err := rows.Scan(&rid); err != nil { return nil, err } out = append(out, rid) } return out, rows.Err() } // MergeClaims is the stateless POST /api/merge: verify each claim (invalid // ones are dropped and counted, never an error for the batch), merge per key // with the §7 rule, and overlay viewer's own votes — those in the request and // those this core holds (claim_by_pid) for the same cids. Returns the merged // fields, the agreeing-signer count per key, and how many claims were rejected. func (c *Client) MergeClaims(claims []claim.Claim, viewer string) (map[string]string, map[string]int, int, error) { byKey := map[string][]Vote{} rejected := 0 cids := map[string]bool{} for i := range claims { cl := &claims[i] if err := cl.Verify(); err != nil { rejected++ continue } cids[cl.CID] = true byKey[cl.Key] = append(byKey[cl.Key], Vote{Uid: cl.UID, Value: cl.Value, TS: cl.TS}) } fields := make(map[string]string, len(byKey)) nsrc := make(map[string]int, len(byKey)) for key, votes := range byKey { if res := Merge(votes, viewer); res.OK { fields[key], nsrc[key] = res.Value, res.Nsrc } } if viewer == "" { return fields, nsrc, rejected, nil } s, err := c.kv() if err != nil { return fields, nsrc, rejected, nil // no local core: request overlay only } if _, err := c.truthStore(); err != nil { return fields, nsrc, rejected, nil } ctx, cancel := opCtx() defer cancel() seen := map[string]bool{} for cid := range cids { rid, _, err := s.Get(ctx, CIDIndexPrefix+cid) if err != nil || rid == "" || seen[rid] { continue } seen[rid] = true over, err := c.viewerOverrides(ctx, s, rid, viewer) if err != nil { return nil, nil, rejected, err } for k, v := range over { fields[k], nsrc[k] = v, 1 } } return fields, nsrc, rejected, nil }