81 lines
3.6 KiB
Go
81 lines
3.6 KiB
Go
package projection
|
|
|
|
import "encoding/json"
|
|
|
|
// Bus framing contract (shared by producer + consumer).
|
|
//
|
|
// A projection stream carries two channels distinguished only by the event's
|
|
// Meta map. The producer sets these keys and the consumer reads them — one
|
|
// definition so the two never drift.
|
|
//
|
|
// - A FULL SNAPSHOT is a framed begin → chunk* → end stream, all tagged with one
|
|
// MetaEpoch. begin/end carry MetaCount (total entries) for completeness
|
|
// validation; chunks carry the entries.
|
|
// - A DELTA has NO MetaSnapshotPhase key, and MetaEpoch names the snapshot base
|
|
// it builds on (consumers ignore a delta older than the base they hold).
|
|
//
|
|
// The framing is domain-neutral. What rides inside each chunk/delta payload is
|
|
// a domain-specific wire schema, decoded by the Decoder the consumer supplies
|
|
// (see Store / Decoder). catalog re-exports these constants for wire-compat.
|
|
const (
|
|
MetaSnapshotPhase = "snapshot" // begin|chunk|end; absent ⇒ delta
|
|
MetaEpoch = "epoch" // int64 (Unix-ms), monotonic across rebuilds
|
|
MetaSeq = "seq" // chunk sequence within a snapshot
|
|
MetaCount = "count" // total entries (begin/end only)
|
|
MetaTenant = "tenant" // namespace tag
|
|
|
|
SnapshotBegin = "begin"
|
|
SnapshotChunk = "chunk"
|
|
SnapshotEnd = "end"
|
|
)
|
|
|
|
// Update is the decoded form of one entry in a snapshot chunk or a delta. Key
|
|
// is the index key; Value is the domain wire value; Deleted marks a removal
|
|
// (the consumer tombstones the key in its overlay, or skips it in a snapshot).
|
|
// This is the in-memory shape the framing layer works with — the on-the-wire
|
|
// JSON schema is the domain's own concern, bridged by a Decoder.
|
|
type Update[W any] struct {
|
|
Key string
|
|
Value W
|
|
Deleted bool
|
|
}
|
|
|
|
// Decoder turns one chunk/delta payload (the event's Payload bytes) into
|
|
// decoded Updates. Each domain defines its own wire schema and supplies the
|
|
// decoder; the Store owns only the framing/assembly around it. A decode error
|
|
// is surfaced from HandleFrame so the caller may NACK.
|
|
type Decoder[W any] func(payload []byte) ([]Update[W], error)
|
|
|
|
// Transform projects a wire value W down to the consumer's stored type T. It is
|
|
// the one place a consumer declares which facts it actually needs (and any
|
|
// derivation), so its local .bin holds only that. Every value entering the
|
|
// store — snapshot, delta, and fallback — passes through it exactly once.
|
|
type Transform[W, T any] func(W) T
|
|
|
|
// Fallback resolves a key the local store doesn't have yet (cold start before
|
|
// the first snapshot, or a key added since the last base). It returns a raw
|
|
// wire value W — the store applies the same Transform to it. Return (_, false)
|
|
// for a genuine miss. May be nil.
|
|
type Fallback[W any] func(key string) (W, bool)
|
|
|
|
// JSONDecode is a ready-made Decoder for domains whose wire schema is just a
|
|
// JSON array of Update[W] (the generic shape — Key/Value/Deleted). Catalog does
|
|
// NOT use this (its wire is a flat Projection for back-compat); identity
|
|
// consumers like tracking popularity can, pairing it with JSONEncode on the
|
|
// emitter side. Marshal errors propagate so HandleFrame can NACK.
|
|
func JSONDecode[W any](payload []byte) ([]Update[W], error) {
|
|
var out []Update[W]
|
|
if err := json.Unmarshal(payload, &out); err != nil {
|
|
return nil, err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// JSONEncode is the inverse of JSONDecode — the EncodeFunc an Emitter uses when
|
|
// the wire schema is a JSON array of Update[W]. A nil/empty slice marshals to a
|
|
// payload JSONDecode reads back as zero updates, which is what begin/end frames
|
|
// need.
|
|
func JSONEncode[W any](items []Update[W]) ([]byte, error) {
|
|
return json.Marshal(items)
|
|
}
|