meta-core/tools/claims-migrate/migrate.go

package main

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"strings"
	"sync"
	"time"

	_ "modernc.org/sqlite"

	"github.com/metazla/meta-core/internal/cid"
	"github.com/metazla/meta-core/internal/claim"
	"github.com/metazla/meta-core/internal/identity"
	"github.com/metazla/meta-core/internal/storage"
)

type options struct {
	oldPath     string
	truthPath   string
	accountsDir string
	dryRun      bool
	now         time.Time // conversion timestamp; zero = time.Now()
	workers     int
	log         io.Writer
}

// counts is a labelled tally, printed sorted.
type counts map[string]int

func (c counts) total() int {
	n := 0
	for _, v := range c {
		n += v
	}
	return n
}

type stats struct {
	LegacyUID string
	ConvTS    uint64

	Rids          int // records migrated (have at least one cid)
	RidsNoClaims  int // migrated records whose only rows were cids / watcher facts
	RecordsNoCID  int // dropped: no cid at all
	FieldClaims   int
	AliasClaims   int
	WatcherFacts  counts // filePath / mtimeNano skipped
	RefusedFields counts // by reason
	Malformed     int    // file: keys without <rid>/<field>

	CIDGroupRows  int
	CIDConflicts  int // cids/<c> on a rid while cid:<c> names another rid
	CIDShared     int // cids/<c> (no cid: row) already kept by an earlier record
	RecordsFolded int // records whose every cid an earlier record kept: merged into it
	DupClaims     int // (cid, key) already claimed by the legacy signer (folded records)
	OrphanCIDRows int // cid:<c> → rid with no file: rows

	UDLMigrated  counts
	UDLDropped   counts
	UDLSigners   int
	GroupsMinted int // cid_group rows minted for UDL cids not in any record

	Scratch counts // service scratch keys dropped (btmagnet, searchcov, …)
	Derived counts // derived bookkeeping skipped (idx, ev, dup, udl:idx, …)

	Principals    int
	ClaimsWritten int
	Elapsed       time.Duration

	canon map[string]string // rid → the canonical cid its claims sit on
}

func newStats() *stats {
	return &stats{
		WatcherFacts: counts{}, RefusedFields: counts{},
		UDLMigrated: counts{}, UDLDropped: counts{},
		Scratch: counts{}, Derived: counts{},
		canon: map[string]string{},
	}
}

// job is one claim to sign with id.
type job struct {
	cl claim.Claim
	id *identity.Identity
}

const batchSize = 4096

func (o *options) logf(format string, args ...any) {
	if o.log != nil {
		fmt.Fprintf(o.log, format+"\n", args...)
	}
}

