platform/catalog/consumer.go

package catalog

// ProjectionStore — the consumer half of the projection seam (see
// docs/catalog-projection.md). A commerce service (cart, checkout, order, UCP)
// feeds it the catalog.projection_published frames it receives off the bus and
// reads catalog facts back by SKU. 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 (e.g. an
// HTTP fetch from the reader) — last-resort only. The access pattern is point
// lookup by SKU; nothing enumerates the catalog.
//
// No shared volume: each consumer builds its own base from the bus. Concurrency:
// HandleFrame is single-writer (drive it from one delivery goroutine); Get is
// safe from many readers.

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

// ProjectionFallback resolves an SKU the local store doesn't have yet (cold
// start before the first snapshot, or an item added since the last base). It
// returns a raw Projection — the store applies the SAME Transform to it, so a
// fallback hit is shaped identically to bus-fed data. Return (_, false) to
// report a genuine miss.
type ProjectionFallback func(sku string) (Projection, bool)

// ProjectionTransform projects the wire Projection down to the consumer's own
// stored type T. It is MANDATORY: it is the one place a consumer declares which
// catalog facts it actually needs (and any derivation, e.g. tax_class → rate),
// so its local .bin holds only that — smaller, and read-ready with no per-Get
// conversion. Every value entering the store — snapshot, delta, and fallback —
// passes through it exactly once.
type ProjectionTransform[T any] func(Projection) T

// ProjectionStore is a consumer's local, cold-start-ready projection cache of T.
type ProjectionStore[T any] struct {
	dir, name string
	transform ProjectionTransform[T]
	fallback  ProjectionFallback
	holder    *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace

	mu      sync.RWMutex
	epoch   int64                      // epoch of the live base
	overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base

	// 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
}

// NewProjectionStore opens a store of T backed by dir/name (its own .bin).
// transform is REQUIRED — it defines the consumer's projection shape (see
// ProjectionTransform). 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 NewProjectionStore[T any](dir, name string, transform ProjectionTransform[T], fallback ProjectionFallback) (*ProjectionStore[T], error) {
	if dir == "" || name == "" {
		return nil, fmt.Errorf("catalog: ProjectionStore needs dir and name")
	}
	if transform == nil {
		return nil, fmt.Errorf("catalog: ProjectionStore needs a transform (declare which catalog facts this consumer stores)")
	}
	s := &ProjectionStore[T]{
		dir:       dir,
		name:      name,
		transform: transform,
		fallback:  fallback,
		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("catalog: ProjectionStore 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("catalog: ProjectionStore ignoring unreadable base %s: %v", path, err)
	}
	return s, nil
}

// HandleFrame applies one catalog.projection_published 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 *ProjectionStore[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 *ProjectionStore[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)
	s.buildingEpoch = epoch
	s.buildingSeen = 0
	return nil
}

func (s *ProjectionStore[T]) chunkSnapshot(epoch int64, payload []byte) error {
	ups, err := DecodeUpdates(payload)
	if err != nil {
		return fmt.Errorf("catalog: 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.SKU == "" {
			continue // a snapshot carries live items only
		}
		// Transform applied here — the base stores the consumer's shape, not the
		// full Projection.
		if err := s.building.Append(u.SKU, s.transform(u.Projection)); err != nil {
			log.Printf("catalog: ProjectionStore skip %q in snapshot: %v", u.SKU, err)
			continue
		}
		s.buildingSeen++
	}
	return nil
}

func (s *ProjectionStore[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 catalog silently is worse than stale.
	if cs := meta[MetaCount]; cs != "" {
		if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen {
			log.Printf("catalog: ProjectionStore 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("catalog: 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 — the store
	// would then mislabel its base and drop the current epoch's deltas on restart.
	writeEpochSidecar(path, epoch)
	if err := s.holder.Reload(path); err != nil {
		return fmt.Errorf("catalog: reload snapshot base: %w", err)
	}
	s.epoch = epoch
	s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch
	log.Printf("catalog: ProjectionStore snapshot applied — base swapped epoch=%d (%d entries)", epoch, s.buildingSeen)
	return nil
}

func (s *ProjectionStore[T]) applyDelta(epoch int64, payload []byte) error {
	ups, err := DecodeUpdates(payload)
	if err != nil {
		return fmt.Errorf("catalog: 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.SKU == "" {
			continue
		}
		if u.Deleted {
			s.overlay[u.SKU] = overlayEntry[T]{deleted: true}
			deletes++
			continue
		}
		// Transform applied here too — overlay holds the consumer's shape.
		s.overlay[u.SKU] = overlayEntry[T]{val: s.transform(u.Projection)}
		upserts++
	}
	if upserts+deletes > 0 {
		log.Printf("catalog: ProjectionStore delta applied epoch=%d (+%d/-%d, overlay=%d)",
			epoch, upserts, deletes, len(s.overlay))
	}
	return nil
}

// Get resolves a value by SKU: overlay (incl. tombstone) → mmap base →
// fallback. A tombstoned SKU returns (zero, false) and never reaches the
// fallback — the catalog explicitly removed it. A fallback hit is run through
// the same Transform, so it's shaped like everything else.
func (s *ProjectionStore[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 p, ok := s.fallback(sku); ok {
			return s.transform(p), true
		}
	}
	return zero, false
}

// IsDeleted reports whether the overlay holds a tombstone for sku — a delete
// seen on the bus since the current base. Lets a caller skip a downstream fetch
// for an SKU the catalog just removed. After a snapshot folds the delete into a
// new base the SKU is simply absent (IsDeleted → false); "absent" is then the
// caller's fetch/miss path, which a deleted item resolves as a downstream 404.
func (s *ProjectionStore[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 *ProjectionStore[T]) Epoch() int64 {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.epoch
}

// Stats returns observability counters (base size, overlay size, epoch).
func (s *ProjectionStore[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 *ProjectionStore[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 — a reader either sees the
// old epoch or the new one, never garbage.
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("catalog: ProjectionStore epoch sidecar write: %v", err)
		return
	}
	if err := os.Rename(tmp, p); err != nil {
		log.Printf("catalog: ProjectionStore 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
}