meta-core/internal/storage/apply.go

package storage

// The apply loop: folds committed truth changes into the cache.
//
// The cache's merged view keeps the v1 record layout — `file:<rid>/<field>`
// rows of the cache kv table — so every reader (record reads, search, the
// field index, FTS, counters, SSE and its snapshot) is unchanged; those rows
// are now derived, never written by a client. Each committed change re-merges
// only the (rid, key) it touched, from the truth votes of every cid in the
// group. Group membership is synthesised: every cid of a group appears as
// `file:<rid>/cids/<cid> = "true"` plus the `cid:<cid> → rid` lookup, whatever
// the cids/* claims say.
//
// Changes arrive in commit order (kv.Tx.OnCommit runs in the truth writer), so
// `meta:applied_seq` only ever covers a prefix of the truth log. A crash
// between the truth commit and the cache apply leaves the cache behind, never
// wrong; Connect catches up from truth.

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"

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

const (
	appliedSeqKey = "meta:applied_seq"
	// rebuiltAtKey holds the stream id at which the cache was last rebuilt.
	// An SSE cursor older than it gets `gap`: the rebuild emitted no
	// per-key events, so the consumer must take a snapshot.
	rebuiltAtKey = "meta:rebuilt_at"
)

var localFields = []string{"filePath", "sizeByte", "mtimeNano"}

type applier struct {
	c *Client

	mu      sync.Mutex
	queue   []change
	applied int64
	notify  chan struct{}

	wake chan struct{}
	stop chan struct{}
	done chan struct{}
}

func newApplier(c *Client, applied int64) *applier {
	a := &applier{
		c: c, applied: applied,
		notify: make(chan struct{}),
		wake:   make(chan struct{}, 1),
		stop:   make(chan struct{}),
		done:   make(chan struct{}),
	}
	go a.run()
	return a
}

func (a *applier) enqueue(ch []change) {
	a.mu.Lock()
	a.queue = append(a.queue, ch...)
	a.mu.Unlock()
	select {
	case a.wake <- struct{}{}:
	default:
	}
}

// wait blocks until the cache has applied every change up to seq.
func (a *applier) wait(ctx context.Context, seq int64) error {
	for {
		a.mu.Lock()
		if a.applied >= seq {
			a.mu.Unlock()
			return nil
		}
		n := a.notify
		a.mu.Unlock()
		select {
		case <-n:
		case <-ctx.Done():
			return ctx.Err()
		case <-a.stop:
			return errors.New("storage closed before the cache applied the write")
		}
	}
}

func (a *applier) close() {
	close(a.stop)
	<-a.done
}

const applyBatchMax = 2000

func (a *applier) run() {
	defer close(a.done)
	for {
		select {
		case <-a.wake:
		case <-a.stop:
			return
		}
		for {
			a.mu.Lock()
			n := min(len(a.queue), applyBatchMax)
			batch := append([]change(nil), a.queue[:n]...)
			a.mu.Unlock()
			if len(batch) == 0 {
				break
			}
			if err := a.c.applyBatch(batch); err != nil {
				log.Printf("[Storage] apply of %d changes failed, retrying: %v", len(batch), err)
				select {
				case <-time.After(time.Second):
					continue
				case <-a.stop:
					return
				}
			}
			a.mu.Lock()
			a.queue = a.queue[n:]
			if s := batch[len(batch)-1].seq; s > a.applied {
				a.applied = s
			}
			close(a.notify)
			a.notify = make(chan struct{})
			a.mu.Unlock()
		}
	}
}

// applyBatch folds committed changes into the cache in one transaction.
func (c *Client) applyBatch(batch []change) error {
	s, err := c.kv()
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
	defer cancel()
	return s.Update(ctx, func(tx *kv.Tx) error {
		return c.applyTx(tx, batch)
	})
}

