platform/catalog/consumer.go

package catalog

// Projection seam — catalog binding over the generic platform/projection
// package. The value-agnostic machinery (snapshot .bin format, mmap reader,
// race-free SnapshotHolder, the begin→chunk*→end consumer state machine with
// epoch gating + delta overlay) lives in platform/projection. This file is the
// thin catalog-specific binding: it fixes the WIRE type to catalog.Projection
// and supplies the decoder for the catalog wire schema, so cart / checkout /
// order / catalog-emit keep their existing catalog.* API unchanged.
//
// See docs/catalog-projection.md for the consumer contract and
// docs/plan-projectionstore-generalization.md for why the split exists.

import (
	"encoding/json"
	"fmt"
	"time"

	"git.k6n.net/mats/platform/event"
	"git.k6n.net/mats/platform/projection"
	amqp "github.com/rabbitmq/amqp091-go"
)

// Generic machinery, re-exported under their historical catalog names. These
// are type aliases — catalog.SnapshotHolder[T] IS projection.SnapshotHolder[T],
// so values flow between the two packages without conversion.
type (
	SnapshotWriter[T any] = projection.SnapshotWriter[T]
	SnapshotReader[T any] = projection.SnapshotReader[T]
	SnapshotHolder[T any] = projection.SnapshotHolder[T]

	// ProjectionStore is a consumer's local, cold-start-ready cache of T, fed the
	// catalog.projection_published frames off the bus. The wire type is fixed to
	// catalog.Projection; T is the consumer's projected shape.
	ProjectionStore[T any] = projection.Store[Projection, T]

	// ProjectionTransform projects the wire Projection down to the consumer's
	// stored type T (the one place a consumer declares which catalog facts it
	// needs). MANDATORY.
	ProjectionTransform[T any] = projection.Transform[Projection, T]

	// ProjectionFallback resolves a SKU the local store doesn't have yet. May be nil.
	ProjectionFallback = projection.Fallback[Projection]
)

// DefaultDrainGrace is the SnapshotHolder default drain period (re-exported).
const DefaultDrainGrace = projection.DefaultDrainGrace

// Snapshot file errors, re-exported so callers can errors.Is against the
// catalog.* names. Identity is preserved (these ARE the projection package's
// error values).
var (
	ErrSnapshotMagic     = projection.ErrSnapshotMagic
	ErrSnapshotVersion   = projection.ErrSnapshotVersion
	ErrSnapshotTruncated = projection.ErrSnapshotTruncated
	ErrSnapshotBadBlob   = projection.ErrSnapshotBadBlob
	ErrSnapshotOverflow  = projection.ErrSnapshotOverflow
	ErrSKUTooLong        = projection.ErrSKUTooLong
	ErrTenantTooLong     = projection.ErrTenantTooLong
)

// NewSnapshotWriter returns an empty snapshot writer for value type T.
func NewSnapshotWriter[T any]() *SnapshotWriter[T] { return projection.NewSnapshotWriter[T]() }

// NewSnapshotReader wraps an already-loaded []byte as a reader of T.
func NewSnapshotReader[T any](data []byte) (*SnapshotReader[T], error) {
	return projection.NewSnapshotReader[T](data)
}

// ReadSnapshotFile opens + parses a snapshot file as a reader of T (mmap on Unix).
func ReadSnapshotFile[T any](path string) (*SnapshotReader[T], error) {
	return projection.ReadSnapshotFile[T](path)
}

// NewSnapshotHolder returns a holder of T with the given drain grace.
func NewSnapshotHolder[T any](grace time.Duration) *SnapshotHolder[T] {
	return projection.NewSnapshotHolder[T](grace)
}

// NewProjectionStore opens a catalog ProjectionStore backed by dir/name (its own
// .bin). transform is REQUIRED (declares the stored shape); fallback may be nil.
// If a base from a previous run is present it is reopened immediately
// (cold-start ready). Close() when done.
func NewProjectionStore[T any](dir, name string, transform ProjectionTransform[T], fallback ProjectionFallback) (*ProjectionStore[T], error) {
	return projection.NewStore[Projection, T](dir, name, decodeProjectionUpdates, transform, fallback)
}

// decodeProjectionUpdates bridges the catalog wire schema ([]ProjectionUpdate —
// a flat Projection plus a `deleted` flag) to the generic projection.Update the
// Store works with. The index key is the Projection's SKU. This is the one
// catalog-specific seam the generic Store can't know: the on-the-wire payload
// shape is a domain decision.
func decodeProjectionUpdates(payload []byte) ([]projection.Update[Projection], error) {
	raw, err := DecodeUpdates(payload)
	if err != nil {
		return nil, err
	}
	out := make([]projection.Update[Projection], len(raw))
	for i, u := range raw {
		out[i] = projection.Update[Projection]{Key: u.SKU, Value: u.Projection, Deleted: u.Deleted}
	}
	return out, nil
}

// MakeAmqpHandler returns a RabbitMQ delivery handler that wraps a ProjectionStore[T].
// It decodes the event.Event envelope, recovers from panics, and applies the frame to the store.
func MakeAmqpHandler[T any](store *ProjectionStore[T]) func(amqp.Delivery) error {
	return func(d amqp.Delivery) (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = fmt.Errorf("projection handler panic recovered: %v", r)
			}
		}()
		var ev event.Event
		if err := json.Unmarshal(d.Body, &ev); err != nil {
			return fmt.Errorf("decode event envelope: %w", err)
		}
		if err := store.HandleFrame(ev.Meta, ev.Payload); err != nil {
			return fmt.Errorf("apply projection frame: %w", err)
		}
		return nil
	}
}

// HandleEvent feeds an already-decoded event.Event directly into the projection store.
func HandleEvent[T any](store *ProjectionStore[T], ev event.Event) error {
	return store.HandleFrame(ev.Meta, ev.Payload)
}