cart actor
This commit is contained in:
+1
-1
@@ -22,7 +22,7 @@
|
||||
############################
|
||||
# Build Stage
|
||||
############################
|
||||
FROM golang:1.25-alpine AS build
|
||||
FROM golang:1.26-alpine AS build
|
||||
WORKDIR /src
|
||||
|
||||
RUN apk add --no-cache git
|
||||
|
||||
+15
-133
@@ -2,36 +2,16 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
actor "git.k6n.net/go-cart-actor/pkg/actor"
|
||||
"git.k6n.net/go-cart-actor/pkg/cart"
|
||||
"git.k6n.net/go-cart-actor/pkg/checkout"
|
||||
"github.com/matst80/go-redis-inventory/pkg/inventory"
|
||||
"github.com/matst80/slask-finder/pkg/messaging"
|
||||
"git.k6n.net/go-cart-actor/pkg/backofficeadmin"
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type CartFileInfo struct {
|
||||
ID string `json:"id"`
|
||||
CartId cart.CartId `json:"cartId"`
|
||||
Size int64 `json:"size"`
|
||||
Modified time.Time `json:"modified"`
|
||||
}
|
||||
|
||||
type CheckoutFileInfo struct {
|
||||
ID string `json:"id"`
|
||||
CheckoutId checkout.CheckoutId `json:"checkoutId"`
|
||||
Size int64 `json:"size"`
|
||||
Modified time.Time `json:"modified"`
|
||||
}
|
||||
|
||||
func envOrDefault(key, def string) string {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v
|
||||
@@ -39,119 +19,22 @@ func envOrDefault(key, def string) string {
|
||||
return def
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
var redisAddress = os.Getenv("REDIS_ADDRESS")
|
||||
var redisPassword = os.Getenv("REDIS_PASSWORD")
|
||||
|
||||
func main() {
|
||||
dataDir := envOrDefault("DATA_DIR", "data")
|
||||
addr := envOrDefault("ADDR", ":8080")
|
||||
amqpURL := os.Getenv("AMQP_URL")
|
||||
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: redisAddress,
|
||||
Password: redisPassword,
|
||||
DB: 0,
|
||||
app, err := backofficeadmin.New(backofficeadmin.Config{
|
||||
DataDir: envOrDefault("DATA_DIR", "data"),
|
||||
CheckoutDataDir: envOrDefault("CHECKOUT_DATA_DIR", "checkout-data"),
|
||||
RedisAddress: os.Getenv("REDIS_ADDRESS"),
|
||||
RedisPassword: os.Getenv("REDIS_PASSWORD"),
|
||||
})
|
||||
inventoryService, err := inventory.NewRedisInventoryService(rdb)
|
||||
if err != nil {
|
||||
log.Fatalf("Error creating inventory service: %v\n", err)
|
||||
log.Fatalf("Error creating backoffice: %v", err)
|
||||
}
|
||||
|
||||
_ = os.MkdirAll(dataDir, 0755)
|
||||
|
||||
reg := cart.NewCartMultationRegistry(cart.NewCartMutationContext(nil))
|
||||
diskStorage := actor.NewDiskStorage[cart.CartGrain](dataDir, reg)
|
||||
|
||||
checkoutDataDir := envOrDefault("CHECKOUT_DATA_DIR", "checkout-data")
|
||||
_ = os.MkdirAll(checkoutDataDir, 0755)
|
||||
|
||||
regCheckout := checkout.NewCheckoutMutationRegistry(checkout.NewCheckoutMutationContext())
|
||||
diskStorageCheckout := actor.NewDiskStorage[checkout.CheckoutGrain](checkoutDataDir, regCheckout)
|
||||
|
||||
fs := NewFileServer(dataDir, checkoutDataDir, diskStorage, diskStorageCheckout)
|
||||
|
||||
hub := NewHub()
|
||||
go hub.Run()
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("GET /carts", fs.CartsHandler)
|
||||
mux.HandleFunc("GET /cart/{id}", fs.CartHandler)
|
||||
mux.HandleFunc("GET /checkouts", fs.CheckoutsHandler)
|
||||
mux.HandleFunc("GET /checkout/{id}", fs.CheckoutHandler)
|
||||
mux.HandleFunc("PUT /inventory/{locationId}/{sku}", func(w http.ResponseWriter, r *http.Request) {
|
||||
inventoryLocationId := inventory.LocationID(r.PathValue("locationId"))
|
||||
inventorySku := inventory.SKU(r.PathValue("sku"))
|
||||
pipe := rdb.Pipeline()
|
||||
var payload struct {
|
||||
Quantity int64 `json:"quantity"`
|
||||
}
|
||||
err := json.NewDecoder(r.Body).Decode(&payload)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid payload", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
inventoryService.UpdateInventory(r.Context(), pipe, inventorySku, inventoryLocationId, payload.Quantity)
|
||||
|
||||
_, err = pipe.Exec(r.Context())
|
||||
if err != nil {
|
||||
http.Error(w, "failed to update inventory", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
err = inventoryService.SendInventoryChanged(r.Context(), inventorySku, inventoryLocationId)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
mux.HandleFunc("/promotions", fs.PromotionsHandler)
|
||||
mux.HandleFunc("/vouchers", fs.VoucherHandler)
|
||||
mux.HandleFunc("/promotion/{id}", fs.PromotionPartHandler)
|
||||
|
||||
mux.HandleFunc("/ws", hub.ServeWS)
|
||||
app.RegisterRoutes(mux)
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte("ok"))
|
||||
@@ -165,7 +48,7 @@ func main() {
|
||||
_, _ = w.Write([]byte("ok"))
|
||||
})
|
||||
|
||||
// Global CORS middleware allowing all origins and handling preflight
|
||||
// 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")
|
||||
@@ -189,22 +72,21 @@ func main() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
var conn *amqp.Connection
|
||||
if amqpURL != "" {
|
||||
conn, err := amqp.Dial(amqpURL)
|
||||
conn, err = amqp.Dial(amqpURL)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to connect to RabbitMQ: %v", err)
|
||||
}
|
||||
if err := startMutationConsumer(ctx, conn, hub); err != nil {
|
||||
}
|
||||
if err := app.Start(ctx, conn); err != nil {
|
||||
log.Printf("AMQP listener disabled: %v", err)
|
||||
} else {
|
||||
} else if conn != nil {
|
||||
log.Printf("AMQP listener connected")
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("backoffice HTTP listening on %s (dataDir=%s)", addr, dataDir)
|
||||
log.Printf("backoffice HTTP listening on %s", addr)
|
||||
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
log.Fatalf("http server error: %v", err)
|
||||
}
|
||||
|
||||
// server stopped
|
||||
}
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
// Package backofficeadmin is the commerce control-plane admin surface for the
|
||||
// cart-actor system: read-only cart/checkout history (replayed from event logs),
|
||||
// inventory CRUD (Redis), promotions/vouchers JSON, and a WebSocket feed of cart
|
||||
// mutations consumed from RabbitMQ.
|
||||
//
|
||||
// It is extracted from the former cmd/backoffice main package so the same admin
|
||||
// surface can run standalone (cmd/backoffice) or be mounted into the all-in-one
|
||||
// backoffice host. Because it imports the cart/checkout domain (and the private
|
||||
// git.k6n.net module), the host only compiles it into its implementation-specific
|
||||
// (commerce) build variant.
|
||||
package backofficeadmin
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
actor "git.k6n.net/go-cart-actor/pkg/actor"
|
||||
"git.k6n.net/go-cart-actor/pkg/cart"
|
||||
"git.k6n.net/go-cart-actor/pkg/checkout"
|
||||
"github.com/matst80/go-redis-inventory/pkg/inventory"
|
||||
"github.com/matst80/slask-finder/pkg/messaging"
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// CartFileInfo is the metadata returned by the cart listing endpoint.
|
||||
type CartFileInfo struct {
|
||||
ID string `json:"id"`
|
||||
CartId cart.CartId `json:"cartId"`
|
||||
Size int64 `json:"size"`
|
||||
Modified time.Time `json:"modified"`
|
||||
}
|
||||
|
||||
// CheckoutFileInfo is the metadata returned by the checkout listing endpoint.
|
||||
type CheckoutFileInfo struct {
|
||||
ID string `json:"id"`
|
||||
CheckoutId checkout.CheckoutId `json:"checkoutId"`
|
||||
Size int64 `json:"size"`
|
||||
Modified time.Time `json:"modified"`
|
||||
}
|
||||
|
||||
// Config parameterizes the commerce admin App.
|
||||
type Config struct {
|
||||
// DataDir holds the cart event logs (default "data").
|
||||
DataDir string
|
||||
// CheckoutDataDir holds the checkout event logs (default "checkout-data").
|
||||
CheckoutDataDir string
|
||||
// RedisAddress / RedisPassword reach the inventory store.
|
||||
RedisAddress string
|
||||
RedisPassword string
|
||||
}
|
||||
|
||||
// App is the commerce admin control plane. Construct with New, register its
|
||||
// HTTP surface with RegisterRoutes, run background work with Start.
|
||||
type App struct {
|
||||
fs *FileServer
|
||||
hub *Hub
|
||||
rdb *redis.Client
|
||||
inv *inventory.RedisInventoryService
|
||||
}
|
||||
|
||||
// New constructs the commerce admin: it opens the Redis inventory service, the
|
||||
// cart/checkout disk event-log stores, the file server, and the WebSocket hub.
|
||||
// It does not register routes (RegisterRoutes), start background work (Start),
|
||||
// or open a listener.
|
||||
func New(cfg Config) (*App, error) {
|
||||
if cfg.DataDir == "" {
|
||||
cfg.DataDir = "data"
|
||||
}
|
||||
if cfg.CheckoutDataDir == "" {
|
||||
cfg.CheckoutDataDir = "checkout-data"
|
||||
}
|
||||
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: cfg.RedisAddress,
|
||||
Password: cfg.RedisPassword,
|
||||
DB: 0,
|
||||
})
|
||||
inventoryService, err := inventory.NewRedisInventoryService(rdb)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
_ = os.MkdirAll(cfg.DataDir, 0755)
|
||||
reg := cart.NewCartMultationRegistry(cart.NewCartMutationContext(nil))
|
||||
diskStorage := actor.NewDiskStorage[cart.CartGrain](cfg.DataDir, reg)
|
||||
|
||||
_ = os.MkdirAll(cfg.CheckoutDataDir, 0755)
|
||||
regCheckout := checkout.NewCheckoutMutationRegistry(checkout.NewCheckoutMutationContext())
|
||||
diskStorageCheckout := actor.NewDiskStorage[checkout.CheckoutGrain](cfg.CheckoutDataDir, regCheckout)
|
||||
|
||||
return &App{
|
||||
fs: NewFileServer(cfg.DataDir, cfg.CheckoutDataDir, diskStorage, diskStorageCheckout),
|
||||
hub: NewHub(),
|
||||
rdb: rdb,
|
||||
inv: inventoryService,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// RegisterRoutes registers the commerce admin HTTP surface on mux. It does not
|
||||
// apply auth or CORS — the composing host (or standalone main) wraps the mux
|
||||
// with those.
|
||||
func (a *App) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("GET /carts", a.fs.CartsHandler)
|
||||
mux.HandleFunc("GET /cart/{id}", a.fs.CartHandler)
|
||||
mux.HandleFunc("GET /checkouts", a.fs.CheckoutsHandler)
|
||||
mux.HandleFunc("GET /checkout/{id}", a.fs.CheckoutHandler)
|
||||
mux.HandleFunc("PUT /inventory/{locationId}/{sku}", a.updateInventory)
|
||||
mux.HandleFunc("/promotions", a.fs.PromotionsHandler)
|
||||
mux.HandleFunc("/vouchers", a.fs.VoucherHandler)
|
||||
mux.HandleFunc("/promotion/{id}", a.fs.PromotionPartHandler)
|
||||
mux.HandleFunc("/ws", a.hub.ServeWS)
|
||||
}
|
||||
|
||||
func (a *App) updateInventory(w http.ResponseWriter, r *http.Request) {
|
||||
locationID := inventory.LocationID(r.PathValue("locationId"))
|
||||
sku := inventory.SKU(r.PathValue("sku"))
|
||||
pipe := a.rdb.Pipeline()
|
||||
var payload struct {
|
||||
Quantity int64 `json:"quantity"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
|
||||
http.Error(w, "invalid payload", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
a.inv.UpdateInventory(r.Context(), pipe, sku, locationID, payload.Quantity)
|
||||
if _, err := pipe.Exec(r.Context()); err != nil {
|
||||
http.Error(w, "failed to update inventory", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if err := a.inv.SendInventoryChanged(r.Context(), sku, locationID); err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
// Start runs the WebSocket hub loop and, when conn is non-nil, the RabbitMQ
|
||||
// mutation consumer that broadcasts cart mutations to connected clients. The
|
||||
// background goroutines stop when ctx is cancelled.
|
||||
func (a *App) Start(ctx context.Context, conn *amqp.Connection) error {
|
||||
go a.hub.Run()
|
||||
if conn != nil {
|
||||
if err := startMutationConsumer(ctx, conn, a.hub); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Shutdown releases the App's resources (the Redis client). The background
|
||||
// goroutines are stopped by cancelling the ctx passed to Start.
|
||||
func (a *App) Shutdown(_ context.Context) error {
|
||||
return a.rdb.Close()
|
||||
}
|
||||
|
||||
// startMutationConsumer subscribes to the cart "mutation" topic and forwards
|
||||
// each message to the hub for broadcast over WebSocket. Best-effort: a full hub
|
||||
// queue drops the message rather than blocking.
|
||||
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.Printf("mutation consumer: channel closed")
|
||||
return
|
||||
}
|
||||
log.Printf("mutation event: %s", string(m.Body))
|
||||
if hub != nil {
|
||||
select {
|
||||
case hub.broadcast <- m.Body:
|
||||
default:
|
||||
// hub queue full: drop to avoid blocking
|
||||
}
|
||||
}
|
||||
if err := m.Ack(false); err != nil {
|
||||
log.Printf("error acknowledging message: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
package main
|
||||
package backofficeadmin
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
@@ -1,4 +1,4 @@
|
||||
package main
|
||||
package backofficeadmin
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
@@ -1,4 +1,4 @@
|
||||
package main
|
||||
package backofficeadmin
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
Reference in New Issue
Block a user