testing new structure

Co-authored-by: matst80 <mats.tornberg@gmail.com>
Reviewed-on: https://git.tornberg.me/mats/go-cart-actor/pulls/5
Co-authored-by: Mats Törnberg <mats@tornberg.me>
Co-committed-by: Mats Törnberg <mats@tornberg.me>

.gitea/workflows/build.yaml

@@ -47,6 +47,9 @@ jobs:
           docker push registry.knatofs.se/go-cart-actor-amd64:${{ needs.Metadata.outputs.version }}
       - name: Apply deployment manifests
         run: kubectl apply -f deployment/deployment.yaml -n cart
+      - name: Rollout amd64 backoffice deployment
+        run: |
+          kubectl rollout restart deployment/cart-backoffice-x86 -n cart
       - name: Rollout amd64 deployment (pin to version)
         run: |
           kubectl set image deployment/cart-actor-x86 -n cart cart-actor-amd64=registry.knatofs.se/go-cart-actor-amd64:${{ needs.Metadata.outputs.version }}

Dockerfile

@@ -59,6 +59,13 @@ RUN --mount=type=cache,target=/go/build-cache \
     -X main.BuildDate=${BUILD_DATE}" \
     -o /out/go-cart-actor ./cmd/cart
 
+RUN --mount=type=cache,target=/go/build-cache \
+    go build -trimpath -ldflags="-s -w \
+    -X main.Version=${VERSION} \
+    -X main.GitCommit=${GIT_COMMIT} \
+    -X main.BuildDate=${BUILD_DATE}" \
+    -o /out/go-cart-backoffice ./cmd/backoffice
+
 ############################
 # Runtime Stage
 ############################
@@ -67,6 +74,7 @@ FROM gcr.io/distroless/static-debian12:nonroot AS runtime
 WORKDIR /
 
 COPY --from=build /out/go-cart-actor /go-cart-actor
+COPY --from=build /out/go-cart-backoffice /go-cart-backoffice
 
 # Document (not expose forcibly) typical ports: 8080 (HTTP), 1337 (gRPC)
 EXPOSE 8080 1337

cmd/backoffice/fileserver.go

@@ -0,0 +1,185 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/cart"
+)
+
+type FileServer struct {
+	// Define fields here
+	dataDir string
+	storage actor.LogStorage[cart.CartGrain]
+}
+
+func NewFileServer(dataDir string) *FileServer {
+	return &FileServer{
+		dataDir: dataDir,
+	}
+}
+
+func isValidId(id string) (uint64, bool) {
+	if nr, err := strconv.ParseUint(id, 10, 64); err == nil {
+		return nr, true
+	}
+	if nr, ok := cart.ParseCartId(id); ok {
+		return uint64(nr), true
+	}
+	return 0, false
+}
+
+func isValidFileId(name string) (uint64, bool) {
+
+	parts := strings.Split(name, ".")
+	if len(parts) > 1 && parts[1] == "events" {
+		idStr := parts[0]
+
+		return isValidId(idStr)
+	}
+	return 0, false
+}
+
+var cartFileRe = regexp.MustCompile(`^(\d+)\.events\.log$`)
+
+func listCartFiles(dir string) ([]CartFileInfo, error) {
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return []CartFileInfo{}, nil
+		}
+		return nil, err
+	}
+	out := make([]CartFileInfo, 0)
+	for _, e := range entries {
+		if e.IsDir() {
+			continue
+		}
+		id, valid := isValidFileId(e.Name())
+		if !valid {
+			continue
+		}
+
+		info, err := e.Info()
+		if err != nil {
+			continue
+		}
+		info.Sys()
+		out = append(out, CartFileInfo{
+			ID:       fmt.Sprintf("%d", id),
+			CartId:   cart.CartId(id),
+			Size:     info.Size(),
+			Modified: info.ModTime(),
+			System:   info.Sys(),
+		})
+	}
+	return out, nil
+}
+
+func readRawLogLines(path string) ([]json.RawMessage, error) {
+	fh, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer fh.Close()
+	lines := make([]json.RawMessage, 0, 64)
+	s := bufio.NewScanner(fh)
+	// increase buffer to handle larger JSON lines
+	buf := make([]byte, 0, 1024*1024)
+	s.Buffer(buf, 1024*1024)
+	for s.Scan() {
+		line := s.Bytes()
+		if line == nil {
+			continue
+		}
+		lines = append(lines, line)
+	}
+	if err := s.Err(); err != nil {
+		return nil, err
+	}
+	return lines, nil
+}
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+func (fs *FileServer) CartsHandler(w http.ResponseWriter, r *http.Request) {
+	list, err := listCartFiles(fs.dataDir)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
+		return
+	}
+	// sort by modified desc
+	sort.Slice(list, func(i, j int) bool { return list[i].Modified.After(list[j].Modified) })
+
+	writeJSON(w, http.StatusOK, map[string]any{
+		"count": len(list),
+		"carts": list,
+	})
+}
+
+type JsonError struct {
+	Error string `json:"error"`
+}
+
+func (fs *FileServer) CartHandler(w http.ResponseWriter, r *http.Request) {
+	idStr := r.PathValue("id")
+	if idStr == "" {
+		writeJSON(w, http.StatusBadRequest, JsonError{Error: "missing id"})
+		return
+	}
+	id, ok := isValidId(idStr)
+	if !ok {
+		writeJSON(w, http.StatusBadRequest, JsonError{Error: "invalid id"})
+	}
+	// reconstruct state from event log if present
+	grain := cart.NewCartGrain(id, time.Now())
+
+	err := fs.storage.LoadEvents(id, grain)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
+		return
+	}
+
+	path := filepath.Join(fs.dataDir, fmt.Sprintf("%d.events.log", id))
+	info, err := os.Stat(path)
+	if err != nil && errors.Is(err, os.ErrNotExist) {
+		writeJSON(w, http.StatusNotFound, JsonError{Error: "cart not found"})
+		return
+	} else if err != nil {
+		writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
+		return
+	}
+	lines, err := readRawLogLines(path)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, JsonError{Error: err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"id":     id,
+		"cartId": cart.CartId(id).String(),
+		"state":  grain,
+		"rawLog": lines,
+		"meta": map[string]any{
+			"size":     info.Size(),
+			"modified": info.ModTime(),
+			"path":     path,
+			"system":   info.Sys(),
+		},
+	})
+}

