diff --git a/Dockerfile b/Dockerfile index 941dcef..4e3e394 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/cmd/backoffice/main.go b/cmd/backoffice/main.go index fd5b13d..c56a5ad 100644 --- a/cmd/backoffice/main.go +++ b/cmd/backoffice/main.go @@ -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 { - log.Printf("AMQP listener disabled: %v", err) - } else { - log.Printf("AMQP listener connected") - } + } + if err := app.Start(ctx, conn); err != nil { + log.Printf("AMQP listener disabled: %v", err) + } 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 } diff --git a/pkg/backofficeadmin/app.go b/pkg/backofficeadmin/app.go new file mode 100644 index 0000000..add617f --- /dev/null +++ b/pkg/backofficeadmin/app.go @@ -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 +} diff --git a/cmd/backoffice/fileserver.go b/pkg/backofficeadmin/fileserver.go similarity index 99% rename from cmd/backoffice/fileserver.go rename to pkg/backofficeadmin/fileserver.go index 49cc32b..e584492 100644 --- a/cmd/backoffice/fileserver.go +++ b/pkg/backofficeadmin/fileserver.go @@ -1,4 +1,4 @@ -package main +package backofficeadmin import ( "bufio" diff --git a/cmd/backoffice/fileserver_test.go b/pkg/backofficeadmin/fileserver_test.go similarity index 99% rename from cmd/backoffice/fileserver_test.go rename to pkg/backofficeadmin/fileserver_test.go index bd7d175..53918bc 100644 --- a/cmd/backoffice/fileserver_test.go +++ b/pkg/backofficeadmin/fileserver_test.go @@ -1,4 +1,4 @@ -package main +package backofficeadmin import ( "math/rand" diff --git a/cmd/backoffice/hub.go b/pkg/backofficeadmin/hub.go similarity index 99% rename from cmd/backoffice/hub.go rename to pkg/backofficeadmin/hub.go index 67be2cd..4a97c21 100644 --- a/cmd/backoffice/hub.go +++ b/pkg/backofficeadmin/hub.go @@ -1,4 +1,4 @@ -package main +package backofficeadmin import ( "bufio"