platform/cmd/catalog-emit/main.go

// 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)
}