func (c *Client) applyTx(tx *kv.Tx, batch []change) error {
	ctx := tx.Context()
	marks := map[string]map[string]bool{}
	mark := func(rid, key string) {
		if marks[rid] == nil {
			marks[rid] = map[string]bool{}
		}
		marks[rid][key] = true
	}
	all := map[string]bool{}
	dirty := map[string]bool{}
	var maxSeq int64
	for _, it := range batch {
		maxSeq = max(maxSeq, it.seq)
		if it.regroup != nil {
			loser, survivor := it.regroup[0], it.regroup[1]
			if _, err := tx.DeletePrefix(recordPrefix(loser), 0); err != nil {
				return err
			}
			if _, err := tx.Delete(rootKey(loser)); err != nil {
				return err
			}
			if err := deleteSearchRowTx(tx, loser); err != nil {
				return err
			}
			delete(marks, loser)
			cids, err := c.truthMembers(ctx, survivor)
			if err != nil {
				return err
			}
			for _, cid := range cids {
				if _, err := syncMember(tx, cid, survivor); err != nil {
					return err
				}
			}
			all[survivor], dirty[survivor] = true, true
			continue
		}
		rid, err := c.truthRid(ctx, it.cid)
		if err != nil {
			return err
		}
		if rid == "" {
			continue
		}
		newRoot, err := syncMember(tx, it.cid, rid)
		if err != nil {
			return err
		}
		if newRoot {
			dirty[rid] = true
		}
		if it.key == "" {
			continue
		}
		if it.live {
			_, err = tx.Exec(`INSERT OR IGNORE INTO claim_by_pid(pid, cid, key) VALUES(?, ?, ?)`, it.pid, it.cid, it.key)
		} else {
			_, err = tx.Exec(`DELETE FROM claim_by_pid WHERE pid = ? AND cid = ? AND key = ?`, it.pid, it.cid, it.key)
		}
		if err != nil {
			return err
		}
		mark(rid, it.key)
	}
	for rid := range all {
		cids, err := cidsForRoot(ctx, tx, rid)
		if err != nil {
			return err
		}
		keys, err := c.claimKeys(ctx, cids)
		if err != nil {
			return err
		}
		for _, k := range append(keys, localFields...) {
			mark(rid, k)
		}
	}
	for rid, keys := range marks {
		cids, err := cidsForRoot(ctx, tx, rid)
		if err != nil {
			return err
		}
		for key := range keys {
			changed, err := c.remergeTx(tx, rid, cids, key)
			if err != nil {
				return err
			}
			if changed && affectsHaystack(key) {
				dirty[rid] = true
			}
		}
	}
	for rid := range dirty {
		if err := refreshSearchTx(tx, rid); err != nil {
			return err
		}
	}
	if maxSeq > 0 {
		return tx.Put(appliedSeqKey, strconv.FormatInt(maxSeq, 10))
	}
	return nil
}

// syncMember makes cid a member of rid in the cache: the lookup row, the
// synthesised key-set member, and the root. Reports whether rid is new.
func syncMember(tx *kv.Tx, cid, rid string) (bool, error) {
	cur, _, err := tx.Get(tx.Context(), CIDIndexPrefix+cid)
	if err != nil {
		return false, err
	}
	if cur != rid {
		if cur != "" {
			if _, err := tx.Delete(recordPrefix(cur) + CIDsKeyPrefix + cid); err != nil {
				return false, err
			}
		}
		if err := tx.Put(CIDIndexPrefix+cid, rid); err != nil {
			return false, err
		}
	}
	if err := tx.Put(recordPrefix(rid)+CIDsKeyPrefix+cid, "true"); err != nil {
		return false, err
	}
	return tx.Set(rootKey(rid), "")
}

// remergeTx recomputes file:<rid>/<key> from the group's votes (plus the
// box-local facts for filePath / sizeByte / mtimeNano) and reports whether
// the merged row changed.
func (c *Client) remergeTx(tx *kv.Tx, rid string, cids []string, key string) (bool, error) {
	if strings.HasPrefix(key, CIDsKeyPrefix) {
		return false, nil // membership is synthesised, never merged
	}
	votes, err := c.votesFor(tx.Context(), cids, key)
	if err != nil {
		return false, err
	}
	res := Merge(votes, "")
	val, ok := res.Value, res.OK
	if isLocalField(key) {
		lv, lok, err := localFact(tx, cids, key)
		if err != nil {
			return false, err
		}
		// The disk is the truth for where and when a file is; a signed
		// sizeByte (a content fact) wins over the on-disk size.
		if lok && (key != "sizeByte" || !ok) {
			val, ok = lv, true
		}
	}
	k := recordPrefix(rid) + key
	if ok {
		return tx.Set(k, val)
	}
	return tx.Delete(k)
}

