better convert

projection/emitter.go

@@ -0,0 +1,154 @@
+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)
+}