cmd/backoffice/hub.go

@@ -0,0 +1,248 @@
+package main
+
+import (
+	"bufio"
+	"crypto/sha1"
+	"encoding/base64"
+	"encoding/binary"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// Hub manages websocket clients and broadcasts messages to them.
+type Hub struct {
+	register   chan *Client
+	unregister chan *Client
+	broadcast  chan []byte
+	clients    map[*Client]bool
+}
+
+// Client represents a single websocket client connection.
+type Client struct {
+	hub  *Hub
+	conn net.Conn
+	send chan []byte
+}
+
+// NewHub constructs a new Hub instance.
+func NewHub() *Hub {
+	return &Hub{
+		register:   make(chan *Client),
+		unregister: make(chan *Client),
+		broadcast:  make(chan []byte, 1024),
+		clients:    make(map[*Client]bool),
+	}
+}
+
+// Run starts the hub event loop.
+func (h *Hub) Run() {
+	for {
+		select {
+		case c := <-h.register:
+			h.clients[c] = true
+		case c := <-h.unregister:
+			if _, ok := h.clients[c]; ok {
+				delete(h.clients, c)
+				close(c.send)
+				_ = c.conn.Close()
+			}
+		case msg := <-h.broadcast:
+			for c := range h.clients {
+				select {
+				case c.send <- msg:
+				default:
+					// Client is slow or dead; drop it.
+					delete(h.clients, c)
+					close(c.send)
+					_ = c.conn.Close()
+				}
+			}
+		}
+	}
+}
+
+// computeAccept computes the Sec-WebSocket-Accept header value.
+func computeAccept(key string) string {
+	const magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+	h := sha1.New()
+	h.Write([]byte(key + magic))
+	return base64.StdEncoding.EncodeToString(h.Sum(nil))
+}
+
+// ServeWS upgrades the HTTP request to a WebSocket connection and registers a client.
+func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request) {
+	if !strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade") || strings.ToLower(r.Header.Get("Upgrade")) != "websocket" {
+		http.Error(w, "upgrade required", http.StatusBadRequest)
+		return
+	}
+	key := r.Header.Get("Sec-WebSocket-Key")
+	if key == "" {
+		http.Error(w, "missing Sec-WebSocket-Key", http.StatusBadRequest)
+		return
+	}
+	accept := computeAccept(key)
+
+	hj, ok := w.(http.Hijacker)
+	if !ok {
+		http.Error(w, "websocket not supported", http.StatusInternalServerError)
+		return
+	}
+	conn, buf, err := hj.Hijack()
+	if err != nil {
+		return
+	}
+
+	// Write the upgrade response
+	response := "HTTP/1.1 101 Switching Protocols\r\n" +
+		"Upgrade: websocket\r\n" +
+		"Connection: Upgrade\r\n" +
+		"Sec-WebSocket-Accept: " + accept + "\r\n" +
+		"\r\n"
+	if _, err := buf.WriteString(response); err != nil {
+		_ = conn.Close()
+		return
+	}
+	if err := buf.Flush(); err != nil {
+		_ = conn.Close()
+		return
+	}
+
+	client := &Client{
+		hub:  h,
+		conn: conn,
+		send: make(chan []byte, 256),
+	}
+	h.register <- client
+	go client.writePump()
+	go client.readPump()
+}
+
+// writeWSFrame writes a single WebSocket frame to the writer.
+func writeWSFrame(w io.Writer, opcode byte, payload []byte) error {
+	// FIN set, opcode as provided
+	header := []byte{0x80 | (opcode & 0x0F)}
+	l := len(payload)
+	switch {
+	case l < 126:
+		header = append(header, byte(l))
+	case l <= 65535:
+		ext := make([]byte, 2)
+		binary.BigEndian.PutUint16(ext, uint16(l))
+		header = append(header, 126)
+		header = append(header, ext...)
+	default:
+		ext := make([]byte, 8)
+		binary.BigEndian.PutUint64(ext, uint64(l))
+		header = append(header, 127)
+		header = append(header, ext...)
+	}
+	if _, err := w.Write(header); err != nil {
+		return err
+	}
+	if l > 0 {
+		if _, err := w.Write(payload); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// readPump handles control frames from the client and discards other incoming frames.
+// This server is broadcast-only to clients.
+func (c *Client) readPump() {
+	defer func() {
+		c.hub.unregister <- c
+	}()
+	reader := bufio.NewReader(c.conn)
+	for {
+		// Read first two bytes
+		b1, err := reader.ReadByte()
+		if err != nil {
+			return
+		}
+		b2, err := reader.ReadByte()
+		if err != nil {
+			return
+		}
+		opcode := b1 & 0x0F
+		masked := (b2 & 0x80) != 0
+		length := int64(b2 & 0x7F)
+		if length == 126 {
+			ext := make([]byte, 2)
+			if _, err := io.ReadFull(reader, ext); err != nil {
+				return
+			}
+			length = int64(binary.BigEndian.Uint16(ext))
+		} else if length == 127 {
+			ext := make([]byte, 8)
+			if _, err := io.ReadFull(reader, ext); err != nil {
+				return
+			}
+			length = int64(binary.BigEndian.Uint64(ext))
+		}
+		var maskKey [4]byte
+		if masked {
+			if _, err := io.ReadFull(reader, maskKey[:]); err != nil {
+				return
+			}
+		}
+
+		// Handle Ping -> Pong
+		if opcode == 0x9 && length <= 125 {
+			payload := make([]byte, length)
+			if _, err := io.ReadFull(reader, payload); err != nil {
+				return
+			}
+			// Unmask if masked
+			if masked {
+				for i := int64(0); i < length; i++ {
+					payload[i] ^= maskKey[i%4]
+				}
+			}
+			_ = writeWSFrame(c.conn, 0xA, payload) // best-effort pong
+			continue
+		}
+
+		// Close frame
+		if opcode == 0x8 {
+			// Drain payload if any, then exit
+			if _, err := io.CopyN(io.Discard, reader, length); err != nil {
+				return
+			}
+			return
+		}
+
+		// For other frames, just discard payload
+		if _, err := io.CopyN(io.Discard, reader, length); err != nil {
+			return
+		}
+	}
+}
+
+// writePump sends queued messages to the client and pings periodically to keep the connection alive.
+func (c *Client) writePump() {
+	ticker := time.NewTicker(30 * time.Second)
+	defer func() {
+		ticker.Stop()
+		_ = c.conn.Close()
+	}()
+	for {
+		select {
+		case msg, ok := <-c.send:
+			if !ok {
+				// try to send close frame
+				_ = writeWSFrame(c.conn, 0x8, nil)
+				return
+			}
+			if err := writeWSFrame(c.conn, 0x1, msg); err != nil {
+				return
+			}
+		case <-ticker.C:
+			// Send a ping to keep connections alive behind proxies
+			_ = writeWSFrame(c.conn, 0x9, []byte("ping"))
+		}
+	}
+}

cmd/backoffice/main.go

@@ -1,5 +1,151 @@
 package main
 
+import (
+	"context"
+	"errors"
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	actor "git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/cart"
+	"github.com/matst80/slask-finder/pkg/messaging"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type CartFileInfo struct {
+	ID       string      `json:"id"`
+	CartId   cart.CartId `json:"cartId"`
+	Size     int64       `json:"size"`
+	Modified time.Time   `json:"modified"`
+	System   any         `json:"system"`
+}
+
+func envOrDefault(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+var globalDisk *actor.DiskStorage[cart.CartGrain]
+
+func startMutationConsumer(ctx context.Context, conn *amqp.Connection, hub *Hub) error {
+	ch, err := conn.Channel()
+	if err != nil {
+		_ = conn.Close()
+		return err
+	}
+	msgs, err := messaging.DeclareBindAndConsume(ch, "cart", "mutation")
+	if err != nil {
+		_ = ch.Close()
+		return err
+	}
+
+	go func() {
+		defer ch.Close()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case m, ok := <-msgs:
+				if !ok {
+					log.Fatalf("connection closed")
+					continue
+				}
+				// Log and broadcast to all websocket clients
+				log.Printf("mutation event: %s", string(m.Body))
+
+				if hub != nil {
+					select {
+					case hub.broadcast <- m.Body:
+					default:
+						// if hub queue is full, drop to avoid blocking
+					}
+				}
+				if err := m.Ack(false); err != nil {
+					log.Printf("error acknowledging message: %v", err)
+				}
+			}
+		}
+	}()
+	return nil
+}
+
 func main() {
-	// Your code here
+	dataDir := envOrDefault("DATA_DIR", "data")
+	addr := envOrDefault("ADDR", ":8080")
+	amqpURL := os.Getenv("AMQP_URL")
+
+	_ = os.MkdirAll(dataDir, 0755)
+
+	reg := cart.NewCartMultationRegistry()
+	globalDisk = actor.NewDiskStorage[cart.CartGrain](dataDir, reg)
+
+	fs := NewFileServer(dataDir)
+
+	hub := NewHub()
+	go hub.Run()
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /carts", fs.CartsHandler)
+	mux.HandleFunc("GET /cart/{id}", fs.CartHandler)
+	mux.HandleFunc("/ws", hub.ServeWS)
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		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"))
+	})
+
+	// Global CORS middleware allowing all origins and handling preflight
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
+		w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With")
+		w.Header().Set("Access-Control-Expose-Headers", "*")
+		if r.Method == http.MethodOptions {
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		mux.ServeHTTP(w, r)
+	})
+
+	srv := &http.Server{
+		Addr:         addr,
+		Handler:      handler,
+		ReadTimeout:  10 * time.Second,
+		WriteTimeout: 30 * time.Second,
+		IdleTimeout:  60 * time.Second,
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	if amqpURL != "" {
+		conn, err := amqp.Dial(amqpURL)
+		if err != nil {
+			log.Fatalf("failed to connect to RabbitMQ: %w", err)
+		}
+		if err := startMutationConsumer(ctx, conn, hub); err != nil {
+			log.Printf("AMQP listener disabled: %v", err)
+		} else {
+			log.Printf("AMQP listener connected")
+		}
+	}
+
+	log.Printf("backoffice HTTP listening on %s (dataDir=%s)", addr, dataDir)
+	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		log.Fatalf("http server error: %v", err)
+	}
+
+	// server stopped
 }