// migrate converts a meta-core v1.2.0 kv store into a v2.0.0 truth database.
func migrate(o options) (st *stats, err error) {
	started := time.Now()
	st = newStats()
	if o.workers <= 0 {
		o.workers = runtime.NumCPU()
	}
	if o.now.IsZero() {
		o.now = time.Now()
	}
	st.ConvTS = uint64(o.now.UnixMilli())

	old, err := openOld(o.oldPath)
	if err != nil {
		return nil, err
	}
	defer old.Close()

	legacy, err := identity.Generate() // in memory only, never saved
	if err != nil {
		return nil, err
	}
	st.LegacyUID = legacy.UID
	o.logf("legacy signer uid: %s (private key kept in memory only)", legacy.UID)

	var out *sql.DB
	if !o.dryRun {
		if out, err = createTruth(o.truthPath); err != nil {
			return nil, err
		}
		defer func() {
			if out != nil {
				out.Close()
			}
			if err != nil {
				removeDB(o.truthPath)
			}
		}()
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Pipeline: producer (this goroutine) → signers → one writer.
	jobs := make(chan []job, o.workers*2)
	signed := make(chan []claim.Claim, o.workers*2)
	var firstErr error
	var errMu sync.Mutex
	fail := func(e error) {
		errMu.Lock()
		if firstErr == nil {
			firstErr = e
		}
		errMu.Unlock()
		cancel()
	}
	var wg sync.WaitGroup
	for range o.workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for batch := range jobs {
				res := make([]claim.Claim, len(batch))
				for i := range batch {
					cl := batch[i].cl
					if e := cl.Sign(batch[i].id); e != nil {
						fail(fmt.Errorf("sign %s/%s: %w", cl.CID, cl.Key, e))
						return
					}
					if e := cl.Verify(); e != nil {
						fail(fmt.Errorf("verify %s/%s: %w", cl.CID, cl.Key, e))
						return
					}
					res[i] = cl
				}
				select {
				case signed <- res:
				case <-ctx.Done():
					return
				}
			}
		}()
	}
	go func() { wg.Wait(); close(signed) }()

	w := &writer{db: out, pids: map[string]int64{}}
	writerDone := make(chan error, 1)
	go func() {
		var e error
		for batch := range signed {
			if e == nil {
				if e = w.write(batch); e != nil {
					fail(e)
				}
			}
		}
		writerDone <- e
	}()

	var pending []job
	send := func(j job) error {
		pending = append(pending, j)
		if len(pending) < batchSize {
			return nil
		}
		b := pending
		pending = nil
		select {
		case jobs <- b:
			return nil
		case <-ctx.Done():
			return errors.New("aborted")
		}
	}

	groups, prodErr := produce(ctx, old, o, st, legacy, send)
	if prodErr == nil && len(pending) > 0 {
		select {
		case jobs <- pending:
		case <-ctx.Done():
			prodErr = errors.New("aborted")
		}
	}
	close(jobs)
	wErr := <-writerDone
	errMu.Lock()
	fe := firstErr
	errMu.Unlock()
	if fe != nil {
		return nil, fe
	}
	if prodErr != nil {
		return nil, prodErr
	}
	if wErr != nil {
		return nil, wErr
	}

	st.CIDGroupRows = len(groups)
	st.Principals = len(w.pids)
	st.ClaimsWritten = int(w.seq)
	st.DupClaims = w.dups
	if !o.dryRun {
		if err = w.finish(groups); err != nil {
			return nil, err
		}
		if _, err = out.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
			return nil, err
		}
		if err = out.Close(); err != nil {
			out = nil
			return nil, err
		}
		out = nil
		if err = checkTruth(o.truthPath); err != nil {
			return nil, fmt.Errorf("re-opening the result with meta-core's storage: %w", err)
		}
	}
	st.Elapsed = time.Since(started)
	return st, nil
}

