fix: scope dunning retry ZSET key by tenant ID (#920)

Author: bjcoombs

services/payment-order/adapters/http/stripe_event_processor.go

@@ -14,6 +14,7 @@ import (
 var (
 	ErrNilRedisClient        = errors.New("redis client cannot be nil")
 	ErrEventAlreadyProcessed = errors.New("stripe event already processed")
+	ErrDunningMissingTenant  = errors.New("tenant ID is required for dunning scheduling")
 )
 
 // Stripe event processor constants.
@@ -24,9 +25,10 @@ const (
 	// processedWebhookTTL is how long processed event IDs are retained in Redis (48 hours).
 	processedWebhookTTL = 48 * time.Hour
 
-	// dunningRetryZSet is the Redis sorted set key for dunning retry scheduling.
-	// This matches the key used by DunningWorker.
-	dunningRetryZSet = "dunning:retries"
+	// dunningRetryZSetPrefix is the Redis sorted set key prefix for dunning retry scheduling.
+	// The full key is "dunning:retries:{tenantID}" for tenant isolation.
+	// This matches the key pattern used by DunningWorker.
+	dunningRetryZSetPrefix = "dunning:retries:"
 
 	// defaultDunningDelay is the default delay before the first dunning retry.
 	defaultDunningDelay = 24 * time.Hour
@@ -99,31 +101,39 @@ func (p *StripeEventProcessor) PreProcess(ctx context.Context, eventID string) e
 	return nil
 }
 
-// ScheduleDunning adds a payment order to the dunning retry sorted set.
+// ScheduleDunning adds a payment order to the tenant-scoped dunning retry sorted set.
 // Called when a payment_intent.payment_failed event is received.
 // The dunning worker will pick up the entry and trigger escalation.
-func (p *StripeEventProcessor) ScheduleDunning(ctx context.Context, paymentOrderID string) error {
+func (p *StripeEventProcessor) ScheduleDunning(ctx context.Context, tenantID, paymentOrderID string) error {
 	if paymentOrderID == "" {
 		p.logger.Warn("cannot schedule dunning: empty payment order ID")
 		return nil
 	}
+	if tenantID == "" {
+		p.logger.Error("cannot schedule dunning: empty tenant ID",
+			"payment_order_id", paymentOrderID)
+		return ErrDunningMissingTenant
+	}
 
 	dueAt := time.Now().Add(defaultDunningDelay)
 	member := redis.Z{
 		Score:  float64(dueAt.Unix()),
 		Member: fmt.Sprintf("stripe:%s", paymentOrderID),
 	}
 
-	err := p.redis.ZAdd(ctx, dunningRetryZSet, member).Err()
+	key := dunningRetryZSetPrefix + tenantID
+	err := p.redis.ZAdd(ctx, key, member).Err()
 	if err != nil {
 		p.logger.Error("failed to schedule dunning retry",
 			"payment_order_id", paymentOrderID,
+			"tenant_id", tenantID,
 			"error", err)
 		return fmt.Errorf("failed to schedule dunning retry: %w", err)
 	}
 
 	p.logger.Info("dunning retry scheduled for failed stripe payment",
 		"payment_order_id", paymentOrderID,
+		"tenant_id", tenantID,
 		"due_at", dueAt)
 
 	return nil

services/payment-order/adapters/http/stripe_event_processor_test.go

@@ -120,15 +120,17 @@ func TestStripeEventProcessor_PreProcess(t *testing.T) {
 
 func TestStripeEventProcessor_ScheduleDunning(t *testing.T) {
 	ctx := context.Background()
+	const testTenantID = "test_tenant"
+	zsetKey := dunningRetryZSetPrefix + testTenantID
 
 	t.Run("schedules dunning for payment order", func(t *testing.T) {
 		proc, client, _ := setupTestEventProcessor(t)
 
-		err := proc.ScheduleDunning(ctx, "po-123")
+		err := proc.ScheduleDunning(ctx, testTenantID, "po-123")
 		assert.NoError(t, err)
 
-		// Verify the entry was added to the ZSET
-		members, err := client.ZRangeByScore(ctx, dunningRetryZSet, &redis.ZRangeBy{
+		// Verify the entry was added to the tenant-scoped ZSET
+		members, err := client.ZRangeByScore(ctx, zsetKey, &redis.ZRangeBy{
 			Min: "-inf",
 			Max: "+inf",
 		}).Result()
@@ -140,25 +142,32 @@ func TestStripeEventProcessor_ScheduleDunning(t *testing.T) {
 	t.Run("empty payment order ID is a no-op", func(t *testing.T) {
 		proc, client, _ := setupTestEventProcessor(t)
 
-		err := proc.ScheduleDunning(ctx, "")
+		err := proc.ScheduleDunning(ctx, testTenantID, "")
 		assert.NoError(t, err)
 
 		// Verify nothing was added to the ZSET
-		count, err := client.ZCard(ctx, dunningRetryZSet).Result()
+		count, err := client.ZCard(ctx, zsetKey).Result()
 		require.NoError(t, err)
 		assert.Equal(t, int64(0), count)
 	})
 
+	t.Run("empty tenant ID returns error", func(t *testing.T) {
+		proc, _, _ := setupTestEventProcessor(t)
+
+		err := proc.ScheduleDunning(ctx, "", "po-123")
+		assert.ErrorIs(t, err, ErrDunningMissingTenant)
+	})
+
 	t.Run("multiple dunning entries are independent", func(t *testing.T) {
 		proc, client, _ := setupTestEventProcessor(t)
 
-		err := proc.ScheduleDunning(ctx, "po-aaa")
+		err := proc.ScheduleDunning(ctx, testTenantID, "po-aaa")
 		assert.NoError(t, err)
 
-		err = proc.ScheduleDunning(ctx, "po-bbb")
+		err = proc.ScheduleDunning(ctx, testTenantID, "po-bbb")
 		assert.NoError(t, err)
 
-		count, err := client.ZCard(ctx, dunningRetryZSet).Result()
+		count, err := client.ZCard(ctx, zsetKey).Result()
 		require.NoError(t, err)
 		assert.Equal(t, int64(2), count)
 	})
@@ -168,7 +177,39 @@ func TestStripeEventProcessor_ScheduleDunning(t *testing.T) {
 
 		mr.Close()
 
-		err := proc.ScheduleDunning(ctx, "po-fail")
+		err := proc.ScheduleDunning(ctx, testTenantID, "po-fail")
 		assert.Error(t, err)
 	})
+
+	t.Run("tenant isolation between tenants", func(t *testing.T) {
+		proc, client, _ := setupTestEventProcessor(t)
+
+		const tenantA = "tenant_a"
+		const tenantB = "tenant_b"
+
+		err := proc.ScheduleDunning(ctx, tenantA, "po-a1")
+		assert.NoError(t, err)
+		err = proc.ScheduleDunning(ctx, tenantB, "po-b1")
+		assert.NoError(t, err)
+
+		// Verify tenant A's key only has tenant A's entry
+		keyA := dunningRetryZSetPrefix + tenantA
+		membersA, err := client.ZRangeByScore(ctx, keyA, &redis.ZRangeBy{
+			Min: "-inf",
+			Max: "+inf",
+		}).Result()
+		require.NoError(t, err)
+		assert.Len(t, membersA, 1)
+		assert.Equal(t, "stripe:po-a1", membersA[0])
+
+		// Verify tenant B's key only has tenant B's entry
+		keyB := dunningRetryZSetPrefix + tenantB
+		membersB, err := client.ZRangeByScore(ctx, keyB, &redis.ZRangeBy{
+			Min: "-inf",
+			Max: "+inf",
+		}).Result()
+		require.NoError(t, err)
+		assert.Len(t, membersB, 1)
+		assert.Equal(t, "stripe:po-b1", membersB[0])
+	})
 }

services/payment-order/adapters/http/stripe_webhook_handler.go

@@ -186,9 +186,10 @@ func (h *StripeWebhookHandler) HandleStripeWebhook(w http.ResponseWriter, r *htt
 
 	// Schedule dunning for failed payments (fire-and-forget, does not affect response)
 	if h.eventProcessor != nil && parsed.Status == "REJECTED" && parsed.PaymentOrderID != "" {
-		if dunningErr := h.eventProcessor.ScheduleDunning(ctx, parsed.PaymentOrderID); dunningErr != nil {
+		if dunningErr := h.eventProcessor.ScheduleDunning(ctx, tenantID.String(), parsed.PaymentOrderID); dunningErr != nil {
 			h.logger.Error("failed to schedule dunning for failed payment",
 				"payment_order_id", parsed.PaymentOrderID,
+				"tenant_id", tenantID.String(),
 				"event_id", parsed.EventID,
 				"error", dunningErr)
 		}

services/payment-order/adapters/http/stripe_webhook_handler_test.go

@@ -408,7 +408,7 @@ func TestStripeWebhookHandler_FailedPaymentTriggersDunning(t *testing.T) {
 	assert.Equal(t, http.StatusOK, rr.Code)
 
 	// Verify dunning was scheduled in the ZSET
-	members, err := redisClient.ZRangeByScore(ctx, dunningRetryZSet, &redis.ZRangeBy{
+	members, err := redisClient.ZRangeByScore(ctx, dunningRetryZSetPrefix+"test-tenant", &redis.ZRangeBy{
 		Min: "-inf",
 		Max: "+inf",
 	}).Result()
@@ -441,7 +441,7 @@ func TestStripeWebhookHandler_SucceededPaymentDoesNotTriggerDunning(t *testing.T
 	assert.Equal(t, http.StatusOK, rr.Code)
 
 	// Verify dunning was NOT scheduled
-	count, err := redisClient.ZCard(ctx, dunningRetryZSet).Result()
+	count, err := redisClient.ZCard(ctx, dunningRetryZSetPrefix+"test-tenant").Result()
 	require.NoError(t, err)
 	assert.Equal(t, int64(0), count)
 }
@@ -474,7 +474,7 @@ func TestStripeWebhookHandler_RefundedDoesNotTriggerDunning(t *testing.T) {
 	assert.Equal(t, http.StatusOK, rr.Code)
 
 	// Verify dunning was NOT scheduled for refunds
-	count, err := redisClient.ZCard(ctx, dunningRetryZSet).Result()
+	count, err := redisClient.ZCard(ctx, dunningRetryZSetPrefix+"test-tenant").Result()
 	require.NoError(t, err)
 	assert.Equal(t, int64(0), count)
 }

services/payment-order/worker/dunning_worker.go

@@ -18,12 +18,14 @@ import (
 
 // Dunning worker errors.
 var (
-	ErrNilDunningCallback = errors.New("dunning callback is required")
-	ErrDunningKeyTooShort = errors.New("dunning retry key too short")
+	ErrNilDunningCallback   = errors.New("dunning callback is required")
+	ErrDunningKeyTooShort   = errors.New("dunning retry key too short")
+	ErrDunningMissingTenant = errors.New("tenant ID is required for dunning retry")
 )
 
-// dunningRetryZSet is the Redis sorted set key for dunning retry scheduling.
-const dunningRetryZSet = "dunning:retries"
+// dunningRetryZSetPrefix is the Redis sorted set key prefix for dunning retry scheduling.
+// The full key is "dunning:retries:{tenantID}" for tenant isolation.
+const dunningRetryZSetPrefix = "dunning:retries:"
 
 // DunningWorkerConfig holds configuration for the dunning retry worker.
 type DunningWorkerConfig struct {
@@ -145,42 +147,75 @@ func (w *DunningWorker) pollLoop(ctx context.Context) error {
 	}
 }
 
-// ScheduleDunningRetry adds a billing run to the sorted set with a score
-// equal to the Unix timestamp when the retry becomes due.
-func (w *DunningWorker) ScheduleDunningRetry(ctx context.Context, billingRunID uuid.UUID, delay time.Duration) error {
+// ScheduleDunningRetry adds a billing run to the tenant-scoped sorted set with
+// a score equal to the Unix timestamp when the retry becomes due.
+func (w *DunningWorker) ScheduleDunningRetry(ctx context.Context, tenantID string, billingRunID uuid.UUID, delay time.Duration) error {
+	if tenantID == "" {
+		return ErrDunningMissingTenant
+	}
 	dueAt := NowFunc().Add(delay)
 	member := redis.Z{
 		Score:  float64(dueAt.Unix()),
 		Member: billingRunID.String(),
 	}
-	err := w.redis.ZAdd(ctx, dunningRetryZSet, member).Err()
+	key := dunningRetryZSetPrefix + tenantID
+	err := w.redis.ZAdd(ctx, key, member).Err()
 	if err != nil {
 		return fmt.Errorf("failed to schedule dunning retry: %w", err)
 	}
 
 	w.logger.Info("dunning retry scheduled",
 		"billing_run_id", billingRunID,
+		"tenant_id", tenantID,
 		"delay", delay,
 		"due_at", dueAt)
 
 	return nil
 }
 
-// processDueRetries queries the sorted set for all members whose score (due
-// timestamp) is at or before the current time, processes each one, and removes
-// it from the set.
+// processDueRetries scans for all tenant-scoped dunning ZSET keys and processes
+// due retries from each. Uses SCAN to discover keys matching "dunning:retries:*".
 func (w *DunningWorker) processDueRetries(ctx context.Context) {
+	// Discover all tenant-scoped dunning keys
+	keys, err := w.scanDunningKeys(ctx)
+	if err != nil {
+		w.logger.Error("failed to scan dunning retry keys", "error", err)
+		return
+	}
+
+	for _, key := range keys {
+		w.processDueRetriesForKey(ctx, key)
+	}
+}
+
+// scanDunningKeys returns all Redis keys matching the dunning retry ZSET pattern.
+func (w *DunningWorker) scanDunningKeys(ctx context.Context) ([]string, error) {
+	var allKeys []string
+	pattern := dunningRetryZSetPrefix + "*"
+	iter := w.redis.Scan(ctx, 0, pattern, 100).Iterator()
+	for iter.Next(ctx) {
+		allKeys = append(allKeys, iter.Val())
+	}
+	if err := iter.Err(); err != nil {
+		return nil, fmt.Errorf("failed to scan dunning keys: %w", err)
+	}
+	return allKeys, nil
+}
+
+// processDueRetriesForKey queries a single tenant's sorted set for all members
+// whose score (due timestamp) is at or before the current time, processes each
+// one, and removes it from the set.
+func (w *DunningWorker) processDueRetriesForKey(ctx context.Context, key string) {
 	now := NowFunc()
 	maxScore := strconv.FormatInt(now.Unix(), 10)
 
-	// Fetch billing run IDs that are due
-	members, err := w.redis.ZRangeByScore(ctx, dunningRetryZSet, &redis.ZRangeBy{
+	members, err := w.redis.ZRangeByScore(ctx, key, &redis.ZRangeBy{
 		Min:   "-inf",
 		Max:   maxScore,
 		Count: 100,
 	}).Result()
 	if err != nil {
-		w.logger.Error("failed to query dunning retries", "error", err)
+		w.logger.Error("failed to query dunning retries", "key", key, "error", err)
 		return
 	}
 
@@ -192,20 +227,19 @@ func (w *DunningWorker) processDueRetries(ctx context.Context) {
 	for _, member := range members {
 		billingRunID, parseErr := uuid.Parse(member)
 		if parseErr != nil {
-			w.logger.Error("invalid billing run ID in dunning set", "member", member, "error", parseErr)
-			w.redis.ZRem(ctx, dunningRetryZSet, member)
+			w.logger.Error("invalid billing run ID in dunning set", "key", key, "member", member, "error", parseErr)
+			w.redis.ZRem(ctx, key, member)
 			continue
 		}
 
 		if w.processRetry(ctx, billingRunID) {
-			// Only remove on success; transient failures retain the member for next poll
-			w.redis.ZRem(ctx, dunningRetryZSet, member)
+			w.redis.ZRem(ctx, key, member)
 			processed++
 		}
 	}
 
 	if processed > 0 {
-		w.logger.Info("processed dunning retries", "count", processed)
+		w.logger.Info("processed dunning retries", "key", key, "count", processed)
 	}
 }
 
@@ -283,8 +317,12 @@ func (w *DunningWorker) executeRetry(ctx context.Context, billingRunID uuid.UUID
 	return true
 }
 
-// CancelDunningRetry removes a billing run from the retry set.
+// CancelDunningRetry removes a billing run from the tenant-scoped retry set.
 // Called when a billing run is resolved (e.g., manual payment succeeds).
-func (w *DunningWorker) CancelDunningRetry(ctx context.Context, billingRunID uuid.UUID) error {
-	return w.redis.ZRem(ctx, dunningRetryZSet, billingRunID.String()).Err()
+func (w *DunningWorker) CancelDunningRetry(ctx context.Context, tenantID string, billingRunID uuid.UUID) error {
+	if tenantID == "" {
+		return ErrDunningMissingTenant
+	}
+	key := dunningRetryZSetPrefix + tenantID
+	return w.redis.ZRem(ctx, key, billingRunID.String()).Err()
 }