cmd/cart/amqp-order-handler.go

@@ -9,42 +9,48 @@ import (
 )
 
 type AmqpOrderHandler struct {
-	Url        string
-	Connection *amqp.Connection
-	Channel    *amqp.Channel
+	conn *amqp.Connection
 }
 
-func (h *AmqpOrderHandler) Connect() error {
-	conn, err := amqp.Dial(h.Url)
-	if err != nil {
-		return fmt.Errorf("failed to connect to RabbitMQ: %w", err)
+func NewAmqpOrderHandler(conn *amqp.Connection) *AmqpOrderHandler {
+	return &AmqpOrderHandler{
+		conn: conn,
 	}
-	h.Connection = conn
+}
 
-	ch, err := conn.Channel()
+func (h *AmqpOrderHandler) DefineTopics() error {
+	ch, err := h.conn.Channel()
 	if err != nil {
 		return fmt.Errorf("failed to open a channel: %w", err)
 	}
-	h.Channel = ch
+	defer ch.Close()
 
-	return nil
-}
+	err = ch.ExchangeDeclare(
+		"orders", // name
+		"direct", // type
+		true,     // durable
+		false,    // auto-deleted
+		false,    // internal
+		false,    // no-wait
+		nil,      // arguments
+	)
+	if err != nil {
+		return fmt.Errorf("failed to declare an exchange: %w", err)
+	}
 
-func (h *AmqpOrderHandler) Close() error {
-	if h.Channel != nil {
-		h.Channel.Close()
-	}
-	if h.Connection != nil {
-		return h.Connection.Close()
-	}
 	return nil
 }
 
 func (h *AmqpOrderHandler) OrderCompleted(body []byte) error {
+	ch, err := h.conn.Channel()
+	if err != nil {
+		return fmt.Errorf("failed to open a channel: %w", err)
+	}
+	defer ch.Close()
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	err := h.Channel.PublishWithContext(ctx,
+	return ch.PublishWithContext(ctx,
 		"orders", // exchange
 		"new",    // routing key
 		false,    // mandatory
@@ -53,9 +59,5 @@ func (h *AmqpOrderHandler) OrderCompleted(body []byte) error {
 			ContentType: "application/json",
 			Body:        body,
 		})
-	if err != nil {
-		return fmt.Errorf("failed to publish a message: %w", err)
-	}
 
-	return nil
 }

cmd/cart/checkout_builder.go

@@ -3,6 +3,8 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+
+	"git.tornberg.me/go-cart-actor/pkg/cart"
 )
 
 // CheckoutMeta carries the external / URL metadata required to build a
@@ -33,7 +35,7 @@ type CheckoutMeta struct {
 //
 // If you later need to support different tax rates per line, you can extend
 // CartItem / Delivery to expose that data and propagate it here.
-func BuildCheckoutOrderPayload(grain *CartGrain, meta *CheckoutMeta) ([]byte, *CheckoutOrder, error) {
+func BuildCheckoutOrderPayload(grain *cart.CartGrain, meta *CheckoutMeta) ([]byte, *CheckoutOrder, error) {
 	if grain == nil {
 		return nil, nil, fmt.Errorf("nil grain")
 	}

cmd/cart/main.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/cart"
 	"git.tornberg.me/go-cart-actor/pkg/discovery"
 	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 	"git.tornberg.me/go-cart-actor/pkg/proxy"
@@ -20,6 +21,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	amqp "github.com/rabbitmq/amqp091-go"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -45,7 +47,7 @@ func init() {
 }
 
 type App struct {
-	pool *actor.SimpleGrainPool[CartGrain]
+	pool *actor.SimpleGrainPool[cart.CartGrain]
 }
 
 var podIp = os.Getenv("POD_IP")
@@ -97,63 +99,24 @@ type MutationContext struct {
 	VoucherService voucher.Service
 }
 
+type CartChangeEvent struct {
+	CartId    cart.CartId         `json:"cartId"`
+	Mutations []actor.ApplyResult `json:"mutations"`
+}
+
 func main() {
 
 	controlPlaneConfig := actor.DefaultServerConfig()
 
-	reg := actor.NewMutationRegistry()
-	reg.RegisterMutations(
-		actor.NewMutation(AddItem, func() *messages.AddItem {
-			return &messages.AddItem{}
-		}),
-		actor.NewMutation(ChangeQuantity, func() *messages.ChangeQuantity {
-			return &messages.ChangeQuantity{}
-		}),
-		actor.NewMutation(RemoveItem, func() *messages.RemoveItem {
-			return &messages.RemoveItem{}
-		}),
-		actor.NewMutation(InitializeCheckout, func() *messages.InitializeCheckout {
-			return &messages.InitializeCheckout{}
-		}),
-		actor.NewMutation(OrderCreated, func() *messages.OrderCreated {
-			return &messages.OrderCreated{}
-		}),
-		actor.NewMutation(RemoveDelivery, func() *messages.RemoveDelivery {
-			return &messages.RemoveDelivery{}
-		}),
-		actor.NewMutation(SetDelivery, func() *messages.SetDelivery {
-			return &messages.SetDelivery{}
-		}),
-		actor.NewMutation(SetPickupPoint, func() *messages.SetPickupPoint {
-			return &messages.SetPickupPoint{}
-		}),
-		actor.NewMutation(ClearCart, func() *messages.ClearCartRequest {
-			return &messages.ClearCartRequest{}
-		}),
-		actor.NewMutation(AddVoucher, func() *messages.AddVoucher {
-			return &messages.AddVoucher{}
-		}),
-		actor.NewMutation(RemoveVoucher, func() *messages.RemoveVoucher {
-			return &messages.RemoveVoucher{}
-		}),
-	)
-	diskStorage := actor.NewDiskStorage[CartGrain]("data", reg)
-	poolConfig := actor.GrainPoolConfig[CartGrain]{
+	reg := cart.NewCartMultationRegistry()
+	diskStorage := actor.NewDiskStorage[cart.CartGrain]("data", reg)
+	poolConfig := actor.GrainPoolConfig[cart.CartGrain]{
 		MutationRegistry: reg,
 		Storage:          diskStorage,
-		Spawn: func(id uint64) (actor.Grain[CartGrain], error) {
+		Spawn: func(id uint64) (actor.Grain[cart.CartGrain], error) {
 			grainSpawns.Inc()
-			ret := &CartGrain{
-				lastItemId:     0,
-				lastDeliveryId: 0,
-				Deliveries:     []*CartDelivery{},
-				Id:             CartId(id),
-				Items:          []*CartItem{},
-				TotalPrice:     NewPrice(),
-			}
+			ret := cart.NewCartGrain(id, time.Now())
 			// Set baseline lastChange at spawn; replay may update it to last event timestamp.
-			ret.lastChange = time.Now()
-			ret.lastAccess = time.Now()
 
 			err := diskStorage.LoadEvents(id, ret)
 
@@ -175,13 +138,27 @@ func main() {
 		pool: pool,
 	}
 
-	grpcSrv, err := actor.NewControlServer[*CartGrain](controlPlaneConfig, pool)
+	conn, err := amqp.Dial(amqpUrl)
+	if err != nil {
+		fmt.Errorf("failed to connect to RabbitMQ: %w", err)
+	}
+
+	amqpListener := actor.NewAmqpListener(conn, func(id uint64, msg []actor.ApplyResult) (any, error) {
+		return &CartChangeEvent{
+			CartId:    cart.CartId(id),
+			Mutations: msg,
+		}, nil
+	})
+	amqpListener.DefineTopics()
+	pool.AddListener(amqpListener)
+
+	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)
+	// go diskStorage.SaveLoop(10 * time.Second)
 
 	go func(hw discovery.Discovery) {
 		if hw == nil {
@@ -211,12 +188,12 @@ func main() {
 		}
 	}(GetDiscovery())
 
-	orderHandler := &AmqpOrderHandler{
-		Url: amqpUrl,
-	}
+	orderHandler := NewAmqpOrderHandler(conn)
+	orderHandler.DefineTopics()
 	klarnaClient := NewKlarnaClient(KlarnaPlaygroundUrl, os.Getenv("KLARNA_API_USERNAME"), os.Getenv("KLARNA_API_PASSWORD"))
 
 	syncedServer := NewPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp), klarnaClient)
+
 	mux := http.NewServeMux()
 	mux.Handle("/cart/", http.StripPrefix("/cart", syncedServer.Serve()))
 	// only for local
@@ -272,14 +249,14 @@ func main() {
 				w.Write([]byte("no cart id to checkout is empty"))
 				return
 			}
-			parsed, ok := ParseCartId(cookie.Value)
+			parsed, ok := cart.ParseCartId(cookie.Value)
 			if !ok {
 				w.WriteHeader(http.StatusBadRequest)
 				w.Write([]byte("invalid cart id format"))
 				return
 			}
 			cartId := parsed
-			syncedServer.ProxyHandler(func(w http.ResponseWriter, r *http.Request, cartId CartId) error {
+			syncedServer.ProxyHandler(func(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
 				order, err = syncedServer.CreateOrUpdateCheckout(r.Host, cartId)
 				if err != nil {
 					return err
@@ -433,7 +410,7 @@ func triggerOrderCompleted(syncedServer *PoolServer, order *CheckoutOrder) error
 		OrderId: order.ID,
 		Status:  order.Status,
 	}
-	cid, ok := ParseCartId(order.MerchantReference1)
+	cid, ok := cart.ParseCartId(order.MerchantReference1)
 	if !ok {
 		return fmt.Errorf("invalid cart id in order reference: %s", order.MerchantReference1)
 	}
@@ -447,11 +424,7 @@ func confirmOrder(order *CheckoutOrder, orderHandler *AmqpOrderHandler) error {
 	if err != nil {
 		return err
 	}
-	err = orderHandler.Connect()
-	if err != nil {
-		return err
-	}
-	defer orderHandler.Close()
+
 	err = orderHandler.OrderCompleted(orderToSend)
 	if err != nil {
 		return err

cmd/cart/pool-server.go

@@ -11,18 +11,19 @@ import (
 	"time"
 
 	"git.tornberg.me/go-cart-actor/pkg/actor"
+	"git.tornberg.me/go-cart-actor/pkg/cart"
 	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 	"git.tornberg.me/go-cart-actor/pkg/voucher"
 	"github.com/gogo/protobuf/proto"
 )
 
 type PoolServer struct {
-	actor.GrainPool[*CartGrain]
+	actor.GrainPool[*cart.CartGrain]
 	pod_name     string
 	klarnaClient *KlarnaClient
 }
 
-func NewPoolServer(pool actor.GrainPool[*CartGrain], pod_name string, klarnaClient *KlarnaClient) *PoolServer {
+func NewPoolServer(pool actor.GrainPool[*cart.CartGrain], pod_name string, klarnaClient *KlarnaClient) *PoolServer {
 	return &PoolServer{
 		GrainPool:    pool,
 		pod_name:     pod_name,
@@ -30,11 +31,11 @@ func NewPoolServer(pool actor.GrainPool[*CartGrain], pod_name string, klarnaClie
 	}
 }
 
-func (s *PoolServer) ApplyLocal(id CartId, mutation ...proto.Message) (*actor.MutationResult[*CartGrain], error) {
+func (s *PoolServer) ApplyLocal(id cart.CartId, mutation ...proto.Message) (*actor.MutationResult[*cart.CartGrain], error) {
 	return s.Apply(uint64(id), mutation...)
 }
 
-func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	grain, err := s.Get(uint64(id))
 	if err != nil {
 		return err
@@ -43,7 +44,7 @@ func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id C
 	return s.WriteResult(w, grain)
 }
 
-func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	sku := r.PathValue("sku")
 	msg, err := GetItemAddMessage(sku, 1, getCountryFromHost(r.Host), nil)
 	if err != nil {
@@ -73,7 +74,7 @@ func (s *PoolServer) WriteResult(w http.ResponseWriter, result any) error {
 	return err
 }
 
-func (s *PoolServer) DeleteItemHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) DeleteItemHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 
 	itemIdString := r.PathValue("itemId")
 	itemId, err := strconv.ParseInt(itemIdString, 10, 64)
@@ -93,7 +94,7 @@ type SetDeliveryRequest struct {
 	PickupPoint *messages.PickupPoint `json:"pickupPoint,omitempty"`
 }
 
-func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 
 	delivery := SetDeliveryRequest{}
 	err := json.NewDecoder(r.Body).Decode(&delivery)
@@ -111,7 +112,7 @@ func (s *PoolServer) SetDeliveryHandler(w http.ResponseWriter, r *http.Request,
 	return s.WriteResult(w, data)
 }
 
-func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 
 	deliveryIdString := r.PathValue("deliveryId")
 	deliveryId, err := strconv.ParseInt(deliveryIdString, 10, 64)
@@ -138,7 +139,7 @@ func (s *PoolServer) SetPickupPointHandler(w http.ResponseWriter, r *http.Reques
 	return s.WriteResult(w, reply)
 }
 
-func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 
 	deliveryIdString := r.PathValue("deliveryId")
 	deliveryId, err := strconv.Atoi(deliveryIdString)
@@ -152,7 +153,7 @@ func (s *PoolServer) RemoveDeliveryHandler(w http.ResponseWriter, r *http.Reques
 	return s.WriteResult(w, reply)
 }
 
-func (s *PoolServer) QuantityChangeHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) QuantityChangeHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	changeQuantity := messages.ChangeQuantity{}
 	err := json.NewDecoder(r.Body).Decode(&changeQuantity)
 	if err != nil {
@@ -197,7 +198,7 @@ func getMultipleAddMessages(items []Item, country string) []proto.Message {
 	return msgs
 }
 
-func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	setCartItems := SetCartItems{}
 	err := json.NewDecoder(r.Body).Decode(&setCartItems)
 	if err != nil {
@@ -215,7 +216,7 @@ func (s *PoolServer) SetCartItemsHandler(w http.ResponseWriter, r *http.Request,
 	return s.WriteResult(w, reply)
 }
 
-func (s *PoolServer) AddMultipleItemHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) AddMultipleItemHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	setCartItems := SetCartItems{}
 	err := json.NewDecoder(r.Body).Decode(&setCartItems)
 	if err != nil {
@@ -236,7 +237,7 @@ type AddRequest struct {
 	StoreId  *string `json:"storeId"`
 }
 
-func (s *PoolServer) AddSkuRequestHandler(w http.ResponseWriter, r *http.Request, id CartId) error {
+func (s *PoolServer) AddSkuRequestHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	addRequest := AddRequest{Quantity: 1}
 	err := json.NewDecoder(r.Body).Decode(&addRequest)
 	if err != nil {
@@ -287,7 +288,7 @@ func getLocale(country string) string {
 	return "sv-se"
 }
 
-func (s *PoolServer) CreateOrUpdateCheckout(host string, id CartId) (*CheckoutOrder, error) {
+func (s *PoolServer) CreateOrUpdateCheckout(host string, id cart.CartId) (*CheckoutOrder, error) {
 	country := getCountryFromHost(host)
 	meta := &CheckoutMeta{
 		Terms:        fmt.Sprintf("https://%s/terms", host),
@@ -319,7 +320,7 @@ func (s *PoolServer) CreateOrUpdateCheckout(host string, id CartId) (*CheckoutOr
 	}
 }
 
-func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id CartId) (*actor.MutationResult[*CartGrain], error) {
+func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id cart.CartId) (*actor.MutationResult[*cart.CartGrain], error) {
 	// Persist initialization state via mutation (best-effort)
 	return s.ApplyLocal(id, &messages.InitializeCheckout{
 		OrderId:           klarnaOrder.ID,
@@ -341,12 +342,12 @@ func (s *PoolServer) ApplyCheckoutStarted(klarnaOrder *CheckoutOrder, id CartId)
 // }
 //
 
-func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
+func CookieCartIdHandler(fn func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
-		var id CartId
+		var id cart.CartId
 		cookie, err := r.Cookie("cartid")
 		if err != nil || cookie.Value == "" {
-			id = MustNewCartId()
+			id = cart.MustNewCartId()
 			http.SetCookie(w, &http.Cookie{
 				Name:     "cartid",
 				Value:    id.String(),
@@ -358,9 +359,9 @@ func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.R
 			})
 			w.Header().Set("Set-Cart-Id", id.String())
 		} else {
-			parsed, ok := ParseCartId(cookie.Value)
+			parsed, ok := cart.ParseCartId(cookie.Value)
 			if !ok {
-				id = MustNewCartId()
+				id = cart.MustNewCartId()
 				http.SetCookie(w, &http.Cookie{
 					Name:     "cartid",
 					Value:    id.String(),
@@ -388,7 +389,7 @@ func CookieCartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.R
 
 // Removed leftover legacy block after CookieCartIdHandler (obsolete code referencing cid/legacy)
 
-func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, cartId CartId) error {
+func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
 	// Clear cart cookie (breaking change: do not issue a new legacy id here)
 	http.SetCookie(w, &http.Cookie{
 		Name:     "cartid",
@@ -403,17 +404,17 @@ func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, ca
 	return nil
 }
 
-func CartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
+func CartIdHandler(fn func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
-		var id CartId
+		var id cart.CartId
 		raw := r.PathValue("id")
 		// If no id supplied, generate a new one
 		if raw == "" {
-			id := MustNewCartId()
+			id := cart.MustNewCartId()
 			w.Header().Set("Set-Cart-Id", id.String())
 		} else {
 			// Parse base62 cart id
-			if parsedId, ok := ParseCartId(raw); !ok {
+			if parsedId, ok := cart.ParseCartId(raw); !ok {
 				w.WriteHeader(http.StatusBadRequest)
 				w.Write([]byte("cart id is invalid"))
 				return
@@ -431,8 +432,8 @@ func CartIdHandler(fn func(cartId CartId, w http.ResponseWriter, r *http.Request
 	}
 }
 
-func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(cartId CartId, w http.ResponseWriter, r *http.Request) error {
-	return func(cartId CartId, w http.ResponseWriter, r *http.Request) error {
+func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error) func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error {
+	return func(cartId cart.CartId, w http.ResponseWriter, r *http.Request) error {
 		if ownerHost, ok := s.OwnerHost(uint64(cartId)); ok {
 			handled, err := ownerHost.Proxy(uint64(cartId), w, r)
 			if err == nil && handled {
@@ -449,7 +450,7 @@ type AddVoucherRequest struct {
 	VoucherCode string `json:"code"`
 }
 
-func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, cartId CartId) error {
+func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
 	data := &AddVoucherRequest{}
 	json.NewDecoder(r.Body).Decode(data)
 	v := voucher.Service{}
@@ -469,7 +470,7 @@ func (s *PoolServer) AddVoucherHandler(w http.ResponseWriter, r *http.Request, c
 	return nil
 }
 
-func (s *PoolServer) RemoveVoucherHandler(w http.ResponseWriter, r *http.Request, cartId CartId) error {
+func (s *PoolServer) RemoveVoucherHandler(w http.ResponseWriter, r *http.Request, cartId cart.CartId) error {
 
 	idStr := r.PathValue("voucherId")
 	id, err := strconv.ParseInt(idStr, 10, 64)

cmd/cart/product-fetcher.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 
+	"git.tornberg.me/go-cart-actor/pkg/cart"
 	messages "git.tornberg.me/go-cart-actor/pkg/messages"
 	"github.com/matst80/slask-finder/pkg/index"
 )
@@ -53,19 +54,19 @@ func ToItemAddMessage(item *index.DataItem, storeId *string, qty int, country st
 		return nil
 	}
 
-	stock := StockStatus(0)
+	stock := cart.StockStatus(0)
 	centralStockValue, ok := item.GetStringFieldValue(3)
 	if storeId == nil {
 		if ok {
 			pureNumber := strings.Replace(centralStockValue, "+", "", -1)
 			if centralStock, err := strconv.ParseInt(pureNumber, 10, 64); err == nil {
-				stock = StockStatus(centralStock)
+				stock = cart.StockStatus(centralStock)
 			}
 		}
 	} else {
 		storeStock, ok := item.Stock.GetStock()[*storeId]
 		if ok {
-			stock = StockStatus(storeStock)
+			stock = cart.StockStatus(storeStock)
 		}
 	}
 
@@ -119,3 +120,10 @@ func getTax(articleType string) int32 {
 		return 2500
 	}
 }
+
+func getInt(data float64, ok bool) (int, error) {
+	if !ok {
+		return 0, fmt.Errorf("invalid type")
+	}
+	return int(data), nil
+}

deployment/deployment.yaml

@@ -9,6 +9,94 @@ type: Opaque
 ---
 apiVersion: apps/v1
 kind: Deployment
+metadata:
+  labels:
+    app: cart-backoffice
+    arch: amd64
+  name: cart-backoffice-x86
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: cart-backoffice
+      arch: amd64
+  template:
+    metadata:
+      labels:
+        app: cart-backoffice
+        arch: amd64
+    spec:
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+              - matchExpressions:
+                  - key: kubernetes.io/arch
+                    operator: NotIn
+                    values:
+                      - arm64
+      volumes:
+        - name: data
+          nfs:
+            path: /i-data/7a8af061/nfs/cart-actor
+            server: 10.10.1.10
+      imagePullSecrets:
+        - name: regcred
+      serviceAccountName: default
+      containers:
+        - image: registry.knatofs.se/go-cart-actor-amd64:latest
+          name: cart-actor-amd64
+          imagePullPolicy: Always
+          command: ["/go-cart-backoffice"]
+          lifecycle:
+            preStop:
+              exec:
+                command: ["sleep", "15"]
+          ports:
+            - containerPort: 8080
+              name: web
+          livenessProbe:
+            httpGet:
+              path: /livez
+              port: web
+            failureThreshold: 1
+            periodSeconds: 10
+          readinessProbe:
+            httpGet:
+              path: /readyz
+              port: web
+            failureThreshold: 2
+            initialDelaySeconds: 2
+            periodSeconds: 10
+          volumeMounts:
+            - mountPath: "/data"
+              name: data
+          resources:
+            limits:
+              memory: "768Mi"
+            requests:
+              memory: "70Mi"
+              cpu: "1200m"
+          env:
+            - name: TZ
+              value: "Europe/Stockholm"
+            - name: KLARNA_API_USERNAME
+              valueFrom:
+                secretKeyRef:
+                  name: klarna-api-credentials
+                  key: username
+            - name: KLARNA_API_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: klarna-api-credentials
+                  key: password
+            - name: AMQP_URL
+              value: "amqp://admin:12bananer@rabbitmq.dev:5672/"
+            # - name: BASE_URL
+            #   value: "https://s10n-no.tornberg.me"
+---
+apiVersion: apps/v1
+kind: Deployment
 metadata:
   labels:
     app: cart-actor
@@ -222,6 +310,17 @@ spec:
     - name: web
       port: 8080
 ---
+kind: Service
+apiVersion: v1
+metadata:
+  name: cart-backoffice
+spec:
+  selector:
+    app: cart-backoffice
+  ports:
+    - name: web
+      port: 8080
+---
 apiVersion: networking.k8s.io/v1
 kind: Ingress
 metadata:
@@ -250,3 +349,27 @@ spec:
                 name: cart-actor
                 port:
                   number: 8080
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: cart-backend-ingress
+  annotations:
+    cert-manager.io/cluster-issuer: letsencrypt-prod
+spec:
+  ingressClassName: nginx
+  tls:
+    - hosts:
+        - slask-cart.tornberg.me
+      secretName: cart-backoffice-actor-tls-secret
+  rules:
+    - host: slask-cart.tornberg.me
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: cart-backoffice
+                port:
+                  number: 8080

pkg/actor/disk_storage.go

@@ -26,7 +26,7 @@ type DiskStorage[V any] struct {
 
 type LogStorage[V any] interface {
 	LoadEvents(id uint64, grain Grain[V]) error
-	AppendEvent(id uint64, msg ...proto.Message) error
+	AppendMutations(id uint64, msg ...proto.Message) error
 }
 
 func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] {
@@ -108,7 +108,7 @@ func (s *DiskStorage[V]) Close() {
 	close(s.done)
 }
 
-func (s *DiskStorage[V]) AppendEvent(id uint64, msg ...proto.Message) error {
+func (s *DiskStorage[V]) AppendMutations(id uint64, msg ...proto.Message) error {
 	if s.queue != nil {
 		queue := make([]QueueEvent, 0)
 		data, found := s.queue.Load(id)

pkg/actor/grpc_server.go

@@ -16,7 +16,6 @@ import (
 // It delegates to a grain pool and cluster operations to a synced pool.
 type ControlServer[V any] struct {
 	messages.UnimplementedControlPlaneServer
-
 	pool GrainPool[V]
 }
 

pkg/actor/log_listerner.go

@@ -0,0 +1,47 @@
+package actor
+
+import (
+	"log"
+
+	"github.com/matst80/slask-finder/pkg/messaging"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type LogListener interface {
+	AppendMutations(id uint64, msg ...ApplyResult)
+}
+
+type AmqpListener struct {
+	conn        *amqp.Connection
+	transformer func(id uint64, msg []ApplyResult) (any, error)
+}
+
+func NewAmqpListener(conn *amqp.Connection, transformer func(id uint64, msg []ApplyResult) (any, error)) *AmqpListener {
+	return &AmqpListener{
+		conn:        conn,
+		transformer: transformer,
+	}
+}
+
+func (l *AmqpListener) DefineTopics() {
+	ch, err := l.conn.Channel()
+	if err != nil {
+		log.Fatalf("Failed to open a channel: %v", err)
+	}
+	defer ch.Close()
+	if err := messaging.DefineTopic(ch, "cart", "mutation"); err != nil {
+		log.Fatalf("Failed to declare topic mutation: %v", err)
+	}
+}
+
+func (l *AmqpListener) AppendMutations(id uint64, msg ...ApplyResult) {
+	data, err := l.transformer(id, msg)
+	if err != nil {
+		log.Printf("Failed to transform mutation event: %v", err)
+		return
+	}
+	err = messaging.SendChange(l.conn, "cart", "mutation", data)
+	if err != nil {
+		log.Printf("Failed to send mutation event: %v", err)
+	}
+}

pkg/actor/simple_grain_pool.go

@@ -17,6 +17,7 @@ type SimpleGrainPool[V any] struct {
 	mutationRegistry MutationRegistry
 	spawn            func(id uint64) (Grain[V], error)
 	spawnHost        func(host string) (Host, error)
+	listeners        []LogListener
 	storage          LogStorage[V]
 	ttl              time.Duration
 	poolSize         int
@@ -66,6 +67,19 @@ func NewSimpleGrainPool[V any](config GrainPoolConfig[V]) (*SimpleGrainPool[V],
 	return p, nil
 }
 
+func (p *SimpleGrainPool[V]) AddListener(listener LogListener) {
+	p.listeners = append(p.listeners, listener)
+}
+
+func (p *SimpleGrainPool[V]) RemoveListener(listener LogListener) {
+	for i, l := range p.listeners {
+		if l == listener {
+			p.listeners = append(p.listeners[:i], p.listeners[i+1:]...)
+			break
+		}
+	}
+}
+
 func (p *SimpleGrainPool[V]) purge() {
 	purgeLimit := time.Now().Add(-p.ttl)
 	purgedIds := make([]uint64, 0, len(p.grains))
@@ -383,11 +397,14 @@ func (p *SimpleGrainPool[V]) Apply(id uint64, mutation ...proto.Message) (*Mutat
 	}
 	if p.storage != nil {
 		go func() {
-			if err := p.storage.AppendEvent(id, mutation...); err != nil {
+			if err := p.storage.AppendMutations(id, mutation...); err != nil {
 				log.Printf("failed to store mutation for grain %d: %v", id, err)
 			}
 		}()
 	}
+	for _, listener := range p.listeners {
+		go listener.AppendMutations(id, mutations...)
+	}
 	result, err := grain.GetCurrentState()
 	if err != nil {
 		return nil, err

pkg/cart/cart-grain.go

@@ -1,8 +1,7 @@
-package main
+package cart
 
 import (
 	"encoding/json"
-	"fmt"
 	"slices"
 	"sync"
 	"time"
@@ -138,6 +137,22 @@ func (v *Voucher) AppliesTo(cart *CartGrain) ([]*CartItem, bool) {
 	return cart.Items, true
 }
 
+func NewCartGrain(id uint64, ts time.Time) *CartGrain {
+	return &CartGrain{
+		lastItemId:     0,
+		lastDeliveryId: 0,
+		lastVoucherId:  0,
+		lastAccess:     ts,
+		lastChange:     ts,
+		TotalDiscount:  NewPrice(),
+		Vouchers:       []*Voucher{},
+		Deliveries:     []*CartDelivery{},
+		Id:             CartId(id),
+		Items:          []*CartItem{},
+		TotalPrice:     NewPrice(),
+	}
+}
+
 func (c *CartGrain) GetId() uint64 {
 	return uint64(c.Id)
 }
@@ -155,22 +170,6 @@ func (c *CartGrain) GetCurrentState() (*CartGrain, error) {
 	return c, nil
 }
 
-func getInt(data float64, ok bool) (int, error) {
-	if !ok {
-		return 0, fmt.Errorf("invalid type")
-	}
-	return int(data), nil
-}
-
-// func (c *CartGrain) AddItem(sku string, qty int, country string, storeId *string) (*CartGrain, error) {
-// 	cartItem, err := getItemData(sku, qty, country)
-// 	if err != nil {
-// 		return nil, err
-// 	}
-// 	cartItem.StoreId = storeId
-// 	return c.Apply(cartItem, false)
-// }
-
 func (c *CartGrain) GetState() ([]byte, error) {
 	return json.Marshal(c)
 }
@@ -260,6 +259,10 @@ func (c *CartGrain) UpdateTotals() {
 	for _, voucher := range c.Vouchers {
 		if _, ok := voucher.AppliesTo(c); ok {
 			value := NewPriceFromIncVat(voucher.Value, 25)
+			if c.TotalPrice.IncVat <= value.IncVat {
+				// don't apply discounts to more than the total price
+				continue
+			}
 
 			c.TotalDiscount.Add(*value)
 			c.TotalPrice.Subtract(*value)

pkg/cart/cart-mutation-helper.go

@@ -0,0 +1,48 @@
+package cart
+
+import (
+	"git.tornberg.me/go-cart-actor/pkg/actor"
+	messages "git.tornberg.me/go-cart-actor/pkg/messages"
+)
+
+func NewCartMultationRegistry() actor.MutationRegistry {
+
+	reg := actor.NewMutationRegistry()
+	reg.RegisterMutations(
+		actor.NewMutation(AddItem, func() *messages.AddItem {
+			return &messages.AddItem{}
+		}),
+		actor.NewMutation(ChangeQuantity, func() *messages.ChangeQuantity {
+			return &messages.ChangeQuantity{}
+		}),
+		actor.NewMutation(RemoveItem, func() *messages.RemoveItem {
+			return &messages.RemoveItem{}
+		}),
+		actor.NewMutation(InitializeCheckout, func() *messages.InitializeCheckout {
+			return &messages.InitializeCheckout{}
+		}),
+		actor.NewMutation(OrderCreated, func() *messages.OrderCreated {
+			return &messages.OrderCreated{}
+		}),
+		actor.NewMutation(RemoveDelivery, func() *messages.RemoveDelivery {
+			return &messages.RemoveDelivery{}
+		}),
+		actor.NewMutation(SetDelivery, func() *messages.SetDelivery {
+			return &messages.SetDelivery{}
+		}),
+		actor.NewMutation(SetPickupPoint, func() *messages.SetPickupPoint {
+			return &messages.SetPickupPoint{}
+		}),
+		actor.NewMutation(ClearCart, func() *messages.ClearCartRequest {
+			return &messages.ClearCartRequest{}
+		}),
+		actor.NewMutation(AddVoucher, func() *messages.AddVoucher {
+			return &messages.AddVoucher{}
+		}),
+		actor.NewMutation(RemoveVoucher, func() *messages.RemoveVoucher {
+			return &messages.RemoveVoucher{}
+		}),
+	)
+	return reg
+
+}

pkg/cart/cart_grain_totals_test.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"testing"

pkg/cart/cart_id.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"crypto/rand"

pkg/cart/cart_id_test.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"encoding/json"

pkg/cart/mutation_add_item.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"
@@ -30,9 +30,13 @@ func AddItem(g *CartGrain, m *messages.AddItem) error {
 
 	// Fast path: merge with existing item having same SKU
 	if existing, found := g.FindItemWithSku(m.Sku); found {
+		if existing.StoreId == m.StoreId {
 			existing.Quantity += int(m.Quantity)
+			existing.Stock = StockStatus(m.Stock)
+			existing.StoreId = m.StoreId
 			return nil
 		}
+	}
 
 	g.mu.Lock()
 	defer g.mu.Unlock()

pkg/cart/mutation_add_voucher.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"slices"

pkg/cart/mutation_change_quantity.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/mutation_initialize_checkout.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/mutation_order_created.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/mutation_remove_delivery.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/mutation_remove_item.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/mutation_set_delivery.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/mutation_set_pickup_point.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"fmt"

pkg/cart/price.go

@@ -1,4 +1,4 @@
-package main
+package cart
 
 import (
 	"encoding/json"

pkg/voucher/service.go

@@ -21,7 +21,6 @@ type Voucher struct {
 
 type Service struct {
 	// Add fields here
-
 }
 
 var ErrInvalidCode = errors.New("invalid vouchercode")