// produce reads the old store and emits every claim; it returns cid_group.
func produce(ctx context.Context, old *sql.DB, o options, st *stats, legacy *identity.Identity, send func(job) error) (map[string]string, error) {
	// 1. cid index.
	index := map[string]string{}
	ridIdx := map[string][]string{}
	rows, err := old.QueryContext(ctx, `SELECT k, v FROM kv WHERE k >= 'cid:' AND k < 'cid;'`)
	if err != nil {
		return nil, err
	}
	for rows.Next() {
		var k, v string
		if err := rows.Scan(&k, &v); err != nil {
			rows.Close()
			return nil, err
		}
		c := k[len("cid:"):]
		if c == "" || v == "" {
			continue
		}
		index[c] = v
		ridIdx[v] = append(ridIdx[v], c)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	o.logf("cid index: %d rows", len(index))

	groups := map[string]string{}

	// 2. records, one rid at a time (the file: range is contiguous per rid).
	rows, err = old.QueryContext(ctx, `SELECT k, v FROM kv WHERE k >= 'file:' AND k < 'file;' ORDER BY k`)
	if err != nil {
		return nil, err
	}
	seen := map[string]bool{}
	cur := ""
	var fields [][2]string
	flush := func() error {
		if cur == "" {
			return nil
		}
		if seen[cur] {
			return fmt.Errorf("file: range for %s is not contiguous", cur)
		}
		seen[cur] = true
		err := emitRecord(o, st, legacy, index, ridIdx, groups, cur, fields, send)
		fields = fields[:0]
		return err
	}
	n := 0
	for rows.Next() {
		var k, v string
		if err := rows.Scan(&k, &v); err != nil {
			rows.Close()
			return nil, err
		}
		rid, field, ok := strings.Cut(k[len("file:"):], "/")
		if !ok || rid == "" || field == "" {
			st.Malformed++
			continue
		}
		if rid != cur {
			if err := flush(); err != nil {
				rows.Close()
				return nil, err
			}
			cur = rid
		}
		fields = append(fields, [2]string{field, v})
		if n++; n%1_000_000 == 0 {
			o.logf("  … %d file: rows, %d records", n, len(seen))
		}
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return nil, err
	}
	rows.Close()
	if err := flush(); err != nil {
		return nil, err
	}
	for c := range index {
		if _, ok := groups[c]; !ok {
			st.OrphanCIDRows++
		}
	}
	o.logf("records: %d migrated, %d without a cid dropped", st.Rids, st.RecordsNoCID)

	// 3. UDL cells.
	if err := emitUDL(ctx, old, o, st, groups, send); err != nil {
		return nil, err
	}

	// 4. everything else: count and drop.
	rows, err = old.QueryContext(ctx, `SELECT k FROM kv
		WHERE NOT (k >= 'file:' AND k < 'file;') AND NOT (k >= 'cid:' AND k < 'cid;')
		  AND NOT (k >= 'udl:rec:' AND k < 'udl:rec;')`)
	if err != nil {
		return nil, err
	}
	for rows.Next() {
		var k string
		if err := rows.Scan(&k); err != nil {
			rows.Close()
			return nil, err
		}
		classifyOther(st, k)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	return groups, nil
}

func classifyOther(st *stats, k string) {
	for _, p := range []string{"idx:", "ev:", "evhwm:", "dup:", "udl:idx:", "meta-core:", "meta:"} {
		if strings.HasPrefix(k, p) {
			st.Derived[p+"*"]++
			return
		}
	}
	if p, _, ok := strings.Cut(k, ":"); ok && p != "" {
		st.Scratch[p+":*"]++
		return
	}
	st.Scratch["(no prefix)"]++
}

// refusedReason mirrors storage.claimKeyAllowed + the write path's length
// checks: keys meta-core's POST /api/claims would refuse.
func refusedReason(key string) string {
	switch {
	case strings.HasPrefix(key, "cid_"):
		return "cid_*"
	case key == "midhash256", key == "canonical_cid", key == "__fields__":
		return key
	case strings.HasPrefix(key, "/"), strings.Contains(key, "\x00"):
		return "malformed key"
	case len(key) > 4096:
		return "key too long"
	}
	return ""
}

func emitRecord(o options, st *stats, legacy *identity.Identity, index map[string]string,
	ridIdx map[string][]string, groups map[string]string, rid string, fields [][2]string, send func(job) error) error {
	// Members: cids/* the cid index does not assign elsewhere, plus every cid
	// the index assigns here.
	member := map[string]bool{}
	var members []string
	for _, f := range fields {
		c, ok := strings.CutPrefix(f[0], storage.CIDsKeyPrefix)
		if !ok || c == "" {
			continue
		}
		if r, ok := index[c]; ok && r != rid {
			st.CIDConflicts++
			continue
		}
		if !member[c] {
			member[c] = true
			members = append(members, c)
		}
	}
	for _, c := range ridIdx[rid] {
		if !member[c] {
			member[c] = true
			members = append(members, c)
		}
	}
	var usable []string
	for _, c := range members {
		if len(c) <= 512 {
			usable = append(usable, c)
		}
	}
	if len(usable) == 0 {
		st.RecordsNoCID++
		return nil
	}
	// A cid with no cid: index row can sit in two records' cids/* (the same
	// content stored twice). The first record keeps it; a record left with none
	// folds into the record owning its first shared cid, and its fields become
	// claims there (the legacy signer's first claim per (cid, key) wins).
	owner, shared := "", ""
	var own []string
	for _, c := range usable {
		if r, ok := groups[c]; ok && r != rid {
			st.CIDShared++
			if owner == "" {
				owner, shared = r, c
			}
			continue
		}
		own = append(own, c)
	}
	var canonical string
	if len(own) == 0 {
		st.RecordsFolded++
		rid = owner
		if canonical = st.canon[owner]; canonical == "" {
			canonical = shared
		}
	} else {
		canonical = own[0]
		for _, c := range own[1:] {
			canonical = cid.Better(canonical, c)
		}
		for _, c := range own {
			groups[c] = rid
		}
		st.canon[rid] = canonical
		st.Rids++
	}
	emitted := 0
	mk := func(key, value string) error {
		emitted++
		return send(job{id: legacy, cl: claim.Claim{
			CID: canonical, UID: legacy.UID, Key: key, Value: claim.Text(value), TS: st.ConvTS,
		}})
	}
	for _, f := range fields {
		key, val := f[0], f[1]
		if c, ok := strings.CutPrefix(key, storage.CIDsKeyPrefix); ok {
			if c != canonical && member[c] && len(c) <= 512 && groups[c] == rid {
				st.AliasClaims++
				if err := mk(key, "true"); err != nil {
					return err
				}
			}
			continue
		}
		if key == "filePath" || key == "mtimeNano" {
			st.WatcherFacts[key]++
			continue
		}
		if r := refusedReason(key); r != "" {
			st.RefusedFields[r]++
			continue
		}
		st.FieldClaims++
		if err := mk(key, val); err != nil {
			return err
		}
	}
	if emitted == 0 {
		st.RidsNoClaims++
	}
	return nil
}

// publicUDLKeys are the plaintext tier (meta-watch record/tiers.rs); every
// other key is private and travels as an envelope.
var publicUDLKeys = map[string]bool{"like": true, "userRating": true, "profile:name": true, "profile:avatar": true}

type udlCellJSON struct {
	Record  string `json:"record"`
	Version int64  `json:"version"`
	Ts      int64  `json:"ts"`
}

func decodeB64(s string) ([]byte, error) {
	for _, enc := range []*base64.Encoding{base64.StdEncoding, base64.RawStdEncoding, base64.URLEncoding, base64.RawURLEncoding} {
		if b, err := enc.DecodeString(s); err == nil {
			return b, nil
		}
	}
	return nil, errors.New("record is not base64")
}

// udlTS converts an MMUR1 ts (seconds; tolerate ms) to MMUR2 milliseconds.
func udlTS(ts int64, fallback uint64) uint64 {
	switch {
	case ts <= 0:
		return fallback
	case ts < 100_000_000_000: // seconds (before year 5138)
		return uint64(ts) * 1000
	default:
		return uint64(ts)
	}
}

type account struct {
	id      *identity.Identity
	aeadKey []byte
}

func emitUDL(ctx context.Context, old *sql.DB, o options, st *stats, groups map[string]string, send func(job) error) error {
	accounts := map[string]*account{}
	loadAccount := func(uid string) (*account, error) {
		if a, ok := accounts[uid]; ok {
			return a, nil
		}
		var a *account
		if o.accountsDir != "" && identity.ValidUID(uid) {
			id, err := identity.LoadByUID(o.accountsDir, uid)
			if err != nil {
				return nil, fmt.Errorf("account %s: %w", uid, err)
			}
			if id != nil {
				if id.UID != uid {
					return nil, fmt.Errorf("account file %s.json holds uid %s", uid, id.UID)
				}
				k, err := identity.DeriveAEADKey(id)
				if err != nil {
					return nil, err
				}
				a = &account{id: id, aeadKey: k}
			}
		}
		accounts[uid] = a
		return a, nil
	}

	type cellKey struct{ uid, cid, key string }
	best := map[cellKey]job{}
	var order []cellKey
	drop := func(reason string) { st.UDLDropped[reason]++ }
	kinds := map[cellKey]string{}

	rows, err := old.QueryContext(ctx, `SELECT k, v FROM kv WHERE k >= 'udl:rec:' AND k < 'udl:rec;' ORDER BY k`)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var k, v string
		if err := rows.Scan(&k, &v); err != nil {
			return err
		}
		parts := strings.SplitN(k[len("udl:rec:"):], "/", 3)
		if len(parts) != 3 {
			drop("malformed cell key")
			continue
		}
		var cell udlCellJSON
		if err := json.Unmarshal([]byte(v), &cell); err != nil || cell.Record == "" {
			drop("malformed cell JSON")
			continue
		}
		raw, err := decodeB64(cell.Record)
		if err != nil {
			drop("record not base64")
			continue
		}
		rec, err := decodeMMUR1(raw)
		if err != nil {
			drop("record not MMUR1 CBOR")
			continue
		}
		if rec.UID != parts[0] || rec.CID != parts[1] || rec.Key != parts[2] {
			drop("record does not match its cell")
			continue
		}
		if !identity.ValidUID(rec.UID) || identity.Verify(rec.UID, rec.Signed, rec.Sig) != nil {
			drop("bad MMUR1 signature")
			continue
		}
		acct, err := loadAccount(rec.UID)
		if err != nil {
			return err
		}
		if acct == nil {
			drop("no account on this box")
			continue
		}
		key := rec.Key
		if key == "rating" {
			key = "userRating" // §14: a user's rating is userRating
		}
		if r := refusedReason(key); r != "" || rec.CID == "" || len(rec.CID) > 512 {
			drop("key or cid refused by meta-core")
			continue
		}
		ts := udlTS(rec.TS, st.ConvTS)
		cl := claim.Claim{CID: rec.CID, UID: rec.UID, Key: key, TS: ts}
		kind := ""
		switch {
		case rec.Value.kind == cNull:
			kind = "tombstone"
		case publicUDLKeys[key]:
			t, ok := rec.Value.wireText()
			if !ok {
				drop("public key holds bytes")
				continue
			}
			cl.Value, kind = claim.Text(t), "public"
		default:
			var plain []byte
			if rec.Value.kind == cBytes {
				plain, err = openEnvelope(acct.aeadKey, rec.Value.bytes, envelopeAAD(rec.CID, rec.UID, rec.Key, rec.Version))
				if err != nil {
					drop("private envelope does not decrypt")
					continue
				}
				kind = "private (re-encrypted)"
			} else {
				// A private-tier key stored in plaintext: encrypt it now, its
				// plaintext being the CBOR of the typed value, as crypto.rs.
				plain = append([]byte(nil), rec.Value.raw...)
				kind = "private (plaintext, now encrypted)"
			}
			nonce := make([]byte, 24)
			if _, err := rand.Read(nonce); err != nil {
				return err
			}
			env, err := sealEnvelope(acct.aeadKey, plain, envelopeAAD(rec.CID, rec.UID, key, ts), nonce)
			if err != nil {
				return err
			}
			cl.Value = claim.Bytes(env)
		}
		ck := cellKey{rec.UID, rec.CID, key}
		if prev, ok := best[ck]; ok {
			drop("superseded (rating + userRating)")
			if prev.cl.TS >= ts {
				continue
			}
		} else {
			order = append(order, ck)
		}
		best[ck], kinds[ck] = job{cl: cl, id: acct.id}, kind
	}
	if err := rows.Err(); err != nil {
		return err
	}
	signers := map[string]bool{}
	for _, ck := range order {
		j := best[ck]
		if _, ok := groups[j.cl.CID]; !ok {
			rid, err := storage.NewUUID()
			if err != nil {
				return err
			}
			groups[j.cl.CID] = rid
			st.GroupsMinted++
		}
		st.UDLMigrated[kinds[ck]]++
		signers[j.cl.UID] = true
		if err := send(j); err != nil {
			return err
		}
	}
	st.UDLSigners = len(signers)
	return nil
}

// writer stores signed claims in arrival order (seq = arrival order).
type writer struct {
	db   *sql.DB
	pids map[string]int64
	seq  int64
	dups int
}

func (w *writer) write(batch []claim.Claim) error {
	if w.db == nil { // dry run
		for _, cl := range batch {
			if _, ok := w.pids[cl.UID]; !ok {
				w.pids[cl.UID] = int64(len(w.pids) + 1)
			}
		}
		w.seq += int64(len(batch))
		return nil
	}
	tx, err := w.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	ins, err := tx.Prepare(`INSERT OR IGNORE INTO claim(cid, key, pid, value, ts, sig, seq) VALUES(?, ?, ?, ?, ?, ?, ?)`)
	if err != nil {
		return err
	}
	defer ins.Close()
	for _, cl := range batch {
		pid, ok := w.pids[cl.UID]
		if !ok {
			if err := tx.QueryRow(`INSERT INTO principal(uid) VALUES(?) RETURNING pid`, cl.UID).Scan(&pid); err != nil {
				return fmt.Errorf("principal %s: %w", cl.UID, err)
			}
			w.pids[cl.UID] = pid
		}
		w.seq++
		res, err := ins.Exec(cl.CID, cl.Key, pid, valueArg(cl.Value), int64(cl.TS), cl.Sig, w.seq)
		if err != nil {
			return fmt.Errorf("claim (%s, %s, %s): %w", cl.CID, cl.Key, cl.UID, err)
		}
		if n, _ := res.RowsAffected(); n == 0 { // a folded record's duplicate
			w.seq--
			w.dups++
		}
	}
	return tx.Commit()
}

// finish writes cid_group and last_seq.
func (w *writer) finish(groups map[string]string) error {
	tx, err := w.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	ins, err := tx.Prepare(`INSERT INTO cid_group(cid, rid) VALUES(?, ?)`)
	if err != nil {
		return err
	}
	defer ins.Close()
	keys := make([]string, 0, len(groups))
	for c := range groups {
		keys = append(keys, c)
	}
	sort.Strings(keys)
	for _, c := range keys {
		if _, err := ins.Exec(c, groups[c]); err != nil {
			return fmt.Errorf("cid_group %s: %w", c, err)
		}
	}
	if _, err := tx.Exec(`INSERT INTO meta(name, value) VALUES('last_seq', ?)
		ON CONFLICT(name) DO UPDATE SET value = excluded.value`, fmt.Sprint(w.seq)); err != nil {
		return err
	}
	return tx.Commit()
}

// valueArg is storage's column encoding: TEXT, BLOB or NULL (STRICT ANY keeps it).
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
}

