update cart

cmd/backoffice/main.go

@@ -24,8 +24,8 @@ func main() {
 	amqpURL := os.Getenv("AMQP_URL")
 
 	app, err := backofficeadmin.New(backofficeadmin.Config{
-		DataDir:         envOrDefault("DATA_DIR", "data"),
-		CheckoutDataDir: envOrDefault("CHECKOUT_DATA_DIR", "checkout-data"),
+		DataDir:         envOrDefault("CART_DIR", "data"),
+		CheckoutDataDir: envOrDefault("CHECKOUT_DIR", "checkout-data"),
 		RedisAddress:    os.Getenv("REDIS_ADDRESS"),
 		RedisPassword:   os.Getenv("REDIS_PASSWORD"),
 	})

cmd/cart/main.go

@@ -18,6 +18,8 @@ import (
 	"git.k6n.net/mats/go-cart-actor/pkg/proxy"
 	"git.k6n.net/mats/go-cart-actor/pkg/voucher"
 	"git.k6n.net/mats/go-redis-inventory/pkg/inventory"
+	"git.k6n.net/mats/slask-finder/pkg/messaging"
+	amqp "github.com/rabbitmq/amqp091-go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -153,7 +155,7 @@ func main() {
 		}),
 	)
 
-	diskStorage := actor.NewDiskStorage[cart.CartGrain]("data", reg)
+	diskStorage := actor.NewDiskStorage[cart.CartGrain](getEnv("CART_DIR", "data"), reg)
 	poolConfig := actor.GrainPoolConfig[cart.CartGrain]{
 		MutationRegistry: reg,
 		Storage:          diskStorage,
@@ -215,6 +217,28 @@ func main() {
 		log.Fatalf("Error creating cart pool: %v\n", err)
 	}
 
+	// 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.
+	if amqpUrl != "" {
+		if conn, derr := amqp.Dial(amqpUrl); 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)")
+		}
+	}
+
 	syncedServer := NewPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp)) //inventoryService, inventoryReservationService)
 
 	app := &App{

cmd/cart/product-fetcher.go

@@ -186,7 +186,7 @@ func ToItemAddMessage(item *ProductItem, parent *ProductItem, storeId *string, q
 
 	// Top-level items price from their own product; children are priced from the
 	// parent product via the accessory-price hook.
-	price := int64(item.Price * 100)
+	price := int64(item.Price)
 	if parent != nil {
 		price = accessoryPrice(parent, item)
 	}

cmd/checkout/main.go

@@ -68,7 +68,11 @@ func main() {
 		log.Fatalf("Error creating inventory service: %v\n", err)
 	}
 
-	diskStorage := actor.NewDiskStorage[checkout.CheckoutGrain]("data", reg)
+	checkoutDir := os.Getenv("CHECKOUT_DIR")
+	if checkoutDir == "" {
+		checkoutDir = "data"
+	}
+	diskStorage := actor.NewDiskStorage[checkout.CheckoutGrain](checkoutDir, reg)
 
 	poolConfig := actor.GrainPoolConfig[checkout.CheckoutGrain]{
 		MutationRegistry: reg,

cmd/inventory/main.go

@@ -132,20 +132,14 @@ func main() {
 		}
 		// items listener
 		err = messaging.ListenToTopic(ch, country, "item_added", func(d amqp.Delivery) error {
-			var items []*index.DataItem
-			err := json.Unmarshal(d.Body, &items)
-			if err == nil {
-				log.Printf("Got upserts %d, message count %d", len(items), d.MessageCount)
-				wg := &sync.WaitGroup{}
-				for _, item := range items {
-					stockhandler.HandleItem(item, wg)
-				}
+			wg := &sync.WaitGroup{}
+			err = index.ForEachRawDataItemInJSONArray(d.Body, func(item *index.RawDataItem) error {
+				stockhandler.HandleItem(item, wg)
 				wg.Wait()
 				log.Print("Batch done...")
-			} else {
-				log.Printf("Failed to unmarshal upsert message %v", err)
-			}
-			return err
+				return err
+			})
+			return nil
 		})
 		if err != nil {
 			log.Fatalf("Failed to listen to item_added topic: %v", err)

cmd/inventory/stockhandler.go

@@ -3,8 +3,6 @@ package main
 import (
 	"context"
 	"log"
-	"strconv"
-	"strings"
 	"sync"
 
 	"git.k6n.net/mats/go-redis-inventory/pkg/inventory"
@@ -23,25 +21,22 @@ func (s *StockHandler) HandleItem(item types.Item, wg *sync.WaitGroup) {
 	wg.Go(func() {
 		ctx := s.ctx
 		pipe := s.rdb.Pipeline()
-		centralStockString, ok := item.GetStringFieldValue(3)
+		centralStock, ok := item.GetNumberFieldValue("inStock")
 		if !ok {
-			centralStockString = "0"
+			centralStock = 0
 		}
-		centralStockString = strings.Replace(centralStockString, "+", "", -1)
-		centralStockString = strings.Replace(centralStockString, "<", "", -1)
-		centralStockString = strings.Replace(centralStockString, ">", "", -1)
-		centralStock, err := strconv.ParseInt(centralStockString, 10, 64)
 
-		if err != nil {
-			log.Printf("unable to parse central stock for item %s: %v", item.GetSku(), err)
+		sku, ok := item.GetStringFieldValue("sku")
+		if !ok {
+			log.Printf("unable to parse central stock for item %s: %v", sku)
 			centralStock = 0
 		} else {
-			s.svc.UpdateInventory(ctx, pipe, inventory.SKU(item.GetSku()), s.MainStockLocationID, int64(centralStock))
+			s.svc.UpdateInventory(ctx, pipe, inventory.SKU(sku), s.MainStockLocationID, int64(centralStock))
 		}
-		for id, value := range item.GetStock() {
-			s.svc.UpdateInventory(ctx, pipe, inventory.SKU(item.GetSku()), inventory.LocationID(id), int64(value))
-		}
-		_, err = pipe.Exec(ctx)
+		// for id, value := range item.GetStock() {
+		// 	s.svc.UpdateInventory(ctx, pipe, inventory.SKU(sku), inventory.LocationID(id), int64(value))
+		// }
+		_, err := pipe.Exec(ctx)
 		if err != nil {
 			log.Printf("unable to update stock: %v", err)
 		}