all the refactor

cmd/cart/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"log"
 	"net"
@@ -10,6 +11,7 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"sync"
 	"time"
 
 	"git.k6n.net/mats/go-cart-actor/internal/ucp"
@@ -21,8 +23,10 @@ import (
 	"git.k6n.net/mats/go-cart-actor/pkg/telemetry"
 	"git.k6n.net/mats/go-cart-actor/pkg/voucher"
 	"git.k6n.net/mats/go-redis-inventory/pkg/inventory"
+	"git.k6n.net/mats/platform/catalog"
 	"git.k6n.net/mats/platform/config"
-	"git.k6n.net/mats/slask-finder/pkg/messaging"
+	"git.k6n.net/mats/platform/event"
+	"git.k6n.net/mats/platform/rabbit"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -116,6 +120,123 @@ func matchesSkuAndLocation(update inventory.InventoryResult, item cart.CartItem)
 	return false
 }
 
+// catalogProjectionCache is the cart's in-process map of SKU → catalog.Projection,
+// updated by reading catalog.projection_published events off the bus. It is the
+// cart-side source of truth for catalog facts (price, tax_class, inventory_tracked
+// flag, currency, image, brand) used at add-to-cart / checkout-reserve decisions.
+// Stock state is NOT cached here — exact qty is still queried synchronously from
+// the inventory service at decision points per docs/inventory-shape.md.
+//
+// Concurrency: read-mostly workload justifies sync.RWMutex.
+//
+// Single-tier storage. The bus is the only writer — the cart does not separately
+// seed from a snapshot file at boot, because the access pattern is point lookup
+// of cart-active SKUs only, and coupling two sources (mmap snapshot + bus-fed
+// map) created duplicated state with subtle drift risk and an mmap-SIGBUS hazard.
+// Cold-grace: the cache may be incomplete for an SKU the cart sees before the
+// first projection_published arrives for it; the cart falls back to the existing
+// PRODUCT_BASE_URL HTTP fetch and overlays the cache fields on top of it.
+//
+// Deletes carry a tombstone in the map (not a plain delete) so a bus delete of
+// an SKU that already fits the cache rules correctly shadows it for Get /
+// IsDeleted lookups — preventing an oversight from re-fetching a deleted SKU
+// before the cache rebuilds. Tombstones have a per-pod lifetime; bus-driven,
+// so the next sweep on the same SKU flips the entry back to live.
+type catalogProjectionCache struct {
+	mu    sync.RWMutex
+	items map[string]projectionEntry
+}
+
+// projectionEntry is a cache slot: a live projection, or a tombstone (deleted)
+// that shadows the live fallback.
+type projectionEntry struct {
+	proj    catalog.Projection
+	deleted bool
+}
+
+func newCatalogProjectionCache() *catalogProjectionCache {
+	return &catalogProjectionCache{items: make(map[string]projectionEntry, 8192)}
+}
+
+func (c *catalogProjectionCache) Apply(updates []catalog.ProjectionUpdate) (upserts, deletes int) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for _, u := range updates {
+		if u.Deleted {
+			c.items[u.SKU] = projectionEntry{deleted: true}
+			deletes++
+			continue
+		}
+		if u.SKU == "" {
+			continue
+		}
+		c.items[u.SKU] = projectionEntry{proj: u.Projection}
+		upserts++
+	}
+	return
+}
+
+func (c *catalogProjectionCache) Get(sku string) (catalog.Projection, bool) {
+	c.mu.RLock()
+	e, ok := c.items[sku]
+	c.mu.RUnlock()
+	if !ok {
+		return catalog.Projection{}, false
+	}
+	if e.deleted {
+		return catalog.Projection{}, false
+	}
+	return e.proj, true
+}
+
+// IsDeleted reports whether a tombstone exists for sku — used by HTTP-fetch
+// call sites to skip the fallback for an SKU the bus has explicitly deleted,
+// avoiding a wasteful round-trip on items the writer has just removed.
+func (c *catalogProjectionCache) IsDeleted(sku string) bool {
+	c.mu.RLock()
+	e, ok := c.items[sku]
+	c.mu.RUnlock()
+	return ok && e.deleted
+}
+
+// Len reports the number of map entries (bus-fed upserts + tombstones). Used
+// for log gauges only.
+func (c *catalogProjectionCache) Len() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return len(c.items)
+}
+
+// projectionDeliveryHandler is the AMQP delivery handler for
+// `catalog.projection_published` on the `catalog` topic exchange. Decodes the
+// []ProjectionUpdate batch and merges it into cache. Returns nil on a batch
+// that doesn't apply locally so the AMQP delivery is acked; returns an error
+// on a decode failure so the broker knows to redeliver / quarantine.
+//
+// Recovery: any panic in a single event is caught so the goroutine keeps
+// draining the queue; the bus stays live across malformed events.
+func projectionDeliveryHandler(cache *catalogProjectionCache) func(amqp.Delivery) error {
+	return func(d amqp.Delivery) (err error) {
+		defer func() {
+			if r := recover(); r != nil {
+				err = fmt.Errorf("projection handler panic recovered: %v", r)
+			}
+		}()
+		var ev event.Event
+		if err := json.Unmarshal(d.Body, &ev); err != nil {
+			return fmt.Errorf("decode event envelope: %w", err)
+		}
+		updates, err := catalog.DecodeUpdates(ev.Payload)
+		if err != nil {
+			return fmt.Errorf("decode projection updates: %w", err)
+		}
+		upserts, deletes := cache.Apply(updates)
+		log.Printf("cart: catalog projection batch applied: upserts=%d deletes=%d total=%d eventID=%s source=%q",
+			upserts, deletes, cache.Len(), ev.ID, ev.Source)
+		return nil
+	}
+}
+
 func main() {
 
 	// cartPort is the bare HTTP port. It drives both the local listener and the
@@ -282,29 +403,48 @@ func main() {
 
 	cartMCP := cartmcp.New(pool)
 
-	// Publish each applied mutation to the "cart"/"mutation" RabbitMQ topic so the
-	// backoffice /commerce live feed (and other consumers) see cart activity.
-	// Best-effort: if AMQP is unreachable the cart still serves; the feed stays empty.
+	// Stream each applied mutation to the shared "mutations" exchange (routing key
+	// mutation.cart) so the backoffice /commerce live feed (and other consumers)
+	// see cart activity. Best-effort: if AMQP is unreachable the cart still serves.
 	if amqpUrl != "" {
-		if conn, derr := amqp.Dial(amqpUrl); derr != nil {
+		if conn, derr := rabbit.Dial(amqpUrl, "cart"); derr != nil {
 			log.Printf("cart: AMQP connect failed, mutation feed disabled: %v", derr)
 		} else {
-			if ch, cerr := conn.Channel(); cerr == nil {
-				_ = messaging.DefineTopic(ch, "cart", "mutation") // idempotent; non-fatal
-				_ = ch.Close()
-			}
-			pool.AddListener(actor.NewAmqpListener(conn, func(id uint64, results []actor.ApplyResult) (any, error) {
-				types := make([]string, 0, len(results))
-				for _, r := range results {
-					types = append(types, r.Type)
-				}
-				return map[string]any{"cartId": id, "mutations": types}, nil
-			}))
-			log.Printf("cart: mutation feed enabled (cart/mutation)")
+			listener := actor.NewAmqpListener(conn.Connection(), "cart", actor.MutationSummary)
+			listener.DefineTopics()
+			pool.AddListener(listener)
+			log.Printf("cart: mutation feed enabled (mutation.cart)")
 		}
 	}
 
-	syncedServer := NewPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp)) //inventoryService, inventoryReservationService)
+	// catalog.projection_published consumer (the cart is the canonical
+	// consumer of the projection wire per docs/inventory-shape.md; inventory
+	// emits level crossings back to finder only, never the reverse). Own
+	// connection keeps the cart's mutation-feed lifecycle decoupled from a
+	// read-write-consume connection that could grow event handlers later
+	// (e.g. promotions reacting to a price drop on a watched SKU).
+	projectionCache := newCatalogProjectionCache()
+	if amqpUrl != "" {
+		if pConn, perr := rabbit.Dial(amqpUrl, "cart-catalog-projection"); perr != nil {
+			log.Printf("cart: catalog projection consumer DISABLED (dial failed): %v", perr)
+		} else {
+			pConn.NotifyOnReconnect(func() {
+				ch, err := pConn.Channel()
+				if err != nil {
+					log.Printf("cart: channel on projection reconnect: %v", err)
+					return
+				}
+				if err := rabbit.ListenToPattern(ch, "catalog", string(event.CatalogProjectionPublished), projectionDeliveryHandler(projectionCache)); err != nil {
+					log.Printf("cart: bind catalog.projection_published on reconnect: %v", err)
+					_ = ch.Close()
+				}
+			})
+			defer func() { _ = pConn.Close() }()
+			log.Printf("cart: catalog projection consumer ENABLED (catalog.projection_published on 'catalog' exchange)")
+		}
+	}
+
+	syncedServer := NewPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp), projectionCache) //inventoryService, inventoryReservationService)
 
 	app := &App{
 		pool:   pool,
@@ -374,6 +514,23 @@ func main() {
 	debugMux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
 	debugMux.HandleFunc("/debug/pprof/trace", pprof.Trace)
 	debugMux.Handle("/metrics", promhttp.Handler())
+
+	// Projection cache probe (dev/verification). GET /debug/projection/{sku} →
+	// the resolved projection + the bus-fed map size. map_entries counts both
+	// live projections and tombstones; a found=true response confirms the bus
+	// consumer delivered the SKU to this pod.
+	debugMux.HandleFunc("/debug/projection/", func(w http.ResponseWriter, r *http.Request) {
+		sku := strings.TrimPrefix(r.URL.Path, "/debug/projection/")
+		p, found := projectionCache.Get(sku)
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"sku":         sku,
+			"found":       found,
+			"projection":  p,
+			"map_entries": projectionCache.Len(),
+			"deleted":     projectionCache.IsDeleted(sku),
+		})
+	})
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
 		// Grain pool health: simple capacity check (mirrors previous GrainHandler.IsHealthy)
 		grainCount, capacity := app.pool.LocalUsage()
@@ -434,19 +591,11 @@ func main() {
 		srvErr <- srv.ListenAndServe()
 	}()
 
-	// listener := inventory.NewInventoryChangeListener(rdb, context.Background(), func(changes []inventory.InventoryChange) {
-	// 	for _, change := range changes {
-	// 		log.Printf("inventory change: %v", change)
-	// 		inventoryPubSub.Publish(change)
-	// 	}
-	// })
-
-	// go func() {
-	// 	err := listener.Start()
-	// 	if err != nil {
-	// 		log.Fatalf("Unable to start inventory listener: %v", err)
-	// 	}
-	// }()
+	// Inventory change consumption used to live here over the bare Redis
+	// `inventory_changed` channel; it is now owned by the cart-inventory
+	// service, which translates quantity changes into inventory.level_changed
+	// bus crossings. The cart reads exact stock synchronously when it needs it.
+	// See docs/inventory-shape.md.
 
 	log.Printf("Server started at %s (debug %s)", httpAddr, debugAddr)