// localFact reads a box-local field from local_file for the group's cids.
func localFact(tx *kv.Tx, cids []string, key string) (string, bool, error) {
	if len(cids) == 0 {
		return "", false, nil
	}
	args := make([]any, len(cids))
	for i, c := range cids {
		args[i] = c
	}
	var path string
	var size, mtime int64
	err := tx.QueryRow(`SELECT path, size, mtime FROM local_file WHERE midhash IN (?`+
		strings.Repeat(",?", len(cids)-1)+`) ORDER BY path LIMIT 1`, args...).Scan(&path, &size, &mtime)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}
	switch key {
	case "filePath":
		return path, true, nil
	case "sizeByte":
		return strconv.FormatInt(size, 10), true, nil
	default:
		return strconv.FormatInt(mtime, 10), true, nil
	}
}

// truthMembers lists every cid of rid's group (scans cid_group: truth has no
// rid index; only regroups and rebuilds call this).
func (c *Client) truthMembers(ctx context.Context, rid string) ([]string, error) {
	rows, err := c.truth.DB().QueryContext(ctx, `SELECT cid FROM cid_group WHERE rid = ?`, rid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var cid string
		if err := rows.Scan(&cid); err != nil {
			return nil, err
		}
		out = append(out, cid)
	}
	return out, rows.Err()
}

