fix: correct error semantics and add idempotency to financial-gateway webhook handler (#2039)

* fix: correct error semantics and add idempotency to webhook handler

- Define ErrMissingWebhookSecret and use it instead of ErrMissingTenantContext
  when webhook secret is absent; the two errors describe distinct failure modes
- Introduce ProcessedEventChecker interface and GormProcessedEventChecker
  implementation that queries event_outbox by causation_id
- Add EventChecker field to WebhookHandlerConfig; when set, duplicate Stripe
  events are skipped with a 200 rather than re-published; check failures fall
  through to publish to avoid silent event drops
- Set CausationID in PublishConfig so outbox entries carry the Stripe event ID,
  enabling the DB-backed idempotency check
- Add tests: correct error body on empty secret, duplicate skip, check-error
  fall-through

* fix: add tenant scoping and causation_id index to idempotency check

- GormProcessedEventChecker.IsProcessed now filters by tenant_id from
  context in addition to causation_id, consistent with the multi-tenant
  outbox pattern and preventing cross-tenant scope issues
- Add migration 20260329000001 with a partial index on event_outbox(causation_id)
  WHERE causation_id IS NOT NULL to avoid sequential scans on idempotency
  lookups as the outbox grows

* fix: use WithGormTenantTransaction for outbox idempotency check

Replace manual tenant_id WHERE clause with WithGormTenantTransaction
to properly set the search_path to the tenant schema, consistent with
the project's multi-tenant outbox pattern.

* fix: enforce atomic idempotency via unique causation_id index

- Make causation_id index UNIQUE to prevent concurrent duplicate inserts
- Handle duplicate key errors on the write path as already-processed
- Default EventChecker in constructor when DB is available

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

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

@@ -13,6 +13,7 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
+	"strings"
 
 	"github.com/google/uuid"
 	"google.golang.org/protobuf/proto"
@@ -21,6 +22,7 @@ import (
 
 	financialgatewayeventsv1 "github.com/meridianhub/meridian/api/proto/meridian/financial_gateway_events/v1"
 	stripeadapter "github.com/meridianhub/meridian/services/financial-gateway/adapters/stripe"
+	db "github.com/meridianhub/meridian/shared/platform/db"
 	"github.com/meridianhub/meridian/shared/platform/events"
 	"github.com/meridianhub/meridian/shared/platform/events/topics"
 	"github.com/meridianhub/meridian/shared/platform/tenant"
@@ -34,8 +36,43 @@ var (
 	ErrNilClientFactory     = errors.New("stripe client factory cannot be nil")
 	ErrNilOutboxPublisher   = errors.New("outbox publisher cannot be nil")
 	ErrMissingTenantContext = errors.New("missing tenant context for stripe webhook")
+	ErrMissingWebhookSecret = errors.New("missing webhook secret configured for tenant")
 )
 
+// ProcessedEventChecker checks whether a provider webhook event has already been processed.
+// Implementations should return (true, nil) if the event was already published to the outbox.
+type ProcessedEventChecker interface {
+	IsProcessed(ctx context.Context, providerEventID string) (bool, error)
+}
+
+// GormProcessedEventChecker implements ProcessedEventChecker by querying the event_outbox
+// table for an existing entry with a matching causation_id.
+type GormProcessedEventChecker struct {
+	db *gorm.DB
+}
+
+// NewGormProcessedEventChecker creates a GormProcessedEventChecker backed by the given DB.
+func NewGormProcessedEventChecker(db *gorm.DB) *GormProcessedEventChecker {
+	return &GormProcessedEventChecker{db: db}
+}
+
+// IsProcessed returns true if an outbox entry with causation_id matching providerEventID exists
+// for the tenant in ctx. The query runs inside a tenant-scoped transaction (WithGormTenantTransaction)
+// so the search_path targets the correct tenant schema, consistent with the project's multi-tenant
+// outbox pattern.
+func (c *GormProcessedEventChecker) IsProcessed(ctx context.Context, providerEventID string) (bool, error) {
+	var count int64
+	err := db.WithGormTenantTransaction(ctx, c.db, func(tx *gorm.DB) error {
+		return tx.Model(&events.EventOutbox{}).
+			Where("causation_id = ?", providerEventID).
+			Count(&count).Error
+	})
+	if err != nil {
+		return false, err
+	}
+	return count > 0, nil
+}
+
 // OutboxEventPublisher is the interface for publishing domain events to the transactional outbox.
 // This abstraction allows for test doubles without requiring a real *gorm.DB.
 type OutboxEventPublisher interface {
@@ -49,6 +86,7 @@ type WebhookHandler struct {
 	publisher     OutboxEventPublisher
 	db            *gorm.DB
 	logger        *slog.Logger
+	eventChecker  ProcessedEventChecker
 }
 
 // WebhookHandlerConfig contains configuration for creating a WebhookHandler.
@@ -66,6 +104,10 @@ type WebhookHandlerConfig struct {
 
 	// Logger is the structured logger. Defaults to slog.Default() if nil.
 	Logger *slog.Logger
+
+	// EventChecker optionally checks whether a provider event has already been processed.
+	// If nil, idempotency checking is skipped. For production, use NewGormProcessedEventChecker.
+	EventChecker ProcessedEventChecker
 }
 
 // NewWebhookHandler creates a new WebhookHandler.
@@ -81,11 +123,16 @@ func NewWebhookHandler(cfg WebhookHandlerConfig) *WebhookHandler {
 	if logger == nil {
 		logger = slog.Default()
 	}
+	eventChecker := cfg.EventChecker
+	if eventChecker == nil && cfg.DB != nil {
+		eventChecker = NewGormProcessedEventChecker(cfg.DB)
+	}
 	return &WebhookHandler{
 		clientFactory: cfg.ClientFactory,
 		publisher:     cfg.OutboxPublisher,
 		db:            cfg.DB,
 		logger:        logger,
+		eventChecker:  eventChecker,
 	}
 }
 
@@ -206,7 +253,7 @@ func (h *WebhookHandler) validateAndParseWebhook(
 	if client.WebhookEndpointSecret == "" {
 		h.logger.Error("no webhook secret for tenant", "tenant_id", tenantID.String())
 		h.writeError(w, http.StatusInternalServerError, "no webhook secret configured for tenant")
-		return stripeadapter.ParsedWebhookEvent{}, ErrMissingTenantContext
+		return stripeadapter.ParsedWebhookEvent{}, ErrMissingWebhookSecret
 	}
 
 	adapter, err := stripeadapter.NewWebhookAdapter(client.WebhookEndpointSecret)
@@ -252,6 +299,24 @@ func (h *WebhookHandler) publishDomainEvent(
 	topic string,
 	tenantID tenant.TenantID,
 ) {
+	if h.eventChecker != nil {
+		already, err := h.eventChecker.IsProcessed(ctx, parsed.EventID)
+		if err != nil {
+			h.logger.Error("failed to check event idempotency",
+				"event_id", parsed.EventID,
+				"error", err,
+			)
+			// Fall through and publish - safer than silently dropping on a check failure.
+		} else if already {
+			h.logger.Info("skipping duplicate stripe webhook event",
+				"event_id", parsed.EventID,
+				"tenant_id", tenantID.String(),
+			)
+			h.writeSuccess(w, "webhook already processed")
+			return
+		}
+	}
+
 	h.logger.Info("publishing stripe webhook domain event",
 		"event_id", parsed.EventID,
 		"gateway_reference_id", parsed.GatewayReferenceID,
@@ -266,7 +331,16 @@ func (h *WebhookHandler) publishDomainEvent(
 		AggregateID:   parsed.PaymentOrderID,
 		AggregateType: "PaymentOrder",
 		PartitionKey:  parsed.PaymentOrderID,
+		CausationID:   parsed.EventID,
 	}); err != nil {
+		if isDuplicateCausationError(err) {
+			h.logger.Info("skipping duplicate stripe webhook event (unique constraint)",
+				"event_id", parsed.EventID,
+				"tenant_id", tenantID.String(),
+			)
+			h.writeSuccess(w, "webhook already processed")
+			return
+		}
 		h.logger.Error("failed to publish domain event to outbox",
 			"event_id", parsed.EventID,
 			"topic", topic,
@@ -410,3 +484,15 @@ func (h *WebhookHandler) writeError(w http.ResponseWriter, statusCode int, messa
 		h.logger.Error("failed to encode error response", "error", err)
 	}
 }
+
+// isDuplicateCausationError returns true if err is a unique constraint violation on the
+// causation_id column, indicating the event was already published by a concurrent request.
+func isDuplicateCausationError(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := strings.ToLower(err.Error())
+	return strings.Contains(s, "duplicate key") ||
+		strings.Contains(s, "unique constraint") ||
+		strings.Contains(s, "23505")
+}

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

@@ -470,6 +470,160 @@ func TestWebhookHandler_MissingPaymentOrderID_AcknowledgesOnly(t *testing.T) {
 	assert.Len(t, stub.published, 0, "should not publish when payment_order_id is missing")
 }
 
+// stubProcessedEventChecker implements ProcessedEventChecker for testing.
+type stubProcessedEventChecker struct {
+	processed map[string]bool
+	err       error
+}
+
+func (s *stubProcessedEventChecker) IsProcessed(_ context.Context, providerEventID string) (bool, error) {
+	if s.err != nil {
+		return false, s.err
+	}
+	return s.processed[providerEventID], nil
+}
+
+func TestWebhookHandler_EmptyWebhookSecret_ReturnsCorrectError(t *testing.T) {
+	stub := &stubOutboxPublisher{}
+	provider := &testTenantConfigProvider{
+		configs: map[string]stripeadapter.TenantConfig{
+			"no-secret-tenant": {
+				ConnectedAccountID:    "acct_test",
+				WebhookEndpointSecret: "",
+			},
+		},
+	}
+	factory, err := stripeadapter.NewClientFactory(stripeadapter.Config{
+		APIKey:             "sk_test_key",
+		TenantCacheSize:    10,
+		TenantCacheTTL:     time.Minute,
+		CircuitBreakerName: "test-cb",
+	}, provider, nil)
+	require.NoError(t, err)
+
+	h := fghttp.NewWebhookHandler(fghttp.WebhookHandlerConfig{
+		ClientFactory:   factory,
+		OutboxPublisher: stub,
+	})
+
+	payload := buildStripePayload(t, "evt_1", "payment_intent.succeeded", map[string]any{})
+	sig := signPayload(t, payload, testWebhookSecret)
+
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe/no-secret-tenant", bytes.NewReader(payload))
+	req.SetPathValue("tenantID", "no-secret-tenant")
+	req.Header.Set("Stripe-Signature", sig)
+
+	rr := httptest.NewRecorder()
+	h.HandleStripeWebhook(rr, req)
+
+	assert.Equal(t, http.StatusInternalServerError, rr.Code)
+	var resp map[string]interface{}
+	require.NoError(t, json.NewDecoder(rr.Body).Decode(&resp))
+	assert.Equal(t, "no webhook secret configured for tenant", resp["error"])
+}
+
+func TestWebhookHandler_DuplicateEvent_SkipsPublish(t *testing.T) {
+	stub := &stubOutboxPublisher{}
+	checker := &stubProcessedEventChecker{
+		processed: map[string]bool{"evt_dup_1": true},
+	}
+
+	provider := &testTenantConfigProvider{
+		configs: map[string]stripeadapter.TenantConfig{
+			"test-tenant": {
+				ConnectedAccountID:    "acct_test",
+				WebhookEndpointSecret: testWebhookSecret,
+			},
+		},
+	}
+	factory, err := stripeadapter.NewClientFactory(stripeadapter.Config{
+		APIKey:             "sk_test_key",
+		TenantCacheSize:    10,
+		TenantCacheTTL:     time.Minute,
+		CircuitBreakerName: "test-cb",
+	}, provider, nil)
+	require.NoError(t, err)
+
+	h := fghttp.NewWebhookHandler(fghttp.WebhookHandlerConfig{
+		ClientFactory:   factory,
+		OutboxPublisher: stub,
+		EventChecker:    checker,
+	})
+
+	payload := buildStripePayload(t, "evt_dup_1", "payment_intent.succeeded", map[string]any{
+		"id":       "pi_dup_123",
+		"object":   "payment_intent",
+		"amount":   5000,
+		"currency": "gbp",
+		"metadata": map[string]string{"payment_order_id": "po-dup-123"},
+		"status":   "succeeded",
+	})
+	sig := signPayload(t, payload, testWebhookSecret)
+
+	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()
+	h.HandleStripeWebhook(rr, req)
+
+	assert.Equal(t, http.StatusOK, rr.Code)
+	assert.Len(t, stub.published, 0, "duplicate event should not be re-published")
+	var resp map[string]interface{}
+	require.NoError(t, json.NewDecoder(rr.Body).Decode(&resp))
+	assert.Equal(t, true, resp["acknowledged"])
+}
+
+func TestWebhookHandler_IdempotencyCheckError_FallsThroughToPublish(t *testing.T) {
+	stub := &stubOutboxPublisher{}
+	checker := &stubProcessedEventChecker{
+		err: errors.New("db unavailable"),
+	}
+
+	provider := &testTenantConfigProvider{
+		configs: map[string]stripeadapter.TenantConfig{
+			"test-tenant": {
+				ConnectedAccountID:    "acct_test",
+				WebhookEndpointSecret: testWebhookSecret,
+			},
+		},
+	}
+	factory, err := stripeadapter.NewClientFactory(stripeadapter.Config{
+		APIKey:             "sk_test_key",
+		TenantCacheSize:    10,
+		TenantCacheTTL:     time.Minute,
+		CircuitBreakerName: "test-cb",
+	}, provider, nil)
+	require.NoError(t, err)
+
+	h := fghttp.NewWebhookHandler(fghttp.WebhookHandlerConfig{
+		ClientFactory:   factory,
+		OutboxPublisher: stub,
+		EventChecker:    checker,
+	})
+
+	payload := buildStripePayload(t, "evt_fallthrough_1", "payment_intent.succeeded", map[string]any{
+		"id":       "pi_fallthrough_123",
+		"object":   "payment_intent",
+		"amount":   2000,
+		"currency": "usd",
+		"metadata": map[string]string{"payment_order_id": "po-fallthrough-123"},
+		"status":   "succeeded",
+	})
+	sig := signPayload(t, payload, testWebhookSecret)
+
+	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()
+	h.HandleStripeWebhook(rr, req)
+
+	// On idempotency check failure, handler falls through and publishes the event.
+	assert.Equal(t, http.StatusOK, rr.Code)
+	assert.Len(t, stub.published, 1, "event should be published when idempotency check fails")
+}
+
 func TestWebhookHandler_NewWebhookHandler_PanicsOnNilClientFactory(t *testing.T) {
 	stub := &stubOutboxPublisher{}
 	assert.PanicsWithValue(t, fghttp.ErrNilClientFactory.Error(), func() {

services/financial-gateway/migrations/20260329000001_add_event_outbox_causation_id_index.sql

@@ -0,0 +1 @@
+CREATE UNIQUE INDEX IF NOT EXISTS idx_event_outbox_causation_id ON event_outbox (causation_id) WHERE causation_id IS NOT NULL;

services/financial-gateway/migrations/atlas.sum

@@ -1,4 +1,5 @@
-h1:jrzgSYnGrSymUsy8xdJA5VT7B2HfJSNptud30RK9KcI=
+h1:2Tz6trZn1stBcMbe9So9gzm1vzV58P6HoqNpcD8pJxg=
 20260304000001_create_event_outbox.sql h1:3qowbd0HiVAqH+82GdkX5/FN4Z1XoiGAGxmK6wXGXLA=
 20260316000001_add_event_outbox_tenant_id.sql h1:nE4czdkI7LnVIV8Z2NS/d0xrK9Wd/XnqzDu17Cu3sf4=
 20260316000002_add_event_outbox_tenant_id_index.sql h1:dT1XLgLUkdkL5dYpU7SrWnUXob597NGELzuJZHFKW34=
+20260329000001_add_event_outbox_causation_id_index.sql h1:hE5bD39isTGGiGrDbrY4K0zKoquml2OqFhgUZbmaWQ4=