meta-core/internal/schema/indexer.go

package schema

import (
	"context"
	"regexp"
	"strconv"
	"strings"
	"time"

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

var langCodeRegex = regexp.MustCompile(`^[a-z]{2,3}$`)

// CounterPrefix is where the schema lives: one counter per (field, key hint,
// primitive, hint), maintained by the storage write path in the same
// transaction as the change (storage's OnChange hook → CounterKey).
const CounterPrefix = "idx:schema/"

const counterSep = "\x1f"

// CounterKey is the counter a record key holding value contributes to, or
// ok=false for keys that are not record properties.
func CounterKey(recordKey, value string) (string, bool) {
	fieldPath, keyHint, ok := parseFieldPath(recordKey)
	if !ok {
		return "", false
	}
	prim, hint := ClassifyValue(value, true)
	return CounterPrefix + fieldPath + counterSep + string(keyHint) + counterSep + string(prim) + counterSep + string(hint), true
}

// fieldState is the per-field accumulator.
type fieldState struct {
	counts         map[string]int
	primitives     map[Primitive]int
	hintCounts     map[Hint]int
	typedCount     int
	undefinedCount int
	languageKeyed  bool
}

func newFieldState() *fieldState {
	return &fieldState{
		counts:     make(map[string]int),
		primitives: make(map[Primitive]int),
		hintCounts: make(map[Hint]int),
	}
}

// Indexer serves the per-field schema.
//
// The schema is a set of exact counters the write path keeps current in the
// same transaction as each change (see CounterKey), so serving it is one range
// read and nothing runs in the background. Two earlier shapes were dropped: a
// consumer of meta:events that remembered every record key it had seen (~2 GB
// at 6.3M keys), then a full rescan of the store every 10 minutes (~80 s of
// CPU each time, whether or not anyone looked).
type Indexer struct {
	store *kv.Store
}

// NewIndexer creates an Indexer over store.
func NewIndexer(store *kv.Store) *Indexer {
	return &Indexer{store: store}
}

// Start and Stop exist for the server's lifecycle; there is nothing to run.
func (i *Indexer) Start() error { return nil }
func (i *Indexer) Stop() error  { return nil }

// Snapshot reads the counters and folds them into the per-field schema.
func (i *Indexer) Snapshot(ctx context.Context) (*SchemaResponse, error) {
	fields := make(map[string]*fieldState)
	err := i.store.ScanPrefix(ctx, CounterPrefix, 5000, func(rows []kv.KV) error {
		for _, row := range rows {
			parts := strings.Split(row.K[len(CounterPrefix):], counterSep)
			if len(parts) != 4 {
				continue
			}
			n, err := strconv.Atoi(row.V)
			if err != nil || n <= 0 {
				continue
			}
			fs := fields[parts[0]]
			if fs == nil {
				fs = newFieldState()
				fields[parts[0]] = fs
			}
			if parts[1] != "" {
				fs.languageKeyed = true
			}
			incContribution(fs, labelPair{prim: Primitive(parts[2]), hint: Hint(parts[3])}, n)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	snap := &SchemaResponse{
		Fields:      make(map[string]*FieldSchema, len(fields)),
		GeneratedAt: time.Now().UTC(),
		Source:      "live",
	}
	for fp, fs := range fields {
		snap.Fields[fp] = computeFieldSchema(fs)
	}
	return snap, nil
}

type labelPair struct {
	prim Primitive
	hint Hint
}

// parseFieldPath turns "file:<rootID>/<propertyPath>" into the logical field
// path used as the map key. rootID is opaque to the indexer — currently a
// UUIDv7, historically a midhash256:abc token; both shapes parse identically
// because the split is on the first '/' after "file:".
//
// When the last segment of propertyPath matches ^[a-z]{2,3}$ it is treated as
// a language code: the field name collapses to the prefix and key_hint is set.
func parseFieldPath(key string) (string, KeyHint, bool) {
	if !strings.HasPrefix(key, "file:") {
		return "", "", false
	}
	rest := strings.TrimPrefix(key, "file:")
	slash := strings.Index(rest, "/")
	if slash < 0 {
		return "", "", false
	}
	property := rest[slash+1:]
	if property == "" {
		return "", "", false
	}
	if last := strings.LastIndex(property, "/"); last >= 0 {
		tail := property[last+1:]
		if langCodeRegex.MatchString(tail) {
			return property[:last], KeyHintLanguageCode, true
		}
	}
	return property, "", true
}

// incContribution adds n values of one (primitive, hint) to a field.
func incContribution(fs *fieldState, lp labelPair, n int) {
	label := breakdownLabel(lp.prim, lp.hint)
	fs.counts[label] += n
	if lp.prim == PrimUndefined {
		fs.undefinedCount += n
		return
	}
	fs.typedCount += n
	fs.primitives[lp.prim] += n
	if lp.hint != "" {
		fs.hintCounts[lp.hint] += n
	}
}

func computeFieldSchema(fs *fieldState) *FieldSchema {
	out := &FieldSchema{
		Breakdown: make(map[string]int, len(fs.counts)),
	}
	for k, v := range fs.counts {
		out.Breakdown[k] = v
	}

	nPrim := len(fs.primitives)
	hasUndef := fs.undefinedCount > 0

	switch {
	case nPrim == 0 && hasUndef:
		out.Type = PrimUndefined
	case nPrim == 1 && !hasUndef:
		for p := range fs.primitives {
			out.Type = p
		}
	default:
		out.Type = PrimMixed
	}

	// Strict hint promotion: exactly one hint observed, and it covers every
	// typed (non-undefined) value. Hints are never set on mixed fields.
	if out.Type != PrimMixed && len(fs.hintCounts) == 1 && fs.typedCount > 0 {
		for h, c := range fs.hintCounts {
			if c == fs.typedCount {
				out.Hint = h
			}
		}
	}

	if fs.languageKeyed {
		out.KeyHint = KeyHintLanguageCode
	}

	return out
}