more flow

pkg/flow/engine.go

@@ -242,6 +242,11 @@ func (e *Engine) Validate(def *Definition) error {
 				if _, ok := e.reg.hook(h.Type); !ok {
 					return fmt.Errorf("flow %q step %q: unknown hook %q", def.Name, s.Name, h.Type)
 				}
+				if h.When != "" {
+					if _, ok := e.reg.predicate(h.When); !ok {
+						return fmt.Errorf("flow %q step %q: hook %q: unknown predicate %q", def.Name, s.Name, h.Type, h.When)
+					}
+				}
 			}
 		}
 	}
@@ -337,8 +342,26 @@ func (e *Engine) compensate(ctx context.Context, res *Result, executed []Step, s
 }
 
 // fireHooks runs every hook for a phase, logging (never propagating) failures.
+// Hooks with a When predicate are evaluated first; the hook is skipped when the
+// predicate returns false.
 func (e *Engine) fireHooks(ctx context.Context, st *State, refs []HookRef, info HookInfo) {
 	for _, ref := range refs {
+		if ref.When != "" {
+			pred, ok := e.reg.predicate(ref.When)
+			if !ok {
+				st.Logger.Error("flow hook skipped: unknown predicate", "hook", ref.Type, "predicate", ref.When, "step", info.Step)
+				continue
+			}
+			run, err := pred(ctx, st)
+			if err != nil {
+				st.Logger.Error("flow hook predicate error", "hook", ref.Type, "predicate", ref.When, "step", info.Step, "err", err)
+				continue
+			}
+			if !run {
+				st.Logger.Debug("flow hook skipped by predicate", "hook", ref.Type, "predicate", ref.When, "step", info.Step)
+				continue
+			}
+		}
 		h, ok := e.reg.hook(ref.Type)
 		if !ok {
 			st.Logger.Error("flow hook skipped: unknown type", "hook", ref.Type, "step", info.Step)

pkg/flow/flow.go

@@ -51,9 +51,11 @@ type Hooks struct {
 	OnError []HookRef `json:"onError,omitempty"`
 }
 
-// HookRef names a registered hook and its params.
+// HookRef names a registered hook and its params. When, if set, names a
+// registered predicate; the hook is skipped when it returns false.
 type HookRef struct {
 	Type   string          `json:"type"`
+	When   string          `json:"when,omitempty"`
 	Params json.RawMessage `json:"params,omitempty"`
 }
 

pkg/order/actions.go

@@ -63,7 +63,7 @@ const PlaceOrderVar = "placeOrder"
 // RegisterFlowActions registers the order lifecycle actions on reg, driving the
 // 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) {
+func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvider, prefCheckers ...EmailPreferenceChecker) {
 	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 {
@@ -243,7 +243,11 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid
 		ExampleParams: json.RawMessage(`{"carrier":"dropshipper","trackingNumber":"TRACK-123","trackingUri":"https://tracking.example/TRACK-123","lines":[{"reference":"l1","quantity":1}]}`),
 	})
 
-	registerPredicates(reg, app)
+	var checker EmailPreferenceChecker
+	if len(prefCheckers) > 0 {
+		checker = prefCheckers[0]
+	}
+	registerPredicates(reg, app, checker)
 }
 
 // registerPredicates adds the order gating predicates a step can reference via
@@ -251,7 +255,7 @@ func RegisterFlowActions(reg *flow.Registry, app Applier, provider PaymentProvid
 // at the moment the step is reached (e.g. only fire a receipt hook once the
 // order is captured). Predicates take no params (the engine's When is a bare
 // name), so each is a fixed, composable boolean.
-func registerPredicates(reg *flow.Registry, app Applier) {
+func registerPredicates(reg *flow.Registry, app Applier, prefChecker EmailPreferenceChecker) {
 	get := func(ctx context.Context, st *flow.State) (*OrderGrain, error) { return app.Get(ctx, st.ID) }
 
 	reg.PredicateWithMeta("has_customer_email", func(ctx context.Context, st *flow.State) (bool, error) {
@@ -289,4 +293,30 @@ func registerPredicates(reg *flow.Registry, app Applier) {
 		}
 		return o.CapturedAmount-o.RefundedAmount > 0, nil
 	}, flow.CapabilityMeta{Description: "Run only when captured amount remains to refund."})
+
+	// Email consent predicates — gate marketing or order steps when the customer
+	// has opted out in their profile email preferences.
+	if prefChecker != nil {
+		reg.PredicateWithMeta("has_marketing_consent", func(ctx context.Context, st *flow.State) (bool, error) {
+			o, err := get(ctx, st)
+			if err != nil {
+				return false, err
+			}
+			if o.CustomerEmail == "" {
+				return false, nil // no email address -> no consent
+			}
+			return prefChecker.Allowed(ctx, o.CustomerEmail, "marketing"), nil
+		}, flow.CapabilityMeta{Description: "Run only when the customer has consented to marketing emails."})
+
+		reg.PredicateWithMeta("has_order_email_consent", func(ctx context.Context, st *flow.State) (bool, error) {
+			o, err := get(ctx, st)
+			if err != nil {
+				return false, err
+			}
+			if o.CustomerEmail == "" {
+				return false, nil // no email address -> cannot receive order emails
+			}
+			return prefChecker.Allowed(ctx, o.CustomerEmail, "order"), nil
+		}, flow.CapabilityMeta{Description: "Run only when the customer has consented to order transactional emails."})
+	}
 }

pkg/order/flow_test.go

@@ -37,10 +37,15 @@ func newPool(t *testing.T) *actor.SimpleGrainPool[OrderGrain] {
 	return pool
 }
 
+// alwaysAllowChecker is a mock EmailPreferenceChecker that allows every email.
+type alwaysAllowChecker struct{}
+
+func (alwaysAllowChecker) Allowed(_ context.Context, _ string, _ string) bool { return true }
+
 func newFlowRegistry(pool Applier, provider PaymentProvider) *flow.Registry {
 	reg := flow.NewRegistry()
 	flow.RegisterBuiltinHooks(reg)
-	RegisterFlowActions(reg, pool, provider)
+	RegisterFlowActions(reg, pool, provider, alwaysAllowChecker{})
 	// place-and-pay references the emit_order_created hook; register it (nil
 	// publisher = no-op) so flow validation passes in tests too.
 	RegisterOrderCreatedEmit(reg, pool, nil, "order", "order")

pkg/order/flows/fulfill-order.json

@@ -8,7 +8,7 @@
       "hooks": {
         "after": [
           { "type": "log", "params": { "message": "fulfillment created" } },
-          { "type": "send_email", "when": "has_customer_email", "params": { "template": "fulfillment-shipped" } },
+          { "type": "send_email", "when": "has_order_email_consent", "params": { "template": "fulfillment-shipped" } },
           { "type": "fulfillment_webhook", "params": { "url": "", "method": "POST" }, "continueOnError": true }
         ]
       }

pkg/order/flows/place-and-pay.json

@@ -8,7 +8,7 @@
       "hooks": {
         "after": [
           { "type": "log", "params": { "message": "order placed" } },
-          { "type": "send_email", "when": "has_customer_email", "params": { "template": "order-confirmation" } }
+          { "type": "send_email", "when": "has_order_email_consent", "params": { "template": "order-confirmation" } }
         ]
       }
     },
@@ -27,7 +27,7 @@
       "hooks": {
         "after": [
           { "type": "log", "params": { "message": "payment captured" } },
-          { "type": "send_email", "when": "has_customer_email", "params": { "template": "payment-receipt" } },
+          { "type": "send_email", "when": "has_order_email_consent", "params": { "template": "payment-receipt" } },
           { "type": "emit_order_created" }
         ]
       }