func openOld(path string) (*sql.DB, error) {
	if _, err := os.Stat(path); err != nil {
		return nil, fmt.Errorf("old store: %w", err)
	}
	db, err := sql.Open("sqlite", "file:"+path+"?mode=ro&_pragma=query_only(1)&_pragma=busy_timeout(10000)")
	if err != nil {
		return nil, err
	}
	var name string
	if err := db.QueryRow(`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'kv'`).Scan(&name); err != nil {
		db.Close()
		return nil, fmt.Errorf("%s has no kv table — not a meta-core v1.x store: %w", path, err)
	}
	var ver string
	err = db.QueryRow(`SELECT v FROM kv WHERE k = 'meta-core:schema-version'`).Scan(&ver)
	switch {
	case errors.Is(err, sql.ErrNoRows):
		// pre-sentinel store: accepted, the key layout is the same
	case err != nil:
		db.Close()
		return nil, err
	case ver != "2":
		db.Close()
		return nil, fmt.Errorf("%s has schema version %q, this tool converts version \"2\"", path, ver)
	}
	return db, nil
}

// createTruth creates the truth file with meta-core's own schema code
// (storage.Client.Connect + EnsureSchemaVersion stamp schema-version 3), then
// reopens it for the bulk load.
func createTruth(path string) (*sql.DB, error) {
	for _, p := range []string{path, path + "-wal", path + "-shm"} {
		if _, err := os.Stat(p); err == nil {
			return nil, fmt.Errorf("%s already exists; refusing to overwrite", p)
		}
	}
	if fi, err := os.Stat(filepath.Dir(path)); err != nil || !fi.IsDir() {
		return nil, fmt.Errorf("output directory %s does not exist", filepath.Dir(path))
	}
	tmp, err := os.MkdirTemp("", "claims-migrate-cache-")
	if err != nil {
		return nil, err
	}
	defer os.RemoveAll(tmp)
	c := storage.NewClient()
	if err := c.Connect(path, filepath.Join(tmp, "cache.db")); err != nil {
		removeDB(path)
		return nil, err
	}
	if err := c.EnsureSchemaVersion(storage.SchemaVersion); err != nil {
		c.Close()
		removeDB(path)
		return nil, err
	}
	if err := c.Close(); err != nil {
		removeDB(path)
		return nil, err
	}
	db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=synchronous(OFF)&_pragma=busy_timeout(30000)&_pragma=cache_size(-262144)")
	if err != nil {
		removeDB(path)
		return nil, err
	}
	db.SetMaxOpenConns(1)
	return db, nil
}

