platform/projection/store.go

package projection

// Store — the consumer half of the projection seam. A service feeds it the
// snapshot/delta frames it receives off the bus and reads facts back by key.
// The store owns a LOCAL read-only projection:
//
//   - A full snapshot stream (begin → chunk* → end, one MetaEpoch) is assembled
//     into the store's OWN .bin (SnapshotWriter), atomically swapped in as the
//     mmap base (SnapshotHolder). The base is the cold-start substrate — it
//     survives a restart (reopened from disk), so the consumer serves reads
//     immediately without waiting for a bus replay.
//   - Per-batch deltas of the SAME epoch are applied to an in-memory overlay on
//     top of the base (overlay wins; a delete leaves a tombstone that shadows the
//     base and the fallback). The next snapshot folds the overlay into a new base.
//
// Read order on Get: overlay → mmap base → optional caller fallback. Scan walks
// overlay + base (overlay wins, tombstones hidden) for enumeration / ranked
// output.
//
// The store is generic over the WIRE value W (what rides on the bus, decoded by
// the supplied Decoder) and the STORED value T (what the Transform projects W
// down to, held in the .bin). For an identity consumer W == T with an identity
// transform; catalog uses W = catalog.Projection.
//
// No shared volume: each consumer builds its own base from the bus. Concurrency:
// HandleFrame is single-writer (drive it from one delivery goroutine); Get/Scan
// are safe from many readers.

import (
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
)

// Store is a consumer's local, cold-start-ready projection cache of T, fed from
// a bus stream of wire values W.
type Store[W, T any] struct {
	dir, name string
	decode    Decoder[W]
	transform Transform[W, T]
	fallback  Fallback[W]
	holder    *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace

	hashKeys bool // build the base + dedup Scan under the hashed key encoding

	mu      sync.RWMutex
	epoch   int64                      // epoch of the live base
	overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base (keyed by ORIGINAL key)

	// in-progress snapshot assembly (one epoch at a time)
	building      *SnapshotWriter[T]
	buildingEpoch int64
	buildingSeen  int
}

type overlayEntry[T any] struct {
	val     T
	deleted bool
}

// StoreOption configures a Store at construction.
type StoreOption func(*storeConfig)

type storeConfig struct{ hashKeys bool }

// WithKeyHashing makes the store build its base under the hashed key encoding,
// so unbounded keys (page paths, query strings) fit the fixed-width index. Get
// still takes the original key (it is hashed internally); Scan yields digests
// (recover the original from the value). Match this to how the producer emits.
func WithKeyHashing() StoreOption { return func(c *storeConfig) { c.hashKeys = true } }

// NewStore opens a store backed by dir/name (its own .bin). decode and
// transform are REQUIRED: decode parses the domain wire schema from a frame
// payload; transform projects each wire value to the stored shape. If a base
// from a previous run is present it is reopened immediately (cold-start ready);
// its epoch is read from the sidecar so same-epoch deltas can resume. fallback
// may be nil. Close() when done.
func NewStore[W, T any](dir, name string, decode Decoder[W], transform Transform[W, T], fallback Fallback[W], opts ...StoreOption) (*Store[W, T], error) {
	if dir == "" || name == "" {
		return nil, fmt.Errorf("projection: Store needs dir and name")
	}
	if decode == nil {
		return nil, fmt.Errorf("projection: Store needs a decoder (declare the domain wire schema)")
	}
	if transform == nil {
		return nil, fmt.Errorf("projection: Store needs a transform (declare which facts this consumer stores)")
	}
	var cfg storeConfig
	for _, o := range opts {
		o(&cfg)
	}
	s := &Store[W, T]{
		dir:       dir,
		name:      name,
		decode:    decode,
		transform: transform,
		fallback:  fallback,
		hashKeys:  cfg.hashKeys,
		holder:    NewSnapshotHolder[T](0),
		overlay:   make(map[string]overlayEntry[T]),
	}
	path := filepath.Join(dir, name)
	if err := s.holder.Open(path); err == nil {
		s.epoch = readEpochSidecar(path)
		log.Printf("projection: Store reopened base %s (epoch=%d, %d entries)", path, s.epoch, s.holder.Len())
	} else if !os.IsNotExist(err) {
		// A corrupt/partial base on disk shouldn't crash boot — start empty and
		// wait for the next snapshot; reads fall back meanwhile.
		log.Printf("projection: Store ignoring unreadable base %s: %v", path, err)
	}
	return s, nil
}

