refactor

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)

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")

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 {

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()

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"`
 }
 

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,
 		})
 	}

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

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

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})
+}

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)

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()

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})
+}

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)),
 	}

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}`),
+	})
+}

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},
+		},
+	}
+}

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

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

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 {