// checkTruth reopens the result the way meta-core boots it (schema gate).
func checkTruth(path string) error {
	tmp, err := os.MkdirTemp("", "claims-migrate-check-")
	if err != nil {
		return err
	}
	defer os.RemoveAll(tmp)
	c := storage.NewClient()
	if err := c.Connect(path, filepath.Join(tmp, "cache.db")); err != nil {
		return err
	}
	defer c.Close()
	return c.EnsureSchemaVersion(storage.SchemaVersion)
}

func removeDB(path string) {
	for _, p := range []string{path, path + "-wal", path + "-shm"} {
		_ = os.Remove(p)
	}
}

// printSummary writes the human report.
func printSummary(w io.Writer, st *stats, dryRun bool) {
	mode := "written"
	if dryRun {
		mode = "would be written (dry run: signed + verified, nothing stored)"
	}
	fmt.Fprintf(w, "\n=== claims-migrate summary ===\n")
	fmt.Fprintf(w, "legacy signer uid        %s\n", st.LegacyUID)
	fmt.Fprintf(w, "conversion ts (ms)       %d\n", st.ConvTS)
	fmt.Fprintf(w, "records (rids) migrated  %d  (%d with no claims: only cids / watcher facts)\n", st.Rids, st.RidsNoClaims)
	fmt.Fprintf(w, "records dropped (no cid) %d\n", st.RecordsNoCID)
	fmt.Fprintf(w, "claims %-17s %d\n", "total", st.ClaimsWritten)
	fmt.Fprintf(w, "  legacy field claims    %d\n", st.FieldClaims)
	fmt.Fprintf(w, "  legacy alias claims    %d  (cids/<other> = \"true\")\n", st.AliasClaims)
	fmt.Fprintf(w, "  UDL claims             %d  (%d signers)\n", st.UDLMigrated.total(), st.UDLSigners)
	printCounts(w, "    ", st.UDLMigrated)
	fmt.Fprintf(w, "  -> %s\n", mode)
	fmt.Fprintf(w, "principals               %d\n", st.Principals)
	fmt.Fprintf(w, "cid_group rows           %d  (%d minted for UDL-only cids)\n", st.CIDGroupRows, st.GroupsMinted)
	fmt.Fprintf(w, "cid conflicts            %d  (cids/<c> on a rid while cid:<c> names another; index wins)\n", st.CIDConflicts)
	fmt.Fprintf(w, "shared cids              %d  (cids/<c> kept by an earlier record; %d records folded into it)\n", st.CIDShared, st.RecordsFolded)
	fmt.Fprintf(w, "duplicate claims skipped %d  (folded records' (cid, key) already claimed; not counted in dry run)\n", st.DupClaims)
	fmt.Fprintf(w, "orphan cid: rows dropped %d  (rid has no file: rows)\n", st.OrphanCIDRows)
	fmt.Fprintf(w, "watcher facts skipped    %d\n", st.WatcherFacts.total())
	printCounts(w, "    ", st.WatcherFacts)
	fmt.Fprintf(w, "fields refused           %d\n", st.RefusedFields.total())
	printCounts(w, "    ", st.RefusedFields)
	fmt.Fprintf(w, "malformed file: keys     %d\n", st.Malformed)
	fmt.Fprintf(w, "UDL cells dropped        %d\n", st.UDLDropped.total())
	printCounts(w, "    ", st.UDLDropped)
	fmt.Fprintf(w, "scratch keys dropped     %d  (refetched / rebuilt in the cache)\n", st.Scratch.total())
	printCounts(w, "    ", st.Scratch)
	fmt.Fprintf(w, "derived keys skipped     %d\n", st.Derived.total())
	printCounts(w, "    ", st.Derived)
	fmt.Fprintf(w, "elapsed                  %s\n", st.Elapsed.Round(time.Millisecond))
}

func printCounts(w io.Writer, indent string, c counts) {
	keys := make([]string, 0, len(c))
	for k := range c {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		fmt.Fprintf(w, "%s%-40s %d\n", indent, k, c[k])
	}
}