// HandleFrame applies one projection event (snapshot frame or delta), routed by
// its Meta. Drive this from a single delivery goroutine. A returned error means
// the payload was undecodable (the caller may NACK); a frame that simply
// doesn't apply (stale epoch, wrong assembly state) returns nil so the delivery
// is acked.
func (s *Store[W, T]) HandleFrame(meta map[string]string, payload []byte) error {
	epoch, _ := strconv.ParseInt(meta[MetaEpoch], 10, 64)
	switch meta[MetaSnapshotPhase] {
	case SnapshotBegin:
		return s.beginSnapshot(epoch, meta[MetaTenant])
	case SnapshotChunk:
		return s.chunkSnapshot(epoch, payload)
	case SnapshotEnd:
		return s.endSnapshot(epoch, meta)
	case "":
		return s.applyDelta(epoch, payload)
	default:
		return nil // unknown phase — ignore forward-compatibly
	}
}

func (s *Store[W, T]) beginSnapshot(epoch int64, tenant string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if epoch <= s.epoch {
		return nil // we already hold this base or newer
	}
	// Start (or restart) assembly for this epoch. A new begin supersedes any
	// half-built snapshot (e.g. a producer that aborted mid-stream).
	s.building = NewSnapshotWriter[T]().WithTenant(tenant)
	if s.hashKeys {
		s.building.WithKeyHashing()
	}
	s.buildingEpoch = epoch
	s.buildingSeen = 0
	return nil
}

