go-cart-actor/pkg/backofficeadmin/app.go

// 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
}