go-cart-actor/cmd/cart/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/http/pprof"
	"os"
	"os/signal"
	"strings"
	"time"

	"git.k6n.net/mats/go-cart-actor/pkg/actor"
	"git.k6n.net/mats/go-cart-actor/pkg/cart"
	cartmcp "git.k6n.net/mats/go-cart-actor/pkg/cart/mcp"
	"git.k6n.net/mats/go-cart-actor/pkg/promotions"
	"git.k6n.net/mats/go-cart-actor/internal/ucp"
	"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"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

var (
	grainSpawns = promauto.NewCounter(prometheus.CounterOpts{
		Name: "cart_grain_spawned_total",
		Help: "The total number of spawned grains",
	})
)

func init() {
	os.Mkdir("data", 0755)
}

type App struct {
	pool   *actor.SimpleGrainPool[cart.CartGrain]
	server *PoolServer
}

var podIp = os.Getenv("POD_IP")
var name = os.Getenv("POD_NAME")
var amqpUrl = os.Getenv("AMQP_URL")
var redisAddress = os.Getenv("REDIS_ADDRESS")
var redisPassword = os.Getenv("REDIS_PASSWORD")

// getEnv returns the value of the environment variable named by key, or def if unset/empty.
func getEnv(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

// normalizeListenAddr accepts either a bare port ("8080") or a full listen
// address (":8080", "0.0.0.0:8080") and returns a value usable by http.Server.Addr.
func normalizeListenAddr(v string) string {
	if strings.Contains(v, ":") {
		return v
	}
	return ":" + v
}

// loadUCPCartSigner loads the UCP ECDSA signing key from the path specified in
// the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable.
func loadUCPCartSigner() *ucp.SigningConfig {
	path := os.Getenv("UCP_SIGNING_KEY_PATH")
	if path == "" {
		return nil
	}
	pemData, err := os.ReadFile(path)
	if err != nil {
		log.Printf("ucp signing: cannot read key %s: %v", path, err)
		return nil
	}
	cfg, err := ucp.NewSigningConfigFromPEM(pemData, "k6n-ecdsa-2026")
	if err != nil {
		log.Printf("ucp signing: invalid key at %s: %v", path, err)
		return nil
	}
	return cfg
}

func getCountryFromHost(host string) string {
	if strings.Contains(strings.ToLower(host), "-no") {
		return "no"
	}
	if strings.Contains(strings.ToLower(host), "-se") {
		return "se"
	}
	return "se"
}

type MutationContext struct {
	VoucherService voucher.Service
}

type CartChangeEvent struct {
	CartId    cart.CartId         `json:"cartId"`
	Mutations []actor.ApplyResult `json:"mutations"`
}

func matchesSkuAndLocation(update inventory.InventoryResult, item cart.CartItem) bool {
	if string(update.SKU) == item.Sku {
		if update.LocationID == "se" && item.StoreId == nil {
			return true
		}
		if item.StoreId == nil {
			return false
		}
		if *item.StoreId == string(update.LocationID) {
			return true
		}
	}
	return false
}

func main() {

	// cartPort is the bare HTTP port. It drives both the local listener and the
	// port used to proxy to peer pods, so a cluster must run all pods on the
	// same CART_PORT.
	cartPort := getEnv("CART_PORT", "8080")
	if i := strings.LastIndex(cartPort, ":"); i >= 0 {
		cartPort = cartPort[i+1:]
	}

	controlPlaneConfig := actor.DefaultServerConfig()

	promotionStore, err := promotions.NewStore("data/promotions.json")
	if err != nil {
		log.Fatalf("Error loading promotions: %v\n", err)
	}

	log.Printf("loaded %d promotions", len(promotionStore.List()))

	promotionService := promotions.NewPromotionService(nil)

	//inventoryPubSub := actor.NewPubSub[inventory.InventoryChange]()

	// rdb := redis.NewClient(&redis.Options{
	// 	Addr:     redisAddress,
	// 	Password: redisPassword,
	// 	DB:       0,
	// })
	// inventoryService, err := inventory.NewRedisInventoryService(rdb)
	// if err != nil {
	// 	log.Fatalf("Error creating inventory service: %v\n", err)
	// }

	// inventoryReservationService, err := inventory.NewRedisCartReservationService(rdb)
	// if err != nil {
	// 	log.Fatalf("Error creating inventory reservation service: %v\n", err)
	// }

	reg := cart.NewCartMultationRegistry(cart.NewCartMutationContext(nil))
	reg.RegisterProcessor(
		actor.NewMutationProcessor(func(ctx context.Context, g *cart.CartGrain) error {
			_, span := tracer.Start(ctx, "Totals and promotions")
			defer span.End()
			g.UpdateTotals()
			// Evaluate active promotions against the freshly-totalled cart and apply
			// any matched actions (e.g. the Volymrabatt volume discount), which adjust
			// TotalPrice/TotalDiscount on top of line items and vouchers.
			promotionCtx := promotions.NewContextFromCart(g, promotions.WithNow(time.Now()))
			results, _ := promotionService.EvaluateAll(promotionStore.Snapshot(), promotionCtx)
			// ApplyResults applies qualifying actions in priority order and records
			// every effect — both applied discounts and pending "spend X more for ..."
			// nudges with their progress — in g.AppliedPromotions.
			promotionService.ApplyResults(g, results, promotionCtx)
			g.Version++
			return nil
		}),
	)

	diskStorage := actor.NewDiskStorage[cart.CartGrain](getEnv("CART_DIR", "data"), reg)
	poolConfig := actor.GrainPoolConfig[cart.CartGrain]{
		MutationRegistry: reg,
		Storage:          diskStorage,
		Spawn: func(ctx context.Context, id uint64) (actor.Grain[cart.CartGrain], error) {
			_, span := tracer.Start(ctx, fmt.Sprintf("Spawn cart id %d", id))
			defer span.End()
			grainSpawns.Inc()
			ret := cart.NewCartGrain(id, time.Now())
			// Set baseline lastChange at spawn; replay may update it to last event timestamp.

			// inventoryPubSub.Subscribe(ret.HandleInventoryChange)
			err := diskStorage.LoadEvents(ctx, id, ret)
			// if err == nil && inventoryService != nil {
			// 	refs := make([]*inventory.InventoryReference, 0)
			// 	for _, item := range ret.Items {
			// 		refs = append(refs, &inventory.InventoryReference{
			// 			SKU:        inventory.SKU(item.Sku),
			// 			LocationID: getLocationId(item),
			// 		})
			// 	}
			// 	_, span := tracer.Start(ctx, "update inventory")
			// 	defer span.End()
			// 	res, err := inventoryService.GetInventoryBatch(ctx, refs...)
			// 	if err != nil {
			// 		log.Printf("unable to update inventory %v", err)
			// 	} else {
			// 		for _, update := range res {
			// 			for _, item := range ret.Items {
			// 				if matchesSkuAndLocation(update, *item) && update.Quantity != uint32(item.Stock) {
			// 					// maybe apply an update to give visibility to the cart
			// 					item.Stock = uint16(update.Quantity)
			// 				}
			// 			}
			// 		}
			// 	}
			// }

			return ret, err
		},
		Destroy: func(grain actor.Grain[cart.CartGrain]) error {
			// cart, err := grain.GetCurrentState()
			// if err != nil {
			// 	return err
			// }
			//inventoryPubSub.Unsubscribe(cart.HandleInventoryChange)

			return nil
		},
		SpawnHost: func(host string) (actor.Host[cart.CartGrain], error) {
			return proxy.NewRemoteHost[cart.CartGrain](host, cartPort)
		},
		TTL:      5 * time.Minute,
		PoolSize: 2 * 65535,
		Hostname: podIp,
	}

	pool, err := actor.NewSimpleGrainPool(poolConfig)
	if err != nil {
		log.Fatalf("Error creating cart pool: %v\n", err)
	}

	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.
	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{
		pool:   pool,
		server: syncedServer,
	}

	mux := http.NewServeMux()
	debugMux := http.NewServeMux()

	grpcSrv, err := actor.NewControlServer[cart.CartGrain](controlPlaneConfig, pool)
	if err != nil {
		log.Fatalf("Error starting control plane gRPC server: %v\n", err)
	}
	defer grpcSrv.GracefulStop()

	// go diskStorage.SaveLoop(10 * time.Second)
	UseDiscovery(pool)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	defer stop()

	otelShutdown, err := setupOTelSDK(ctx)
	if err != nil {
		log.Fatalf("Unable to start otel %v", err)
	}

	syncedServer.Serve(mux)

	// Cart MCP edge: inspect and mutate carts — get_cart, get_cart_items,
	// update_item_quantity, remove_cart_item, clear_cart, apply_voucher, etc.
	// Exposes 11 tools on the same grain pool, mounted at /cart-mcp so it does
	// not conflict with the Promotions MCP at /promotions-mcp.
	mux.Handle("/cart-mcp", cartMCP.Handler())
	mux.Handle("/cart-mcp/", cartMCP.Handler())

	// UCP REST adapter — Universal Commerce Protocol
	cartUCP := ucp.CartHandler(pool)
	if signer := loadUCPCartSigner(); signer != nil {
		cartUCP = ucp.WithSigning(cartUCP, signer)
		log.Print("ucp signing enabled")
	}
	// StripPrefix is required because the sub-mux (CartHandler) registers
	// routes like "GET /{id}" which expect a relative path; Go's ServeMux
	// passes the full request path (e.g. /ucp/v1/carts/123) without stripping
	// the prefix, so the sub-mux would never match.
	//
	// Only the subtree pattern (/ucp/v1/carts/) is registered, NOT the exact
	// pattern (/ucp/v1/carts), because registering both causes Go's ServeMux
	// to redirect the exact match to the subtree root, and StripPrefix
	// interferes with the redirect target (producing a 307 to "/" instead
	// of "/ucp/v1/carts/"). Consumers should use the trailing-slash variant.
	mux.Handle("/ucp/v1/carts/", http.StripPrefix("/ucp/v1/carts", cartUCP))

	// Stateless promotion evaluation: POST a (possibly partial) eval context and
	// get back the resulting totals plus the applied/pending effect breakdown,
	// without creating or mutating a real cart.
	mux.HandleFunc("POST /promotions/evaluate", newPromotionEvaluateHandler(promotionStore, promotionService))

	// only for local
	mux.HandleFunc("GET /add/remote/{host}", func(w http.ResponseWriter, r *http.Request) {
		pool.AddRemote(r.PathValue("host"))
	})

	debugMux.HandleFunc("/debug/pprof/", pprof.Index)
	debugMux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
	debugMux.HandleFunc("/debug/pprof/profile", pprof.Profile)
	debugMux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
	debugMux.HandleFunc("/debug/pprof/trace", pprof.Trace)
	debugMux.Handle("/metrics", promhttp.Handler())
	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()
		if grainCount >= capacity {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte("grain pool at capacity"))
			return
		}
		if !pool.IsHealthy() {
			w.WriteHeader(http.StatusInternalServerError)
			w.Write([]byte("control plane not healthy"))
			return
		}
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})
	mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("ok"))
	})

	mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		w.Write([]byte("1.0.0"))
	})

	mux.HandleFunc("/openapi.json", ServeEmbeddedOpenAPI)

	httpAddr := normalizeListenAddr(cartPort)
	debugAddr := normalizeListenAddr(getEnv("CART_DEBUG_PORT", "8081"))

	srv := &http.Server{
		Addr:        httpAddr,
		BaseContext: func(net.Listener) context.Context { return ctx },
		ReadTimeout: 10 * time.Second,
		// Close idle keep-alive connections so a load test with many short-lived
		// clients doesn't pin file handles open indefinitely.
		IdleTimeout:  60 * time.Second,
		WriteTimeout: 20 * time.Second,
		Handler:      otelhttp.NewHandler(mux, "/"),
	}

	defer func() {

		fmt.Println("Shutting down due to signal")
		otelShutdown(context.Background())
		diskStorage.Close()
		pool.Close()

	}()

	srvErr := make(chan error, 1)
	go func() {
		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)
	// 	}
	// }()

	log.Printf("Server started at %s (debug %s)", httpAddr, debugAddr)

	go http.ListenAndServe(debugAddr, debugMux)

	select {
	case err = <-srvErr:
		// Error when starting HTTP server.
		log.Fatalf("Unable to start server: %v", err)
	case <-ctx.Done():
		// Wait for first CTRL+C.
		// Stop receiving signal notifications as soon as possible.
		stop()
	}

}