func (s *Store[W, T]) chunkSnapshot(epoch int64, payload []byte) error {
	ups, err := s.decode(payload)
	if err != nil {
		return fmt.Errorf("projection: decode snapshot chunk: %w", err)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.building == nil || epoch != s.buildingEpoch {
		return nil // not assembling this epoch
	}
	for _, u := range ups {
		if u.Deleted || u.Key == "" {
			continue // a snapshot carries live entries only
		}
		// Transform applied here — the base stores the consumer's shape.
		if err := s.building.Append(u.Key, s.transform(u.Value)); err != nil {
			log.Printf("projection: Store skip %q in snapshot: %v", u.Key, err)
			continue
		}
		s.buildingSeen++
	}
	return nil
}

func (s *Store[W, T]) endSnapshot(epoch int64, meta map[string]string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.building == nil || epoch != s.buildingEpoch {
		return nil
	}
	w := s.building
	s.building = nil

	// Completeness guard: if begin/end advertised a count and we appended fewer
	// (a lost chunk), DON'T swap — keep the old base and wait for the next
	// snapshot. Serving an incomplete projection silently is worse than stale.
	if cs := meta[MetaCount]; cs != "" {
		if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen {
			log.Printf("projection: Store snapshot epoch=%d incomplete (%d/%d) — discarding, keeping base epoch=%d",
				epoch, s.buildingSeen, want, s.epoch)
			return nil
		}
	}

	path, err := w.Write(s.dir, s.name)
	if err != nil {
		return fmt.Errorf("projection: write snapshot base: %w", err)
	}
	// Persist the epoch sidecar (atomic) BEFORE swapping the in-memory reader, so
	// a reopened store's epoch always matches the .bin already on disk (w.Write
	// renamed the new file into place above). Writing it after Reload would leave
	// a crash window where the new .bin carries a stale/absent epoch.
	writeEpochSidecar(path, epoch)
	if err := s.holder.Reload(path); err != nil {
		return fmt.Errorf("projection: reload snapshot base: %w", err)
	}
	s.epoch = epoch
	s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch
	log.Printf("projection: Store snapshot applied — base swapped epoch=%d (%d entries)", epoch, s.buildingSeen)
	return nil
}

func (s *Store[W, T]) applyDelta(epoch int64, payload []byte) error {
	ups, err := s.decode(payload)
	if err != nil {
		return fmt.Errorf("projection: decode delta: %w", err)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	switch {
	case s.building != nil:
		// Under FIFO delivery a delta won't interleave a snapshot stream; if it
		// does, the snapshot is authoritative for its epoch, so drop the delta.
		return nil
	case epoch < s.epoch:
		return nil // stale — predates our base
	case epoch > s.epoch:
		// CONTRACT: a future-epoch delta is DROPPED, not buffered. Under FIFO
		// per-queue delivery begin(N) always precedes delta(N), so this only
		// happens after a reconnect that missed the whole snapshot — which a
		// buffer couldn't reassemble anyway. Recovery is the next snapshot of
		// this epoch; the consumer's fallback covers reads until then.
		return nil
	}
	var upserts, deletes int
	for _, u := range ups {
		if u.Key == "" {
			continue
		}
		if u.Deleted {
			s.overlay[u.Key] = overlayEntry[T]{deleted: true}
			deletes++
			continue
		}
		// Transform applied here too — overlay holds the consumer's shape.
		s.overlay[u.Key] = overlayEntry[T]{val: s.transform(u.Value)}
		upserts++
	}
	if upserts+deletes > 0 {
		log.Printf("projection: Store delta applied epoch=%d (+%d/-%d, overlay=%d)",
			epoch, upserts, deletes, len(s.overlay))
	}
	return nil
}

// Get resolves a value by key: overlay (incl. tombstone) → mmap base →
// fallback. A tombstoned key returns (zero, false) and never reaches the
// fallback. A fallback hit is run through the same Transform.
func (s *Store[W, T]) Get(sku string) (T, bool) {
	var zero T
	s.mu.RLock()
	e, ok := s.overlay[sku]
	s.mu.RUnlock()
	if ok {
		if e.deleted {
			return zero, false
		}
		return e.val, true
	}
	if v, ok := s.holder.Get(sku); ok {
		return v, true
	}
	if s.fallback != nil {
		if w, ok := s.fallback(sku); ok {
			return s.transform(w), true
		}
	}
	return zero, false
}

// Scan walks every live (key, value) — overlay first (overlay wins; tombstones
// hidden), then the mmap base minus any key the overlay already covered. Return
// false from fn to stop early. The fallback is NOT consulted (it resolves
// single misses, not enumeration). Callers that need top-N-by-value collect and
// sort in memory — on-disk order is by key. Held under RLock for the whole
// walk, so it blocks a concurrent snapshot swap; keep fn cheap.
func (s *Store[W, T]) Scan(fn func(key string, v T) bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	// The base (holder) is keyed by stored key — a digest in hashed mode. To
	// dedup overlay against it and emit one consistent key space, project the
	// overlay's original keys through the same encoding.
	storeKey := func(k string) string {
		if s.hashKeys {
			return hashKey(k)
		}
		return k
	}
	seen := make(map[string]struct{}, len(s.overlay))
	for k, e := range s.overlay {
		sk := storeKey(k)
		seen[sk] = struct{}{}
		if e.deleted {
			continue
		}
		if !fn(sk, e.val) {
			return
		}
	}
	s.holder.Scan(func(k string, v T) bool {
		if _, ok := seen[k]; ok {
			return true // overlay already emitted (or tombstoned) this key
		}
		return fn(k, v)
	})
}

// IsDeleted reports whether the overlay holds a tombstone for key — a delete
// seen on the bus since the current base.
func (s *Store[W, T]) IsDeleted(sku string) bool {
	s.mu.RLock()
	defer s.mu.RUnlock()
	e, ok := s.overlay[sku]
	return ok && e.deleted
}

// Epoch returns the epoch of the current base (0 if none yet).
func (s *Store[W, T]) Epoch() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.epoch
}

// Stats returns observability counters (base size, overlay size, epoch).
func (s *Store[W, T]) Stats() (epoch int64, baseCount, overlayCount int) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.epoch, s.holder.Len(), len(s.overlay)
}

// Close releases the mmap base.
func (s *Store[W, T]) Close() error { return s.holder.Close() }

// ── epoch sidecar ────────────────────────────────────────────────────────────
// The .bin header has no room for the epoch, so it's persisted next to it as
// "<name>.epoch" — read on boot so a reopened base can resume same-epoch deltas.

func epochSidecarPath(binPath string) string { return binPath + ".epoch" }

// writeEpochSidecar writes the epoch atomically (temp+rename) so a crash mid-write
// can't leave a torn file that parses as a bogus epoch.
func writeEpochSidecar(binPath string, epoch int64) {
	p := epochSidecarPath(binPath)
	tmp := p + ".tmp"
	if err := os.WriteFile(tmp, []byte(strconv.FormatInt(epoch, 10)), 0o644); err != nil {
		log.Printf("projection: Store epoch sidecar write: %v", err)
		return
	}
	if err := os.Rename(tmp, p); err != nil {
		log.Printf("projection: Store epoch sidecar rename: %v", err)
		_ = os.Remove(tmp)
	}
}

func readEpochSidecar(binPath string) int64 {
	b, err := os.ReadFile(epochSidecarPath(binPath))
	if err != nil {
		return 0
	}
	n, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
	return n
}