platform/projection/emitter.go

package projection

import (
	"errors"
	"strconv"
)

// Emitter is the producer half of the seam: it frames a full projection as a
// begin → chunk* → end snapshot stream (and per-batch deltas) and hands each
// frame to a caller-supplied publish function. It is transport-agnostic — it
// owns only the framing (the Meta* keys, chunking, epoch tagging); the caller
// wires PublishFunc to its bus (rabbit.PublishEvent, etc.) and EncodeFunc to
// its wire schema. This is the symmetric counterpart to Store/Decoder.
//
// One authoritative producer per projection: receivers' Stores assemble their
// own .bin from this stream. Two producers publishing the same projection would
// make receivers thrash between epochs — don't.
//
// catalog's finder writeradmin predates this and keeps its own framing for now;
// new domains (tracking popularity/suggestions) use this.

// PublishFunc publishes one framed event: meta carries the framing keys
// (snapshot phase, epoch, seq, count, tenant) already set; payload is the
// chunk/delta body. The caller may add its own keys to meta before sending
// (e.g. a country tag). A non-nil error aborts the current snapshot stream.
type PublishFunc func(meta map[string]string, payload []byte) error

// EncodeFunc serialises a batch of wire values into a chunk/delta payload — the
// inverse of Decoder, and the one place the domain wire schema lives on the
// producer side. It MUST encode a nil/empty slice as a payload the matching
// Decoder reads back as zero updates (begin/end frames carry no entries).
// JSONEncode satisfies this for the generic Update[W] schema.
type EncodeFunc[W any] func([]Update[W]) ([]byte, error)

// ErrEmptySnapshot is returned by Snapshot when handed no entries. An empty
// snapshot would swap a receiver's base to nothing; refusing it mirrors the
// catalog producer, which never propagates an empty catalog.
var ErrEmptySnapshot = errors.New("projection: refusing to emit an empty snapshot (would wipe receivers)")

// DefaultChunkSize is the entries-per-chunk used when WithChunkSize isn't set.
// Mirrors catalog's projectionSnapshotChunk.
const DefaultChunkSize = 500

// Emitter frames and publishes a projection of wire type W.
type Emitter[W any] struct {
	encode    EncodeFunc[W]
	publish   PublishFunc
	tenant    string
	chunkSize int
}

// emitterConfig is set via options then copied onto the typed Emitter (options
// can't be generic over W, so they carry the un-typed knobs).
type emitterConfig struct {
	tenant    string
	chunkSize int
}

// WithTenant tags every frame with a tenant namespace (lands in MetaTenant and,
// for snapshots, in the receiver's base header).
func WithTenant(tenant string) func(*emitterConfig) {
	return func(c *emitterConfig) { c.tenant = tenant }
}

// WithChunkSize overrides the entries-per-chunk (default DefaultChunkSize).
// Bounds per-message size; smaller = more frames.
func WithChunkSize(n int) func(*emitterConfig) {
	return func(c *emitterConfig) {
		if n > 0 {
			c.chunkSize = n
		}
	}
}

// NewEmitter builds an emitter. encode and publish are REQUIRED. Options:
// WithTenant, WithChunkSize.
func NewEmitter[W any](encode EncodeFunc[W], publish PublishFunc, opts ...func(*emitterConfig)) *Emitter[W] {
	cfg := emitterConfig{chunkSize: DefaultChunkSize}
	for _, o := range opts {
		o(&cfg)
	}
	return &Emitter[W]{
		encode:    encode,
		publish:   publish,
		tenant:    cfg.tenant,
		chunkSize: cfg.chunkSize,
	}
}

// Snapshot publishes the full entry set as a begin → chunk* → end stream tagged
// with epoch (a monotonic Unix-ms stamp the caller provides). Returns the
// number of chunk frames emitted. ErrEmptySnapshot if items is empty; a
// publish/encode error aborts the stream (receivers keep their previous base)
// and is returned. The same epoch must later tag any Delta built on this base.
func (e *Emitter[W]) Snapshot(epoch int64, items []Update[W]) (chunks int, err error) {
	if len(items) == 0 {
		return 0, ErrEmptySnapshot
	}
	if err := e.frame(epoch, SnapshotBegin, 0, len(items), nil); err != nil {
		return 0, err
	}
	seq := 0
	for start := 0; start < len(items); start += e.chunkSize {
		end := start + e.chunkSize
		if end > len(items) {
			end = len(items)
		}
		if err := e.frame(epoch, SnapshotChunk, seq, 0, items[start:end]); err != nil {
			return seq, err
		}
		seq++
	}
	if err := e.frame(epoch, SnapshotEnd, seq, len(items), nil); err != nil {
		return seq, err
	}
	return seq, nil
}

// Delta publishes one delta batch tagged with the base epoch it builds on.
// Entries with Deleted=true tombstone the key in receivers' overlays. A
// receiver ignores a delta whose epoch predates the base it holds. Empty batch
// is a no-op (no error).
func (e *Emitter[W]) Delta(epoch int64, items []Update[W]) error {
	if len(items) == 0 {
		return nil
	}
	return e.frame(epoch, "", 0, 0, items)
}

// frame encodes items and publishes one frame. phase "" ⇒ delta (no snapshot
// phase key). begin/end carry MetaCount; chunks/deltas carry the entries.
func (e *Emitter[W]) frame(epoch int64, phase string, seq, count int, items []Update[W]) error {
	// Always encode — even begin/end with no entries. encode(nil) must yield a
	// payload the Decoder reads as zero updates (see EncodeFunc), so a
	// framing-unaware decode never sees an empty/undecodable body.
	payload, err := e.encode(items)
	if err != nil {
		return err
	}
	meta := map[string]string{
		MetaEpoch: strconv.FormatInt(epoch, 10),
	}
	if e.tenant != "" {
		meta[MetaTenant] = e.tenant
	}
	if phase != "" {
		meta[MetaSnapshotPhase] = phase // presence marks a snapshot frame vs a delta
		meta[MetaSeq] = strconv.Itoa(seq)
	}
	if phase == SnapshotBegin || phase == SnapshotEnd {
		meta[MetaCount] = strconv.Itoa(count)
	}
	return e.publish(meta, payload)
}