// truthLastSeq reads the newest committed seq.
func (c *Client) truthLastSeq(ctx context.Context) (int64, error) {
	var v string
	err := c.truth.DB().QueryRowContext(ctx, `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)
}

// catchUp brings the cache level with truth at boot: nothing to do, a replay
// of the changes after applied_seq (found by a full scan — there is no seq
// index, and this only runs after an unclean shutdown), or a full rebuild
// when the cache is missing or a group changed while it was behind.
func (c *Client) catchUp() (int64, error) {
	s := c.store
	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
	defer cancel()
	last, err := c.truthLastSeq(ctx)
	if err != nil {
		return 0, err
	}
	appliedStr, has, err := s.Get(ctx, appliedSeqKey)
	if err != nil {
		return 0, err
	}
	applied, _ := strconv.ParseInt(appliedStr, 10, 64)
	if has && applied >= last {
		return applied, nil
	}
	if !has {
		if last == 0 {
			return 0, nil // fresh store
		}
		log.Printf("[Storage] cache is empty, rebuilding from truth (last_seq %d)", last)
		return last, c.rebuild(ctx, last)
	}
	rows, err := c.truth.DB().QueryContext(ctx,
		`SELECT cid, key, pid, value IS NOT NULL, seq FROM claim WHERE seq > ? ORDER BY seq`, applied)
	if err != nil {
		return 0, err
	}
	var missed []change
	regroup := false
	for rows.Next() {
		var ch change
		if err := rows.Scan(&ch.cid, &ch.key, &ch.pid, &ch.live, &ch.seq); err != nil {
			rows.Close()
			return 0, err
		}
		if strings.HasPrefix(ch.key, CIDsKeyPrefix) {
			regroup = true
		}
		missed = append(missed, ch)
	}
	if err := rows.Close(); err != nil {
		return 0, err
	}
	if regroup {
		log.Printf("[Storage] cache behind by %d changes including a group link, rebuilding", len(missed))
		return last, c.rebuild(ctx, last)
	}
	log.Printf("[Storage] cache behind (applied %d < last %d), replaying %d changes", applied, last, len(missed))
	for start := 0; start < len(missed); start += applyBatchMax {
		end := min(start+applyBatchMax, len(missed))
		if err := c.applyBatch(missed[start:end]); err != nil {
			return 0, err
		}
	}
	if err := s.Update(ctx, func(tx *kv.Tx) error { return tx.Put(appliedSeqKey, strconv.FormatInt(last, 10)) }); err != nil {
		return 0, err
	}
	return last, nil
}

// Rebuild drops every derived row of the cache and recomputes it from truth.
// Exposed for tests and operators; boot calls it when the cache is missing.
func (c *Client) Rebuild() error {
	if _, err := c.truthStore(); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
	defer cancel()
	last, err := c.truthLastSeq(ctx)
	if err != nil {
		return err
	}
	return c.rebuild(ctx, last)
}

const rebuildBatch = 500

func (c *Client) rebuild(ctx context.Context, last int64) error {
	s := c.store
	started := time.Now()
	for _, prefix := range []string{recordNS, CIDIndexPrefix, rootsNS, ftsNS} {
		for {
			var n int64
			if err := s.Update(ctx, func(tx *kv.Tx) error {
				tx.Quiet()
				var err error
				n, err = tx.DeletePrefix(prefix, bulkBatch)
				return err
			}); err != nil {
				return fmt.Errorf("clear %s: %w", prefix, err)
			}
			if n < bulkBatch {
				break
			}
		}
	}
	if err := s.Update(ctx, func(tx *kv.Tx) error {
		if _, err := tx.Exec(`DELETE FROM fts`); err != nil {
			return err
		}
		_, err := tx.Exec(`DELETE FROM claim_by_pid`)
		return err
	}); err != nil {
		return err
	}

	groups := map[string][]string{}
	var order []string
	rows, err := c.truth.DB().QueryContext(ctx, `SELECT cid, rid FROM cid_group`)
	if err != nil {
		return err
	}
	for rows.Next() {
		var cid, rid string
		if err := rows.Scan(&cid, &rid); err != nil {
			rows.Close()
			return err
		}
		if _, ok := groups[rid]; !ok {
			order = append(order, rid)
		}
		groups[rid] = append(groups[rid], cid)
	}
	if err := rows.Close(); err != nil {
		return err
	}
	for start := 0; start < len(order); start += rebuildBatch {
		end := min(start+rebuildBatch, len(order))
		if err := s.Update(ctx, func(tx *kv.Tx) error {
			tx.Quiet()
			for _, rid := range order[start:end] {
				cids := groups[rid]
				for _, cid := range cids {
					if _, err := syncMember(tx, cid, rid); err != nil {
						return err
					}
				}
				keys, err := c.claimKeys(ctx, cids)
				if err != nil {
					return err
				}
				for _, key := range append(keys, localFields...) {
					if _, err := c.remergeTx(tx, rid, cids, key); err != nil {
						return err
					}
				}
				if err := refreshSearchTx(tx, rid); err != nil {
					return err
				}
			}
			return nil
		}); err != nil {
			return fmt.Errorf("rebuild records: %w", err)
		}
	}
	if err := c.rebuildClaimIndex(ctx); err != nil {
		return err
	}
	if err := c.RebuildCounters(); err != nil {
		return err
	}
	rebuiltAt := strconv.FormatInt(time.Now().UnixMilli(), 10) + "-0"
	if err := s.Update(ctx, func(tx *kv.Tx) error {
		if err := tx.Put(rebuiltAtKey, rebuiltAt); err != nil {
			return err
		}
		return tx.Put(appliedSeqKey, strconv.FormatInt(last, 10))
	}); err != nil {
		return err
	}
	log.Printf("[Storage] cache rebuilt: %d records in %s", len(order), time.Since(started).Round(time.Millisecond))
	return nil
}

// rebuildClaimIndex refills claim_by_pid from truth, a page at a time.
func (c *Client) rebuildClaimIndex(ctx context.Context) error {
	type row struct {
		pid      int64
		cid, key string
	}
	var after row
	first := true
	for {
		q := `SELECT pid, cid, key FROM claim WHERE value IS NOT NULL`
		var args []any
		if !first {
			q += ` AND (cid, key, pid) > (?, ?, ?)`
			args = append(args, after.cid, after.key, after.pid)
		}
		q += ` ORDER BY cid, key, pid LIMIT 5000`
		rs, err := c.truth.DB().QueryContext(ctx, q, args...)
		if err != nil {
			return err
		}
		var page []row
		for rs.Next() {
			var r row
			if err := rs.Scan(&r.pid, &r.cid, &r.key); err != nil {
				rs.Close()
				return err
			}
			page = append(page, r)
		}
		if err := rs.Close(); err != nil {
			return err
		}
		if len(page) == 0 {
			return nil
		}
		if err := c.store.Update(ctx, func(tx *kv.Tx) error {
			for _, r := range page {
				if _, err := tx.Exec(`INSERT OR IGNORE INTO claim_by_pid(pid, cid, key) VALUES(?, ?, ?)`, r.pid, r.cid, r.key); err != nil {
					return err
				}
			}
			return nil
		}); err != nil {
			return err
		}
		after, first = page[len(page)-1], false
	}
}

// RebuiltAt is the stream id of the last cache rebuild ("" if never).
func (c *Client) RebuiltAt() string {
	s, err := c.kv()
	if err != nil {
		return ""
	}
	ctx, cancel := opCtx()
	defer cancel()
	v, _, _ := s.Get(ctx, rebuiltAtKey)
	return v
}