fix: address CodeRabbit review comments for webhook migration

- Fix silent proto deserialization corruption: use two separate
  kafka.ProtoConsumer instances (one per topic) with the correct
  typed msgFactory, preventing PaymentFailedEvent bytes from being
  decoded as PaymentCapturedEvent

- Fix missing tenant context: change route to POST /webhooks/stripe/{tenantID}
  and extract tenant via r.PathValue("tenantID") instead of relying on
  middleware that was never wired; inject into ctx via tenant.WithTenant

- Fix nil OutboxPublisher not validated: panic in NewWebhookHandler
  when OutboxPublisher is nil, consistent with ClientFactory guard

- Fix gRPC listener resource leak: add listenerClosed flag with deferred
  close so listener is not abandoned on startup failures after bind

- Fix serverErrors channel buffer too small: increase from 2 to 3
  to accommodate gRPC + HTTP + payment event consumer goroutines

- Use static sentinel errors (ErrUnexpectedCapturedMessageType,
  ErrUnexpectedFailedMessageType) for type assertion failures in
  consumers to satisfy err113 linter

- Update webhook_handler_test.go to use r.SetPathValue("tenantID")
  instead of tenant.WithTenant context injection, matching new handler
  behaviour; remove unused tenant import

- Update payment-order/cmd/main.go to use new Start(capturedTopic,
  failedTopic string) signature for PaymentEventConsumer

Author: bjcoombs

services/financial-gateway/adapters/http/webhook_handler.go

@@ -69,11 +69,14 @@ type WebhookHandlerConfig struct {
 }
 
 // NewWebhookHandler creates a new WebhookHandler.
