platform

cmd/catalog-emit/main.go

@@ -0,0 +1,571 @@
+// Package main — `catalog-emit` is the producer-side harness for the
+// platform/catalog wire. It has two modes:
+//
+//  1. One-shot (no -rabbit flag): synthesise / load items per tenant, write
+//     one snapshot file per tenant under -out-dir, probe a randomised sample
+//     of SKUs to verify the format round-trips, then exit. Useful for
+//     seed-and-verify runs and CI.
+//
+//  2. Watch (-rabbit amqp://...): emit a baseline snapshot per tenant (so the
+//     start state matches what the bus will then incrementally update), bind
+//     to the AMQP topic exchange and consume `catalog.projection_published`
+//     events. Each event carries a `[]ProjectionUpdate` batch (upsert+delete
+//     per SKU). Per-tenant live state is reflected in a per-tenant
+//     SnapshotHolder; reload-on-event uses the same drain-grace atomic-swap
+//     pattern as the consumer side. A periodic probe validates that live Get
+//     traffic continues to hit under concurrent reloads.
+//
+// Note on naming: "producer wire-up" here means the harness produces per-tenant
+// snapshot BIN files autosaved on every bus batch — it does NOT publish
+// catalog.projection_published events itself (it only consumes them, in the
+// same role as cart / checkout / order).
+//
+// Usage (one-shot):
+//
+//	catalog-emit \
+//	  -schemas /etc/platform/catalog/schemas.json \
+//	  [-items ./items.json]    \
+//	  -out-dir ./catalog        \
+//	  -tenants default[,bygglagret]
+//
+// Usage (watch):
+//
+//	catalog-emit \
+//	  -schemas /etc/platform/catalog/schemas.json \
+//	  -rabbit amqp://guest:guest@localhost:5672/ \
+//	  -exchange platform \
+//	  -out-dir ./catalog        \
+//	  -tenants default[,bygglagret]
+//
+// Exit code 0 = every sampled lookup matched; 2 = bad CLI args; 1 = a probe
+// miss happened in one-shot mode (in watch mode, misses are reported but the
+// process keeps running for further evaluation).
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"math/rand"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"git.k6n.net/mats/platform/catalog"
+	"git.k6n.net/mats/platform/cmd/internal/catalogtest"
+	"git.k6n.net/mats/platform/event"
+	"git.k6n.net/mats/platform/rabbit"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+// TenantState is one tenant's live projection state + a SnapshotHolder that
+// hot-reloads a freshly-written snapshot file on every update batch.
+//
+// The items map is the live projection set; rebuilding it from a snapshot
+// is intentionally cheap because the written file IS the on-disk truth
+// after every rebuild — but the live map is what bus batches merge into.
+//
+// Counters are atomic.Uint64 because they are read by the SIGINT shutdown
+// path concurrently with writes from both the bus-handler goroutine and
+// the probe goroutine; a mutex-protected uint64 would either contend on
+// the probe hot path or race with the shutdown summary.
+type TenantState struct {
+	tenant string
+	schema catalog.Schema
+
+	mu    sync.RWMutex
+	items map[string]catalog.Projection
+
+	holder *catalog.SnapshotHolder
+	path   string // <out-dir>/<tenant>.bin
+
+	rebuildCount atomic.Uint64
+	probeCount   atomic.Uint64
+	probeMisses  atomic.Uint64
+}
+
+// newTenantState wires a brand-new tenant. Caller is responsible for
+// calling rebuild() (or Open via the initial state pipeline) before any Get.
+func newTenantState(tenant string, sch catalog.Schema, holder *catalog.SnapshotHolder, path string) *TenantState {
+	return &TenantState{
+		tenant: tenant,
+		schema: sch,
+		items:  make(map[string]catalog.Projection, 1024),
+		holder: holder,
+		path:   path,
+	}
+}
+
+// rebuild merges batch atomically under the write lock, captures a snapshot
+// of the items map, releases the lock, and does IO (Write + Reload) without
+// holding it. The snapshot capture is the only step that touches the items
+// map; everything after the unlock races freely with concurrent
+// rebuilds / probes, and drain-grace handles the prior-reader lifecycle.
+//
+// Race analysis: between unlock and Write completion, a second event for
+// the same tenant could mutate items and trigger a parallel rebuild that
+// races to the disk. Resolve in favour of the second event — its Write
+// wins via atomic-rename; the first Write either finishes last (writes the
+// older state, harmless because items already moved) or its tmp file is
+// overwritten by the second's tmp + rename. The fact that the second's
+// reload is the latest Set means downstream Get sees the newest state;
+// the first's Reload, if it lands second, demotes a newer reader to
+// older — drain-grace then queues the newer one for reap after grace,
+// and the older one serves drops via atomic.Pointer swap. Both ends are
+// eventually consistent; not a correctness break, an arrival-ordering race
+// in IO. For a producer harness verification role this is acceptable; a
+// high-volume producer should use a per-tenant coalescing channel.
+func (ts *TenantState) rebuild(batch []catalog.ProjectionUpdate) error {
+	ts.mu.Lock()
+	if batch != nil {
+		for _, u := range batch {
+			if u.Deleted {
+				delete(ts.items, u.SKU)
+				continue
+			}
+			if u.SKU == "" {
+				continue
+			}
+			ts.items[u.SKU] = u.Projection
+		}
+	}
+	snapshot := make([]catalog.Projection, 0, len(ts.items))
+	for _, p := range ts.items {
+		snapshot = append(snapshot, p)
+	}
+	ts.rebuildCount.Add(1)
+	ts.mu.Unlock()
+
+	return ts.writeSnapshot(snapshot)
+}
+
+// writeSnapshot emits snapshot to disk (atomic temp+rename) and reloads
+// the holder. Called WITHOUT holding ts.mu.
+func (ts *TenantState) writeSnapshot(snap []catalog.Projection) error {
+	w := catalog.NewSnapshotWriter().WithTenant(ts.tenant)
+	for _, p := range snap {
+		if err := w.Append(p); err != nil {
+			return fmt.Errorf("append %s: %w", p.SKU, err)
+		}
+	}
+	out, err := w.Write(filepath.Dir(ts.path), filepath.Base(ts.path))
+	if err != nil {
+		return fmt.Errorf("snapshot write %s: %w", ts.path, err)
+	}
+	if err := ts.holder.Reload(out); err != nil {
+		return fmt.Errorf("holder reload %s: %w", out, err)
+	}
+	return nil
+}
+
+// upsert inserts or replaces a projection in the live map. Used by the
+// initial-seed pipeline; the bus-update path goes through rebuild().
+func (ts *TenantState) upsert(p catalog.Projection) {
+	ts.mu.Lock()
+	defer ts.mu.Unlock()
+	ts.items[p.SKU] = p
+}
+
+// count returns the current item-set size under read lock.
+func (ts *TenantState) count() int {
+	ts.mu.RLock()
+	defer ts.mu.RUnlock()
+	return len(ts.items)
+}
+
+// sample uniformly picks up to n SKUs at random from the live map. Used by
+// the probe paths to validate atomic-reload under drain-grace.
+func (ts *TenantState) sample(rng *rand.Rand, n int) []string {
+	ts.mu.RLock()
+	defer ts.mu.RUnlock()
+	if len(ts.items) == 0 {
+		return nil
+	}
+	skus := make([]string, 0, len(ts.items))
+	for sku := range ts.items {
+		skus = append(skus, sku)
+	}
+	if n >= len(skus) {
+		return skus
+	}
+	out := make([]string, 0, n)
+	picked := make(map[int]struct{}, n)
+	for len(out) < n {
+		idx := rng.Intn(len(skus))
+		if _, dup := picked[idx]; dup {
+			continue
+		}
+		picked[idx] = struct{}{}
+		out = append(out, skus[idx])
+	}
+	return out
+}
+
+// get is a thin wrapper around holder.Get. Counts probes and misses
+// atomically — readable from the SIGINT shutdown path without a lock.
+func (ts *TenantState) get(sku string) (catalog.Projection, bool) {
+	ts.probeCount.Add(1)
+	p, ok := ts.holder.Get(sku)
+	if !ok {
+		ts.probeMisses.Add(1)
+	}
+	return p, ok
+}
+
+func main() {
+	schemasPath := flag.String("schemas", "", "JSON file mapping tenant -> catalog.Schema")
+	itemsPath := flag.String("items", "", "JSON file mapping tenant -> []source-item (optional; loadOrSynthesize falls back to -n synth per tenant if absent)")
+	outDir := flag.String("out-dir", "./out", "output directory; per-tenant files <out-dir>/<tenant>.bin")
+	sampleCount := flag.Int("sample", 200, "number of random SKUs to verify per probe call")
+	probeTick := flag.Duration("probe-tick", 2*time.Second, "watch mode: probe interval; one-shot ignores it")
+	tenantsCSV := flag.String("tenants", "default", "comma-separated tenant names to operate on")
+	emitN := flag.Int("n", 10000, "synthesised item count per tenant when -items is omitted")
+	seed := flag.Int64("seed", 42, "RNG seed for synthesised items / probe sample")
+	rabbitURL := flag.String("rabbit", "", "if set, watch mode consumes catalog.projection_published from this RabbitMQ URL")
+	exchange := flag.String("exchange", "catalog", "RabbitMQ topic exchange to bind on (default 'catalog' to match the writeradmin event family)")
+	drainGrace := flag.Duration("drain-grace", catalog.DefaultDrainGrace, "SnapshotHolder drain grace (override for tight tests)")
+	flag.Parse()
+
+	if *schemasPath == "" {
+		fmt.Fprintln(os.Stderr, "ERROR: -schemas is required")
+		flag.Usage()
+		os.Exit(2)
+	}
+
+	tenants := splitCSV(*tenantsCSV)
+	if len(tenants) == 0 {
+		fail("no tenants parsed from -tenants %q", *tenantsCSV)
+	}
+
+	reg := catalog.NewRegistry()
+	if err := reg.LoadFile(*schemasPath); err != nil {
+		fail("load schemas: %v", err)
+	}
+
+	if err := os.MkdirAll(*outDir, 0o755); err != nil {
+		fail("mkdir out-dir %q: %v", *outDir, err)
+	}
+
+	// Per-tenant state. Built once and shared across seed/bus/probe paths.
+	states := make(map[string]*TenantState, len(tenants))
+	for _, t := range tenants {
+		sch := reg.SchemaFor(t)
+		if sch.Tenant == "" {
+			sch.Tenant = t
+		}
+		path := filepath.Join(*outDir, t+".bin")
+		holder := catalog.NewSnapshotHolder(*drainGrace)
+		states[t] = newTenantState(t, sch, holder, path)
+	}
+	// Register cleanup for both the defer path (graceful shutdown / SIGINT)
+	// and the fail path (mid-run error). Defers run LIFO: rc.Close() (added
+	// later, in watch mode) fires first; this hook runs second. With rc
+	// nil, the extra teardown is harmless.
+	cleanup = func() {
+		for _, ts := range states {
+			_ = ts.holder.Close()
+		}
+	}
+	defer cleanup()
+	// Watch-mode rc is appended after this point; in watch its Close is
+	// added to the defer stack too, separately.
+
+	// Initial state pipeline: synthesise / load items, Apply per tenant's
+	// registered schema, upsert into each tenant's map, then rebuild() the
+	// snapshots so the SnapshotHolders are populated before any Get.
+	sources, err := catalogtest.LoadOrSynthesize(*itemsPath, tenants, *emitN, *seed, reg, "B")
+	if err != nil {
+		fail("load items: %v", err)
+	}
+	skipped := 0
+	for _, t := range tenants {
+		sch := reg.SchemaFor(t)
+		ts := states[t]
+		for _, src := range sources[t] {
+			p, err := catalog.Apply(sch, src)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "skip %s: %v\n", t, err)
+				skipped++
+				continue
+			}
+			ts.upsert(p)
+		}
+	}
+	for _, t := range tenants {
+		ts := states[t]
+		if err := ts.rebuild(nil); err != nil {
+			fail("initial rebuild %s: %v", t, err)
+		}
+		fmt.Printf("initial %s: %d items -> %s\n", t, ts.holder.Len(), ts.path)
+	}
+	fmt.Printf("seeded %d items across %d tenants, skipped %d\n", sumCounts(states), len(tenants), skipped)
+
+	watch := *rabbitURL != ""
+
+	if !watch {
+		if err := probe(states, tenants, *sampleCount, *seed+1); err != nil {
+			fail("probe miss: %v", err)
+		}
+		return
+	}
+
+	// Watch mode.
+	fmt.Printf("connecting to RabbitMQ: %s (exchange=%s)\n", *rabbitURL, *exchange)
+	rc, err := rabbit.Dial(*rabbitURL, "catalog-emit")
+	if err != nil {
+		fail("rabbit dial: %v", err)
+	}
+	defer rc.Close()
+
+	pattern := string(event.CatalogProjectionPublished)
+	bindOnce := func() {
+		ch, err := rc.Channel()
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "watch: open channel: %v\n", err)
+			return
+		}
+		if err := rabbit.ListenToPattern(ch, *exchange, pattern, makeHandler(states)); err != nil {
+			fmt.Fprintf(os.Stderr, "watch: bind pattern %q: %v\n", pattern, err)
+			_ = ch.Close()
+		}
+	}
+	rc.NotifyOnReconnect(bindOnce)
+
+	fmt.Printf("watching %q on exchange %q\n", pattern, *exchange)
+
+	go probeLoop(states, tenants, *sampleCount, *probeTick)
+
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+	sig := <-sigCh
+	fmt.Printf("received signal %s; shutting down (rebuilds=%d probes=%d misses=%d)\n",
+		sig,
+		sumRebuilds(states),
+		sumProbes(states),
+		sumMisses(states),
+	)
+}
+
+// routedEvent is the result of routeTenant — `tenant` is the key into the
+// states map; `fallback` tells the handler it should warn (a real
+// default-tenant publisher via Meta["tenant"]="default" does NOT set
+// fallback, so legitimate default traffic stays silent).
+type routedEvent struct {
+	tenant   string
+	fallback bool // no Meta[tenant] and no Source "catalog:<x>" prefix
+}
+
+// makeHandler returns the AMQP delivery handler that consumes
+// `catalog.projection_published`. It parses the event, picks the tenant
+// (Meta["tenant"] -> Source[:N] suffix -> "default"), decodes the
+// ProjectionUpdate batch, and feeds it to the per-tenant rebuild path.
+//
+// A deferred recover wraps the body so a panic in any single event
+// doesn't kill the consumer goroutine; the queue keeps draining on the
+// next delivery, and the reconnect loop re-binds cleanly on the next
+// broker hiccup.
+func makeHandler(states map[string]*TenantState) func(amqp.Delivery) error {
+	return func(d amqp.Delivery) (err error) {
+		defer func() {
+			if r := recover(); r != nil {
+				err = fmt.Errorf("handler panic recovered: %v", r)
+			}
+		}()
+		var ev event.Event
+		if err := json.Unmarshal(d.Body, &ev); err != nil {
+			return fmt.Errorf("decode event: %w", err)
+		}
+		r := routeTenant(&ev)
+		if r.fallback {
+			// Loud-warn so a misrouted publisher shows up in ops logs.
+			fmt.Fprintf(os.Stderr,
+				"watch WARN: event id=%s source=%s type=%s — no Meta[%q] and no \"catalog:<tenant>\" Source prefix — routing to default\n",
+				ev.ID, ev.Source, ev.Type, "tenant")
+		}
+		ts, ok := states[r.tenant]
+		if !ok {
+			// The publisher-decorated tenant exists but isn't in this
+			// harness's -tenants list. The event is silently dropped — a
+			// half-configured test environment would look like a broken
+			// wire. Surface loudly so the misconfig is visible from
+			// stdout.
+			fmt.Fprintf(os.Stderr,
+				"watch WARN: event id=%s source=%s type=%s routed to tenant=%q but this harness only tracks %v — dropping (check -tenants on the harness vs Meta[tenant]/Meta[country] on the publisher)\n",
+				ev.ID, ev.Source, ev.Type, r.tenant, trackedTenants(states))
+			return nil
+		}
+		updates, err := catalog.DecodeUpdates(ev.Payload)
+		if err != nil {
+			return fmt.Errorf("decode payload as []ProjectionUpdate: %w", err)
+		}
+		if err := ts.rebuild(updates); err != nil {
+			return fmt.Errorf("rebuild %s: %w", r.tenant, err)
+		}
+		return nil
+	}
+}
+
+// routeTenant extracts the tenant for a delivered event. Meta and Source
+// values are TrimSpace'd so a publisher that emits " default " or
+// "catalog: bygglagret " still lands in the right bucket. Lookup order:
+//
+//  1. Meta["tenant"] — explicit per-event tenant (preferred).
+//  2. Meta["country"] — compat fallback for finder writeradmin today
+//     (single-tenant-per-country writers).
+//  3. Source split — "catalog:<tenant>" → tenant portion.
+//  4. "default" — final fallback so an undecorated publisher still lands
+//     somewhere the harness can see it. Marked via `fallback: true` so the
+//     caller can log a warning, distinguishing a real default-tenant
+//     publisher from a missing-decoration one.
+func routeTenant(ev *event.Event) routedEvent {
+	if t, ok := ev.Meta["tenant"]; ok {
+		if t = strings.TrimSpace(t); t != "" {
+			return routedEvent{tenant: t}
+		}
+	}
+	if t, ok := ev.Meta["country"]; ok {
+		// Legacy finder writers (writeradmin today) only know country. Treat
+		// country as tenant in the single-tenant-per-country model; multi-
+		// tenant writers should set Meta["tenant"] explicitly.
+		if t = strings.TrimSpace(t); t != "" {
+			return routedEvent{tenant: t}
+		}
+	}
+	if strings.HasPrefix(ev.Source, "catalog:") {
+		if t := strings.TrimSpace(strings.TrimPrefix(ev.Source, "catalog:")); t != "" {
+			return routedEvent{tenant: t}
+		}
+	}
+	return routedEvent{tenant: "default", fallback: true}
+}
+
+// trackedTenants returns the sorted list of tenant names configured on
+// this harness. Used in misroute WARN lines so a half-configured test
+// environment is visible from stdout.
+func trackedTenants(states map[string]*TenantState) []string {
+	out := make([]string, 0, len(states))
+	for t := range states {
+		out = append(out, t)
+	}
+	return out
+}
+
+// sumCounts / sumRebuilds / sumProbes / sumMisses aggregate per-tenant
+// counters for ops visibility at shutdown. Reads are atomic; safe to call
+// concurrently with rebuild() / get().
+func sumCounts(states map[string]*TenantState) int {
+	n := 0
+	for _, ts := range states {
+		n += ts.count()
+	}
+	return n
+}
+
+func sumRebuilds(states map[string]*TenantState) uint64 {
+	var n uint64
+	for _, ts := range states {
+		n += ts.rebuildCount.Load()
+	}
+	return n
+}
+
+func sumProbes(states map[string]*TenantState) uint64 {
+	var n uint64
+	for _, ts := range states {
+		n += ts.probeCount.Load()
+	}
+	return n
+}
+
+func sumMisses(states map[string]*TenantState) uint64 {
+	var n uint64
+	for _, ts := range states {
+		n += ts.probeMisses.Load()
+	}
+	return n
+}
+
+// probe picks `sampleCount` random SKUs across all tenanted state and
+// looks them up via the SnapshotHolder. Misses are reported and accumulated;
+// on a miss, an error is returned so the caller can fail loud (one-shot)
+// or log (watch).
+func probe(states map[string]*TenantState, tenants []string, sampleCount int, seed int64) error {
+	rng := rand.New(rand.NewSource(seed))
+	misses := uint64(0)
+	hits := uint64(0)
+	start := time.Now()
+	for _, t := range tenants {
+		ts := states[t]
+		skus := ts.sample(rng, sampleCount)
+		for _, sku := range skus {
+			if _, ok := ts.get(sku); ok {
+				hits++
+			} else {
+				fmt.Fprintf(os.Stderr, "MISS: %s/%s\n", t, sku)
+				misses++
+			}
+		}
+	}
+	elapsed := time.Since(start)
+	total := int(hits + misses)
+	if total == 0 {
+		return nil
+	}
+	fmt.Printf("probe: %d hits, %d misses, %v (%v/lookup)\n",
+		hits, misses, elapsed, elapsed/time.Duration(total))
+	if misses > 0 {
+		return fmt.Errorf("%d probe misses", misses)
+	}
+	return nil
+}
+
+// probeLoop runs probe() on every tick; logs misses but does NOT exit. It
+// is the watch-mode heartbeat that validates the holder under live reloads.
+func probeLoop(states map[string]*TenantState, tenants []string, sampleCount int, tick time.Duration) {
+	t := time.NewTicker(tick)
+	defer t.Stop()
+	seed := int64(1)
+	for range t.C {
+		_ = probe(states, tenants, sampleCount, seed)
+		seed++
+	}
+}
+
+// splitCSV trims and splits a comma-separated tenant list into a slice.
+func splitCSV(s string) []string {
+	if s == "" {
+		return nil
+	}
+	out := make([]string, 0, 2)
+	for _, p := range strings.Split(s, ",") {
+		p = strings.TrimSpace(p)
+		if p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+// cleanup runs the registered shutdown hooks (holder file mmap unmaps, AMQP
+// connection close) before exit. main() populates `cleanup` with a function
+// that closes all per-tenant SnapshotHolders and rc.Close if rc was opened;
+// fail() invokes the same hook before os.Exit so a mid-run failure does not
+// leak mmap mappings or AMQP sockets.
+//
+// We can't just rely on defers because os.Exit skips them — and the holder
+// close is the only defer that's registered before several possible fail()
+// sites (initial rebuild, one-shot probe miss).
+var cleanup = func() {}
+
+// fail prints a FATAL line, runs registered cleanup hooks, then exits 1.
+func fail(format string, a ...any) {
+	fmt.Fprintf(os.Stderr, "FATAL: "+format+"\n", a...)
+	cleanup()
+	os.Exit(1)
+}

cmd/event-emit-publisher/main.go

@@ -0,0 +1,314 @@
+// Package main — `event-emit-publisher` is the test-producer for the
+// catalog.projection_published wire. It drives the same shape the
+// slask-finder writeradmin emits, so cmd/catalog-emit (consumer harness)
+// and go-cart-actor/inventory (consumer bridge) can be exercised
+// end-to-end without a live writer admin in the loop.
+//
+// Two modes:
+//
+//  1. Tick (default): every -tick interval, build a batch of -batch-size
+//     upserts (mixed with a small fraction of deletes for the cart bridge to
+//     see), publish one catalog.projection_published event per tenant,
+//     repeat.
+//  2. One-shot (-oneshot): publish a single batch and exit — useful for CI
+//     and one-line smoke checks.
+//
+// Usage:
+//
+//	event-emit-publisher \
+//	  -schemas /etc/platform/catalog/schemas.json \
+//	  -tenants default,bygglagret[,se] \
+//	  -rabbit amqp://admin:devpass@localhost:5672/ \
+//	  -exchange catalog \
+//	  -batch-size 200 -tick 5s -jitter 1s [-oneshot]
+package main
+
+import (
+	"flag"
+	"fmt"
+	"math/rand"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"git.k6n.net/mats/platform/catalog"
+	"git.k6n.net/mats/platform/cmd/internal/catalogtest"
+	"git.k6n.net/mats/platform/event"
+	"git.k6n.net/mats/platform/rabbit"
+)
+
+// perTenantPublisher holds the batches+counter for one tenant so we can
+// emit one event per tenant per tick without rebuilding the schema work.
+type perTenantPublisher struct {
+	tenant string
+	schema catalog.Schema
+	count  atomic.Uint64 // batches published for this tenant
+}
+
+func main() {
+	schemasPath := flag.String("schemas", "", "JSON file mapping tenant -> catalog.Schema")
+	itemsPath := flag.String("items", "", "JSON file mapping tenant -> []source-item (optional; falls back to synthetic items when absent)")
+	tenantsCSV := flag.String("tenants", "default", "comma-separated tenant names (also used as Meta[tenant]")
+	emitN := flag.Int("n", 10000, "synthesised total item count per tenant when -items is omitted")
+	batchSize := flag.Int("batch-size", 100, "items per published batch per tenant")
+	deleteRatio := flag.Float64("delete-ratio", 0.05, "fraction of each batch marked deleted (0..1)")
+	rabbitURL := flag.String("rabbit", "amqp://admin:devpass@localhost:5672/", "RabbitMQ URL")
+	exchange := flag.String("exchange", "catalog", "RabbitMQ topic exchange to publish on (default 'catalog')")
+	tickDuration := flag.Duration("tick", 5*time.Second, "interval between batches (0 disables ticking)")
+	jitter := flag.Duration("jitter", 1*time.Second, "max random delay added on top of tick")
+	seed := flag.Int64("seed", 42, "RNG seed")
+	oneshot := flag.Bool("oneshot", false, "publish a single batch per tenant and exit")
+	maxBatches := flag.Int("max-batches", 0, "stop after this many batches per tenant (0 = unlimited)")
+	eventIDPrefix := flag.String("event-id-prefix", "pub-", "prefix for emitted event IDs")
+	flag.Parse()
+
+	if *schemasPath == "" {
+		fmt.Fprintln(os.Stderr, "ERROR: -schemas is required")
+		flag.Usage()
+		os.Exit(2)
+	}
+	if *batchSize <= 0 {
+		fail("batch-size must be > 0, got %d", *batchSize)
+	}
+	if *deleteRatio < 0 || *deleteRatio > 1 {
+		fail("delete-ratio must be in [0,1], got %v", *deleteRatio)
+	}
+
+	tenants := splitCSV(*tenantsCSV)
+	if len(tenants) == 0 {
+		fail("no tenants parsed from -tenants %q", *tenantsCSV)
+	}
+
+	reg := catalog.NewRegistry()
+	if err := reg.LoadFile(*schemasPath); err != nil {
+		fail("load schemas: %v", err)
+	}
+
+	sources, err := catalogtest.LoadOrSynthesize(*itemsPath, tenants, *emitN, *seed, reg, "P")
+	if err != nil {
+		fail("load items: %v", err)
+	}
+
+	// Build per-tenant schema and pre-applied projections once (the cart
+	// and the harness hot path both want the typed Projection shape).
+	pubs := make([]*perTenantPublisher, 0, len(tenants))
+	for _, t := range tenants {
+		sch := reg.SchemaFor(t)
+		if sch.Tenant == "" {
+			sch.Tenant = t
+		}
+		if _, ok := sources[t]; !ok {
+			fail("tenant %q has no source items (check -tenants and -items / -n)", t)
+		}
+		pubs = append(pubs, &perTenantPublisher{tenant: t, schema: sch})
+	}
+
+	// Dial RabbitMQ. We use the rabbit.Conn wrapper so the publisher can run
+	// across broker hiccups.
+	rc, err := rabbit.Dial(*rabbitURL, "event-emit-publisher")
+	if err != nil {
+		fail("rabbit dial %q: %v", *rabbitURL, err)
+	}
+	// Register cleanup so a later fail() won't leak the AMQP socket, and
+	// also so graceful shutdown closes the connection cleanly.
+	cleanup = func() { _ = rc.Close() }
+	defer cleanup()
+
+	if err := declareExchange(rc, *exchange); err != nil {
+		fail("declare exchange %q: %v", *exchange, err)
+	}
+	fmt.Printf("publishing to %s on exchange %q (tenants=%v batch=%d tick=%v jitter=%v oneshot=%v)\n",
+		*rabbitURL, *exchange, tenants, *batchSize, *tickDuration, *jitter, *oneshot)
+
+	rng := rand.New(rand.NewSource(*seed))
+
+	publishOnce := func() {
+		for _, pub := range pubs {
+			updates, skipped := buildBatch(pub, sources[pub.tenant], *batchSize, *deleteRatio, rng)
+			ev := buildEvent(pub, updates, *eventIDPrefix)
+			if err := publishEvent(rc, *exchange, ev); err != nil {
+				fmt.Fprintf(os.Stderr, "publish %s: %v\n", pub.tenant, err)
+				continue
+			}
+			n := pub.count.Add(1)
+			fmt.Printf("tenant=%s batch=%d upserts=%d skipped=%d eventID=%s\n",
+				pub.tenant, n, len(updates), skipped, ev.ID)
+		}
+	}
+
+	if *oneshot {
+		publishOnce()
+		return
+	}
+
+	if *tickDuration <= 0 {
+		fail("watch-mode requires -tick > 0 (use -oneshot to emit one batch and exit)")
+	}
+
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-sigCh
+		fmt.Fprintln(os.Stderr, "received signal; shutting down")
+		// bulk-total counter for the SIGINT summary
+		for _, pub := range pubs {
+			fmt.Printf("summary tenant=%s batches=%d\n", pub.tenant, pub.count.Load())
+		}
+		os.Exit(0)
+	}()
+
+	t := time.NewTicker(*tickDuration)
+	defer t.Stop()
+	batches := 0
+	for {
+		select {
+		case <-t.C:
+			publishOnce()
+			batches++
+			if *maxBatches > 0 && batches >= *maxBatches {
+				fmt.Printf("reached max-batches=%d per tenant, exiting\n", *maxBatches)
+				return
+			}
+			if *jitter > 0 {
+				// slop the next tick by a small random fraction of the jitter
+				time.Sleep(time.Duration(rng.Int63n(int64(*jitter))))
+			}
+		}
+	}
+}
+
+// buildBatch picks batchSize items from sources, applies the schema, and
+// returns a []ProjectionUpdate batch (upserts and deletes). deleteRatio of
+// items become deletion markers (with the same Schema on apply but
+// upsert=false) so the cart bridge sees both shapes.
+//
+// skipped counts items that failed Apply — leave them out rather than
+// poisoning the batch (mirrors writeradmin's behavior).
+func buildBatch(pub *perTenantPublisher, sourceItems []map[string]any, batchSize int, deleteRatio float64, rng *rand.Rand) ([]catalog.ProjectionUpdate, int) {
+	updates := make([]catalog.ProjectionUpdate, 0, batchSize)
+	skipped := 0
+	n := len(sourceItems)
+	if n == 0 {
+		return updates, 0
+	}
+	for i := 0; i < batchSize; i++ {
+		idx := rng.Intn(n)
+		src := sourceItems[idx]
+		isDelete := rng.Float64() < deleteRatio
+		if isDelete {
+			sku := catalogtest.StringField(src, pub.schema.SKUField)
+			if sku == "" {
+				skipped++
+				continue
+			}
+			updates = append(updates, catalog.ProjectionUpdate{
+				Projection: catalog.Projection{
+					SKU: sku,
+					ID:  pub.schema.IDPrefix + sku,
+				},
+				Deleted: true,
+			})
+			continue
+		}
+		p, err := catalog.Apply(pub.schema, src)
+		if err != nil {
+			skipped++
+			continue
+		}
+		updates = append(updates, catalog.ProjectionUpdate{Projection: p})
+	}
+	return updates, skipped
+}
+
+// buildEvent wraps a batch of updates into one envelope.Event matching the
+// publisher conventions used by slask-finder/writeradmin/projection-emit.go.
+//
+// Source uses the "catalog:<tenant>" prefix (consumer's routeTenant splits
+// on ':'), and Meta carries both "tenant" AND "country" (the latter for
+// compat with single-tenant-per-country models). Push a stable unique ID
+// so a replay tool can dedupe.
+func buildEvent(pub *perTenantPublisher, updates []catalog.ProjectionUpdate, idPrefix string) event.Event {
+	payload, err := catalog.EncodeUpdates(updates)
+	if err != nil {
+		// EncodeUpdates only errors on a fundamentally broken batch; surface loud.
+		fail("encode updates for %s: %v", pub.tenant, err)
+	}
+	return event.Event{
+		ID:      idPrefix + pub.tenant + "-" + strconv.FormatInt(time.Now().UnixNano(), 36),
+		Type:    event.CatalogProjectionPublished,
+		Source:  "catalog:" + pub.tenant,
+		Subject: pub.tenant,
+		Payload: payload,
+		Time:    time.Now(),
+		Meta: map[string]string{
+			"tenant":  pub.tenant,
+			"country": pub.tenant, // single-tenant-per-country model
+			"emitter": "event-emit-publisher",
+		},
+	}
+}
+
+// publishEvent publishes the envelope via rabbit.PublishEvent (which
+// opens a channel per call — fine for a low-rate test publisher). Routing
+// key inside PublishEvent is implied as string(ev.Type); here we want
+// "catalog.projection_published" which is exactly that. We pass the
+// underlying *amqp.Connection that rc.Connection() returns.
+//
+// Note: PublishEvent transiently opens + closes a channel per publish.
+// At -tick ≈ 5s that's negligible, but if this CLI is ever pushed to
+// high rates, swap to a single long-lived channel + Publish().
+func publishEvent(rc *rabbit.Conn, exchange string, ev event.Event) error {
+	return rabbit.PublishEvent(rc.Connection(), exchange, ev)
+}
+
+// declareExchange opens a channel and issue-s a topic exchange declare so a
+// fresh broker has the topology the harness's reader and the cart bridge
+// expect. Idempotent.
+func declareExchange(rc *rabbit.Conn, exchange string) error {
+	ch, err := rc.Channel()
+	if err != nil {
+		return fmt.Errorf("open channel: %w", err)
+	}
+	defer ch.Close()
+	if err := ch.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
+		return fmt.Errorf("exchange declare: %w", err)
+	}
+	return nil
+}
+
+// splitCSV trims and splits a comma-separated tenant list.
+func splitCSV(s string) []string {
+	if s == "" {
+		return nil
+	}
+	out := make([]string, 0, 2)
+	for _, p := range strings.Split(s, ",") {
+		p = strings.TrimSpace(p)
+		if p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+// cleanup runs the registered shutdown hooks before os.Exit so a mid-run
+// failure does not leak the AMQP connection (or any other resource
+// registered later). main() assigns a closure to cleanup that closes the
+// rabbit.Conn; fail() invokes it before exiting.
+//
+// We can't rely solely on defer cleanup() because os.Exit skips defers; the
+// fail path is reachable from several sites before a defer would normally
+// fire, plus the connection may be the only system resource the binary
+// opened. Same pattern as platform/cmd/catalog-emit's cleanup hook.
+var cleanup = func() {}
+
+// fail prints a FATAL line, runs registered cleanup, and exits with code 1.
+func fail(format string, a ...any) {
+	fmt.Fprintf(os.Stderr, "FATAL: "+format+"\n", a...)
+	cleanup()
+	os.Exit(1)
+}

cmd/internal/catalogtest/items.go

@@ -0,0 +1,138 @@
+// Package catalogtest holds shared helpers for the platform catalog test
+// harnesses — cmd/catalog-emit (consumer harness) and cmd/event-emit-publisher
+// (test publisher) both use these to seed a source-item set with the same
+// shape conventions. Anything in this package is a test-only utility; do
+// not import from production code.
+//
+// Scope: this package is parked under platform/cmd/internal/ on purpose.
+// Go's internal-package rule restricts imports to packages within the
+// parent directory's subtree, so only platform/cmd/* binaries can import
+// it. That keeps the test-helper API out of the public platform surface
+// (e.g. platform/event, platform/rabbit must not import catalogtest —
+// they're cross-module seams, not test scaffolding).
+package catalogtest
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"os"
+	"strconv"
+
+	"git.k6n.net/mats/platform/catalog"
+)
+
+// LoadOrSynthesize returns one source-batch per tenant name. If itemsPath
+// is non-empty, it parses a {"tenant":[...]}-keyed JSON map and returns
+// it unchanged. Otherwise it generates -n synthetic items per tenant
+// using the tenant's registered Schema's field names, with skuPrefix
+// prepended to each SKU.
+//
+// schemaFor is supplied so callers must already have a Registry built
+// (production code paths usually have a Registry at composition root;
+// publishing this helper to test code keeps the hot path clean).
+//
+// skuPrefix is a single-character prefix that lets log lines / probe
+// output attribute synthetic items to the binary that produced them
+// ("B" for catalog-emit, "P" for publisher). Substituting per-binary
+// retains recognisability in mixed test runs.
+//
+// Determinism: synth items are content-version-identical between runs at
+// the same (schema, seed, i). UpdatedAt is Unix-ms seeded from `i`.
+func LoadOrSynthesize(itemsPath string, tenants []string, n int, seed int64, reg *catalog.Registry, skuPrefix string) (map[string][]map[string]any, error) {
+	if itemsPath != "" {
+		data, err := os.ReadFile(itemsPath)
+		if err != nil {
+			return nil, err
+		}
+		var raw map[string][]map[string]any
+		if err := json.Unmarshal(data, &raw); err != nil {
+			return nil, fmt.Errorf("parse items: %w", err)
+		}
+		return raw, nil
+	}
+	rng := rand.New(rand.NewSource(seed))
+	out := make(map[string][]map[string]any, len(tenants))
+	for _, t := range tenants {
+		s := reg.SchemaFor(t)
+		items := make([]map[string]any, n)
+		for i := 0; i < n; i++ {
+			items[i] = SynthItem(s, rng, i, skuPrefix)
+		}
+		out[t] = items
+	}
+	return out, nil
+}
+
+// SynthItem builds one source-map whose key names match the supplied
+// schema's field names. catalog.Apply on (s, item) succeeds for any
+// schema that uses the same field-name conventions as DefaultSchema() /
+// BygglagretSchema() / a Registry-supplied override.
+//
+// skuPrefix is prepended to each SKU (e.g. "B" or "P"); reasonable
+// titles and prices are derived from a stable per-i seed plus the rng.
+// Title text is intentionally uniform across binaries so log lines
+// diff-cleanly between catalog-emit and event-emit-publisher runs.
+//
+// UpdatedAt is deterministic per i (Unix-ms, stable across runs) so the
+// synthesised projection is content-version-identical between runs at
+// the same (schema, seed, i).
+func SynthItem(s catalog.Schema, rng *rand.Rand, i int, skuPrefix string) map[string]any {
+	sku := fmt.Sprintf("%s%07d", skuPrefix, i)
+	m := map[string]any{}
+	if s.SKUField != "" {
+		m[s.SKUField] = sku
+	}
+	if s.TitleField != "" {
+		m[s.TitleField] = fmt.Sprintf("Catalog Item %07d", i)
+	}
+	if s.PriceField != "" {
+		m[s.PriceField] = int64(99 + rng.Intn(9000))
+	}
+	if s.CurrencyField != "" {
+		m[s.CurrencyField] = "SEK"
+	}
+	if s.TaxClassField != "" {
+		m[s.TaxClassField] = "standard"
+	}
+	if s.InventoryTrackedF != "" {
+		m[s.InventoryTrackedF] = rng.Intn(2) == 0
+	}
+	if s.DropShipF != "" {
+		m[s.DropShipF] = rng.Intn(8) == 0
+	}
+	if s.CategoryField != "" {
+		m[s.CategoryField] = fmt.Sprintf("c-%d", rng.Intn(50))
+	}
+	if s.BrandField != "" {
+		m[s.BrandField] = "Acme"
+	}
+	if s.UpdatedAtField != "" {
+		m[s.UpdatedAtField] = int64(1700000000000 + i)
+	}
+	return m
+}
+
+// StringField returns the string value of a source key, accepting
+// strings or integral-number source values (mirrors writeradmin's
+// helper of the same name). Used by delete-marker detection where
+// the body field may be either JSON-string "999" or JSON-number 999.
+//
+// Empty key returns ""; unknown type returns "". Caller decides
+// what "" means (in writeradmin / publisher buildBatch it means
+// "skip this delete — no SKU to delete on").
+func StringField(src map[string]any, key string) string {
+	if key == "" {
+		return ""
+	}
+	switch v := src[key].(type) {
+	case string:
+		return v
+	case float64:
+		if v == float64(int64(v)) {
+			return strconv.FormatInt(int64(v), 10)
+		}
+		return strconv.FormatFloat(v, 'f', -1, 64)
+	}
+	return ""
+}