test: End-to-end Stripe Connect integration test (#918)

* test: End-to-end Stripe Connect integration test

Add E2E tests exercising the full Stripe payment flow from billing
trigger through webhook processing to ledger posting:

- Happy path: payment order -> saga -> Stripe webhook (succeeded) ->
  ledger entries -> lien execution
- Failure path: payment decline -> dunning escalation (Redis ZSET)
- Webhook idempotency: replay same Stripe event ID -> no duplicate
  processing (verified via Redis and call counts)
- Multiple distinct webhooks on same payment order
- Successful payment does not trigger dunning

Uses CockroachDB testcontainer + miniredis for persistence parity.
Extends E2E test infra with sagaOrchestrationEnabled option and
saga_executions schema.

* fix: resolve ineffassign lint issues in stripe e2e test

Move lien call count recording after the await settle to avoid
redundant assignment, and discard the intentional timeout error
with blank identifier.

* fix: address CodeRabbit review comments on PR #918

1. Assert saga_executions DDL succeeds instead of silently discarding
   errors (e2e_test.go:430).

2. TestStripeE2E_SagaExecutionPersisted now queries the saga_executions
   table directly in CockroachDB and verifies records exist with correct
   saga_name, status, and payment_order_id. Also auto-wire
   SagaExecutionRepository into the service when saga orchestration is
   enabled.

* fix: remove unused withSagaExecutionLogger option function

Auto-wire SagaExecutionRepository directly when saga orchestration
is enabled, eliminating the need for an explicit option.

---------

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

Author: bjcoombs

services/payment-order/e2e/e2e_test.go

@@ -133,6 +133,12 @@ func setupE2E(t *testing.T, opts ...e2eOption) *E2ETestEnvironment {
 
 	repo := persistence.NewPaymentOrderRepository(db)
 
+	// Auto-wire saga execution logger when saga orchestration is enabled.
+	var sagaExecLogger domain.SagaExecutionLogger
+	if cfg.sagaOrchestrationEnabled {
+		sagaExecLogger = persistence.NewSagaExecutionRepository(db)
+	}
+
 	svc, err := service.NewServiceWithConfig(service.Config{
 		Repository:                repo,
 		CurrentAccountClient:      caClient,
@@ -143,6 +149,8 @@ func setupE2E(t *testing.T, opts ...e2eOption) *E2ETestEnvironment {
 		IdempotencyService:        newMockIdempotencyService(),
 		Logger:                    logger,
 		SagaTimeout:               sagaTimeout,
+		SagaOrchestrationEnabled:  cfg.sagaOrchestrationEnabled,
+		SagaExecutionLogger:       sagaExecLogger,
 	})
 	require.NoError(t, err)
 
@@ -161,11 +169,12 @@ func setupE2E(t *testing.T, opts ...e2eOption) *E2ETestEnvironment {
 
 // e2eConfig holds configuration options for E2E test setup.
 type e2eConfig struct {
-	gatewayApprove    bool
-	gatewayReject     bool
-	gatewayDelay      time.Duration
-	insufficientFunds bool
-	sagaTimeout       time.Duration
+	gatewayApprove           bool
+	gatewayReject            bool
+	gatewayDelay             time.Duration
+	insufficientFunds        bool
+	sagaTimeout              time.Duration
+	sagaOrchestrationEnabled bool
 }
 
 type e2eOption func(*e2eConfig)
@@ -195,6 +204,12 @@ func withSagaTimeout(d time.Duration) e2eOption {
 	}
 }
 
+func withSagaOrchestration() e2eOption {
+	return func(c *e2eConfig) {
+		c.sagaOrchestrationEnabled = true
+	}
+}
+
 // ============================================================================
 // Schema Setup
 // ============================================================================
@@ -416,6 +431,26 @@ func applyPaymentOrderSchema(t *testing.T, db *gorm.DB, schemaName string) {
 			client_ip VARCHAR(45),
 			user_agent TEXT
 		)`, auditTable))
+
+	// Create saga_executions table for saga audit trail
+	sagaExecTable := fmt.Sprintf("%s.saga_executions", pq.QuoteIdentifier(schemaName))
+	_, err = sqlDB.Exec(fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			payment_order_id UUID NOT NULL,
+			saga_name VARCHAR(128) NOT NULL,
+			saga_version INT NOT NULL DEFAULT 0,
+			status VARCHAR(32) NOT NULL DEFAULT 'RUNNING',
+			correlation_id VARCHAR(128) NOT NULL DEFAULT '',
+			input JSONB NOT NULL DEFAULT '{}',
+			output JSONB NOT NULL DEFAULT '{}',
+			error_message TEXT NOT NULL DEFAULT '',
+			step_count INT NOT NULL DEFAULT 0,
+			duration_ms BIGINT NOT NULL DEFAULT 0,
+			started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+			completed_at TIMESTAMPTZ
+		)`, sagaExecTable))
+	require.NoError(t, err)
 }
 
 // ============================================================================