-// Panics if ClientFactory is nil to fail fast during initialization.
+// Panics if ClientFactory or OutboxPublisher is nil to fail fast during initialization.
 func NewWebhookHandler(cfg WebhookHandlerConfig) *WebhookHandler {
 	if cfg.ClientFactory == nil {
 		panic(ErrNilClientFactory.Error())
 	}
+	if cfg.OutboxPublisher == nil {
+		panic(ErrNilOutboxPublisher.Error())
+	}
 	logger := cfg.Logger
 	if logger == nil {
 		logger = slog.Default()
@@ -95,9 +98,13 @@ type webhookResponse struct {
 
 // HandleStripeWebhook processes an incoming Stripe webhook.
 //
+// The route must be registered as POST /webhooks/stripe/{tenantID} so that
+// r.PathValue("tenantID") returns the tenant identifier. Stripe is configured
+// to call the tenant-specific URL (e.g. /webhooks/stripe/acme-corp).
+//
 // Flow:
 //  1. Validate HTTP method and read body
-//  2. Extract tenant context
+//  2. Extract tenant ID from URL path and inject into context
 //  3. Resolve per-tenant Stripe client (contains webhook secret)
 //  4. Validate Stripe-Signature using the tenant-specific secret
 //  5. Map the Stripe event to a domain event (PaymentCapturedEvent or PaymentFailedEvent)
@@ -124,12 +131,16 @@ func (h *WebhookHandler) HandleStripeWebhook(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
-	tenantID, ok := tenant.FromContext(ctx)
-	if !ok {
-		h.logger.Warn("missing tenant context for stripe webhook")
+	// Extract tenant ID from the URL path segment {tenantID}.
+	// The route must be registered as: POST /webhooks/stripe/{tenantID}
+	rawTenantID := r.PathValue("tenantID")
+	if rawTenantID == "" {
+		h.logger.Warn("missing tenant ID in stripe webhook URL path")
 		h.writeError(w, http.StatusBadRequest, ErrMissingTenantContext.Error())
 		return
 	}
+	tenantID := tenant.TenantID(rawTenantID)
+	ctx = tenant.WithTenant(ctx, tenantID)
 
 	client, err := h.clientFactory.NewClient(ctx)
 	if err != nil {

services/financial-gateway/adapters/http/webhook_handler_test.go

@@ -20,7 +20,6 @@ import (
 	fghttp "github.com/meridianhub/meridian/services/financial-gateway/adapters/http"
 	stripeadapter "github.com/meridianhub/meridian/services/financial-gateway/adapters/stripe"
 	"github.com/meridianhub/meridian/shared/platform/events"
-	"github.com/meridianhub/meridian/shared/platform/tenant"
 )
 
 const testWebhookSecret = "whsec_test_webhook_secret_for_financial_gateway"
@@ -132,7 +131,8 @@ func TestWebhookHandler_MissingTenantContext(t *testing.T) {
 	payload := buildStripePayload(t, "evt_1", "payment_intent.succeeded", map[string]any{})
 	sig := signPayload(t, payload, testWebhookSecret)
 
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
+	// No {tenantID} path value set — simulates request without tenant in URL.
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/", bytes.NewReader(payload))
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()
@@ -145,9 +145,8 @@ func TestWebhookHandler_MissingSignature(t *testing.T) {
 	h := setupHandler(t, nil)
 
 	payload := buildStripePayload(t, "evt_2", "payment_intent.succeeded", map[string]any{})
-	ctx := tenant.WithTenant(context.Background(), "test-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/test-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "test-tenant")
 
 	rr := httptest.NewRecorder()
 	h.HandleStripeWebhook(rr, req)
@@ -161,9 +160,8 @@ func TestWebhookHandler_InvalidSignature(t *testing.T) {
 	payload := buildStripePayload(t, "evt_3", "payment_intent.succeeded", map[string]any{})
 	sig := signPayload(t, payload, "whsec_wrong_secret")
 
-	ctx := tenant.WithTenant(context.Background(), "test-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/test-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "test-tenant")
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()
@@ -180,9 +178,8 @@ func TestWebhookHandler_UnsupportedEvent_Returns200(t *testing.T) {
 	})
 	sig := signPayload(t, payload, testWebhookSecret)
 
-	ctx := tenant.WithTenant(context.Background(), "test-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/test-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "test-tenant")
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()
@@ -208,9 +205,8 @@ func TestWebhookHandler_PaymentCaptured_PublishesToOutbox(t *testing.T) {
 	})
 	sig := signPayload(t, payload, testWebhookSecret)
 
-	ctx := tenant.WithTenant(context.Background(), "test-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/test-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "test-tenant")
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()
@@ -247,9 +243,8 @@ func TestWebhookHandler_PaymentFailed_PublishesToOutbox(t *testing.T) {
 	})
 	sig := signPayload(t, payload, testWebhookSecret)
 
-	ctx := tenant.WithTenant(context.Background(), "test-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/test-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "test-tenant")
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()
@@ -281,9 +276,8 @@ func TestWebhookHandler_OutboxPublishFails_Returns500(t *testing.T) {
 	})
 	sig := signPayload(t, payload, testWebhookSecret)
 
-	ctx := tenant.WithTenant(context.Background(), "test-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/test-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "test-tenant")
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()
@@ -313,9 +307,8 @@ func TestWebhookHandler_TenantNotFound_Returns500(t *testing.T) {
 	payload := buildStripePayload(t, "evt_notfound_1", "payment_intent.succeeded", map[string]any{})
 	sig := signPayload(t, payload, testWebhookSecret)
 
-	ctx := tenant.WithTenant(context.Background(), "unknown-tenant")
-	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
-	req = req.WithContext(ctx)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/unknown-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "unknown-tenant")
 	req.Header.Set("Stripe-Signature", sig)
 
 	rr := httptest.NewRecorder()

services/financial-gateway/cmd/main.go

@@ -144,6 +144,13 @@ func run(logger *slog.Logger) error {
 	if err != nil {
 		return fmt.Errorf("failed to listen on %s: %w", grpcAddress, err)
 	}
+	// Close the listener if any subsequent setup fails (before grpcServer.Serve takes ownership).
+	listenerClosed := false
+	defer func() {
+		if !listenerClosed {
+			_ = listener.Close()
+		}
+	}()
 
 	// --- Stripe webhook receiver setup ---
 
@@ -220,7 +227,10 @@ func run(logger *slog.Logger) error {
 
 	// Create HTTP mux and register the Stripe webhook endpoint.
 	mux := http.NewServeMux()
-	mux.HandleFunc("/webhooks/stripe", webhookHandler.HandleStripeWebhook)
+	// Tenant ID is embedded in the URL path so Stripe can be configured to call
+	// per-tenant endpoints (e.g. /webhooks/stripe/acme-corp). The handler
+	// extracts the tenant from r.PathValue("tenantID") and injects it into ctx.
+	mux.HandleFunc("POST /webhooks/stripe/{tenantID}", webhookHandler.HandleStripeWebhook)
 
 	httpAddress := fmt.Sprintf(":%s", cfg.HTTPPort)
 	httpServer := &http.Server{
@@ -232,6 +242,9 @@ func run(logger *slog.Logger) error {
 	serverErrors := make(chan error, 2)
 
 	// Start gRPC server in background.
+	// grpcServer.Serve takes ownership of the listener; mark it so the deferred
+	// close does not double-close on normal shutdown.
+	listenerClosed = true
 	go func() {
 		logger.Info("starting gRPC server", "address", grpcAddress)
 		if err := grpcServer.Serve(listener); err != nil {

services/payment-order/adapters/messaging/payment_event_consumer.go

@@ -18,6 +18,12 @@ import (
 // created without Kafka (use NewPaymentEventConsumerWithKafka for production).
 var ErrConsumerNotConfigured = errors.New("consumer not configured with Kafka — use NewPaymentEventConsumerWithKafka")
 
+// ErrUnexpectedCapturedMessageType is returned when the payment-captured consumer receives a message that is not *PaymentCapturedEvent.
+var ErrUnexpectedCapturedMessageType = errors.New("unexpected message type for payment-captured topic")
+
+// ErrUnexpectedFailedMessageType is returned when the payment-failed consumer receives a message that is not *PaymentFailedEvent.
+var ErrUnexpectedFailedMessageType = errors.New("unexpected message type for payment-failed topic")
+
 // PaymentOrderUpdater is the interface for updating payment order status.
 // Implemented by the payment-order gRPC service.
 type PaymentOrderUpdater interface {
@@ -30,10 +36,16 @@ type PaymentOrderUpdater interface {
 // This consumer subscribes to:
 //   - financial-gateway.payment-captured.v1 → marks payment order as SETTLED
 //   - financial-gateway.payment-failed.v1   → marks payment order as REJECTED
+//
+// Two separate ProtoConsumers are used so that each topic uses the correct
+// proto message factory for deserialization. A single consumer with a shared
+// msgFactory would cause PaymentFailedEvent bytes to be deserialized as
+// PaymentCapturedEvent, silently corrupting the event.
 type PaymentEventConsumer struct {
-	updater *kafka.ProtoConsumer
-	svc     PaymentOrderUpdater
-	logger  *slog.Logger
+	capturedConsumer *kafka.ProtoConsumer
+	failedConsumer   *kafka.ProtoConsumer
+	svc              PaymentOrderUpdater
+	logger           *slog.Logger
 }
 
 // NewPaymentEventConsumer creates a consumer that handles domain events from financial-gateway.
@@ -45,8 +57,8 @@ func NewPaymentEventConsumer(svc PaymentOrderUpdater) *PaymentEventConsumer {
 }
 
 // NewPaymentEventConsumerWithKafka creates a PaymentEventConsumer wired to real Kafka topics.
-// The returned consumer handles both payment-captured and payment-failed events by routing
-// on the event type within a single consumer group.
+// Two separate consumers are created — one per topic — so each uses the correct proto
+// message factory for deserialization.
 //
 // Call Start() to begin consuming and Stop()/Close() for graceful shutdown.
 func NewPaymentEventConsumerWithKafka(
@@ -63,66 +75,113 @@ func NewPaymentEventConsumerWithKafka(
 		logger: logger,
 	}
 
-	// The message factory returns PaymentCapturedEvent by default;
-	// we dispatch based on the proto type in the handler.
-	// Since the two event types share the same consumer group and topics,
-	// we use a wrapper that attempts to unmarshal as each type.
-	msgFactory := func() proto.Message {
-		// Return a placeholder; actual type detection happens in the handler
-		// by attempting unmarshal of each event type.
-		return &financialgatewayeventsv1.PaymentCapturedEvent{}
-	}
-
-	handler := func(ctx context.Context, key []byte, msg proto.Message) error {
-		return c.dispatch(ctx, key, msg)
+	// Consumer for payment-captured events uses PaymentCapturedEvent as the proto type.
+	capturedConfig := kafkaConfig
+	capturedConfig.GroupID = kafkaConfig.GroupID + "-captured"
+	capturedConfig.ClientID = kafkaConfig.ClientID + "-captured"
+	capturedConsumer, err := kafka.NewProtoConsumer(
+		capturedConfig,
+		func() proto.Message { return &financialgatewayeventsv1.PaymentCapturedEvent{} },
+		func(ctx context.Context, key []byte, msg proto.Message) error {
+			evt, ok := msg.(*financialgatewayeventsv1.PaymentCapturedEvent)
+			if !ok {
+				return fmt.Errorf("%w: %T", ErrUnexpectedCapturedMessageType, msg)
+			}
+			return c.HandlePaymentCapturedEvent(ctx, key, evt)
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create payment-captured kafka consumer: %w", err)
 	}
 
-	consumer, err := kafka.NewProtoConsumer(kafkaConfig, msgFactory, handler)
+	// Consumer for payment-failed events uses PaymentFailedEvent as the proto type.
+	failedConfig := kafkaConfig
+	failedConfig.GroupID = kafkaConfig.GroupID + "-failed"
+	failedConfig.ClientID = kafkaConfig.ClientID + "-failed"
+	failedConsumer, err := kafka.NewProtoConsumer(
+		failedConfig,
+		func() proto.Message { return &financialgatewayeventsv1.PaymentFailedEvent{} },
+		func(ctx context.Context, key []byte, msg proto.Message) error {
+			evt, ok := msg.(*financialgatewayeventsv1.PaymentFailedEvent)
+			if !ok {
+				return fmt.Errorf("%w: %T", ErrUnexpectedFailedMessageType, msg)
+			}
+			return c.HandlePaymentFailedEvent(ctx, key, evt)
+		},
+	)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create payment event kafka consumer: %w", err)
+		_ = capturedConsumer.Close()
+		return nil, fmt.Errorf("failed to create payment-failed kafka consumer: %w", err)
 	}
 
-	c.updater = consumer
+	c.capturedConsumer = capturedConsumer
+	c.failedConsumer = failedConsumer
 	return c, nil
 }
 
 // Start subscribes to the financial-gateway payment topics and begins consuming.
-// Blocks until Stop() is called or an unrecoverable error occurs.
-func (c *PaymentEventConsumer) Start(topicList []string) error {
-	if c.updater == nil {
+// Starts both consumers in goroutines and blocks until both have stopped.
+// Returns the first error encountered, or nil if both stop cleanly.
+func (c *PaymentEventConsumer) Start(capturedTopic, failedTopic string) error {
+	if c.capturedConsumer == nil || c.failedConsumer == nil {
 		return ErrConsumerNotConfigured
 	}
-	c.logger.Info("starting payment event consumer", "topics", topicList)
-	return c.updater.Subscribe(topicList)
+	c.logger.Info("starting payment event consumers",
+		"captured_topic", capturedTopic,
+		"failed_topic", failedTopic,
+	)
+
+	errs := make(chan error, 2)
+
+	go func() {
+		if err := c.capturedConsumer.Subscribe([]string{capturedTopic}); err != nil {
+			errs <- fmt.Errorf("payment-captured consumer: %w", err)
+		} else {
+			errs <- nil
+		}
+	}()
+
+	go func() {
+		if err := c.failedConsumer.Subscribe([]string{failedTopic}); err != nil {
+			errs <- fmt.Errorf("payment-failed consumer: %w", err)
+		} else {
+			errs <- nil
+		}
+	}()
+
+	// Wait for both consumers to stop.
+	var firstErr error
+	for range 2 {
+		if err := <-errs; err != nil && firstErr == nil {
+			firstErr = err
+		}
+	}
+	return firstErr
 }
 
-// Stop gracefully stops the Kafka consumer.
+// Stop gracefully stops both Kafka consumers.
 func (c *PaymentEventConsumer) Stop() {
-	if c.updater != nil {
-		c.updater.Stop()
+	if c.capturedConsumer != nil {
+		c.capturedConsumer.Stop()
+	}
+	if c.failedConsumer != nil {
+		c.failedConsumer.Stop()
 	}
 }
 
-// Close closes the consumer and releases resources.
+// Close closes both consumers and releases resources.
 func (c *PaymentEventConsumer) Close() error {
-	if c.updater != nil {
-		return c.updater.Close()
+	var capturedErr, failedErr error
+	if c.capturedConsumer != nil {
+		capturedErr = c.capturedConsumer.Close()
 	}
-	return nil
-}
-
-// dispatch routes an incoming Kafka message to the correct handler based on its proto type.
-// Since both captured and failed events flow through the same consumer, we attempt
-// to identify which event type was received.
-func (c *PaymentEventConsumer) dispatch(ctx context.Context, key []byte, msg proto.Message) error {
-	switch evt := msg.(type) {
-	case *financialgatewayeventsv1.PaymentCapturedEvent:
-		return c.HandlePaymentCapturedEvent(ctx, key, evt)
-	default:
-		c.logger.Warn("received unexpected message type in payment event consumer",
-			"type", fmt.Sprintf("%T", msg))
-		return nil
+	if c.failedConsumer != nil {
+		failedErr = c.failedConsumer.Close()
+	}
+	if capturedErr != nil {
+		return capturedErr
 	}
+	return failedErr
 }
 
 // HandlePaymentCapturedEvent processes a PaymentCapturedEvent by marking the