diff --git a/cmd/backoffice/main.go b/cmd/backoffice/main.go index 4982e24..7cc87eb 100644 --- a/cmd/backoffice/main.go +++ b/cmd/backoffice/main.go @@ -20,8 +20,6 @@ func main() { app, err := backofficeadmin.New(backofficeadmin.Config{ DataDir: config.EnvString("CART_DIR", "data"), CheckoutDataDir: config.EnvString("CHECKOUT_DIR", "checkout-data"), - RedisAddress: os.Getenv("REDIS_ADDRESS"), - RedisPassword: os.Getenv("REDIS_PASSWORD"), }) if err != nil { log.Fatalf("Error creating backoffice: %v", err) diff --git a/cmd/cart/main.go b/cmd/cart/main.go index 64feba5..07add85 100644 --- a/cmd/cart/main.go +++ b/cmd/cart/main.go @@ -104,21 +104,6 @@ type CartChangeEvent struct { 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 -} - // catalogProjectionCache is the cart's per-pod, cold-start-ready catalog // projection cache, backed by platform/catalog.ProjectionStore. It consumes the // catalog.projection_published wire — a full snapshot (begin/chunk/end, framed by @@ -232,18 +217,13 @@ func main() { Password: redisPassword, DB: 0, }) - inventoryService, err := inventory.NewRedisInventoryService(rdb) - if err != nil { - log.Fatalf("Error creating inventory service: %v\n", err) - } - _ = inventoryService - inventoryReservationService, err := inventory.NewRedisCartReservationService(rdb) if err != nil { log.Fatalf("Error creating inventory reservation service: %v\n", err) } + reservationPolicy := inventory.NewReservationPolicyAdapter(inventoryReservationService) - reg := cart.NewCartMultationRegistry(cart.NewCartMutationContext(inventoryReservationService)) + reg := cart.NewCartMultationRegistry(cart.NewCartMutationContext(reservationPolicy)) reg.RegisterProcessor( actor.NewMutationProcessor(func(ctx context.Context, g *cart.CartGrain) error { _, span := tracer.Start(ctx, "Totals and promotions") diff --git a/cmd/checkout/klarna-handlers.go b/cmd/checkout/klarna-handlers.go index 7848e7d..32b45c9 100644 --- a/cmd/checkout/klarna-handlers.go +++ b/cmd/checkout/klarna-handlers.go @@ -11,7 +11,7 @@ import ( "git.k6n.net/mats/go-cart-actor/pkg/cart" "git.k6n.net/mats/go-cart-actor/pkg/checkout" messages "git.k6n.net/mats/go-cart-actor/proto/checkout" - "git.k6n.net/mats/go-redis-inventory/pkg/inventory" + "git.k6n.net/mats/platform/inventory" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -244,7 +244,7 @@ func createOrderFromCheckout(ctx context.Context, s *CheckoutPoolServer, grain * func getLocationId(item *cart.CartItem) inventory.LocationID { if item.StoreId == nil || *item.StoreId == "" { - return "se" + return inventory.LocationID("se") } return inventory.LocationID(*item.StoreId) } @@ -259,23 +259,6 @@ func shouldTrackInventory(item *cart.CartItem) bool { return false } -func getInventoryRequests(items []*cart.CartItem) []inventory.ReserveRequest { - var requests []inventory.ReserveRequest - for _, item := range items { - if item == nil || !shouldTrackInventory(item) { - continue - } - requests = append(requests, inventory.ReserveRequest{ - InventoryReference: &inventory.InventoryReference{ - SKU: inventory.SKU(item.Sku), - LocationID: getLocationId(item), - }, - Quantity: uint32(item.Quantity), - }) - } - return requests -} - func (a *CheckoutPoolServer) getGrainFromKlarnaOrder(ctx context.Context, order *CheckoutOrder) (*checkout.CheckoutGrain, error) { cartId, ok := cart.ParseCartId(order.MerchantReference1) if !ok { diff --git a/cmd/checkout/main.go b/cmd/checkout/main.go index f85769e..ad7dd90 100644 --- a/cmd/checkout/main.go +++ b/cmd/checkout/main.go @@ -15,7 +15,7 @@ import ( "git.k6n.net/mats/go-cart-actor/pkg/checkout" "git.k6n.net/mats/go-cart-actor/pkg/proxy" "git.k6n.net/mats/go-cart-actor/pkg/telemetry" - "git.k6n.net/mats/go-redis-inventory/pkg/inventory" + redisinv "git.k6n.net/mats/go-redis-inventory/pkg/inventory" "git.k6n.net/mats/platform/config" "git.k6n.net/mats/platform/rabbit" "git.k6n.net/mats/platform/tax" @@ -102,15 +102,12 @@ func main() { Password: redisPassword, DB: 0, }) - inventoryService, err := inventory.NewRedisInventoryService(rdb) - if err != nil { - log.Fatalf("Error creating inventory service: %v\n", err) - } - reservationService, err := inventory.NewRedisCartReservationService(rdb) + reservationService, err := redisinv.NewRedisCartReservationService(rdb) if err != nil { log.Fatalf("Error creating reservation service: %v\n", err) } + reservationPolicy := redisinv.NewReservationPolicyAdapter(reservationService) checkoutDir := os.Getenv("CHECKOUT_DIR") if checkoutDir == "" { @@ -194,8 +191,7 @@ func main() { pool.AddListener(checkoutFeed) syncedServer = NewCheckoutPoolServer(pool, fmt.Sprintf("%s, %s", name, podIp), klarnaClient, cartClient, adyenClient, orderClient, orderHandler) - syncedServer.inventoryService = inventoryService - syncedServer.reservationService = reservationService + syncedServer.reservationPolicy = reservationPolicy syncedServer.taxProvider = selectTaxProvider() mux := http.NewServeMux() diff --git a/cmd/checkout/order_client.go b/cmd/checkout/order_client.go index 312c298..3d5b431 100644 --- a/cmd/checkout/order_client.go +++ b/cmd/checkout/order_client.go @@ -66,6 +66,7 @@ type OrderLine struct { Quantity int32 `json:"quantity"` UnitPrice int64 `json:"unitPrice"` TaxRate int32 `json:"taxRate"` + DropShip bool `json:"drop_ship,omitempty"` Location string `json:"location,omitempty"` } diff --git a/cmd/checkout/order_create.go b/cmd/checkout/order_create.go index b695585..2aefb7d 100644 --- a/cmd/checkout/order_create.go +++ b/cmd/checkout/order_create.go @@ -58,6 +58,7 @@ func buildOrderLinesFromGrain(grain *checkout.CheckoutGrain) []OrderLine { // CartItem.Tax and OrderLine.TaxRate share the platform basis-point scale // (2500 = 25%), so the rate passes through unchanged. TaxRate: int32(it.Tax), + DropShip: it.DropShip, Location: location, }) } diff --git a/cmd/checkout/pool-server.go b/cmd/checkout/pool-server.go index 5d31e30..b252db0 100644 --- a/cmd/checkout/pool-server.go +++ b/cmd/checkout/pool-server.go @@ -16,8 +16,8 @@ import ( "git.k6n.net/mats/platform/tax" messages "git.k6n.net/mats/go-cart-actor/proto/checkout" + "git.k6n.net/mats/platform/inventory" adyen "github.com/adyen/adyen-go-api-library/v21/src/adyen" - "git.k6n.net/mats/go-redis-inventory/pkg/inventory" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" @@ -52,9 +52,8 @@ type CheckoutPoolServer struct { cartClient *CartClient orderClient *OrderClient orderHandler *AmqpOrderHandler - inventoryService *inventory.RedisInventoryService - reservationService *inventory.RedisCartReservationService - taxProvider tax.Provider + reservationPolicy inventory.ReservationPolicy + taxProvider tax.Provider } func NewCheckoutPoolServer(pool actor.GrainPool[checkout.CheckoutGrain], pod_name string, klarnaClient *KlarnaClient, cartClient *CartClient, adyenClient *adyen.APIClient, orderClient *OrderClient, orderHandler *AmqpOrderHandler) *CheckoutPoolServer { @@ -556,14 +555,14 @@ func (s *CheckoutPoolServer) Serve(mux *http.ServeMux) { } func (s *CheckoutPoolServer) releaseCartReservations(ctx context.Context, grain *checkout.CheckoutGrain) { - if s.reservationService == nil || grain.CartState == nil { + if s.reservationPolicy == nil || grain.CartState == nil { return } for _, item := range grain.CartState.Items { if item == nil || !shouldTrackInventory(item) { continue } - if err := s.reservationService.ReleaseForCart(ctx, inventory.SKU(item.Sku), getLocationId(item), inventory.CartID(grain.CartId.String())); err != nil { + if err := s.reservationPolicy.Release(ctx, inventory.SKU(item.Sku), getLocationId(item), inventory.CartID(grain.CartId.String())); err != nil { logger.WarnContext(ctx, "failed to release cart reservation", "sku", item.Sku, "err", err) } } diff --git a/cmd/checkout/utils.go b/cmd/checkout/utils.go index d016b6b..f52729f 100644 --- a/cmd/checkout/utils.go +++ b/cmd/checkout/utils.go @@ -2,6 +2,7 @@ package main import ( "context" + "fmt" "log" "net/http" "strings" @@ -9,6 +10,7 @@ import ( "git.k6n.net/mats/go-cart-actor/pkg/cart" "git.k6n.net/mats/go-cart-actor/pkg/checkout" + "git.k6n.net/mats/platform/inventory" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" ) @@ -77,12 +79,23 @@ func getCountryFromHost(host string) string { } func (a *CheckoutPoolServer) reserveInventory(ctx context.Context, grain *checkout.CheckoutGrain) error { - if a.inventoryService != nil { - inventoryRequests := getInventoryRequests(grain.CartState.Items) - _, err := a.inventoryService.ReservationCheck(ctx, inventoryRequests...) + if a.reservationPolicy == nil { + return nil + } + for _, item := range grain.CartState.Items { + if item == nil || !shouldTrackInventory(item) { + continue + } + loc := inventory.LocationID("se") + if item.StoreId != nil && *item.StoreId != "" { + loc = inventory.LocationID(*item.StoreId) + } + avail, err := a.reservationPolicy.Available(ctx, inventory.SKU(item.Sku), loc) if err != nil { - logger.WarnContext(ctx, "placeorder inventory check failed") - return err + return fmt.Errorf("inventory check failed for SKU %s: %w", item.Sku, err) + } + if int64(item.Quantity) > avail { + return fmt.Errorf("insufficient inventory for SKU %s: have %d, need %d", item.Sku, avail, item.Quantity) } } return nil diff --git a/cmd/inventory/batch.go b/cmd/inventory/batch.go new file mode 100644 index 0000000..1fc7be8 --- /dev/null +++ b/cmd/inventory/batch.go @@ -0,0 +1,100 @@ +package main + +import ( + "encoding/json" + "io" + "net/http" + + "git.k6n.net/mats/go-redis-inventory/pkg/inventory" +) + +// batchItem is one stock set in a batch request. LocationID is optional; an +// empty value defaults to the service's country location (the same location the +// catalog-feed listener writes to). +type batchItem struct { + SKU string `json:"sku"` + LocationID string `json:"locationId,omitempty"` + Quantity int64 `json:"quantity"` +} + +// batchResult mirrors one input item with the location actually written and an +// optional per-item error (empty on success). +type batchResult struct { + SKU string `json:"sku"` + LocationID string `json:"locationId"` + Quantity int64 `json:"quantity"` + Error string `json:"error,omitempty"` +} + +// batchInventoryHandler SETS stock to absolute values for many SKUs in one +// pipelined call — for re-seeding/import and for testing the inventory + reserve +// paths directly. Same write as the catalog-feed listener (UpdateInventory + +// level crossing), so a value set here is authoritative until the next +// catalog.item_changed for that SKU overwrites it (catalog is the source of +// truth). Body is a JSON array of {sku, locationId?, quantity}. +func (srv *Server) batchInventoryHandler(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(io.LimitReader(r.Body, 16<<20)) // 16 MiB + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + var items []batchItem + if err := json.Unmarshal(body, &items); err != nil { + http.Error(w, "body must be a JSON array of {sku, locationId?, quantity}: "+err.Error(), http.StatusBadRequest) + return + } + if len(items) == 0 { + http.Error(w, "empty batch", http.StatusBadRequest) + return + } + + ctx := r.Context() + results := make([]batchResult, len(items)) + pipe := srv.rdb.Pipeline() + for i, it := range items { + loc := it.LocationID + if loc == "" { + loc = country + } + results[i] = batchResult{SKU: it.SKU, LocationID: loc, Quantity: it.Quantity} + if it.SKU == "" { + results[i].Error = "sku is required" + continue + } + srv.inventoryService.UpdateInventory(ctx, pipe, inventory.SKU(it.SKU), inventory.LocationID(loc), it.Quantity) + } + + hadError := false + if _, err := pipe.Exec(ctx); err != nil { + // Exec reports the first failing command; flag the whole batch rather than + // guess which items committed. + hadError = true + for i := range results { + if results[i].Error == "" { + results[i].Error = err.Error() + } + } + } else { + // Emit level crossings for the items that were actually written, so the + // finder/level listeners react the same way they do for catalog feeds. + for i := range results { + if results[i].Error == "" { + emitLevelIfChanged(ctx, srv.rdb, srv.conn, country, lowWatermark, results[i].SKU, results[i].LocationID, results[i].Quantity) + } + } + } + for i := range results { + if results[i].Error != "" { + hadError = true + break + } + } + + status := http.StatusOK + if hadError { + status = http.StatusMultiStatus // 207 + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(map[string]any{"count": len(results), "results": results}) +} diff --git a/cmd/inventory/commit.go b/cmd/inventory/commit.go index ea0854f..c332750 100644 --- a/cmd/inventory/commit.go +++ b/cmd/inventory/commit.go @@ -40,6 +40,11 @@ func commitOrder(ctx context.Context, rdb *redis.Client, svc *inventory.RedisInv if line.SKU == "" || line.Quantity <= 0 { continue } + // Drop-ship items are fulfilled by the supplier — never decrement + // local stock for them. + if line.DropShip { + continue + } loc := inventory.LocationID(defaultLoc) if line.Location != "" { loc = inventory.LocationID(line.Location) diff --git a/cmd/inventory/main.go b/cmd/inventory/main.go index 99a6f86..44b605e 100644 --- a/cmd/inventory/main.go +++ b/cmd/inventory/main.go @@ -24,6 +24,11 @@ import ( type Server struct { inventoryService *inventory.RedisInventoryService reservationService *inventory.RedisCartReservationService + + // rdb + conn back the batch write path (pipelined UpdateInventory + level + // crossing). conn is nil when RABBIT_HOST is unset; emit is then skipped. + rdb *redis.Client + conn *rabbit.Conn } func (srv *Server) livezHandler(w http.ResponseWriter, r *http.Request) { @@ -124,13 +129,16 @@ func main() { return } - server := &Server{inventoryService: s, reservationService: r} + server := &Server{inventoryService: s, reservationService: r, rdb: rdb} // Set up HTTP routes http.HandleFunc("/livez", server.livezHandler) http.HandleFunc("/readyz", server.readyzHandler) http.HandleFunc("/inventory/{sku}/{locationId}", server.getInventoryHandler) http.HandleFunc("/reservations/{sku}/{locationId}", server.getReservationHandler) + http.HandleFunc("POST /reservations", server.reserveInventoryHandler) + http.HandleFunc("POST /reservations/release", server.releaseReservationHandler) + http.HandleFunc("POST /inventory/batch", server.batchInventoryHandler) stockhandler := &StockHandler{ MainStockLocationID: inventory.LocationID(country), @@ -151,6 +159,9 @@ func main() { // The catalog-feed handler emits inventory.level_changed crossings // directly on this connection as it writes stock. stockhandler.conn = conn + // Share the bus connection with the batch handler so its writes emit + // inventory.level_changed crossings too. + server.conn = conn // Reconnecting consumer: re-listen on reconnect. conn.NotifyOnReconnect(func() { ch, err := conn.Channel() diff --git a/cmd/inventory/reservations.go b/cmd/inventory/reservations.go new file mode 100644 index 0000000..6977348 --- /dev/null +++ b/cmd/inventory/reservations.go @@ -0,0 +1,98 @@ +package main + +import ( + "encoding/json" + "errors" + "net/http" + "time" + + "git.k6n.net/mats/go-redis-inventory/pkg/inventory" +) + +type reservationLineReq struct { + SKU string `json:"sku"` + LocationID string `json:"locationId"` + Quantity uint32 `json:"quantity"` +} + +type reservationBatchReq struct { + HolderID string `json:"holderId"` + TTLSeconds int64 `json:"ttlSeconds,omitempty"` + Lines []reservationLineReq `json:"lines"` +} + +func (srv *Server) reserveInventoryHandler(w http.ResponseWriter, r *http.Request) { + var req reservationBatchReq + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + if req.HolderID == "" { + http.Error(w, "missing holderId", http.StatusBadRequest) + return + } + if len(req.Lines) == 0 { + http.Error(w, "missing lines", http.StatusBadRequest) + return + } + ttl := 15 * time.Minute + if req.TTLSeconds > 0 { + ttl = time.Duration(req.TTLSeconds) * time.Second + } + for _, line := range req.Lines { + if line.SKU == "" || line.LocationID == "" || line.Quantity == 0 { + http.Error(w, "each line must include sku, locationId, and quantity", http.StatusBadRequest) + return + } + if err := srv.reservationService.ReserveForCart(r.Context(), inventory.CartReserveRequest{ + InventoryReference: &inventory.InventoryReference{ + SKU: inventory.SKU(line.SKU), + LocationID: inventory.LocationID(line.LocationID), + }, + CartID: inventory.CartID(req.HolderID), + Quantity: line.Quantity, + TTL: ttl, + }); err != nil { + status := http.StatusInternalServerError + if errors.Is(err, inventory.ErrInsufficientInventory) { + status = http.StatusConflict + } + http.Error(w, err.Error(), status) + return + } + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"status": "reserved", "holderId": req.HolderID}) +} + +func (srv *Server) releaseReservationHandler(w http.ResponseWriter, r *http.Request) { + var req reservationBatchReq + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + if req.HolderID == "" { + http.Error(w, "missing holderId", http.StatusBadRequest) + return + } + if len(req.Lines) == 0 { + http.Error(w, "missing lines", http.StatusBadRequest) + return + } + for _, line := range req.Lines { + if line.SKU == "" || line.LocationID == "" { + http.Error(w, "each line must include sku and locationId", http.StatusBadRequest) + return + } + if err := srv.reservationService.ReleaseForCart(r.Context(), + inventory.SKU(line.SKU), + inventory.LocationID(line.LocationID), + inventory.CartID(req.HolderID), + ); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"status": "released", "holderId": req.HolderID}) +} diff --git a/cmd/order/amqp_test.go b/cmd/order/amqp_test.go index 1e8897c..49a4bec 100644 --- a/cmd/order/amqp_test.go +++ b/cmd/order/amqp_test.go @@ -10,8 +10,8 @@ import ( "time" "git.k6n.net/mats/go-cart-actor/pkg/actor" - "git.k6n.net/mats/go-cart-actor/pkg/idempotency" "git.k6n.net/mats/go-cart-actor/pkg/flow" + "git.k6n.net/mats/go-cart-actor/pkg/idempotency" "git.k6n.net/mats/go-cart-actor/pkg/order" ) @@ -41,10 +41,7 @@ func testServer(t *testing.T) *server { applier := &orderedApplier{pool: pool, storage: storage} provider := order.NewPassthroughProvider("legacy", "ref", 15000) - freg := flow.NewRegistry() - flow.RegisterBuiltinHooks(freg) - order.RegisterFlowActions(freg, applier, provider) - order.RegisterEmitHook(freg, nil) + freg := buildFlowRegistry(applier, provider, nil, nil, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) engine := flow.NewEngine(freg, slog.New(slog.NewTextHandler(io.Discard, nil))) def, err := order.EmbeddedFlow("place-and-pay") @@ -60,7 +57,7 @@ func testServer(t *testing.T) *server { return &server{ pool: pool, applier: applier, storage: storage, reg: reg, engine: engine, freg: freg, - flows: &flowStore{defs: map[string]*flow.Definition{def.Name: def}}, + flows: &flowStore{defs: map[string]*flow.Definition{def.Name: def}}, provider: provider, taxProvider: order.NewStaticTaxProvider(), defaultFlow: def.Name, idem: idem, logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } diff --git a/cmd/order/custom_hooks.go b/cmd/order/custom_hooks.go new file mode 100644 index 0000000..faa9177 --- /dev/null +++ b/cmd/order/custom_hooks.go @@ -0,0 +1,68 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + + "git.k6n.net/mats/go-cart-actor/pkg/flow" + "git.k6n.net/mats/go-cart-actor/pkg/order" +) + +type projectFulfillmentHookParams struct { + Integration string `json:"integration,omitempty"` + Target string `json:"target,omitempty"` + Dropship bool `json:"dropship,omitempty"` +} + +// registerProjectFlowHooks is the project-layer seam for customer-specific flow +// integrations. These belong here rather than pkg/order so core order logic +// stays generic while deployments can register their own fulfillment tools. +func registerProjectFlowHooks(reg *flow.Registry, app *orderedApplier, logger *slog.Logger) { + if logger == nil { + logger = slog.Default() + } + reg.HookWithMeta("project_fulfillment_integration", func(ctx context.Context, st *flow.State, info flow.HookInfo, params json.RawMessage) error { + var p projectFulfillmentHookParams + if len(params) > 0 { + if err := json.Unmarshal(params, &p); err != nil { + return fmt.Errorf("project_fulfillment_integration: bad params: %w", err) + } + } + if p.Integration == "" { + p.Integration = "order-integration" + } + + o, err := app.Get(ctx, st.ID) + if err != nil { + return err + } + if len(o.Fulfillments) == 0 { + return fmt.Errorf("project_fulfillment_integration: order %s has no fulfillment yet", order.OrderId(st.ID).String()) + } + last := o.Fulfillments[len(o.Fulfillments)-1] + + log := st.Logger + if log == nil { + log = logger + } + log.Info("project fulfillment integration", + "orderId", order.OrderId(st.ID).String(), + "orderReference", o.OrderReference, + "customerEmail", o.CustomerEmail, + "integration", p.Integration, + "target", p.Target, + "dropship", p.Dropship, + "step", info.Step, + "phase", info.Phase, + "fulfillmentId", last.ID, + "carrier", last.Carrier, + "trackingNumber", last.TrackingNumber, + ) + return nil + }, flow.CapabilityMeta{ + Description: "Project-specific fulfillment integration seam for local or testing adapters.", + ExampleParams: json.RawMessage(`{"integration":"dropship-test","target":"erp-sandbox","dropship":true}`), + }) +} diff --git a/cmd/order/custom_hooks_test.go b/cmd/order/custom_hooks_test.go new file mode 100644 index 0000000..170bfdf --- /dev/null +++ b/cmd/order/custom_hooks_test.go @@ -0,0 +1,87 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "strings" + "testing" + + "git.k6n.net/mats/go-cart-actor/pkg/flow" + "git.k6n.net/mats/go-cart-actor/pkg/order" + messages "git.k6n.net/mats/go-cart-actor/proto/order" +) + +func TestProjectFulfillmentIntegrationHookIsRegistered(t *testing.T) { + s := testServer(t) + if !strings.Contains(strings.Join(s.freg.Capabilities().Hooks, ","), "project_fulfillment_integration") { + t.Fatalf("hooks = %v, want project_fulfillment_integration", s.freg.Capabilities().Hooks) + } +} + +func TestProjectFulfillmentIntegrationLogs(t *testing.T) { + s := testServer(t) + var logs bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logs, nil)) + + provider := order.NewPassthroughProvider("legacy", "ref", 25000) + freg := buildFlowRegistry(s.applier, provider, nil, nil, nil, nil, logger) + engine := flow.NewEngine(freg, logger) + + st := flow.NewState(801, logger) + po := orderTestPlaceMsg() + po.CustomerEmail = "dropship@example.com" + st.Vars[order.PlaceOrderVar] = po + + def, err := order.EmbeddedFlow("place-and-pay") + if err != nil { + t.Fatal(err) + } + if _, err := engine.Run(context.Background(), def, st); err != nil { + t.Fatal(err) + } + + ship := &flow.Definition{Name: "ship", Steps: []flow.Step{{ + Name: "fulfill", + Action: "create_fulfillment", + Params: json.RawMessage(`{"carrier":"dropshipper","trackingNumber":"DS-123","lines":[{"reference":"l1","quantity":1}]}`), + Hooks: flow.Hooks{After: []flow.HookRef{{ + Type: "project_fulfillment_integration", + Params: json.RawMessage(`{"integration":"dropship-test","target":"erp-sandbox","dropship":true}`), + }}}, + }}} + if _, err := engine.Run(context.Background(), ship, flow.NewState(801, logger)); err != nil { + t.Fatal(err) + } + + got := logs.String() + for _, want := range []string{ + `"msg":"project fulfillment integration"`, + `"integration":"dropship-test"`, + `"target":"erp-sandbox"`, + `"dropship":true`, + `"trackingNumber":"DS-123"`, + } { + if !strings.Contains(got, want) { + t.Fatalf("log output missing %s: %s", want, got) + } + } +} + +func orderTestPlaceMsg() *messages.PlaceOrder { + return &messages.PlaceOrder{ + OrderReference: "ref-1", + CartId: "cart-1", + Currency: "SEK", + Country: "SE", + CustomerName: "Dropship Customer", + TotalAmount: 25000, + TotalTax: 5000, + PlacedAtMs: 1718877600000, + Lines: []*messages.OrderLine{ + {Reference: "l1", Sku: "ABC", Name: "Widget", Quantity: 2, UnitPrice: 10000, TaxRate: 25, TotalAmount: 20000}, + {Reference: "l2", Sku: "DEF", Name: "Gizmo", Quantity: 1, UnitPrice: 5000, TaxRate: 25, TotalAmount: 5000}, + }, + } +} diff --git a/cmd/order/flow_registry.go b/cmd/order/flow_registry.go new file mode 100644 index 0000000..8781224 --- /dev/null +++ b/cmd/order/flow_registry.go @@ -0,0 +1,29 @@ +package main + +import ( + "log/slog" + + "git.k6n.net/mats/go-cart-actor/pkg/flow" + "git.k6n.net/mats/go-cart-actor/pkg/order" +) + +func buildFlowRegistry( + applier *orderedApplier, + provider order.PaymentProvider, + inventoryReservations order.InventoryReservationService, + emitPub order.Publisher, + emailSender order.EmailSender, + emailTemplates order.EmailTemplateStore, + logger *slog.Logger, +) *flow.Registry { + freg := flow.NewRegistry() + flow.RegisterBuiltinHooks(freg) + order.RegisterFlowActions(freg, applier, provider) + order.RegisterInventoryReservationActions(freg, applier, inventoryReservations) + order.RegisterEmitHook(freg, emitPub) + order.RegisterEmailHook(freg, applier, emailSender, emailTemplates) + order.RegisterFulfillmentWebhookHook(freg, applier, nil) + order.RegisterOrderCreatedEmit(freg, applier, emitPub, "order", "order") + registerProjectFlowHooks(freg, applier, logger) + return freg +} diff --git a/cmd/order/handlers_checkout.go b/cmd/order/handlers_checkout.go index b6ab39e..503ca42 100644 --- a/cmd/order/handlers_checkout.go +++ b/cmd/order/handlers_checkout.go @@ -18,9 +18,9 @@ import ( // via Klarna/Adyen on the checkout grain; the order service records the result // in its event-sourced grain. type fromCheckoutReq struct { - CheckoutId string `json:"checkoutId"` - IdempotencyKey string `json:"idempotencyKey"` - Flow string `json:"flow,omitempty"` + CheckoutId string `json:"checkoutId"` + IdempotencyKey string `json:"idempotencyKey"` + Flow string `json:"flow,omitempty"` CartId string `json:"cartId"` Currency string `json:"currency"` @@ -120,11 +120,7 @@ func (s *server) createOrderFromCheckoutInternal(ctx context.Context, req *fromC return 0, nil, nil, false, nil, fmt.Errorf("unknown flow %q", flowName) } - freg := flow.NewRegistry() - flow.RegisterBuiltinHooks(freg) - order.RegisterFlowActions(freg, s.applier, provider) - order.RegisterEmitHook(freg, nil) - + freg := buildFlowRegistry(s.applier, provider, s.inventoryReservations, nil, nil, nil, s.logger) engine := flow.NewEngine(freg, s.logger) st := flow.NewState(ordID, s.logger) st.Vars[order.PlaceOrderVar] = po @@ -178,6 +174,7 @@ func buildFromCheckoutPlaceOrder(id order.OrderId, req *fromCheckoutReq) *messag TotalAmount: lineTotal, TotalTax: lineTax, Location: l.Location, + DropShip: l.DropShip, }) } po.TotalAmount = total diff --git a/cmd/order/main.go b/cmd/order/main.go index d9ba9da..5c6ce72 100644 --- a/cmd/order/main.go +++ b/cmd/order/main.go @@ -12,6 +12,7 @@ import ( "log" "log/slog" "net/http" + "net/mail" "os" "path/filepath" "strconv" @@ -36,19 +37,20 @@ import ( ) type server struct { - pool *actor.SimpleGrainPool[order.OrderGrain] - applier *orderedApplier - storage *actor.DiskStorage[order.OrderGrain] - reg actor.MutationRegistry - engine *flow.Engine - freg *flow.Registry - flows *flowStore - provider order.PaymentProvider - taxProvider tax.Provider - defaultFlow string - dataDir string - idem *idempotency.Store - logger *slog.Logger + pool *actor.SimpleGrainPool[order.OrderGrain] + applier *orderedApplier + storage *actor.DiskStorage[order.OrderGrain] + reg actor.MutationRegistry + engine *flow.Engine + freg *flow.Registry + flows *flowStore + provider order.PaymentProvider + taxProvider tax.Provider + defaultFlow string + dataDir string + idem *idempotency.Store + logger *slog.Logger + inventoryReservations order.InventoryReservationService } func main() { @@ -101,6 +103,18 @@ func main() { // the broker (publisher is nil without AMQP — the hook then errors at run // time, which is logged, not fatal). amqpURL := os.Getenv("AMQP_URL") + emailTemplates, err := order.LoadEmailTemplates(config.EnvString("ORDER_EMAIL_TEMPLATES", "data/order-email-templates")) + if err != nil { + log.Fatalf("load email templates: %v", err) + } + emailSender, err := selectEmailSender(logger) + if err != nil { + log.Fatalf("configure email sender: %v", err) + } + inventoryReservations, err := selectInventoryReservationService() + if err != nil { + log.Fatalf("configure inventory reservation service: %v", err) + } var emitPub order.Publisher if amqpURL != "" { // The amqp_emit hook writes to a durable outbox rather than the broker @@ -138,13 +152,7 @@ func main() { } } } - freg := flow.NewRegistry() - flow.RegisterBuiltinHooks(freg) - order.RegisterFlowActions(freg, applier, provider) - order.RegisterEmitHook(freg, emitPub) - // order.created domain event → durable outbox → "order" topic exchange. - // Reactors: inventory commit, fulfillment allocation. - order.RegisterOrderCreatedEmit(freg, applier, emitPub, "order", "order") + freg := buildFlowRegistry(applier, provider, inventoryReservations, emitPub, emailSender, emailTemplates, logger) engine := flow.NewEngine(freg, logger) def, err := order.EmbeddedFlow("place-and-pay") @@ -171,6 +179,7 @@ func main() { engine: engine, freg: freg, flows: flows, provider: provider, taxProvider: taxProvider, defaultFlow: def.Name, dataDir: dataDir, idem: idem, logger: logger, + inventoryReservations: inventoryReservations, } mux := http.NewServeMux() @@ -276,8 +285,9 @@ type lineReq struct { Sku string `json:"sku"` Name string `json:"name"` Quantity int32 `json:"quantity"` - UnitPrice int64 `json:"unitPrice"` // inc-vat, minor units - TaxRate int32 `json:"taxRate"` // basis points (2500 = 25%) + UnitPrice int64 `json:"unitPrice"` // inc-vat, minor units + TaxRate int32 `json:"taxRate"` // basis points (2500 = 25%) + DropShip bool `json:"drop_ship,omitempty"` Location string `json:"location,omitempty"` // inventory commit location / store id } @@ -384,6 +394,7 @@ func (s *server) buildPlaceOrder(id order.OrderId, req *checkoutReq) *messages.P TaxRate: l.TaxRate, TotalAmount: lineTotal, TotalTax: lineTax, + DropShip: l.DropShip, }) } po.TotalAmount = total @@ -502,7 +513,6 @@ func selectTaxProvider() tax.Provider { // the dependency-free mock; set PAYMENT_PROVIDER=stripe + STRIPE_SECRET_KEY for // real Stripe (STRIPE_API_BASE / STRIPE_PAYMENT_METHOD optionally override). func selectProvider(logger *slog.Logger) order.PaymentProvider { - if os.Getenv("PAYMENT_PROVIDER") == "stripe" { key := os.Getenv("STRIPE_SECRET_KEY") if key == "" { @@ -515,6 +525,48 @@ func selectProvider(logger *slog.Logger) order.PaymentProvider { return order.NewMockProvider() } +// selectEmailSender picks the flow-email transport. SMTP is the only sender +// implementation for now; when no SMTP host is configured the hook still exists +// in capabilities but errors at run time if a flow tries to use it. +func selectEmailSender(logger *slog.Logger) (order.EmailSender, error) { + kind := config.EnvString("ORDER_EMAIL_SENDER", "") + if kind == "" && os.Getenv("ORDER_EMAIL_SMTP_HOST") == "" { + return nil, nil + } + if kind == "" { + kind = "smtp" + } + if kind != "smtp" { + return nil, fmt.Errorf("unsupported ORDER_EMAIL_SENDER %q", kind) + } + + fromAddr := strings.TrimSpace(os.Getenv("ORDER_EMAIL_FROM_ADDRESS")) + if fromAddr == "" { + return nil, fmt.Errorf("ORDER_EMAIL_FROM_ADDRESS is required when email sender is configured") + } + from := mail.Address{ + Name: strings.TrimSpace(os.Getenv("ORDER_EMAIL_FROM_NAME")), + Address: fromAddr, + } + sender, err := order.NewSMTPEmailSender(order.SMTPEmailSenderConfig{ + Host: config.EnvString("ORDER_EMAIL_SMTP_HOST", ""), + Port: config.EnvInt("ORDER_EMAIL_SMTP_PORT", 587, func(n int) bool { return n > 0 }), + Username: os.Getenv("ORDER_EMAIL_SMTP_USERNAME"), + Password: os.Getenv("ORDER_EMAIL_SMTP_PASSWORD"), + DefaultFrom: from, + Timeout: config.EnvDuration("ORDER_EMAIL_SMTP_TIMEOUT", 10*time.Second), + }) + if err != nil { + return nil, err + } + logger.Info("email sender enabled", "kind", kind, "host", config.EnvString("ORDER_EMAIL_SMTP_HOST", ""), "from", from.Address) + return sender, nil +} + +func selectInventoryReservationService() (order.InventoryReservationService, error) { + return order.NewHTTPInventoryReservationService(config.EnvString("INVENTORY_URL", "http://localhost:8080"), nil) +} + // loadUCPOrderSigner 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 loadUCPOrderSigner() *ucp.SigningConfig { diff --git a/go.work.sum b/go.work.sum index d66ceba..d1ce332 100644 --- a/go.work.sum +++ b/go.work.sum @@ -127,6 +127,7 @@ github.com/golang/glog v1.2.5/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwm github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-pkcs11 v0.3.0/go.mod h1:6eQoGcuNJpa7jnd5pMGdkSaQpNDYvPlXWMcjXXThLlY= github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ= github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So= @@ -152,8 +153,14 @@ github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtX github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= github.com/zeebo/errs v1.4.0/go.mod h1:sgbWHsvVuTPHcqJJGQ1WhI5KbWlHYz+2+2C/LSEtCw4= go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo= +golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= +golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ= +golang.org/x/net v0.54.0/go.mod h1:Sj4oj8jK6XmHpBZU/zWHw3BV3abl4Kvi+Ut7cQcY+cQ= golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/telemetry v0.0.0-20260409153401-be6f6cb8b1fa/go.mod h1:kHjTxDEnAu6/Nl9lDkzjWpR+bmKfxeiRuSDlsMb70gE= +golang.org/x/telemetry v0.0.0-20260508192327-42602be52be6/go.mod h1:Eqhaxk/wZsWEH8CRxLwj6xzEJbz7k1EFGqx7nyCoabE= +golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0= golang.org/x/tools/go/expect v0.1.0-deprecated/go.mod h1:eihoPOH+FgIqa3FpoTwguz/bVUSGBlGQU67vpBeOrBY= golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated/go.mod h1:RVAQXBGNv1ib0J382/DPCRS/BPnsGebyM1Gj5VSDpG8= google.golang.org/appengine v1.6.8/go.mod h1:1jJ3jBArFh5pcgW8gCtRJnepW8FzD1V44FJffLiz/Ds= diff --git a/internal/ucp/checkout.go b/internal/ucp/checkout.go index 05cd664..ccbd94b 100644 --- a/internal/ucp/checkout.go +++ b/internal/ucp/checkout.go @@ -451,6 +451,8 @@ type orderLine struct { Quantity int32 `json:"quantity"` UnitPrice money.Cents `json:"unitPrice"` TaxRate int32 `json:"taxRate"` + DropShip bool `json:"drop_ship,omitempty"` + Location string `json:"location,omitempty"` } type orderPayment struct { @@ -549,6 +551,10 @@ func buildOrderLines(g *checkout.CheckoutGrain) []orderLine { if it.Meta != nil { name = it.Meta.Name } + location := "" + if it.StoreId != nil { + location = *it.StoreId + } lines = append(lines, orderLine{ Reference: it.Sku, Sku: it.Sku, @@ -558,6 +564,8 @@ func buildOrderLines(g *checkout.CheckoutGrain) []orderLine { // CartItem.Tax and OrderLine.TaxRate share the platform basis-point scale // (2500 = 25%), so the rate passes through unchanged. TaxRate: int32(it.Tax), + DropShip: it.DropShip, + Location: location, }) } for _, d := range g.Deliveries { diff --git a/pkg/backofficeadmin/app.go b/pkg/backofficeadmin/app.go index 1615f83..d35b7ba 100644 --- a/pkg/backofficeadmin/app.go +++ b/pkg/backofficeadmin/app.go @@ -12,7 +12,6 @@ package backofficeadmin import ( "context" - "encoding/json" "log" "net/http" "os" @@ -22,10 +21,8 @@ import ( "git.k6n.net/mats/go-cart-actor/pkg/cart" "git.k6n.net/mats/go-cart-actor/pkg/checkout" "git.k6n.net/mats/go-cart-actor/pkg/profile" - "git.k6n.net/mats/go-redis-inventory/pkg/inventory" "git.k6n.net/mats/platform/rabbit" amqp "github.com/rabbitmq/amqp091-go" - "github.com/redis/go-redis/v9" ) // CartFileInfo is the metadata returned by the cart listing endpoint. @@ -52,9 +49,6 @@ type Config struct { CheckoutDataDir string // ProfileDataDir holds the profile event logs (optional, default "" — customer endpoints disabled). ProfileDataDir string - // RedisAddress / RedisPassword reach the inventory store. - RedisAddress string - RedisPassword string } // App is the commerce admin control plane. Construct with New, register its @@ -62,15 +56,15 @@ type Config struct { type App struct { fs *FileServer hub *Hub - rdb *redis.Client - inv *inventory.RedisInventoryService cs *CustomerServer } -// 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. +// New constructs the commerce admin: it opens 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. +// +// Inventory writes no longer live here — they go through the standalone inventory +// service's HTTP API (/api/inventory/batch), reverse-proxied by the host. func New(cfg Config) (*App, error) { if cfg.DataDir == "" { cfg.DataDir = "data" @@ -79,16 +73,6 @@ func New(cfg Config) (*App, error) { 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) @@ -109,8 +93,6 @@ func New(cfg Config) (*App, error) { return &App{ fs: NewFileServer(cfg.DataDir, cfg.CheckoutDataDir, diskStorage, diskStorageCheckout), hub: NewHub(), - rdb: rdb, - inv: inventoryService, cs: customerSrv, }, nil } @@ -123,7 +105,6 @@ func (a *App) RegisterRoutes(mux *http.ServeMux) { 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) @@ -134,28 +115,7 @@ func (a *App) RegisterRoutes(mux *http.ServeMux) { } } -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 @@ -169,10 +129,10 @@ func (a *App) Start(ctx context.Context, conn *rabbit.Conn) error { return nil } -// Shutdown releases the App's resources (the Redis client). The background -// goroutines are stopped by cancelling the ctx passed to Start. +// Shutdown releases the App's resources. The background goroutines are stopped +// by cancelling the ctx passed to Start. func (a *App) Shutdown(_ context.Context) error { - return a.rdb.Close() + return nil } // startMutationConsumer subscribes to the cart "mutation" topic and forwards diff --git a/pkg/cart/cart-mutation-helper.go b/pkg/cart/cart-mutation-helper.go index f1a5c7a..371f23e 100644 --- a/pkg/cart/cart-mutation-helper.go +++ b/pkg/cart/cart-mutation-helper.go @@ -5,14 +5,14 @@ import ( "time" "git.k6n.net/mats/go-cart-actor/pkg/actor" - "git.k6n.net/mats/go-redis-inventory/pkg/inventory" + "git.k6n.net/mats/platform/inventory" ) type CartMutationContext struct { - reservationService inventory.CartReservationService + reservationService inventory.ReservationPolicy } -func NewCartMutationContext(reservationService inventory.CartReservationService) *CartMutationContext { +func NewCartMutationContext(reservationService inventory.ReservationPolicy) *CartMutationContext { return &CartMutationContext{ reservationService: reservationService, } @@ -30,7 +30,7 @@ func (c *CartMutationContext) ReserveItem(ctx context.Context, cartId CartId, sk } ttl := time.Minute * 15 endTime := time.Now().Add(ttl) - err := c.reservationService.ReserveForCart(ctx, inventory.CartReserveRequest{ + err := c.reservationService.Reserve(ctx, inventory.CartReserveRequest{ CartID: inventory.CartID(cartId.String()), InventoryReference: &inventory.InventoryReference{ SKU: inventory.SKU(sku), @@ -57,7 +57,7 @@ func (c *CartMutationContext) UseReservations(item *CartItem) bool { if item.InventoryTracked { return true } - return item.Cgm == "55010" + return false } func (c *CartMutationContext) ReleaseItem(ctx context.Context, cartId CartId, sku string, locationId *string) error { @@ -68,7 +68,7 @@ func (c *CartMutationContext) ReleaseItem(ctx context.Context, cartId CartId, sk if locationId != nil { l = inventory.LocationID(*locationId) } - return c.reservationService.ReleaseForCart(ctx, inventory.SKU(sku), l, inventory.CartID(cartId.String())) + return c.reservationService.Release(ctx, inventory.SKU(sku), l, inventory.CartID(cartId.String())) } func NewCartMultationRegistry(context *CartMutationContext) actor.MutationRegistry { diff --git a/pkg/flow/engine.go b/pkg/flow/engine.go index 168c6e0..19739e3 100644 --- a/pkg/flow/engine.go +++ b/pkg/flow/engine.go @@ -54,42 +54,78 @@ type HookInfo struct { // — so observability/notification hooks can't break the business transaction. type Hook func(ctx context.Context, st *State, info HookInfo, params json.RawMessage) error +// CapabilityMeta is the editor-facing documentation attached at registration +// time. The server generates `/sagas/capabilities` from these registrations, so +// UIs should consume this payload instead of hardcoding per-tool help. +// +// Register every action/predicate/hook with the matching *WithMeta helper when +// it should be discoverable in editors. ExampleParams is arbitrary JSON shown as +// a starter payload for params-capable tools. +type CapabilityMeta struct { + Description string `json:"description,omitempty"` + ExampleParams json.RawMessage `json:"exampleParams,omitempty"` +} + // Registry holds the named actions, predicates and hooks a flow can reference. type Registry struct { - mu sync.RWMutex - actions map[string]Action - predicates map[string]Predicate - hooks map[string]Hook + mu sync.RWMutex + actions map[string]Action + actionMeta map[string]CapabilityMeta + predicates map[string]Predicate + predicateMeta map[string]CapabilityMeta + hooks map[string]Hook + hookMeta map[string]CapabilityMeta } // NewRegistry returns an empty registry. func NewRegistry() *Registry { return &Registry{ - actions: map[string]Action{}, - predicates: map[string]Predicate{}, - hooks: map[string]Hook{}, + actions: map[string]Action{}, + actionMeta: map[string]CapabilityMeta{}, + predicates: map[string]Predicate{}, + predicateMeta: map[string]CapabilityMeta{}, + hooks: map[string]Hook{}, + hookMeta: map[string]CapabilityMeta{}, } } // Action registers an action under name (overwrites an existing one). func (r *Registry) Action(name string, fn Action) { + r.ActionWithMeta(name, fn, CapabilityMeta{}) +} + +// ActionWithMeta registers an action plus the metadata editors should display. +func (r *Registry) ActionWithMeta(name string, fn Action, meta CapabilityMeta) { r.mu.Lock() defer r.mu.Unlock() r.actions[name] = fn + r.actionMeta[name] = meta } // Predicate registers a predicate under name. func (r *Registry) Predicate(name string, fn Predicate) { + r.PredicateWithMeta(name, fn, CapabilityMeta{}) +} + +// PredicateWithMeta registers a predicate plus the metadata editors should display. +func (r *Registry) PredicateWithMeta(name string, fn Predicate, meta CapabilityMeta) { r.mu.Lock() defer r.mu.Unlock() r.predicates[name] = fn + r.predicateMeta[name] = meta } // Hook registers a hook under name. func (r *Registry) Hook(name string, fn Hook) { + r.HookWithMeta(name, fn, CapabilityMeta{}) +} + +// HookWithMeta registers a hook plus the metadata editors should display. +func (r *Registry) HookWithMeta(name string, fn Hook, meta CapabilityMeta) { r.mu.Lock() defer r.mu.Unlock() r.hooks[name] = fn + r.hookMeta[name] = meta } func (r *Registry) action(name string) (Action, bool) { @@ -116,9 +152,12 @@ func (r *Registry) hook(name string) (Hook, bool) { // Capabilities is the set of registered names, surfaced so an editor UI can // offer exactly the actions/predicates/hooks a flow may reference. type Capabilities struct { - Actions []string `json:"actions"` - Predicates []string `json:"predicates"` - Hooks []string `json:"hooks"` + Actions []string `json:"actions"` + ActionMeta map[string]CapabilityMeta `json:"actionMeta,omitempty"` + Predicates []string `json:"predicates"` + PredicateMeta map[string]CapabilityMeta `json:"predicateMeta,omitempty"` + Hooks []string `json:"hooks"` + HookMeta map[string]CapabilityMeta `json:"hookMeta,omitempty"` } // Capabilities returns the sorted registered names. @@ -126,12 +165,23 @@ func (r *Registry) Capabilities() Capabilities { r.mu.RLock() defer r.mu.RUnlock() return Capabilities{ - Actions: sortedKeys(r.actions), - Predicates: sortedKeys(r.predicates), - Hooks: sortedKeys(r.hooks), + Actions: sortedKeys(r.actions), + ActionMeta: cloneMeta(r.actionMeta), + Predicates: sortedKeys(r.predicates), + PredicateMeta: cloneMeta(r.predicateMeta), + Hooks: sortedKeys(r.hooks), + HookMeta: cloneMeta(r.hookMeta), } } +func cloneMeta(in map[string]CapabilityMeta) map[string]CapabilityMeta { + out := make(map[string]CapabilityMeta, len(in)) + for k, v := range in { + out[k] = v + } + return out +} + func sortedKeys[V any](m map[string]V) []string { out := make([]string, 0, len(m)) for k := range m { diff --git a/pkg/flow/hooks.go b/pkg/flow/hooks.go index e95a759..173e329 100644 --- a/pkg/flow/hooks.go +++ b/pkg/flow/hooks.go @@ -13,9 +13,17 @@ import ( // (structured log line) and "webhook" (HTTP POST of the event). Domain packages // register their own hooks (e.g. "amqp_emit") on the same registry. func RegisterBuiltinHooks(reg *Registry) { - reg.Hook("log", logHook) - reg.Hook("noop", func(context.Context, *State, HookInfo, json.RawMessage) error { return nil }) - reg.Hook("webhook", webhookHook(&http.Client{Timeout: 10 * time.Second})) + reg.HookWithMeta("log", logHook, CapabilityMeta{ + Description: "Write a structured log line for the current step and phase.", + ExampleParams: json.RawMessage(`{"message":"payment captured"}`), + }) + reg.HookWithMeta("noop", func(context.Context, *State, HookInfo, json.RawMessage) error { return nil }, CapabilityMeta{ + Description: "Do nothing. Useful as a placeholder while shaping a flow.", + }) + reg.HookWithMeta("webhook", webhookHook(&http.Client{Timeout: 10 * time.Second}), CapabilityMeta{ + Description: "POST the flow event payload to an external HTTP endpoint.", + ExampleParams: json.RawMessage(`{"url":"https://example.test/order-events","headers":{"X-Flow":"place-and-pay"}}`), + }) } // logHook emits a structured log line for the step/phase. Params (optional): diff --git a/pkg/order/actions.go b/pkg/order/actions.go index f896a50..7fc47ab 100644 --- a/pkg/order/actions.go +++ b/pkg/order/actions.go @@ -64,7 +64,7 @@ const PlaceOrderVar = "placeOrder" // grain through app and taking payment via provider. Compose with // flow.RegisterBuiltinHooks(reg) to get the "log"/"webhook" hooks too. func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvider) { - reg.Action("place_order", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { + reg.ActionWithMeta("place_order", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { po, ok := st.Vars[PlaceOrderVar].(*messages.PlaceOrder) if !ok || po == nil { return fmt.Errorf("place_order: flow var %q is not a *PlaceOrder", PlaceOrderVar) @@ -78,9 +78,9 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid st.Vars["currency"] = po.GetCurrency() st.Vars["orderReference"] = po.GetOrderReference() return nil - }) + }, flow.CapabilityMeta{Description: "Create the order from the flow state variable `placeOrder`."}) - reg.Action("authorize_payment", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { + reg.ActionWithMeta("authorize_payment", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { o, err := app.Get(ctx, st.ID) if err != nil { return err @@ -101,9 +101,9 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid st.Vars["authProvider"] = auth.Provider st.Vars["authAmount"] = auth.Amount return nil - }) + }, flow.CapabilityMeta{Description: "Authorize payment with the configured provider."}) - reg.Action("capture_payment", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { + reg.ActionWithMeta("capture_payment", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { authRef, _ := st.Vars["authRef"].(string) amount, _ := st.Vars["authAmount"].(int64) if amount == 0 { @@ -121,11 +121,11 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid Reference: capture.Reference, AtMs: nowMs(), }) - }) + }, flow.CapabilityMeta{Description: "Capture the authorized payment."}) // void_payment is the compensation for authorize_payment: void with the // provider and cancel the order if it is still in a cancellable state. - reg.Action("void_payment", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { + reg.ActionWithMeta("void_payment", func(ctx context.Context, st *flow.State, _ json.RawMessage) error { if authRef, ok := st.Vars["authRef"].(string); ok && authRef != "" { if err := provider.Void(ctx, authRef); err != nil { return err @@ -134,9 +134,9 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid // Best-effort cancel; ignore if no longer cancellable. _, _ = app.Apply(ctx, st.ID, &messages.CancelOrder{Reason: "compensation: payment voided", AtMs: nowMs()}) return nil - }) + }, flow.CapabilityMeta{Description: "Void the authorization and cancel the order as compensation."}) - reg.Action("cancel_order", func(ctx context.Context, st *flow.State, params json.RawMessage) error { + reg.ActionWithMeta("cancel_order", func(ctx context.Context, st *flow.State, params json.RawMessage) error { reason := "cancelled" if len(params) > 0 { var p struct { @@ -147,9 +147,12 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid } } return applyOne(ctx, app, st.ID, &messages.CancelOrder{Reason: reason, AtMs: nowMs()}) + }, flow.CapabilityMeta{ + Description: "Cancel the order with an optional reason.", + ExampleParams: json.RawMessage(`{"reason":"customer requested cancellation"}`), }) - reg.Action("issue_refund", func(ctx context.Context, st *flow.State, params json.RawMessage) error { + reg.ActionWithMeta("issue_refund", func(ctx context.Context, st *flow.State, params json.RawMessage) error { o, err := app.Get(ctx, st.ID) if err != nil { return err @@ -180,6 +183,64 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid Reference: ref.Reference, AtMs: nowMs(), }) + }, flow.CapabilityMeta{ + Description: "Refund the remaining or specified captured amount.", + ExampleParams: json.RawMessage(`{"amount":2500}`), + }) + + reg.ActionWithMeta("create_fulfillment", func(ctx context.Context, st *flow.State, params json.RawMessage) error { + var p struct { + ID string `json:"id"` + Carrier string `json:"carrier,omitempty"` + TrackingNumber string `json:"trackingNumber,omitempty"` + TrackingURI string `json:"trackingUri,omitempty"` + AtMs int64 `json:"atMs,omitempty"` + Lines []struct { + Reference string `json:"reference"` + Quantity int32 `json:"quantity"` + } `json:"lines"` + } + if err := json.Unmarshal(params, &p); err != nil { + return fmt.Errorf("create_fulfillment: bad params: %w", err) + } + if len(p.Lines) == 0 { + return fmt.Errorf("create_fulfillment: missing lines") + } + id := p.ID + if id == "" { + fid, err := NewOrderId() + if err != nil { + return err + } + id = "f_" + fid.String() + } + atMs := p.AtMs + if atMs == 0 { + atMs = nowMs() + } + msg := &messages.CreateFulfillment{ + Id: id, + Carrier: p.Carrier, + TrackingNumber: p.TrackingNumber, + TrackingUri: p.TrackingURI, + AtMs: atMs, + } + for _, line := range p.Lines { + if line.Reference == "" { + return fmt.Errorf("create_fulfillment: line missing reference") + } + if line.Quantity <= 0 { + return fmt.Errorf("create_fulfillment: line %q has invalid quantity %d", line.Reference, line.Quantity) + } + msg.Lines = append(msg.Lines, &messages.FulfillmentLine{ + Reference: line.Reference, + Quantity: line.Quantity, + }) + } + return applyOne(ctx, app, st.ID, msg) + }, flow.CapabilityMeta{ + Description: "Create a shipment / fulfillment and advance the order fulfillment status.", + ExampleParams: json.RawMessage(`{"carrier":"dropshipper","trackingNumber":"TRACK-123","trackingUri":"https://tracking.example/TRACK-123","lines":[{"reference":"l1","quantity":1}]}`), }) registerPredicates(reg, app) @@ -193,39 +254,39 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid func registerPredicates(reg *flow.Registry, app Applier) { get := func(ctx context.Context, st *flow.State) (*OrderGrain, error) { return app.Get(ctx, st.ID) } - reg.Predicate("has_customer_email", func(ctx context.Context, st *flow.State) (bool, error) { + reg.PredicateWithMeta("has_customer_email", func(ctx context.Context, st *flow.State) (bool, error) { o, err := get(ctx, st) if err != nil { return false, err } return o.CustomerEmail != "", nil - }) - reg.Predicate("has_items", func(ctx context.Context, st *flow.State) (bool, error) { + }, flow.CapabilityMeta{Description: "Run only when the order has a customer email address."}) + reg.PredicateWithMeta("has_items", func(ctx context.Context, st *flow.State) (bool, error) { o, err := get(ctx, st) if err != nil { return false, err } return len(o.Lines) > 0, nil - }) - reg.Predicate("is_captured", func(ctx context.Context, st *flow.State) (bool, error) { + }, flow.CapabilityMeta{Description: "Run only when the order contains at least one line item."}) + reg.PredicateWithMeta("is_captured", func(ctx context.Context, st *flow.State) (bool, error) { o, err := get(ctx, st) if err != nil { return false, err } return o.Status == StatusCaptured, nil - }) - reg.Predicate("not_captured", func(ctx context.Context, st *flow.State) (bool, error) { + }, flow.CapabilityMeta{Description: "Run only after payment has been captured."}) + reg.PredicateWithMeta("not_captured", func(ctx context.Context, st *flow.State) (bool, error) { o, err := get(ctx, st) if err != nil { return false, err } return o.Status != StatusCaptured, nil - }) - reg.Predicate("has_open_balance", func(ctx context.Context, st *flow.State) (bool, error) { + }, flow.CapabilityMeta{Description: "Run only before payment is captured."}) + reg.PredicateWithMeta("has_open_balance", func(ctx context.Context, st *flow.State) (bool, error) { o, err := get(ctx, st) if err != nil { return false, err } return o.CapturedAmount-o.RefundedAmount > 0, nil - }) + }, flow.CapabilityMeta{Description: "Run only when captured amount remains to refund."}) } diff --git a/pkg/order/capabilities_test.go b/pkg/order/capabilities_test.go index ca33d71..b7fdd9d 100644 --- a/pkg/order/capabilities_test.go +++ b/pkg/order/capabilities_test.go @@ -3,19 +3,28 @@ package order import ( "context" "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/mail" "slices" "strings" "testing" + "time" "git.k6n.net/mats/go-cart-actor/pkg/flow" ) func fullRegistry(t *testing.T, pub Publisher) *flow.Registry { t.Helper() + pool := newPool(t) reg := flow.NewRegistry() flow.RegisterBuiltinHooks(reg) - RegisterFlowActions(reg, newPool(t), NewMockProvider()) + RegisterFlowActions(reg, pool, NewMockProvider()) + RegisterInventoryReservationActions(reg, pool, nil) RegisterEmitHook(reg, pub) + RegisterEmailHook(reg, pool, nil, nil) + RegisterFulfillmentWebhookHook(reg, pool, nil) return reg } @@ -30,6 +39,92 @@ func TestCapabilitiesIncludesPredicatesAndEmit(t *testing.T) { if !slices.Contains(caps.Hooks, "amqp_emit") { t.Errorf("hooks missing amqp_emit (got %v)", caps.Hooks) } + if !slices.Contains(caps.Hooks, "send_email") { + t.Errorf("hooks missing send_email (got %v)", caps.Hooks) + } + if !slices.Contains(caps.Hooks, "fulfillment_webhook") { + t.Errorf("hooks missing fulfillment_webhook (got %v)", caps.Hooks) + } + if !slices.Contains(caps.Actions, "create_fulfillment") { + t.Errorf("actions missing create_fulfillment (got %v)", caps.Actions) + } + if !slices.Contains(caps.Actions, "reserve_inventory") { + t.Errorf("actions missing reserve_inventory (got %v)", caps.Actions) + } + if !slices.Contains(caps.Actions, "release_inventory") { + t.Errorf("actions missing release_inventory (got %v)", caps.Actions) + } + if caps.ActionMeta["create_fulfillment"].Description == "" { + t.Fatalf("action meta missing for create_fulfillment: %#v", caps.ActionMeta["create_fulfillment"]) + } + if len(caps.HookMeta["fulfillment_webhook"].ExampleParams) == 0 { + t.Fatalf("hook meta missing example params for fulfillment_webhook: %#v", caps.HookMeta["fulfillment_webhook"]) + } + if caps.PredicateMeta["has_customer_email"].Description == "" { + t.Fatalf("predicate meta missing for has_customer_email: %#v", caps.PredicateMeta["has_customer_email"]) + } +} + +type fakeInventoryReservationService struct { + reserved []InventoryReservationRequest + released []InventoryReservationRequest +} + +func (f *fakeInventoryReservationService) Reserve(_ context.Context, req InventoryReservationRequest) error { + f.reserved = append(f.reserved, req) + return nil +} + +func (f *fakeInventoryReservationService) Release(_ context.Context, req InventoryReservationRequest) error { + f.released = append(f.released, req) + return nil +} + +func TestReserveInventoryActionBuildsReservationRequest(t *testing.T) { + pool := newPool(t) + reg := flow.NewRegistry() + RegisterFlowActions(reg, pool, NewMockProvider()) + fake := &fakeInventoryReservationService{} + RegisterInventoryReservationActions(reg, pool, fake) + eng := flow.NewEngine(reg, nil) + + const id = 703 + st := flow.NewState(id, nil) + po := placeMsg() + st.Vars[PlaceOrderVar] = po + def, err := EmbeddedFlow("place-and-pay") + if err != nil { + t.Fatal(err) + } + if _, err := eng.Run(context.Background(), def, st); err != nil { + t.Fatal(err) + } + + reserve := &flow.Definition{Name: "reserve", Steps: []flow.Step{{ + Name: "reserve", + Action: "reserve_inventory", + Params: json.RawMessage(`{"ttlSeconds":600,"lines":[{"reference":"l1","quantity":1,"locationId":"se"}]}`), + Compensate: &flow.ActionRef{ + Action: "release_inventory", + Params: json.RawMessage(`{"lines":[{"reference":"l1","quantity":1,"locationId":"se"}]}`), + }, + }}} + if _, err := eng.Run(context.Background(), reserve, flow.NewState(id, nil)); err != nil { + t.Fatal(err) + } + if len(fake.reserved) != 1 { + t.Fatalf("reserved = %d, want 1", len(fake.reserved)) + } + got := fake.reserved[0] + if got.HolderID != "order:"+OrderId(id).String() { + t.Fatalf("holderId = %q", got.HolderID) + } + if got.TTL != 10*time.Minute { + t.Fatalf("ttl = %s", got.TTL) + } + if len(got.Lines) != 1 || got.Lines[0].SKU != "ABC" || got.Lines[0].Quantity != 1 { + t.Fatalf("lines = %+v", got.Lines) + } } func TestIsCapturedPredicateGatesStep(t *testing.T) { @@ -94,3 +189,148 @@ func TestAmqpEmitHookPublishes(t *testing.T) { t.Fatalf("expected one message routed to k, got %v", fp.msgs) } } + +type fakeEmailSender struct { + defaultFrom mail.Address + msgs []EmailMessage +} + +func (f *fakeEmailSender) DefaultFrom() mail.Address { return f.defaultFrom } + +func (f *fakeEmailSender) Send(_ context.Context, msg EmailMessage) error { + f.msgs = append(f.msgs, msg) + return nil +} + +func TestSendEmailHookRendersFulfillmentTemplate(t *testing.T) { + pool := newPool(t) + reg := flow.NewRegistry() + RegisterFlowActions(reg, pool, NewMockProvider()) + + sender := &fakeEmailSender{defaultFrom: mail.Address{Name: "Ops", Address: "ops@example.com"}} + templates := newEmailTemplateCatalog(map[string]EmailTemplate{ + "ship": { + Subject: "Order {{ .Order.OrderReference }} shipped", + Text: "Tracking {{ .Fulfillment.TrackingNumber }} for {{ .Order.CustomerEmail }}", + }, + }) + RegisterEmailHook(reg, pool, sender, templates) + eng := flow.NewEngine(reg, nil) + + const id = 701 + st := flow.NewState(id, nil) + po := placeMsg() + po.CustomerEmail = "alice@example.com" + po.CustomerName = "Alice" + st.Vars[PlaceOrderVar] = po + + def, err := EmbeddedFlow("place-and-pay") + if err != nil { + t.Fatal(err) + } + if _, err := eng.Run(context.Background(), def, st); err != nil { + t.Fatal(err) + } + + ship := &flow.Definition{Name: "ship", Steps: []flow.Step{{ + Name: "fulfill", + Action: "create_fulfillment", + Params: json.RawMessage(`{"carrier":"postnord","trackingNumber":"TRACK-123","trackingUri":"https://track.example/TRACK-123","lines":[{"reference":"l1","quantity":1}]}`), + Hooks: flow.Hooks{After: []flow.HookRef{{Type: "send_email", Params: json.RawMessage(`{"template":"ship"}`)}}}, + }}} + if _, err := eng.Run(context.Background(), ship, flow.NewState(id, nil)); err != nil { + t.Fatal(err) + } + + if len(sender.msgs) != 1 { + t.Fatalf("sent %d emails, want 1", len(sender.msgs)) + } + got := sender.msgs[0] + if got.From.Address != "ops@example.com" { + t.Fatalf("from = %q, want ops@example.com", got.From.Address) + } + if len(got.To) != 1 || got.To[0].Address != "alice@example.com" { + t.Fatalf("to = %+v, want alice@example.com", got.To) + } + if got.Subject != "Order ref-1 shipped" { + t.Fatalf("subject = %q", got.Subject) + } + if !strings.Contains(got.TextBody, "TRACK-123") { + t.Fatalf("text body = %q, want tracking number", got.TextBody) + } +} + +func TestFulfillmentWebhookPostsOrderPayload(t *testing.T) { + pool := newPool(t) + reg := flow.NewRegistry() + RegisterFlowActions(reg, pool, NewMockProvider()) + + var gotMethod, gotPath, gotHeader string + var gotBody map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod = r.Method + gotPath = r.URL.Path + gotHeader = r.Header.Get("X-Integration") + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Fatalf("decode webhook body: %v", err) + } + w.WriteHeader(http.StatusAccepted) + })) + defer srv.Close() + + RegisterFulfillmentWebhookHook(reg, pool, srv.Client()) + eng := flow.NewEngine(reg, nil) + + const id = 702 + st := flow.NewState(id, nil) + po := placeMsg() + po.CustomerEmail = "dropship@example.com" + st.Vars[PlaceOrderVar] = po + + def, err := EmbeddedFlow("place-and-pay") + if err != nil { + t.Fatal(err) + } + if _, err := eng.Run(context.Background(), def, st); err != nil { + t.Fatal(err) + } + + ship := &flow.Definition{Name: "ship", Steps: []flow.Step{{ + Name: "fulfill", + Action: "create_fulfillment", + Params: json.RawMessage(`{"carrier":"dropshipper","trackingNumber":"DS-123","lines":[{"reference":"l1","quantity":1}]}`), + Hooks: flow.Hooks{After: []flow.HookRef{{ + Type: "fulfillment_webhook", + Params: json.RawMessage(fmt.Sprintf(`{ + "url": %q, + "method": "POST", + "headers": { + "X-Integration": "dropship {{ .Order.OrderReference }}" + } + }`, srv.URL+`/hooks/{{ .OrderID }}`)), + }}}, + }}} + if _, err := eng.Run(context.Background(), ship, flow.NewState(id, nil)); err != nil { + t.Fatal(err) + } + + if gotMethod != http.MethodPost { + t.Fatalf("method = %q, want POST", gotMethod) + } + if gotPath != "/hooks/"+OrderId(id).String() { + t.Fatalf("path = %q", gotPath) + } + if gotHeader != "dropship ref-1" { + t.Fatalf("header = %q", gotHeader) + } + if gotBody["orderId"] != OrderId(id).String() { + t.Fatalf("body orderId = %v", gotBody["orderId"]) + } + fulfillment, ok := gotBody["fulfillment"].(map[string]any) + if !ok { + t.Fatalf("body fulfillment missing: %#v", gotBody["fulfillment"]) + } + if fulfillment["trackingNumber"] != "DS-123" { + t.Fatalf("trackingNumber = %v", fulfillment["trackingNumber"]) + } +} diff --git a/pkg/order/email.go b/pkg/order/email.go new file mode 100644 index 0000000..2c8be98 --- /dev/null +++ b/pkg/order/email.go @@ -0,0 +1,508 @@ +package order + +import ( + "bytes" + "context" + "crypto/tls" + "embed" + "encoding/json" + "fmt" + htmltmpl "html/template" + "io/fs" + "mime" + "mime/quotedprintable" + "net" + "net/mail" + "net/smtp" + "os" + "path/filepath" + "strings" + texttmpl "text/template" + "time" + + "git.k6n.net/mats/go-cart-actor/pkg/flow" +) + +// EmailMessage is the rendered mail payload a sender delivers. +type EmailMessage struct { + From mail.Address + To []mail.Address + Subject string + TextBody string + HTMLBody string +} + +// EmailSender delivers an already-rendered mail and exposes its default sender. +type EmailSender interface { + Send(ctx context.Context, msg EmailMessage) error + DefaultFrom() mail.Address +} + +// SMTPEmailSenderConfig configures the SMTP mail transport. +type SMTPEmailSenderConfig struct { + Host string + Port int + Username string + Password string + DefaultFrom mail.Address + Timeout time.Duration +} + +type smtpEmailSender struct { + host string + addr string + auth smtp.Auth + defaultFrom mail.Address + timeout time.Duration +} + +// NewSMTPEmailSender returns an SMTP-backed sender. It supports plain SMTP with +// opportunistic STARTTLS, which is the only configured transport for now. +func NewSMTPEmailSender(cfg SMTPEmailSenderConfig) (EmailSender, error) { + if strings.TrimSpace(cfg.Host) == "" { + return nil, fmt.Errorf("smtp sender: missing host") + } + if cfg.Port <= 0 { + return nil, fmt.Errorf("smtp sender: invalid port %d", cfg.Port) + } + if cfg.Timeout <= 0 { + cfg.Timeout = 10 * time.Second + } + if strings.TrimSpace(cfg.DefaultFrom.Address) == "" { + return nil, fmt.Errorf("smtp sender: missing default from address") + } + s := &smtpEmailSender{ + host: cfg.Host, + addr: net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port)), + defaultFrom: cfg.DefaultFrom, + timeout: cfg.Timeout, + } + if cfg.Username != "" { + s.auth = smtp.PlainAuth("", cfg.Username, cfg.Password, cfg.Host) + } + return s, nil +} + +func (s *smtpEmailSender) DefaultFrom() mail.Address { return s.defaultFrom } + +func (s *smtpEmailSender) Send(ctx context.Context, msg EmailMessage) error { + if strings.TrimSpace(msg.From.Address) == "" { + return fmt.Errorf("smtp sender: missing from address") + } + if len(msg.To) == 0 { + return fmt.Errorf("smtp sender: missing recipients") + } + if strings.TrimSpace(msg.Subject) == "" { + return fmt.Errorf("smtp sender: missing subject") + } + body, err := buildSMTPMessage(msg) + if err != nil { + return err + } + + conn, err := (&net.Dialer{Timeout: s.timeout}).DialContext(ctx, "tcp", s.addr) + if err != nil { + return fmt.Errorf("smtp sender: dial %s: %w", s.addr, err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(s.timeout)) + + client, err := smtp.NewClient(conn, s.host) + if err != nil { + return fmt.Errorf("smtp sender: create client: %w", err) + } + defer client.Close() + + if ok, _ := client.Extension("STARTTLS"); ok { + if err := client.StartTLS(&tls.Config{ServerName: s.host, MinVersion: tls.VersionTLS12}); err != nil { + return fmt.Errorf("smtp sender: starttls: %w", err) + } + } + if s.auth != nil { + if ok, _ := client.Extension("AUTH"); !ok { + return fmt.Errorf("smtp sender: server does not support auth") + } + if err := client.Auth(s.auth); err != nil { + return fmt.Errorf("smtp sender: auth: %w", err) + } + } + if err := client.Mail(msg.From.Address); err != nil { + return fmt.Errorf("smtp sender: mail from %s: %w", msg.From.Address, err) + } + for _, rcpt := range msg.To { + if err := client.Rcpt(rcpt.Address); err != nil { + return fmt.Errorf("smtp sender: rcpt to %s: %w", rcpt.Address, err) + } + } + w, err := client.Data() + if err != nil { + return fmt.Errorf("smtp sender: data: %w", err) + } + if _, err := w.Write(body); err != nil { + _ = w.Close() + return fmt.Errorf("smtp sender: write body: %w", err) + } + if err := w.Close(); err != nil { + return fmt.Errorf("smtp sender: close body: %w", err) + } + if err := client.Quit(); err != nil { + return fmt.Errorf("smtp sender: quit: %w", err) + } + return nil +} + +func buildSMTPMessage(msg EmailMessage) ([]byte, error) { + if strings.TrimSpace(msg.TextBody) == "" && strings.TrimSpace(msg.HTMLBody) == "" { + return nil, fmt.Errorf("email: missing body") + } + + var buf bytes.Buffer + writeHeader := func(name, value string) { + buf.WriteString(name) + buf.WriteString(": ") + buf.WriteString(value) + buf.WriteString("\r\n") + } + writeHeader("From", msg.From.String()) + toList := make([]string, 0, len(msg.To)) + for _, rcpt := range msg.To { + toList = append(toList, rcpt.String()) + } + writeHeader("To", strings.Join(toList, ", ")) + writeHeader("Subject", mime.QEncoding.Encode("utf-8", msg.Subject)) + writeHeader("Date", time.Now().Format(time.RFC1123Z)) + writeHeader("MIME-Version", "1.0") + + switch { + case msg.TextBody != "" && msg.HTMLBody != "": + boundary := fmt.Sprintf("alt-%d", time.Now().UnixNano()) + writeHeader("Content-Type", fmt.Sprintf(`multipart/alternative; boundary="%s"`, boundary)) + buf.WriteString("\r\n") + writeMIMEPart(&buf, boundary, "text/plain; charset=utf-8", msg.TextBody) + writeMIMEPart(&buf, boundary, "text/html; charset=utf-8", msg.HTMLBody) + buf.WriteString("--") + buf.WriteString(boundary) + buf.WriteString("--\r\n") + case msg.HTMLBody != "": + writeHeader("Content-Type", `text/html; charset=utf-8`) + writeHeader("Content-Transfer-Encoding", "quoted-printable") + buf.WriteString("\r\n") + if err := writeQuotedPrintable(&buf, msg.HTMLBody); err != nil { + return nil, fmt.Errorf("email: encode html body: %w", err) + } + default: + writeHeader("Content-Type", `text/plain; charset=utf-8`) + writeHeader("Content-Transfer-Encoding", "quoted-printable") + buf.WriteString("\r\n") + if err := writeQuotedPrintable(&buf, msg.TextBody); err != nil { + return nil, fmt.Errorf("email: encode text body: %w", err) + } + } + return buf.Bytes(), nil +} + +func writeMIMEPart(buf *bytes.Buffer, boundary, contentType, body string) { + buf.WriteString("--") + buf.WriteString(boundary) + buf.WriteString("\r\n") + buf.WriteString("Content-Type: ") + buf.WriteString(contentType) + buf.WriteString("\r\n") + buf.WriteString("Content-Transfer-Encoding: quoted-printable\r\n\r\n") + _ = writeQuotedPrintable(buf, body) + buf.WriteString("\r\n") +} + +func writeQuotedPrintable(buf *bytes.Buffer, body string) error { + w := quotedprintable.NewWriter(buf) + if _, err := w.Write([]byte(strings.ReplaceAll(body, "\n", "\r\n"))); err != nil { + return err + } + return w.Close() +} + +// EmailTemplate is a named mail template, typically selected by a flow hook. +type EmailTemplate struct { + Subject string `json:"subject"` + Text string `json:"text,omitempty"` + HTML string `json:"html,omitempty"` +} + +func (t EmailTemplate) validate(name string) error { + if strings.TrimSpace(t.Subject) == "" { + return fmt.Errorf("email template %q: missing subject", name) + } + if strings.TrimSpace(t.Text) == "" && strings.TrimSpace(t.HTML) == "" { + return fmt.Errorf("email template %q: missing text/html body", name) + } + return nil +} + +// EmailTemplateStore resolves a template by name. +type EmailTemplateStore interface { + Get(name string) (EmailTemplate, error) +} + +type emailTemplateCatalog struct { + templates map[string]EmailTemplate +} + +func newEmailTemplateCatalog(seed map[string]EmailTemplate) *emailTemplateCatalog { + out := &emailTemplateCatalog{templates: make(map[string]EmailTemplate, len(seed))} + for name, tmpl := range seed { + out.templates[name] = tmpl + } + return out +} + +func (c *emailTemplateCatalog) Get(name string) (EmailTemplate, error) { + tmpl, ok := c.templates[name] + if !ok { + return EmailTemplate{}, fmt.Errorf("unknown template %q", name) + } + return tmpl, nil +} + +//go:embed email_templates/*.json +var embeddedEmailTemplateFS embed.FS + +// LoadEmailTemplates returns the embedded templates overlaid by any JSON files in +// dir (same precedence model as flow definitions: on-disk wins). +func LoadEmailTemplates(dir string) (EmailTemplateStore, error) { + templates := map[string]EmailTemplate{} + + matches, err := fs.Glob(embeddedEmailTemplateFS, "email_templates/*.json") + if err != nil { + return nil, fmt.Errorf("list embedded email templates: %w", err) + } + for _, match := range matches { + data, err := embeddedEmailTemplateFS.ReadFile(match) + if err != nil { + return nil, fmt.Errorf("read embedded email template %s: %w", match, err) + } + name := strings.TrimSuffix(filepath.Base(match), ".json") + tmpl, err := parseEmailTemplate(name, data) + if err != nil { + return nil, err + } + templates[name] = tmpl + } + + if dir != "" { + if _, err := os.Stat(dir); err == nil { + files, err := filepath.Glob(filepath.Join(dir, "*.json")) + if err != nil { + return nil, fmt.Errorf("list email templates in %s: %w", dir, err) + } + for _, file := range files { + data, err := os.ReadFile(file) + if err != nil { + return nil, fmt.Errorf("read email template %s: %w", file, err) + } + name := strings.TrimSuffix(filepath.Base(file), ".json") + tmpl, err := parseEmailTemplate(name, data) + if err != nil { + return nil, err + } + templates[name] = tmpl + } + } else if !os.IsNotExist(err) { + return nil, fmt.Errorf("stat email template dir %s: %w", dir, err) + } + } + + return newEmailTemplateCatalog(templates), nil +} + +func parseEmailTemplate(name string, data []byte) (EmailTemplate, error) { + var tmpl EmailTemplate + if err := json.Unmarshal(data, &tmpl); err != nil { + return EmailTemplate{}, fmt.Errorf("parse email template %q: %w", name, err) + } + if err := tmpl.validate(name); err != nil { + return EmailTemplate{}, err + } + return tmpl, nil +} + +type emailHookParams struct { + Template string `json:"template"` + To string `json:"to,omitempty"` + From string `json:"from,omitempty"` + Subject string `json:"subject,omitempty"` +} + +type flowTemplateData struct { + Order *OrderGrain + OrderID string + Step string + Phase flow.Phase + Error string + Vars map[string]any + Fulfillment *Fulfillment + ShippingAddress any + BillingAddress any +} + +// RegisterEmailHook registers the "send_email" flow hook. It reads the current +// order for template data, renders the named template, and delivers it through +// the configured sender. Hook params: +// +// { +// "template": "fulfillment-shipped", +// "to": "{{ .Order.CustomerEmail }}", +// "from": "Warehouse " +// } +func RegisterEmailHook(reg *flow.Registry, app Applier, sender EmailSender, templates EmailTemplateStore) { + reg.HookWithMeta("send_email", func(ctx context.Context, st *flow.State, info flow.HookInfo, params json.RawMessage) error { + if sender == nil { + return fmt.Errorf("send_email: no sender configured") + } + if templates == nil { + return fmt.Errorf("send_email: no template store configured") + } + var p emailHookParams + if len(params) > 0 { + if err := json.Unmarshal(params, &p); err != nil { + return fmt.Errorf("send_email: bad params: %w", err) + } + } + if p.Template == "" { + return fmt.Errorf("send_email: missing template") + } + o, err := app.Get(ctx, st.ID) + if err != nil { + return err + } + tmpl, err := templates.Get(p.Template) + if err != nil { + return fmt.Errorf("send_email: %w", err) + } + data := buildFlowTemplateData(o, st, info) + + subjectSource := tmpl.Subject + if p.Subject != "" { + subjectSource = p.Subject + } + subject, err := renderTextTemplate("email-subject", subjectSource, data) + if err != nil { + return fmt.Errorf("send_email: render subject: %w", err) + } + textBody, err := renderOptionalTextTemplate("email-text", tmpl.Text, data) + if err != nil { + return fmt.Errorf("send_email: render text body: %w", err) + } + htmlBody, err := renderOptionalHTMLTemplate("email-html", tmpl.HTML, data) + if err != nil { + return fmt.Errorf("send_email: render html body: %w", err) + } + + toSource := p.To + if strings.TrimSpace(toSource) == "" { + toSource = o.CustomerEmail + } + toRendered, err := renderTextTemplate("email-to", toSource, data) + if err != nil { + return fmt.Errorf("send_email: render recipient: %w", err) + } + recipients, err := mail.ParseAddressList(toRendered) + if err != nil { + return fmt.Errorf("send_email: parse recipient %q: %w", toRendered, err) + } + to := make([]mail.Address, 0, len(recipients)) + for _, rcpt := range recipients { + to = append(to, *rcpt) + } + + from := sender.DefaultFrom() + if strings.TrimSpace(p.From) != "" { + fromRendered, err := renderTextTemplate("email-from", p.From, data) + if err != nil { + return fmt.Errorf("send_email: render sender: %w", err) + } + parsed, err := mail.ParseAddress(strings.TrimSpace(fromRendered)) + if err != nil { + return fmt.Errorf("send_email: parse sender %q: %w", fromRendered, err) + } + from = *parsed + } + + return sender.Send(ctx, EmailMessage{ + From: from, + To: to, + Subject: subject, + TextBody: textBody, + HTMLBody: htmlBody, + }) + }, flow.CapabilityMeta{ + Description: "Render a named email template and send it through the configured sender.", + ExampleParams: json.RawMessage(`{"template":"fulfillment-shipped","to":"{{ .Order.CustomerEmail }}","from":"Store "}`), + }) +} + +func buildFlowTemplateData(o *OrderGrain, st *flow.State, info flow.HookInfo) flowTemplateData { + errStr := "" + if info.Err != nil { + errStr = info.Err.Error() + } + var lastFulfillment *Fulfillment + if n := len(o.Fulfillments); n > 0 { + lastFulfillment = &o.Fulfillments[n-1] + } + return flowTemplateData{ + Order: o, + OrderID: OrderId(st.ID).String(), + Step: info.Step, + Phase: info.Phase, + Error: errStr, + Vars: st.Vars, + Fulfillment: lastFulfillment, + ShippingAddress: decodeEmailTemplateJSON(o.ShippingAddress), + BillingAddress: decodeEmailTemplateJSON(o.BillingAddress), + } +} + +func decodeEmailTemplateJSON(raw json.RawMessage) any { + if len(raw) == 0 { + return nil + } + var v any + if err := json.Unmarshal(raw, &v); err != nil { + return string(raw) + } + return v +} + +func renderTextTemplate(name, src string, data any) (string, error) { + tmpl, err := texttmpl.New(name).Option("missingkey=error").Parse(src) + if err != nil { + return "", err + } + var buf bytes.Buffer + if err := tmpl.Execute(&buf, data); err != nil { + return "", err + } + return strings.TrimSpace(buf.String()), nil +} + +func renderOptionalTextTemplate(name, src string, data any) (string, error) { + if strings.TrimSpace(src) == "" { + return "", nil + } + return renderTextTemplate(name, src, data) +} + +func renderOptionalHTMLTemplate(name, src string, data any) (string, error) { + if strings.TrimSpace(src) == "" { + return "", nil + } + tmpl, err := htmltmpl.New(name).Option("missingkey=error").Parse(src) + if err != nil { + return "", err + } + var buf bytes.Buffer + if err := tmpl.Execute(&buf, data); err != nil { + return "", err + } + return strings.TrimSpace(buf.String()), nil +} diff --git a/pkg/order/email_templates/fulfillment-shipped.json b/pkg/order/email_templates/fulfillment-shipped.json new file mode 100644 index 0000000..72e3c2b --- /dev/null +++ b/pkg/order/email_templates/fulfillment-shipped.json @@ -0,0 +1,5 @@ +{ + "subject": "Your order {{ .Order.OrderReference }} has shipped", + "text": "Hi {{ if .Order.CustomerName }}{{ .Order.CustomerName }}{{ else }}there{{ end }},\n\nYour order {{ .Order.OrderReference }} has shipped.\n{{ with .Fulfillment }}Carrier: {{ .Carrier }}\nTracking number: {{ .TrackingNumber }}\n{{ if .TrackingURI }}Track your shipment: {{ .TrackingURI }}\n{{ end }}{{ end }}\nOrder id: {{ .OrderID }}\n", + "html": "

Hi {{ if .Order.CustomerName }}{{ .Order.CustomerName }}{{ else }}there{{ end }},

Your order {{ .Order.OrderReference }} has shipped.

{{ with .Fulfillment }}{{ end }}

Order id: {{ .OrderID }}

" +} diff --git a/pkg/order/emit_created.go b/pkg/order/emit_created.go index b7e89ae..a522a71 100644 --- a/pkg/order/emit_created.go +++ b/pkg/order/emit_created.go @@ -21,7 +21,7 @@ import ( // retries delivery on its own. With a nil publisher (dev without AMQP) it is a // no-op. Attach it to the "after" hooks of the final paid step, e.g. capture. func RegisterOrderCreatedEmit(reg *flow.Registry, app Applier, pub Publisher, exchange, source string) { - reg.Hook("emit_order_created", func(ctx context.Context, st *flow.State, _ flow.HookInfo, _ json.RawMessage) error { + reg.HookWithMeta("emit_order_created", func(ctx context.Context, st *flow.State, _ flow.HookInfo, _ json.RawMessage) error { if pub == nil { return nil // no broker configured (dev) } @@ -51,19 +51,21 @@ func RegisterOrderCreatedEmit(reg *flow.Registry, app Applier, pub Publisher, ex return err } return pub.Publish(exchange, string(event.OrderCreated), body) + }, flow.CapabilityMeta{ + Description: "Emit the order.created domain event after a paid order is finalized.", }) } // buildOrderCreated projects the grain onto the exported order.created contract. -// Lines carry only SKU + quantity (what reactors need); inventory commits at the -// order's Country location. +// Lines carry SKU, quantity, the inventory commit location, and the drop-ship +// flag so the inventory reactor can skip the stock decrement for supplier-fulfilled items. func buildOrderCreated(o *OrderGrain) contract.Created { lines := make([]contract.Line, 0, len(o.Lines)) for _, l := range o.Lines { if l.Sku == "" || l.Quantity <= 0 { continue } - lines = append(lines, contract.Line{SKU: l.Sku, Quantity: l.Quantity, Location: l.Location}) + lines = append(lines, contract.Line{SKU: l.Sku, Quantity: l.Quantity, Location: l.Location, DropShip: l.DropShip}) } return contract.Created{ OrderID: OrderId(o.Id).String(), diff --git a/pkg/order/fulfillment_webhook.go b/pkg/order/fulfillment_webhook.go new file mode 100644 index 0000000..cb86793 --- /dev/null +++ b/pkg/order/fulfillment_webhook.go @@ -0,0 +1,109 @@ +package order + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "git.k6n.net/mats/go-cart-actor/pkg/flow" +) + +type fulfillmentWebhookParams struct { + URL string `json:"url"` + Method string `json:"method,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + ContentType string `json:"contentType,omitempty"` +} + +// RegisterFulfillmentWebhookHook registers the "fulfillment_webhook" flow hook. +// It posts the current order + latest fulfillment to an external integration, +// with templated URL/headers so flows can target dropship or ERP adapters +// without custom Go code. Params: +// +// { +// "url": "https://example.test/hooks/{{ .OrderID }}", +// "method": "POST", +// "headers": { +// "X-Integration": "dropship", +// "Authorization": "Bearer {{ index .Vars \"token\" }}" +// } +// } +func RegisterFulfillmentWebhookHook(reg *flow.Registry, app Applier, client *http.Client) { + if client == nil { + client = &http.Client{Timeout: 10 * time.Second} + } + reg.HookWithMeta("fulfillment_webhook", func(ctx context.Context, st *flow.State, info flow.HookInfo, params json.RawMessage) error { + var p fulfillmentWebhookParams + if err := json.Unmarshal(params, &p); err != nil { + return fmt.Errorf("fulfillment_webhook: bad params: %w", err) + } + if strings.TrimSpace(p.URL) == "" { + return fmt.Errorf("fulfillment_webhook: missing url") + } + method := strings.ToUpper(strings.TrimSpace(p.Method)) + if method == "" { + method = http.MethodPost + } + if p.ContentType == "" { + p.ContentType = "application/json" + } + + o, err := app.Get(ctx, st.ID) + if err != nil { + return err + } + data := buildFlowTemplateData(o, st, info) + if data.Fulfillment == nil { + return fmt.Errorf("fulfillment_webhook: order %s has no fulfillment yet", data.OrderID) + } + + url, err := renderTextTemplate("fulfillment-webhook-url", p.URL, data) + if err != nil { + return fmt.Errorf("fulfillment_webhook: render url: %w", err) + } + + body, err := json.Marshal(map[string]any{ + "orderId": data.OrderID, + "step": data.Step, + "phase": data.Phase, + "error": data.Error, + "order": data.Order, + "fulfillment": data.Fulfillment, + "billingAddress": data.BillingAddress, + "shippingAddress": data.ShippingAddress, + "vars": data.Vars, + }) + if err != nil { + return fmt.Errorf("fulfillment_webhook: encode body: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, method, url, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("fulfillment_webhook: build request: %w", err) + } + req.Header.Set("Content-Type", p.ContentType) + for key, raw := range p.Headers { + value, err := renderTextTemplate("fulfillment-webhook-header-"+key, raw, data) + if err != nil { + return fmt.Errorf("fulfillment_webhook: render header %s: %w", key, err) + } + req.Header.Set(key, value) + } + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("fulfillment_webhook: post %s: %w", url, err) + } + defer resp.Body.Close() + if resp.StatusCode >= 300 { + return fmt.Errorf("fulfillment_webhook: %s returned %d", url, resp.StatusCode) + } + return nil + }, flow.CapabilityMeta{ + Description: "POST the order and latest fulfillment to an external dropship or ERP endpoint.", + ExampleParams: json.RawMessage(`{"url":"https://dropship.example/hooks/{{ .OrderID }}","method":"POST","headers":{"X-Integration":"dropship {{ .Order.OrderReference }}","Authorization":"Bearer {{ index .Vars \"token\" }}"}}`), + }) +} diff --git a/pkg/order/hooks.go b/pkg/order/hooks.go index a39b435..17e7dbf 100644 --- a/pkg/order/hooks.go +++ b/pkg/order/hooks.go @@ -26,7 +26,7 @@ type Publisher interface { // still registered (so the editor lists it) but errors at run time — and since // hook errors never abort a flow, that failure is logged, not fatal. func RegisterEmitHook(reg *flow.Registry, pub Publisher) { - reg.Hook("amqp_emit", func(_ context.Context, st *flow.State, info flow.HookInfo, params json.RawMessage) error { + reg.HookWithMeta("amqp_emit", func(_ context.Context, st *flow.State, info flow.HookInfo, params json.RawMessage) error { if pub == nil { return fmt.Errorf("amqp_emit: no publisher configured") } @@ -54,5 +54,8 @@ func RegisterEmitHook(reg *flow.Registry, pub Publisher) { "vars": st.Vars, }) return pub.Publish(p.Exchange, p.RoutingKey, body) + }, flow.CapabilityMeta{ + Description: "Publish the flow event payload to RabbitMQ.", + ExampleParams: json.RawMessage(`{"exchange":"order","routingKey":"order-events"}`), }) } diff --git a/pkg/order/inventory_reservation.go b/pkg/order/inventory_reservation.go new file mode 100644 index 0000000..6774965 --- /dev/null +++ b/pkg/order/inventory_reservation.go @@ -0,0 +1,179 @@ +package order + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "git.k6n.net/mats/go-cart-actor/pkg/flow" +) + +type InventoryReservationLine struct { + SKU string `json:"sku"` + LocationID string `json:"locationId"` + Quantity uint32 `json:"quantity"` +} + +type InventoryReservationRequest struct { + HolderID string `json:"holderId"` + TTL time.Duration `json:"-"` + Lines []InventoryReservationLine `json:"lines"` +} + +type InventoryReservationService interface { + Reserve(ctx context.Context, req InventoryReservationRequest) error + Release(ctx context.Context, req InventoryReservationRequest) error +} + +type HTTPInventoryReservationService struct { + baseURL string + client *http.Client +} + +func NewHTTPInventoryReservationService(baseURL string, client *http.Client) (*HTTPInventoryReservationService, error) { + baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/") + if baseURL == "" { + return nil, fmt.Errorf("inventory reservation service: missing base url") + } + if client == nil { + client = &http.Client{Timeout: 10 * time.Second} + } + return &HTTPInventoryReservationService{baseURL: baseURL, client: client}, nil +} + +func (s *HTTPInventoryReservationService) Reserve(ctx context.Context, req InventoryReservationRequest) error { + return s.post(ctx, "/reservations", req) +} + +func (s *HTTPInventoryReservationService) Release(ctx context.Context, req InventoryReservationRequest) error { + return s.post(ctx, "/reservations/release", req) +} + +func (s *HTTPInventoryReservationService) post(ctx context.Context, path string, req InventoryReservationRequest) error { + body := map[string]any{ + "holderId": req.HolderID, + "lines": req.Lines, + } + if req.TTL > 0 { + body["ttlSeconds"] = int64(req.TTL / time.Second) + } + raw, err := json.Marshal(body) + if err != nil { + return err + } + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, s.baseURL+path, bytes.NewReader(raw)) + if err != nil { + return err + } + httpReq.Header.Set("Content-Type", "application/json") + resp, err := s.client.Do(httpReq) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 300 { + return fmt.Errorf("inventory reservation service: %s returned %d", path, resp.StatusCode) + } + return nil +} + +type reserveInventoryParams struct { + HolderID string `json:"holderId,omitempty"` + TTLSeconds int64 `json:"ttlSeconds,omitempty"` + Lines []struct { + Reference string `json:"reference"` + Quantity int32 `json:"quantity"` + LocationID string `json:"locationId,omitempty"` + } `json:"lines"` +} + +func RegisterInventoryReservationActions(reg *flow.Registry, app Applier, svc InventoryReservationService) { + reg.ActionWithMeta("reserve_inventory", func(ctx context.Context, st *flow.State, params json.RawMessage) error { + if svc == nil { + return fmt.Errorf("reserve_inventory: no reservation service configured") + } + req, err := buildInventoryReservationRequest(ctx, app, st, params, "reserve_inventory") + if err != nil { + return err + } + return svc.Reserve(ctx, req) + }, flow.CapabilityMeta{ + Description: "Reserve inventory for specific order lines through the inventory service.", + ExampleParams: json.RawMessage(`{"ttlSeconds":900,"lines":[{"reference":"l1","quantity":1,"locationId":"se"}]}`), + }) + + reg.ActionWithMeta("release_inventory", func(ctx context.Context, st *flow.State, params json.RawMessage) error { + if svc == nil { + return fmt.Errorf("release_inventory: no reservation service configured") + } + req, err := buildInventoryReservationRequest(ctx, app, st, params, "release_inventory") + if err != nil { + return err + } + return svc.Release(ctx, req) + }, flow.CapabilityMeta{ + Description: "Release a prior inventory reservation for specific order lines.", + ExampleParams: json.RawMessage(`{"lines":[{"reference":"l1","quantity":1,"locationId":"se"}]}`), + }) +} + +func buildInventoryReservationRequest(ctx context.Context, app Applier, st *flow.State, params json.RawMessage, action string) (InventoryReservationRequest, error) { + var p reserveInventoryParams + if err := json.Unmarshal(params, &p); err != nil { + return InventoryReservationRequest{}, fmt.Errorf("%s: bad params: %w", action, err) + } + if len(p.Lines) == 0 { + return InventoryReservationRequest{}, fmt.Errorf("%s: missing lines", action) + } + o, err := app.Get(ctx, st.ID) + if err != nil { + return InventoryReservationRequest{}, err + } + holderID := p.HolderID + if holderID == "" { + holderID = "order:" + OrderId(st.ID).String() + } + req := InventoryReservationRequest{HolderID: holderID} + if p.TTLSeconds > 0 { + req.TTL = time.Duration(p.TTLSeconds) * time.Second + } + for _, lineReq := range p.Lines { + if lineReq.Reference == "" { + return InventoryReservationRequest{}, fmt.Errorf("%s: line missing reference", action) + } + if lineReq.Quantity <= 0 { + return InventoryReservationRequest{}, fmt.Errorf("%s: line %q has invalid quantity %d", action, lineReq.Reference, lineReq.Quantity) + } + line := o.findLine(lineReq.Reference) + if line == nil { + return InventoryReservationRequest{}, fmt.Errorf("%s: unknown line %q", action, lineReq.Reference) + } + if line.Sku == "" { + return InventoryReservationRequest{}, fmt.Errorf("%s: line %q has no sku", action, lineReq.Reference) + } + remaining := line.Quantity - line.Fulfilled + if int(lineReq.Quantity) > remaining { + return InventoryReservationRequest{}, fmt.Errorf("%s: line %q quantity %d exceeds remaining fulfillable %d", action, lineReq.Reference, lineReq.Quantity, remaining) + } + locationID := strings.TrimSpace(lineReq.LocationID) + if locationID == "" { + locationID = strings.TrimSpace(line.Location) + } + if locationID == "" { + locationID = strings.TrimSpace(o.Country) + } + if locationID == "" { + return InventoryReservationRequest{}, fmt.Errorf("%s: line %q has no location", action, lineReq.Reference) + } + req.Lines = append(req.Lines, InventoryReservationLine{ + SKU: line.Sku, + LocationID: locationID, + Quantity: uint32(lineReq.Quantity), + }) + } + return req, nil +} diff --git a/pkg/order/mutations.go b/pkg/order/mutations.go index af221d1..90fa126 100644 --- a/pkg/order/mutations.go +++ b/pkg/order/mutations.go @@ -60,6 +60,7 @@ func HandlePlaceOrder(o *OrderGrain, m *messages.PlaceOrder) error { TotalAmount: money.Cents(l.GetTotalAmount()), TotalTax: money.Cents(l.GetTotalTax()), Location: l.GetLocation(), + DropShip: l.GetDropShip(), }) } o.Status = StatusPending diff --git a/pkg/order/order-grain.go b/pkg/order/order-grain.go index 4c62552..1c90a6a 100644 --- a/pkg/order/order-grain.go +++ b/pkg/order/order-grain.go @@ -66,6 +66,9 @@ type Line struct { // Location is the inventory location / store id to commit against (empty = // order country). Carried through to the order.created event for commit. Location string `json:"location,omitempty"` + // DropShip marks items fulfilled by the supplier; the inventory reactor + // must skip the stock decrement. Set from the cart item at order creation. + DropShip bool `json:"drop_ship,omitempty"` // Fulfilled tracks how many units of this line have shipped. Fulfilled int `json:"fulfilled"` } diff --git a/proto/order/order.pb.go b/proto/order/order.pb.go index 04061bd..f0869c2 100644 --- a/proto/order/order.pb.go +++ b/proto/order/order.pb.go @@ -32,6 +32,7 @@ type OrderLine struct { TotalAmount int64 `protobuf:"varint,7,opt,name=total_amount,json=totalAmount,proto3" json:"total_amount,omitempty"` TotalTax int64 `protobuf:"varint,8,opt,name=total_tax,json=totalTax,proto3" json:"total_tax,omitempty"` Location string `protobuf:"bytes,9,opt,name=location,proto3" json:"location,omitempty"` // inventory location / store id for commit; empty = order country + DropShip bool `protobuf:"varint,10,opt,name=drop_ship,json=dropShip,proto3" json:"drop_ship,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -129,6 +130,13 @@ func (x *OrderLine) GetLocation() string { return "" } +func (x *OrderLine) GetDropShip() bool { + if x != nil { + return x.DropShip + } + return false +} + // PlaceOrder creates the order (only legal when the order does not yet exist). type PlaceOrder struct { state protoimpl.MessageState `protogen:"open.v1"`