Stripe Sync #283

BRO3886 · 2025-06-10T20:27:35Z

BRO3886
Jun 10, 2025

Stripe Integration - High Level Design

System Overview

The Stripe Integration system enables bidirectional usage synchronization between FlexPrice and Stripe, allowing customers to maintain their existing Stripe billing infrastructure while leveraging FlexPrice's advanced usage-based billing capabilities. The system acts as a bridge, intelligently deciding which usage events should be synchronized to Stripe based on billing logic and business rules.

Key Objectives

Non-disruptive Integration: Customers can continue using Stripe for payments while gaining FlexPrice's advanced metering
Intelligent Sync: Only billable events are synced to Stripe; free tier and credited usage remains in FlexPrice
Reliability: Temporal-based workflows ensure exactly-once delivery with comprehensive retry mechanisms
Scalability: Batch aggregation prevents rate limit issues while maintaining accuracy
Auditability: Complete tracking of sync decisions and states for compliance and debugging

Architecture Overview

graph TB
    subgraph "Customer Systems"
        CS[Customer Application]
        CDB[Customer Database]
    end
    
    subgraph "FlexPrice Platform"
        subgraph "API Layer"
            API[REST API]
            WH[Webhook Endpoints]
        end
        
        subgraph "Core Services"
            ES[Event Service]
            BS[Billing Service]
            IS[Integration Service]
            CS_FP[Customer Service]
        end
        
        subgraph "Integration Layer"
            ISS[Integration Sync State]
            CM[Customer Mapping]
            PC[Provider Config]
        end
        
        subgraph "Workflow Engine"
            TW[Temporal Workflows]
            TA[Temporal Activities]
        end
        
        subgraph "Data Layer"
            PG[(PostgreSQL)]
            CH[(ClickHouse)]
            RD[(Redis)]
        end
    end
    
    subgraph "External Providers"
        ST[Stripe API]
        HB[HubSpot API]
        RZ[RazorPay API]
    end
    
    CS -->|Usage Events| API
    API --> ES
    ES --> CH
    ES --> BS
    BS --> PG
    IS --> ISS
    IS --> CM
    IS --> PC
    TW --> TA
    TA --> ST
    WH -->|Customer Sync| IS
    IS --> TW
    
    classDef external fill:#e1f5fe
    classDef core fill:#f3e5f5
    classDef data fill:#e8f5e8
    classDef integration fill:#fff3e0
    
    class CS,CDB,ST,HB,RZ external
    class ES,BS,IS,CS_FP core
    class PG,CH,RD data
    class ISS,CM,PC,TW,TA integration

Core Components

1. Event Processing Pipeline

sequenceDiagram
    participant CA as Customer App
    participant FP as FlexPrice API
    participant ES as Event Service
    participant CH as ClickHouse
    participant BS as Billing Service
    participant PG as PostgreSQL
    
    CA->>FP: Send Usage Events
    FP->>ES: Validate & Process
    ES->>CH: Store Raw Events
    ES->>BS: Trigger Billing Processing
    BS->>CH: Query Event Data
    BS->>PG: Store Processed Events
    Note over PG: events_processed table<br/>with qty_billable, qty_free

2. Integration Sync Decision Engine

flowchart TD
    A[Raw Event] --> B{Processed by<br/>FlexPrice Billing?}
    B -->|No| C[Mark as Pending]
    B -->|Yes| D{Billable Quantity > 0?}
    D -->|No| E{Free Tier Applied?}
    D -->|Yes| F{Meter Configured<br/>for Sync?}
    E -->|Yes| G[Skip - Free Tier]
    E -->|No| H[Skip - Credited]
    F -->|No| I[Skip - Not Configured]
    F -->|Yes| J[Mark as Billable]
    
    C --> K[integration_sync_state]
    G --> K
    H --> K
    I --> K
    J --> K
    
    classDef billable fill:#c8e6c9
    classDef skipped fill:#ffecb3
    classDef pending fill:#e1f5fe
    
    class J billable
    class G,H,I skipped
    class C pending

3. Temporal Workflow Architecture

graph TB
    subgraph "Temporal Workflows"
        CW[Cron Sync Workflow<br/>Runs Hourly]
        SW[Stripe Sync Workflow<br/>Main Orchestrator]
        RW[Retry Workflow<br/>Failed Events]
        BW[Backfill Workflow<br/>Historical Data]
    end
    
    subgraph "Activities"
        SDA[Sync Decision Activity<br/>Process Unsynced Events]
        AGA[Aggregation Activity<br/>Group by Customer+Meter]
        SSA[Stripe Sync Activity<br/>API Calls]
        RFA[Retry Failed Activity<br/>Retry Logic]
    end
    
    CW --> SW
    SW --> SDA
    SW --> AGA
    SW --> SSA
    RW --> RFA
    BW --> SW
    
    SDA --> ISS[(Integration Sync State)]
    AGA --> ISS
    SSA --> ST[Stripe API]
    SSA --> ISS

Data Flow Architecture

1. Customer Onboarding Flow

sequenceDiagram
    participant C as Customer
    participant S as Stripe
    participant FP as FlexPrice
    participant WH as Webhook Handler
    participant IS as Integration Service
    
    Note over C,S: Customer creates Stripe customer
    C->>S: Create Customer
    S-->>C: customer_id: cus_stripe_123
    
    Note over C,FP: Customer creates FlexPrice customer
    C->>FP: Create Customer
    FP-->>C: customer_id: cust_fp_456
    
    Note over C,WH: Customer maps the two systems
    C->>WH: POST /webhook/customer-sync
    Note right of WH: Payload: {<br/>  stripe_customer_id: "cus_stripe_123",<br/>  flexprice_customer_id: "cust_fp_456"<br/>}
    WH->>IS: Create Customer Mapping
    IS->>IS: Store in customer_integration_mappings
    
    Note over C: Customer can now send usage events<br/>FlexPrice will sync billable usage to Stripe

2. Usage Event Sync Flow

sequenceDiagram
    participant CA as Customer App
    participant FP as FlexPrice
    participant TW as Temporal Workflow
    participant ST as Stripe
    
    Note over CA,FP: 1. Usage Events Flow to FlexPrice
    CA->>FP: Send Usage Events<br/>(customer_id: external_123)
    FP->>FP: Process Events<br/>Calculate Billable Usage
    
    Note over TW: 2. Hourly Sync Workflow
    TW->>TW: Cron: Every Hour
    TW->>FP: Find Unprocessed Events
    FP-->>TW: Events List
    
    TW->>TW: Make Sync Decisions<br/>(billable vs skipped)
    TW->>TW: Aggregate Billable Events<br/>by Customer + Meter
    
    Note over TW,ST: 3. Sync to Stripe
    loop For each Customer+Meter
        TW->>ST: Create Meter Event<br/>{value: aggregated_usage}
        ST-->>TW: meter_event_id
        TW->>FP: Mark as Synced<br/>(batch_id, stripe_event_id)
    end

3. Sync Decision Matrix

graph TD
    E[Event] --> P{Processed?}
    P -->|No| W[Wait for Processing]
    P -->|Yes| Q{qty_billable > 0?}
    
    Q -->|No| F{qty_free > 0?}
    Q -->|Yes| M{Meter Configured?}
    
    F -->|Yes| SF[Status: skipped_free<br/>Reason: free_tier_applied]
    F -->|No| SC[Status: skipped_credited<br/>Reason: credit_applied]
    
    M -->|No| SM[Status: skipped_config<br/>Reason: meter_not_configured]
    M -->|Yes| SB[Status: billable<br/>Ready for Sync]
    
    W --> P
    SF --> END[Store Decision]
    SC --> END
    SM --> END
    SB --> AGG[Add to Aggregation Queue]
    AGG --> END
    
    classDef billable fill:#c8e6c9
    classDef skipped fill:#ffecb3
    classDef waiting fill:#f8bbd9
    
    class SB,AGG billable
    class SF,SC,SM skipped
    class W waiting

Customer Journey

Phase 1: Initial Setup

Customer has existing Stripe setup with subscriptions and billing
Customer signs up for FlexPrice for advanced usage metering
Integration mapping - Customer maps their Stripe customers to FlexPrice customers
Meter configuration - Customer configures which FlexPrice meters should sync to Stripe

Phase 2: Usage Tracking

Customer sends usage events to FlexPrice with their internal customer IDs
FlexPrice processes events through its advanced billing engine
FlexPrice applies pricing logic - free tiers, credits, entitlements, complex pricing models
Only billable usage (after free tiers/credits) is marked for Stripe sync

Phase 3: Automated Sync

Temporal workflows run hourly to process unsynced events
Events are aggregated by customer and meter to avoid rate limits
Billable usage is synced to Stripe as meter events
Stripe generates invoices using the synced usage data
Customer maintains their existing payment flows through Stripe

Phase 4: Monitoring & Management

Customer monitors sync status through FlexPrice dashboard
Failed syncs are automatically retried with exponential backoff
Historical data can be backfilled using CLI tools or API
Complete audit trail available for debugging and compliance

Integration Patterns

1. Three-Layer ID Mapping

graph LR
    subgraph "Customer System"
        CID[Customer Internal ID<br/>user_12345]
    end
    
    subgraph "FlexPrice"
        FPID[FlexPrice Customer ID<br/>cust_fp_789]
        EID[Event.external_customer_id<br/>user_12345]
    end
    
    subgraph "Stripe"
        SID[Stripe Customer ID<br/>cus_stripe_456]
    end
    
    subgraph "Mapping Table"
        MT[customer_integration_mappings<br/>fp_id ↔ stripe_id]
    end
    
    CID -.->|Usage Events| EID
    EID -.->|Links to| FPID
    FPID -.->|Maps via| MT
    MT -.->|Resolves to| SID
    
    classDef system fill:#e3f2fd
    classDef mapping fill:#fff3e0
    
    class CID,FPID,SID system
    class MT,EID mapping

2. Event Aggregation Strategy

gantt
    title Hourly Aggregation Window (3 PM - 4 PM)
    dateFormat HH:mm
    axisFormat %H:%M
    
    section Customer A Events
    100 API calls    :active, 15:05, 15:06
    200 API calls    :active, 15:15, 15:16  
    150 API calls    :active, 15:30, 15:31
    
    section Customer B Events
    50 API calls     :active, 15:05, 15:06
    25 API calls     :active, 15:30, 15:31
    
    section Customer C Events
    75 API calls     :active, 15:15, 15:16
    125 API calls    :active, 15:45, 15:46
    
    section Aggregation
    Workflow Trigger :milestone, 16:00, 16:00
    Customer A Total 450 :crit, 16:01, 16:02
    Customer B Total 75  :crit, 16:01, 16:02
    Customer C Total 200 :crit, 16:01, 16:02
    Send to Stripe   :milestone, 16:05, 16:05

3. Provider Extensibility

graph TB
    subgraph "Core Integration Engine"
        ISE[Integration Sync Engine]
        CM[Customer Mapping]
        PC[Provider Config]
    end
    
    subgraph "Provider Adapters"
        SA[Stripe Adapter]
        HA[HubSpot Adapter]
        RA[RazorPay Adapter]
        CA[Custom Adapter]
    end
    
    subgraph "External APIs"
        ST[Stripe API]
        HB[HubSpot API]
        RZ[RazorPay API]
        EX[External System]
    end
    
    ISE --> SA
    ISE --> HA
    ISE --> RA
    ISE --> CA
    
    SA --> ST
    HA --> HB
    RA --> RZ
    CA --> EX
    
    CM -.-> ISE
    PC -.-> ISE
    
    classDef core fill:#e8f5e8
    classDef adapter fill:#fff3e0
    classDef external fill:#e1f5fe
    
    class ISE,CM,PC core
    class SA,HA,RA,CA adapter
    class ST,HB,RZ,EX external

Error Handling & Resilience

1. Retry Strategy

stateDiagram-v2
    [*] --> Pending
    Pending --> Processing: Workflow Triggered
    Processing --> Billable: qty_billable > 0
    Processing --> Skipped: qty_billable = 0
    Billable --> Syncing: Send to Stripe
    Syncing --> Synced: Success
    Syncing --> Failed: API Error
    Failed --> Retrying: Exponential Backoff
    Retrying --> Synced: Retry Success
    Retrying --> Failed: Retry Failed
    Failed --> ManualReview: Max Attempts Reached
    Skipped --> [*]
    Synced --> [*]
    ManualReview --> [*]
    
    state Failed {
        [*] --> Attempt1: 1min delay
        Attempt1 --> Attempt2: 2min delay
        Attempt2 --> Attempt3: 4min delay
        Attempt3 --> Attempt4: 8min delay
        Attempt4 --> Attempt5: 16min delay
        Attempt5 --> [*]: Max attempts
    }

2. Circuit Breaker Pattern

stateDiagram-v2
    [*] --> Closed: Normal Operation
    Closed --> Open: Failure Threshold Reached
    Open --> HalfOpen: Timeout Expired
    HalfOpen --> Closed: Test Request Success
    HalfOpen --> Open: Test Request Failed
    
    note right of Closed : All requests pass through\nTrack failure rate
    note right of Open : All requests fail fast\nNo calls to Stripe API
    note right of HalfOpen : Limited test requests\nGradual recovery

3. Failure Categories & Responses

Failure Type	Response Strategy	Recovery Action
Network Timeout	Automatic Retry	Exponential backoff, max 5 attempts
Rate Limit	Backoff & Retry	Wait for rate limit reset + jitter
Authentication	Alert & Stop	Manual API key verification needed
Invalid Data	Skip & Log	Mark as permanently failed, alert ops
Stripe Downtime	Circuit Breaker	Fail fast, resume when healthy

Monitoring & Observability

1. Key Metrics Dashboard

graph TB
    subgraph "Sync Metrics"
        SM1[Events Processed/Hour]
        SM2[Events Synced/Hour]
        SM3[Sync Success Rate]
        SM4[Average Sync Latency]
    end
    
    subgraph "Error Metrics"
        EM1[Failed Events Count]
        EM2[Retry Attempts]
        EM3[Circuit Breaker State]
        EM4[Error Rate by Type]
    end
    
    subgraph "Business Metrics"
        BM1[Free Tier Usage %]
        BM2[Billable Events %]
        BM3[Revenue Impact]
        BM4[Customer Adoption Rate]
    end
    
    subgraph "Performance Metrics"
        PM1[Temporal Workflow Health]
        PM2[API Response Times]
        PM3[Database Performance]
        PM4[Queue Depths]
    end
    
    classDef sync fill:#e8f5e8
    classDef error fill:#ffebee
    classDef business fill:#e3f2fd
    classDef performance fill:#fff3e0
    
    class SM1,SM2,SM3,SM4 sync
    class EM1,EM2,EM3,EM4 error
    class BM1,BM2,BM3,BM4 business
    class PM1,PM2,PM3,PM4 performance

2. Alert Hierarchy

graph TD
    A[System Health Check] --> B{Sync Success Rate}
    B -->|< 95%| C[Warning Alert]
    B -->|< 85%| D[Critical Alert]
    B -->|< 70%| E[Emergency Alert]
    
    F[Individual Event] --> G{Retry Count}
    G -->|> 3 attempts| H[Error Log]
    G -->|> 5 attempts| I[Support Ticket]
    
    J[Temporal Workflow] --> K{Workflow Status}
    K -->|Failed| L[Immediate Alert]
    K -->|Delayed| M[Warning]
    
    C --> N[Slack Notification]
    D --> O[PagerDuty + Slack]
    E --> P[PagerDuty + Phone + Slack]
    
    classDef warning fill:#fff3e0
    classDef critical fill:#ffebee
    classDef emergency fill:#f3e5f5
    
    class C,H,M warning
    class D,I critical
    class E,L emergency

Security & Compliance

1. Data Security

graph TB
    subgraph "Data in Transit"
        TLS[TLS 1.3 Encryption]
        API[API Key Authentication]
        SIGN[Request Signing]
    end
    
    subgraph "Data at Rest"
        ENC[Database Encryption]
        VAULT[Secret Management]
        BACKUP[Encrypted Backups]
    end
    
    subgraph "Access Control"
        RBAC[Role-Based Access]
        MFA[Multi-Factor Auth]
        AUDIT[Audit Logging]
    end
    
    subgraph "Compliance"
        SOC[SOC 2 Type II]
        GDPR[GDPR Compliance]
        PCI[PCI DSS]
    end
    
    TLS --> ENC
    API --> VAULT
    SIGN --> RBAC
    ENC --> SOC
    VAULT --> GDPR
    RBAC --> PCI
    
    classDef security fill:#e8f5e8
    classDef compliance fill:#e3f2fd
    
    class TLS,API,SIGN,ENC,VAULT,BACKUP,RBAC,MFA,AUDIT security
    class SOC,GDPR,PCI compliance

2. API Security Model

Component	Security Measure	Implementation
Customer Data	Tenant Isolation	Row-level security, tenant_id filtering
API Keys	Rotation & Encryption	Vault storage, 90-day rotation
Webhook Verification	HMAC Signatures	SHA-256 request signing
Rate Limiting	Per-tenant Limits	Redis-based sliding window
Audit Trail	Complete Logging	All actions logged with timestamps

Performance & Scalability

1. Scalability Targets

Metric	Current	Target	Scale Strategy
Events/Second	1,000	10,000	Horizontal scaling, batch processing
Customers	1,000	100,000	Database sharding, caching
Sync Latency	< 5 min	< 2 min	Parallel processing, optimization
API Calls/Min	500	5,000	Connection pooling, rate limiting

2. Performance Optimization

graph TB
    subgraph "Database Optimization"
        IDX[Strategic Indexes]
        PART[Table Partitioning]
        CACHE[Redis Caching]
    end
    
    subgraph "API Optimization"
        POOL[Connection Pooling]
        BATCH[Batch Processing]
        COMP[Response Compression]
    end
    
    subgraph "Workflow Optimization"
        PAR[Parallel Activities]
        CHUNK[Chunked Processing]
        PRIO[Priority Queues]
    end
    
    IDX --> POOL
    PART --> BATCH
    CACHE --> COMP
    POOL --> PAR
    BATCH --> CHUNK
    COMP --> PRIO
    
    classDef db fill:#e8f5e8
    classDef api fill:#e3f2fd
    classDef workflow fill:#fff3e0
    
    class IDX,PART,CACHE db
    class POOL,BATCH,COMP api
    class PAR,CHUNK,PRIO workflow

Deployment Architecture

1. Infrastructure Layout

graph TB
    subgraph "Load Balancer"
        LB[Application Load Balancer]
    end
    
    subgraph "Application Tier"
        APP1[FlexPrice Instance 1]
        APP2[FlexPrice Instance 2]
        APP3[FlexPrice Instance 3]
    end
    
    subgraph "Temporal Cluster"
        TW1[Temporal Worker 1]
        TW2[Temporal Worker 2]
        TS[Temporal Server]
    end
    
    subgraph "Data Tier"
        PG_M[PostgreSQL Primary]
        PG_S[PostgreSQL Replica]
        CH_C[ClickHouse Cluster]
        RD_C[Redis Cluster]
    end
    
    subgraph "External Services"
        ST[Stripe API]
        MON[Monitoring Stack]
    end
    
    LB --> APP1
    LB --> APP2
    LB --> APP3
    
    APP1 --> TW1
    APP2 --> TW2
    APP3 --> TS
    
    TW1 --> PG_M
    TW2 --> PG_M
    TS --> PG_M
    
    APP1 --> CH_C
    APP2 --> CH_C
    APP3 --> RD_C
    
    TW1 --> ST
    TW2 --> ST
    
    APP1 --> MON
    TW1 --> MON
    
    classDef app fill:#e8f5e8
    classDef temporal fill:#fff3e0
    classDef data fill:#e3f2fd
    classDef external fill:#ffebee
    
    class APP1,APP2,APP3 app
    class TW1,TW2,TS temporal
    class PG_M,PG_S,CH_C,RD_C data
    class ST,MON external

2. Deployment Strategy

timeline
    title Blue-Green Deployment Process
    
    Planning : Prepare new environment
             : Run integration tests
             : Verify Stripe connectivity
    
    Deployment : Deploy to Green environment
               : Run smoke tests
               : Validate Temporal workflows
    
    Validation : Process test transactions
               : Verify sync functionality
               : Monitor error rates
    
    Switch : Update load balancer
           : Route traffic to Green
           : Monitor system health
    
    Cleanup : Keep Blue as fallback
            : Clean up old resources
            : Update monitoring

Risk Assessment & Mitigation

1. Technical Risks

Risk	Impact	Probability	Mitigation
Stripe API Downtime	High	Low	Circuit breaker, retry queue
Data Sync Inconsistency	High	Medium	Reconciliation jobs, audit trails
Rate Limit Exceeded	Medium	Medium	Batch aggregation, backoff logic
Temporal Workflow Failure	High	Low	Redundant workers, monitoring

2. Business Risks

Risk	Impact	Probability	Mitigation
Customer Invoice Errors	High	Low	Comprehensive testing, validation
Revenue Recognition Issues	High	Low	Audit trails, reconciliation
Customer Migration Complexity	Medium	High	Detailed documentation, support
Vendor Lock-in	Medium	Medium	Provider abstraction layer

Success Criteria

Technical Success

✅ 99.9% sync success rate
✅ < 2 minute average sync latency
✅ Zero data loss or duplication
✅ Complete audit trail for all operations

Business Success

✅ Seamless customer migration from Stripe-only to FlexPrice+Stripe
✅ Reduced time-to-value for new customers
✅ Expansion to multiple billing providers
✅ Improved customer retention through advanced features

Operational Success

✅ Automated deployment and rollback procedures
✅ Comprehensive monitoring and alerting
✅ Self-healing error recovery
✅ 24/7 system availability

This high-level design provides a comprehensive framework for implementing a robust, scalable, and maintainable Stripe integration that serves as a foundation for multi-provider billing synchronization.

BRO3886 · 2025-06-10T20:39:15Z

BRO3886
Jun 10, 2025
Author

Stripe Integration - Low Level Design

Sequence Diagrams & State Machine Definitions

1. Core Event Flow Sequences

1.1 Complete Usage Event Sync Flow

sequenceDiagram
    participant CA as Customer App
    participant FP as FlexPrice API
    participant ES as Event Service
    participant CH as ClickHouse
    participant BS as Billing Service
    participant PG as PostgreSQL
    participant TW as Temporal Worker
    participant IS as Integration Service
    participant ST as Stripe API

    Note over CA,ST: Complete End-to-End Usage Event Sync Flow

    %% Event Ingestion
    CA->>FP: POST /events (external_customer_id: "client_123")
    FP->>ES: Validate & Store Event
    ES->>CH: Store Raw Event
    ES->>BS: Trigger Billing Processing
    
    %% FlexPrice Billing Processing
    BS->>CH: Query Event Data
    BS->>PG: Apply Pricing Rules
    Note over BS,PG: Calculate free tier, credits, tiers<br/>Result: qty_billable, qty_free
    BS->>PG: Store in events_processed
    
    %% Temporal Sync Workflow (Hourly)
    Note over TW: Cron Trigger: Every Hour
    TW->>IS: Start Sync Decision Activity
    IS->>PG: Query Unprocessed Events
    
    loop For Each Unprocessed Event
        IS->>IS: Make Sync Decision
        alt qty_billable > 0
            IS->>PG: Mark as 'billable' in integration_sync_state
        else qty_free > 0
            IS->>PG: Mark as 'skipped_free'
        else
            IS->>PG: Mark as 'skipped_credited'
        end
    end
    
    %% Event Aggregation
    TW->>IS: Start Aggregation Activity
    IS->>PG: Query Billable Events (1 hour window)
    IS->>IS: Group by customer_id + meter_id
    IS->>PG: Mark events with batch_id
    
    %% Stripe Sync
    TW->>IS: Start Stripe Sync Activity
    loop For Each Aggregated Batch
        IS->>PG: Lookup Customer Mapping
        IS->>ST: POST /v1/billing/meter_events
        ST-->>IS: meter_event_id
        IS->>PG: Update sync_status = 'synced'
    end

1.2 Customer Onboarding Sequence

sequenceDiagram
    participant C as Customer
    participant ST as Stripe
    participant CA as Customer App
    participant FP as FlexPrice
    participant WH as Webhook Handler
    participant IS as Integration Service

    Note over C,IS: Customer Onboarding & Mapping Flow

    %% Phase 1: Stripe Setup
    C->>ST: Create Customer
    ST-->>C: stripe_customer_id: "cus_abc123"
    C->>ST: Create Subscription
    ST-->>C: subscription_id: "sub_def456"
    
    %% Phase 2: FlexPrice Setup
    C->>FP: Create Customer (external_id: "client_789")
    FP-->>C: flexprice_customer_id: "fp_ghi123"
    
    %% Phase 3: Integration Mapping
    C->>WH: POST /webhook/customer-sync
    Note right of WH: Payload:<br/>{<br/>  "stripe_customer_id": "cus_abc123",<br/>  "flexprice_customer_id": "fp_ghi123",<br/>  "external_customer_id": "client_789"<br/>}
    
    WH->>IS: Create Customer Mapping
    IS->>IS: Validate IDs
    IS->>PG: Insert customer_integration_mapping
    IS-->>WH: Mapping Created
    WH-->>C: Success
    
    %% Phase 4: Usage Events Can Now Flow
    Note over C: Ready to send usage events<br/>with external_customer_id: "client_789"

1.3 Sync Decision Processing Sequence

sequenceDiagram
    participant TW as Temporal Worker
    participant IS as Integration Service
    participant PG as PostgreSQL
    participant CH as ClickHouse

    Note over TW,CH: Sync Decision Processing for Unprocessed Events

    TW->>IS: ProcessSyncDecisionsActivity(windowStart, windowEnd)
    
    %% Query Unprocessed Events
    IS->>PG: Query events needing sync decision
    Note right of PG: SELECT e.*, ep.qty_billable, ep.qty_free<br/>FROM events e<br/>JOIN events_processed ep ON e.id = ep.event_id<br/>LEFT JOIN integration_sync_state iss ON e.id = iss.event_id<br/>WHERE iss.id IS NULL
    
    PG-->>IS: List of UnsyncedEvent objects
    
    loop For Each Event
        IS->>IS: Analyze Event Processing Results
        
        alt qty_billable > 0
            Note over IS: Event has billable usage after free tier/credits
            IS->>PG: Check Provider Config for meter_id
            alt Meter configured for Stripe sync
                IS->>PG: CREATE integration_sync_state<br/>(status='billable', should_sync=true)
            else
                IS->>PG: CREATE integration_sync_state<br/>(status='skipped_config', skip_reason='meter_not_configured')
            end
        else qty_free > 0
            Note over IS: Event covered by free tier
            IS->>PG: CREATE integration_sync_state<br/>(status='skipped_free', skip_reason='free_tier_applied')
        else
            Note over IS: Event covered by credits
            IS->>PG: CREATE integration_sync_state<br/>(status='skipped_credited', skip_reason='credit_applied')
        end
    end
    
    IS-->>TW: SyncDecisionResult{EventsProcessed: count}

1.4 Event Aggregation Sequence

sequenceDiagram
    participant TW as Temporal Worker
    participant IS as Integration Service
    participant PG as PostgreSQL

    Note over TW,PG: Event Aggregation for Batch Sync

    TW->>IS: AggregateBillableEventsActivity(windowStart, windowEnd)
    
    %% Query Billable Events
    IS->>PG: Query billable events in time window
    Note right of PG: SELECT * FROM integration_sync_state iss<br/>JOIN events e ON iss.event_id = e.id<br/>WHERE sync_status = 'billable'<br/>AND aggregation_window_start BETWEEN ? AND ?<br/>AND batch_id IS NULL
    
    PG-->>IS: List of billable sync states
    
    %% Group and Aggregate
    IS->>IS: Group by (customer_id, meter_id, provider_meter_id)
    
    loop For Each Customer+Meter Group
        IS->>IS: Sum billable_quantity
        IS->>IS: Collect event_ids in group
        IS->>IS: Generate batch_id = uuid()
        
        %% Mark events as batched
        IS->>PG: UPDATE integration_sync_state<br/>SET batch_id = ?, updated_at = NOW()<br/>WHERE event_id IN (...)
        
        IS->>IS: Create AggregatedUsage object
        Note right of IS: {<br/>  customer_id,<br/>  provider_customer_id,<br/>  meter_id,<br/>  provider_meter_id,<br/>  total_usage: sum(billable_quantity),<br/>  event_ids: [...]<br/>}
    end
    
    IS-->>TW: AggregationResult{AggregatedUsage: [...], TotalEvents: count}

1.5 Stripe API Sync Sequence

sequenceDiagram
    participant TW as Temporal Worker
    participant IS as Integration Service
    participant PG as PostgreSQL
    participant SC as Stripe Client
    participant ST as Stripe API

    Note over TW,ST: Stripe API Synchronization

    TW->>IS: SyncToStripeActivity(aggregationResult)
    
    loop For Each AggregatedUsage
        %% Prepare Stripe Request
        IS->>PG: Lookup Customer Mapping
        PG-->>IS: provider_customer_id
        
        IS->>IS: Build Stripe Meter Event Request
        Note right of IS: event_name: meter_name<br/>payload.stripe_customer_id: provider_customer_id<br/>payload.value: total_usage
        
        %% Make Stripe API Call
        IS->>SC: CreateMeterEvent(request)
        SC->>ST: POST /v1/billing/meter_events
        ST-->>SC: meter_event_id: mtr_evt_abc123
        SC-->>IS: stripe_event_id
        
        %% Update Sync State
        IS->>PG: UPDATE integration_sync_state SET sync_status = synced
        Note right of PG: Updates all event_ids in batch<br/>with provider_event_id and timestamps
    end
    
    IS-->>TW: StripeSyncActivityResult{SyncedCount: count}

1.6 Error Handling & Retry Sequence

sequenceDiagram
    participant TW as Temporal Worker
    participant IS as Integration Service
    participant PG as PostgreSQL
    participant SC as Stripe Client
    participant ST as Stripe API
    participant RW as Retry Workflow

    Note over TW,RW: Error Handling & Retry Flow

    TW->>IS: SyncToStripeActivity(aggregationResult)
    IS->>SC: CreateMeterEvent(request)
    SC->>ST: POST /v1/billing/meter_events
    
    alt Stripe API Error
        ST-->>SC: 429 Rate Limit Error
        SC-->>IS: RateLimitError
        
        IS->>IS: Calculate exponential backoff delay
        IS->>PG: UPDATE integration_sync_state<br/>SET sync_status = 'retrying',<br/>sync_attempt_count = sync_attempt_count + 1,<br/>next_retry_at = NOW() + delay,<br/>error_message = 'Rate limit exceeded'
        
        IS-->>TW: RetryableError
        TW->>TW: Schedule Retry Workflow
        TW->>RW: Start RetryWorkflow(eventIds, delay)
        
        %% Retry Workflow
        RW->>RW: Sleep(delay)
        RW->>IS: RetryFailedEventsActivity(eventIds)
        IS->>SC: CreateMeterEvent(request) [Retry]
        
        alt Retry Success
            SC-->>IS: stripe_event_id
            IS->>PG: UPDATE sync_status = 'synced'
        else Retry Fails Again
            IS->>PG: UPDATE sync_attempt_count++, next_retry_at
            alt Max Attempts Reached
                IS->>PG: UPDATE sync_status = 'failed'
                IS->>IS: Send Alert to Operations
            end
        end
    end

2. State Machine Definitions

2.1 Integration Sync State Machine

stateDiagram-v2
    [*] --> Pending: Event received
    
    Pending --> Billable: qty_billable > 0 AND meter_configured
    Pending --> SkippedFree: qty_billable = 0 AND qty_free > 0
    Pending --> SkippedCredited: qty_billable = 0 AND qty_free = 0
    Pending --> SkippedConfig: qty_billable > 0 AND meter_not_configured
    
    Billable --> Syncing: Added to batch
    Syncing --> Synced: Stripe API success
    Syncing --> Failed: Stripe API error
    
    Failed --> Retrying: retry_attempts < max_attempts
    Retrying --> Synced: Retry success
    Retrying --> Failed: Retry failed
    Failed --> ManualReview: retry_attempts >= max_attempts
    
    SkippedFree --> [*]
    SkippedCredited --> [*]
    SkippedConfig --> [*]
    Synced --> [*]
    ManualReview --> [*]
    
    state Failed {
        [*] --> Attempt1: Initial failure
        Attempt1 --> Attempt2: delay=1min
        Attempt2 --> Attempt3: delay=2min  
        Attempt3 --> Attempt4: delay=4min
        Attempt4 --> Attempt5: delay=8min
        Attempt5 --> MaxAttemptsReached: delay=16min
    }

2.2 Customer Mapping State Machine

stateDiagram-v2
    [*] --> PendingMapping: Customer created in provider
    
    PendingMapping --> Active: Mapping created successfully
    PendingMapping --> ValidationFailed: Invalid customer IDs
    
    ValidationFailed --> PendingMapping: Retry with corrected data
    ValidationFailed --> Rejected: Permanent validation failure
    
    Active --> Suspended: Temporary sync disable
    Active --> Updated: Mapping data changed
    Active --> Deleted: Customer relationship ended
    
    Suspended --> Active: Re-enable sync
    Updated --> Active: Changes applied
    
    Deleted --> [*]
    Rejected --> [*]
    
    state Active {
        [*] --> Syncing
        Syncing --> Healthy: All events syncing successfully
        Healthy --> Degraded: Some sync failures
        Degraded --> Healthy: Error resolved
        Degraded --> Critical: High failure rate
        Critical --> Healthy: Recovery completed
    }

2.3 Temporal Workflow State Machine

stateDiagram-v2
    [*] --> Scheduled: Cron trigger
    
    Scheduled --> DecisionProcessing: Start sync workflow
    DecisionProcessing --> Aggregating: Sync decisions complete
    DecisionProcessing --> Failed: Decision processing error
    
    Aggregating --> Syncing: Events aggregated
    Aggregating --> Completed: No billable events
    Aggregating --> Failed: Aggregation error
    
    Syncing --> Completed: All batches synced
    Syncing --> PartialSuccess: Some batches failed
    Syncing --> Failed: All batches failed
    
    PartialSuccess --> RetryScheduled: Schedule retry for failed batches
    Failed --> RetryScheduled: Schedule retry workflow
    
    RetryScheduled --> Retrying: Retry delay elapsed
    Retrying --> Completed: Retry successful
    Retrying --> Failed: Retry failed
    
    Failed --> AlertGenerated: Max retries exceeded
    AlertGenerated --> [*]
    Completed --> [*]
    
    state DecisionProcessing {
        [*] --> QueryingEvents
        QueryingEvents --> AnalyzingBilling: Events retrieved
        AnalyzingBilling --> CreatingSyncStates: Decisions made
        CreatingSyncStates --> [*]: States persisted
    }
    
    state Syncing {
        [*] --> BatchProcessing
        BatchProcessing --> StripeAPICalls: Batches prepared
        StripeAPICalls --> UpdatingStates: API calls complete
        UpdatingStates --> [*]: States updated
    }

2.4 Provider Configuration State Machine

stateDiagram-v2
    [*] --> Draft: Configuration created
    
    Draft --> Validating: Submit for validation
    Validating --> Active: Validation passed
    Validating --> Invalid: Validation failed
    
    Invalid --> Draft: Fix configuration
    Invalid --> Rejected: Permanent failure
    
    Active --> Paused: Temporarily disable sync
    Active --> Updating: Modify configuration
    Active --> Deleted: Remove configuration
    
    Paused --> Active: Resume sync
    Updating --> Validating: Validate changes
    
    Deleted --> [*]
    Rejected --> [*]
    
    state Active {
        entry / enable_meter_sync
        exit / disable_meter_sync
        
        [*] --> Monitoring
        Monitoring --> Healthy: Sync working normally
        Healthy --> Warning: Sync performance degraded
        Warning --> Healthy: Performance recovered
        Warning --> Critical: Sync failures increasing
        Critical --> Healthy: Issues resolved
        
        state Critical {
            entry / alert_high_failure_rate
            do / throttle_sync_attempts
        }
    }
    
    state Validating {
        [*] --> CheckingMeterExists
        CheckingMeterExists --> CheckingProviderMeter: FlexPrice meter valid
        CheckingProviderMeter --> CheckingFieldMappings: Provider meter valid
        CheckingFieldMappings --> [*]: All validations passed
        
        CheckingMeterExists --> [*]: Meter not found
        CheckingProviderMeter --> [*]: Provider meter invalid
        CheckingFieldMappings --> [*]: Field mapping invalid
    }

2.5 Batch Processing State Machine

stateDiagram-v2
    [*] --> Collecting: Gather billable events
    
    Collecting --> Grouping: Events collected for window
    Collecting --> Empty: No billable events found
    
    Grouping --> Batched: Events grouped by customer+meter
    Batched --> Submitting: Submit batches to Stripe
    
    Submitting --> AllSuccessful: All API calls succeeded
    Submitting --> PartialFailure: Some API calls failed
    Submitting --> AllFailed: All API calls failed
    
    AllSuccessful --> Completed: Mark events as synced
    PartialFailure --> RetryQueued: Queue failed batches for retry
    AllFailed --> RetryQueued: Queue all batches for retry
    
    RetryQueued --> [*]: Retry workflow scheduled
    Empty --> [*]
    Completed --> [*]
    
    state Grouping {
        [*] --> SortingByCustomer
        SortingByCustomer --> SortingByMeter: Customers grouped
        SortingByMeter --> AggregatingUsage: Meters grouped
        AggregatingUsage --> GeneratingBatchIDs: Usage summed
        GeneratingBatchIDs --> [*]: Batch IDs assigned
    }
    
    state Submitting {
        [*] --> PreparingRequests
        PreparingRequests --> CallingStripeAPI: Requests formatted
        CallingStripeAPI --> ProcessingResponses: API calls made
        ProcessingResponses --> [*]: Responses handled
    }

3. State Transition Rules

3.1 Sync Status Transitions

From State	To State	Trigger	Condition	Action
`pending`	`billable`	Billing complete	`qty_billable > 0 AND meter_configured`	Set aggregation window
`pending`	`skipped_free`	Billing complete	`qty_billable = 0 AND qty_free > 0`	Log skip reason
`pending`	`skipped_credited`	Billing complete	`qty_billable = 0 AND qty_free = 0`	Log skip reason
`billable`	`syncing`	Batch created	Event added to batch	Update batch_id
`syncing`	`synced`	Stripe success	API call succeeds	Store provider_event_id
`syncing`	`failed`	Stripe error	API call fails	Increment attempt_count
`failed`	`retrying`	Retry trigger	`attempt_count < max_attempts`	Schedule next_retry_at
`retrying`	`synced`	Retry success	Retry API succeeds	Clear error fields
`retrying`	`failed`	Retry failure	Retry API fails	Increment attempt_count

3.2 Workflow Transition Rules

From State	To State	Trigger	Condition	Action
`scheduled`	`decision_processing`	Workflow start	Cron trigger fires	Query unprocessed events
`decision_processing`	`aggregating`	Processing complete	All decisions made	Query billable events
`aggregating`	`syncing`	Aggregation complete	Batches created	Start Stripe sync
`syncing`	`completed`	Sync complete	All batches processed	Update final states
Any state	`failed`	Error occurred	Unrecoverable error	Schedule retry workflow

3.3 Error Recovery Rules

Error Type	Recovery Action	Max Retries	Backoff Strategy
Rate Limit (429)	Exponential backoff	5	1min, 2min, 4min, 8min, 16min
Network Timeout	Linear backoff	3	30sec, 1min, 2min
Auth Error (401)	Immediate failure	0	Alert operations
Invalid Data (400)	Immediate failure	0	Mark as permanently failed
Server Error (5xx)	Exponential backoff	3	1min, 4min, 16min

0 replies

BRO3886 · 2025-06-10T20:39:58Z

BRO3886
Jun 10, 2025
Author

Customer Integration and Sync Design

FlexPrice Stripe Integration - Customer Management

Overview

This document outlines the design for customer integration and synchronization between FlexPrice and external billing providers (starting with Stripe). The solution enables FlexPrice to work as an enhancement layer for existing Stripe billing infrastructure while maintaining proper customer identity mapping.

Problem Statement

Use Case

Client already has customers in Stripe
Client wants to use FlexPrice for advanced usage-based billing
Client wants to maintain existing Stripe subscription/payment infrastructure
Need to synchronize customer data between systems

Key Requirements

Bulk import existing Stripe customers into FlexPrice
Real-time sync for new customers created in Stripe
Three-way ID mapping: FlexPrice ID ↔ Client Internal ID ↔ Stripe Customer ID
Support for future provider integrations (HubSpot, RazorPay, ChargeBee)

Database Design

Customer Integration Mapping Table

CREATE TABLE customer_integration_mappings (
    customer_id UUID NOT NULL,
    provider_type VARCHAR(50) NOT NULL,
    provider_customer_id VARCHAR(255) NOT NULL,
    tenant_id UUID NOT NULL,
    environment_id UUID NOT NULL,
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    
    -- Composite primary key (no artificial UUID)
    PRIMARY KEY (customer_id, provider_type, tenant_id, environment_id),
    
    -- Foreign key to customers table
    CONSTRAINT fk_customer FOREIGN KEY (customer_id) REFERENCES customers(id) ON DELETE CASCADE,
    
    -- Unique constraint for reverse lookup
    CONSTRAINT uk_provider_customer UNIQUE (provider_type, provider_customer_id, tenant_id, environment_id)
);

-- Index for provider-to-customer lookup
CREATE INDEX idx_provider_customer_lookup ON customer_integration_mappings(provider_type, provider_customer_id, tenant_id, environment_id);

Ent Schema (Go)

// ent/schema/customerintegrationmapping.go
func (CustomerIntegrationMapping) Fields() []ent.Field {
    return []ent.Field{
        field.UUID("customer_id", uuid.UUID{}),
        field.String("provider_type").MaxLen(50),
        field.String("provider_customer_id").MaxLen(255),
        field.UUID("tenant_id", uuid.UUID{}),
        field.UUID("environment_id", uuid.UUID{}),
        field.JSON("metadata", map[string]interface{}{}).Optional(),
        field.Time("created_at").Default(time.Now),
        field.Time("updated_at").Default(time.Now).UpdateDefault(time.Now),
    }
}

func (CustomerIntegrationMapping) Indexes() []ent.Index {
    return []ent.Index{
        index.Fields("customer_id", "provider_type", "tenant_id", "environment_id").Unique(),
        index.Fields("provider_type", "provider_customer_id", "tenant_id", "environment_id").Unique(),
    }
}

Key Design Decisions

Composite Primary Key: Uses natural business key instead of artificial UUID
Provider Type Enum: Supports multiple providers (stripe, hubspot, razorpay, chargebee)
Metadata Field: JSONB for provider-specific data storage
Multi-tenant: Includes tenant_id and environment_id for proper isolation

ID Mapping Strategy

Three-Layer Mapping

FlexPrice Customer ID ↔ Client Internal ID ↔ Provider Customer ID
      (UUID)               (external_id)        (cus_xyz)

Implementation

FlexPrice Customer.external_id = Client's internal customer ID
CustomerIntegrationMapping links FlexPrice customer to provider customer IDs
One-to-many relationship: One FlexPrice customer can map to multiple providers

Lookup Patterns

-- Find customer by client internal ID
SELECT * FROM customers WHERE external_id = '123' AND tenant_id = ?

-- Find provider mappings for customer
SELECT * FROM customer_integration_mappings WHERE customer_id = ? AND tenant_id = ?

-- Find customer by provider ID
SELECT c.* FROM customers c 
JOIN customer_integration_mappings cim ON c.id = cim.customer_id 
WHERE cim.provider_type = 'stripe' 
AND cim.provider_customer_id = 'cus_xyz' 
AND cim.tenant_id = ?

Customer Sync Approaches

Evaluated Options

1. API Pull + Polling ❌

Approach: Periodically poll Stripe API to fetch new customers

Problems:

Race Condition: Customer created in Stripe → Subscription setup → Usage events sent → Customer not yet pulled
Complex Backfilling: Need to handle orphaned events and retroactive customer linking
Timing Uncertainty: Polling frequency vs real-time needs

Verdict: Rejected due to race condition complexity

2. Just-In-Time (JIT) Fallback ❌

Approach: Fetch customer from Stripe when usage event arrives for unknown customer

Problems:

Usage events contain client's external_id (e.g., "123")
Stripe API requires Stripe customer ID (e.g., "cus_xyz")
No way to reverse-lookup Stripe customer ID from client's internal ID

Verdict: Impossible due to ID mapping limitations

3. Webhook-Driven Sync ✅

Approach: Client triggers FlexPrice sync after creating customer in Stripe

Benefits:

No Race Conditions: Client controls timing
Simple Logic: Deterministic order of operations
No Backfilling: Customer always exists before events
Leverages Existing Infrastructure: Uses FlexPrice's integration system

Verdict: Recommended approach

Recommended Implementation

Bulk Import (One-time Migration)

POST /integrations/stripe/customers/bulk-import
Authorization: Bearer {api_key}

Implementation:

Use existing Stripe integration credentials from FlexPrice
Fetch all customers via Stripe API (/v1/customers)
For each Stripe customer:
- Create FlexPrice customer with external_id = client's internal customer ID
- Create integration mapping entry linking FlexPrice customer to Stripe customer ID
- Include address data for invoice generation

Real-time Sync (Ongoing)

POST /integrations/stripe/customers/sync
Content-Type: application/json

{
  "external_id": "123",           // Client's internal customer ID
  "stripe_customer_id": "cus_abc" // Stripe customer ID
}

Required Client Flow:

Client creates customer in Stripe → receives cus_xyz
Client MUST call FlexPrice sync webhook before sending usage events
FlexPrice creates customer + integration mapping
Only then client can send usage events with external_id

API Endpoints

// Bulk import existing customers
POST /integrations/stripe/customers/bulk-import

// Sync individual customer
POST /integrations/stripe/customers/sync

// Get customer integrations
GET /customers/{id}/integrations

// Add integration mapping
POST /customers/{id}/integrations

// Find customer by provider ID
GET /customers/by-provider/{provider}/{provider_id}

Critical Requirements

Mandatory Webhook Ordering

Customer sync webhook must be called before sending usage events
No fallback mechanisms available due to ID mapping constraints
Client responsibility to maintain proper ordering

Error Handling

Duplicate customer detection and resolution
Failed Stripe API calls during bulk import
Webhook retry mechanisms for sync failures

Data Requirements

Customer basic info: name, email, external_id
Address data for invoice generation
Stripe customer ID for integration mapping

Future Extensibility

Multi-Provider Support

The schema supports additional providers by changing provider_type:

stripe → Stripe customer ID
hubspot → HubSpot contact ID
razorpay → RazorPay customer ID
chargebee → ChargeBee customer ID

Provider-Specific Metadata

Use metadata JSONB field for provider-specific data:

{
  "stripe": {
    "subscription_id": "sub_xyz",
    "payment_method": "card_abc"
  }
}

Implementation Timeline

MVP (Week 1)

Database schema creation
Basic sync webhook endpoint
Integration with existing FlexPrice integration system
Bulk import endpoint

Future Enhancements

Additional provider support
Advanced error handling and retry logic
Webhook event tracking and monitoring
Customer data synchronization beyond basic info

0 replies

BRO3886 · 2025-06-10T20:42:17Z

BRO3886
Jun 10, 2025
Author

PRD: Stripe Usage Sync Integration

Overview

Implement a bidirectional usage event synchronization system between FlexPrice and Stripe, enabling customers to maintain existing Stripe billing infrastructure while leveraging FlexPrice's advanced usage-based billing capabilities. The system will sync only billable usage events (excluding free tier and credited usage) using Temporal workflows for reliability and batch aggregation to avoid rate limits.

Architecture

Core Components

Temporal Workflows: Orchestrate sync operations with retry logic and error handling
Integration Sync State: Track sync decisions and status for each event
Customer Mapping: Map FlexPrice customers to Stripe customers
Batch Aggregation: Aggregate events by time windows to avoid Stripe rate limits
Provider Configuration: Configure which meters sync to which external providers

Database Schema Implementation

Task 1: Create Integration Sync State Ent Schema

File Path: /ent/schema/integration_sync_state.go

Create Ent schema for integration sync state:

package schema

import (
    "entgo.io/ent"
    "entgo.io/ent/schema/edge"
    "entgo.io/ent/schema/field"
    "entgo.io/ent/schema/index"
    "github.com/shopspring/decimal"
)

type IntegrationSyncState struct {
    ent.Schema
}

func (IntegrationSyncState) Fields() []ent.Field {
    return []ent.Field{
        field.String("id").Unique(),
        field.String("event_id"),
        field.String("tenant_id"),
        field.String("environment_id"),
        field.String("provider_type"),
        field.String("sync_status"),
        field.Bool("should_sync").Default(false),
        field.String("skip_reason").Optional(),
        field.String("subscription_id").Optional(),
        field.String("meter_id").Optional(),
        field.Other("billable_quantity", decimal.Decimal{}).SchemaType(map[string]string{
            dialect.Postgres: "numeric",
        }).Optional(),
        field.Other("free_quantity", decimal.Decimal{}).SchemaType(map[string]string{
            dialect.Postgres: "numeric",
        }).Optional(),
        field.Other("total_quantity", decimal.Decimal{}).SchemaType(map[string]string{
            dialect.Postgres: "numeric",
        }).Optional(),
        field.String("provider_event_id").Optional(),
        field.String("provider_meter_id").Optional(),
        field.String("provider_customer_id").Optional(),
        field.JSON("sync_payload", map[string]interface{}{}).Optional(),
        field.Int("sync_attempt_count").Default(0),
        field.Time("last_sync_attempt").Optional(),
        field.Time("next_retry_at").Optional(),
        field.String("error_message").Optional(),
        field.String("error_code").Optional(),
        field.String("batch_id").Optional(),
        field.Time("aggregation_window_start").Optional(),
        field.Time("aggregation_window_end").Optional(),
        field.Time("created_at").Default(time.Now),
        field.Time("updated_at").Default(time.Now).UpdateDefault(time.Now),
    }
}

func (IntegrationSyncState) Indexes() []ent.Index {
    return []ent.Index{
        index.Fields("event_id", "provider_type", "tenant_id", "environment_id").Unique(),
        index.Fields("tenant_id", "environment_id", "provider_type", "sync_status"),
        index.Fields("next_retry_at", "sync_status"),
        index.Fields("batch_id", "provider_type"),
    }
}

Task 2: Create Provider Config Ent Schema

File Path: /ent/schema/integration_provider_config.go

Create Ent schema for provider configurations:

package schema

import (
    "entgo.io/ent"
    "entgo.io/ent/schema/field"
    "entgo.io/ent/schema/index"
)

type IntegrationProviderConfig struct {
    ent.Schema
}

func (IntegrationProviderConfig) Fields() []ent.Field {
    return []ent.Field{
        field.String("id").Unique(),
        field.String("tenant_id"),
        field.String("environment_id"),
        field.String("provider_type"),
        field.String("meter_id"),
        field.String("provider_meter_id"),
        field.Bool("sync_enabled").Default(true),
        field.String("sync_frequency").Default("1h"),
        field.String("aggregation_window").Default("1h"),
        field.JSON("field_mappings", map[string]interface{}{}).Optional(),
        field.Time("created_at").Default(time.Now),
        field.Time("updated_at").Default(time.Now).UpdateDefault(time.Now),
    }
}

func (IntegrationProviderConfig) Indexes() []ent.Index {
    return []ent.Index{
        index.Fields("tenant_id", "environment_id", "provider_type", "meter_id").Unique(),
    }
}

Task 3: Create Customer Integration Mapping Ent Schema

File Path: /ent/schema/customer_integration_mapping.go

Create Ent schema for customer mappings:

package schema

import (
    "entgo.io/ent"
    "entgo.io/ent/schema/field"
    "entgo.io/ent/schema/index"
)

type CustomerIntegrationMapping struct {
    ent.Schema
}

func (CustomerIntegrationMapping) Fields() []ent.Field {
    return []ent.Field{
        field.String("customer_id"),
        field.String("provider_type"),
        field.String("provider_customer_id"),
        field.String("tenant_id"),
        field.String("environment_id"),
        field.JSON("metadata", map[string]interface{}{}).Optional(),
        field.Time("created_at").Default(time.Now),
        field.Time("updated_at").Default(time.Now).UpdateDefault(time.Now),
    }
}

func (CustomerIntegrationMapping) Indexes() []ent.Index {
    return []ent.Index{
        index.Fields("customer_id", "provider_type", "tenant_id", "environment_id").Unique(),
    }
}

Task 4: Generate Ent Code

File Path: /Makefile

Add Ent generation command to existing Makefile:

# Add to existing Makefile
generate-ent:
	go generate ./ent

Run: make generate-ent to generate the Ent client code.

Task 3: Create Customer Integration Mapping Table

File Path: /migrations/[timestamp]_create_customer_integration_mappings.sql

Create table for mapping FlexPrice customers to external provider customers:

CREATE TABLE customer_integration_mappings (
    customer_id STRING,
    provider_type STRING,
    provider_customer_id STRING,
    tenant_id STRING,
    environment_id STRING,
    metadata JSONB,
    created_at TIMESTAMP,
    updated_at TIMESTAMP,
    PRIMARY KEY (customer_id, provider_type, tenant_id, environment_id)
);

Domain Models Implementation

Task 4: Create Integration Sync Domain Model

File Path: /internal/domain/integration/model.go

Define the domain model for integration sync state with validation and business logic:

package integration

import (
    "time"
    "github.com/flexprice/flexprice/internal/types"
    "github.com/shopspring/decimal"
)

type SyncStatus string

const (
    SyncStatusBillable        SyncStatus = "billable"
    SyncStatusSkippedFree     SyncStatus = "skipped_free"
    SyncStatusSkippedCredits  SyncStatus = "skipped_credited"
    SyncStatusSkippedConfig   SyncStatus = "skipped_config"
    SyncStatusSynced          SyncStatus = "synced"
    SyncStatusFailed          SyncStatus = "failed"
    SyncStatusRetrying        SyncStatus = "retrying"
)

type IntegrationSyncState struct {
    ID              string
    EventID         string
    ProviderType    string
    SyncStatus      SyncStatus
    ShouldSync      bool
    SkipReason      string
    BillableQuantity decimal.Decimal
    // ... other fields
}

type ProviderConfig struct {
    ID               string
    MeterID          string
    ProviderMeterID  string
    SyncEnabled      bool
    SyncFrequency    string
    AggregationWindow string
    // ... other fields
}

type CustomerMapping struct {
    CustomerID         string
    ProviderType       string
    ProviderCustomerID string
    // ... other fields
}

Task 5: Create Integration Repository Interface

File Path: /internal/domain/integration/repository.go

Define repository interface for integration operations:

package integration

import "context"

type Repository interface {
    CreateSyncState(ctx context.Context, syncState *IntegrationSyncState) error
    UpdateSyncState(ctx context.Context, syncState *IntegrationSyncState) error
    GetSyncStateByEventID(ctx context.Context, eventID string, providerType string) (*IntegrationSyncState, error)
    GetBillableEventsForWindow(ctx context.Context, params *GetBillableEventsParams) ([]*IntegrationSyncState, error)
    GetEventsNeedingSyncDecision(ctx context.Context, params *GetUnsyncedEventsParams) ([]*UnsyncedEvent, error)
    
    CreateProviderConfig(ctx context.Context, config *ProviderConfig) error
    GetProviderConfig(ctx context.Context, meterID string, providerType string) (*ProviderConfig, error)
    
    CreateCustomerMapping(ctx context.Context, mapping *CustomerMapping) error
    GetCustomerMapping(ctx context.Context, customerID string, providerType string) (*CustomerMapping, error)
}

Repository Implementation

Task 6: Create Integration Repository Implementation

File Path: /internal/repository/ent/integration.go

Implement the repository interface using FlexPrice's existing Ent patterns:

package ent

import (
    "context"
    "github.com/flexprice/flexprice/internal/domain/integration"
    "github.com/flexprice/flexprice/ent"
)

type IntegrationRepository struct {
    client *ent.Client
}

func NewIntegrationRepository(client *ent.Client) integration.Repository {
    return &IntegrationRepository{client: client}
}

func (r *IntegrationRepository) GetEventsNeedingSyncDecision(ctx context.Context, params *integration.GetUnsyncedEventsParams) ([]*integration.UnsyncedEvent, error) {
    // Query events that exist in events table, have been processed by FlexPrice,
    // but don't have sync decision yet
    query := `
        SELECT e.id, e.event_name, e.external_customer_id, ep.qty_billable, ep.qty_free_applied
        FROM events e
        JOIN events_processed ep ON e.id = ep.id 
        LEFT JOIN integration_sync_state iss ON e.id = iss.event_id 
            AND iss.provider_type = ? AND iss.tenant_id = e.tenant_id
        WHERE e.tenant_id = ? AND e.environment_id = ?
            AND e.timestamp BETWEEN ? AND ?
            AND iss.id IS NULL
        ORDER BY e.timestamp ASC LIMIT ?
    `
    // Implementation...
}

func (r *IntegrationRepository) GetBillableEventsForWindow(ctx context.Context, params *integration.GetBillableEventsParams) ([]*integration.IntegrationSyncState, error) {
    // Query billable events ready for sync within aggregation window
    query := `
        SELECT iss.*, e.event_name, e.properties
        FROM integration_sync_state iss
        JOIN events e ON iss.event_id = e.id
        WHERE iss.tenant_id = ? AND iss.environment_id = ?
            AND iss.provider_type = ? AND iss.sync_status = 'billable'
            AND iss.aggregation_window_start BETWEEN ? AND ?
            AND iss.batch_id IS NULL
        ORDER BY e.timestamp ASC
    `
    // Implementation...
}

Temporal Workflow Implementation

Task 7: Create Stripe Sync Workflow

File Path: /internal/temporal/workflows/stripe_sync_workflow.go

Implement the main workflow that orchestrates the sync process using FlexPrice's existing Temporal patterns:

package workflows

import (
    "time"
    "go.temporal.io/sdk/workflow"
    "github.com/flexprice/flexprice/internal/temporal/models"
)

func StripeSyncWorkflow(ctx workflow.Context, input models.StripeSyncInput) (*models.StripeSyncResult, error) {
    logger := workflow.GetLogger(ctx)
    logger.Info("Starting Stripe sync workflow", "windowStart", input.WindowStart, "windowEnd", input.WindowEnd)

    activityOptions := workflow.ActivityOptions{
        StartToCloseTimeout: time.Minute * 5,
        RetryPolicy: &temporalsdk.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    time.Minute * 2,
            MaximumAttempts:    3,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, activityOptions)

    // Step 1: Process sync decisions for unprocessed events
    var decisionResult models.SyncDecisionResult
    err := workflow.ExecuteActivity(ctx, ProcessSyncDecisionsActivity, input).Get(ctx, &decisionResult)
    if err != nil {
        logger.Error("Failed to process sync decisions", "error", err)
        return nil, err
    }

    // Step 2: Aggregate billable events for sync
    var aggregationResult models.AggregationResult
    err = workflow.ExecuteActivity(ctx, AggregateBillableEventsActivity, input).Get(ctx, &aggregationResult)
    if err != nil {
        logger.Error("Failed to aggregate billable events", "error", err)
        return nil, err
    }

    // Step 3: Sync aggregated usage to Stripe
    if len(aggregationResult.AggregatedUsage) > 0 {
        var syncResult models.StripeSyncActivityResult
        err = workflow.ExecuteActivity(ctx, SyncToStripeActivity, aggregationResult).Get(ctx, &syncResult)
        if err != nil {
            logger.Error("Failed to sync to Stripe", "error", err)
            return nil, err
        }
    }

    return &models.StripeSyncResult{
        EventsProcessed: decisionResult.EventsProcessed,
        EventsSynced:    aggregationResult.TotalEvents,
        BatchesSynced:   len(aggregationResult.AggregatedUsage),
    }, nil
}

func CronStripeSyncWorkflow(ctx workflow.Context, input models.CronSyncInput) error {
    logger := workflow.GetLogger(ctx)
    
    // Run sync for the last hour
    now := workflow.Now(ctx)
    windowStart := now.Add(-1 * time.Hour)
    windowEnd := now
    
    syncInput := models.StripeSyncInput{
        WindowStart: windowStart,
        WindowEnd:   windowEnd,
        ProviderType: "stripe",
    }
    
    childWorkflowOptions := workflow.ChildWorkflowOptions{
        WorkflowID: fmt.Sprintf("stripe-sync-%d", now.Unix()),
    }
    childCtx := workflow.WithChildOptions(ctx, childWorkflowOptions)
    
    var result models.StripeSyncResult
    future := workflow.ExecuteChildWorkflow(childCtx, StripeSyncWorkflow, syncInput)
    return future.Get(ctx, &result)
}

Task 8: Create Stripe Sync Activities

File Path: /internal/temporal/activities/stripe_sync_activities.go

Implement activities for sync decision processing, aggregation, and Stripe API calls:

package activities

import (
    "context"
    "fmt"
    "github.com/flexprice/flexprice/internal/domain/integration"
    "github.com/flexprice/flexprice/internal/temporal/models"
)

type StripeSyncActivities struct {
    integrationRepo integration.Repository
    stripeClient    StripeClient
}

func (a *StripeSyncActivities) ProcessSyncDecisionsActivity(ctx context.Context, input models.StripeSyncInput) (*models.SyncDecisionResult, error) {
    // Get events that need sync decision
    params := &integration.GetUnsyncedEventsParams{
        WindowStart:  input.WindowStart,
        WindowEnd:    input.WindowEnd,
        ProviderType: input.ProviderType,
        BatchSize:    1000,
    }
    
    unsyncedEvents, err := a.integrationRepo.GetEventsNeedingSyncDecision(ctx, params)
    if err != nil {
        return nil, err
    }

    var processed int
    for _, event := range unsyncedEvents {
        // Determine sync decision based on processed event data
        syncState := a.decideSyncStatus(event)
        
        // Save sync decision
        err := a.integrationRepo.CreateSyncState(ctx, syncState)
        if err != nil {
            return nil, err
        }
        processed++
    }

    return &models.SyncDecisionResult{EventsProcessed: processed}, nil
}

func (a *StripeSyncActivities) AggregateBillableEventsActivity(ctx context.Context, input models.StripeSyncInput) (*models.AggregationResult, error) {
    // Get billable events for the window
    params := &integration.GetBillableEventsParams{
        WindowStart:  input.WindowStart,
        WindowEnd:    input.WindowEnd,
        ProviderType: input.ProviderType,
    }
    
    billableEvents, err := a.integrationRepo.GetBillableEventsForWindow(ctx, params)
    if err != nil {
        return nil, err
    }

    // Aggregate by customer + meter
    aggregatedUsage := a.aggregateEventsByCustomerAndMeter(billableEvents)
    
    return &models.AggregationResult{
        AggregatedUsage: aggregatedUsage,
        TotalEvents:     len(billableEvents),
    }, nil
}

func (a *StripeSyncActivities) SyncToStripeActivity(ctx context.Context, input models.AggregationResult) (*models.StripeSyncActivityResult, error) {
    batchID := generateBatchID()
    var syncedCount int
    
    for _, usage := range input.AggregatedUsage {
        // Create Stripe meter event
        stripeEventID, err := a.stripeClient.CreateMeterEvent(ctx, &StripeEventRequest{
            EventName: usage.EventName,
            Payload: map[string]interface{}{
                "stripe_customer_id": usage.ProviderCustomerID,
                "value": usage.TotalUsage.String(),
            },
        })
        
        if err != nil {
            // Mark events as failed
            a.markEventsFailed(ctx, usage.EventIDs, err.Error())
            continue
        }
        
        // Mark events as synced
        err = a.markEventsSynced(ctx, usage.EventIDs, batchID, stripeEventID)
        if err != nil {
            return nil, err
        }
        
        syncedCount++
    }
    
    return &models.StripeSyncActivityResult{SyncedCount: syncedCount}, nil
}

func (a *StripeSyncActivities) decideSyncStatus(event *integration.UnsyncedEvent) *integration.IntegrationSyncState {
    // If billable quantity is zero (covered by free tier/credits)
    if event.QtyBillable.IsZero() {
        if event.QtyFreeApplied.GreaterThan(decimal.Zero) {
            return &integration.IntegrationSyncState{
                EventID:      event.EventID,
                SyncStatus:   integration.SyncStatusSkippedFree,
                SkipReason:   "free_tier_applied",
                ShouldSync:   false,
            }
        }
        return &integration.IntegrationSyncState{
            EventID:      event.EventID,
            SyncStatus:   integration.SyncStatusSkippedCredits,
            SkipReason:   "credit_applied",
            ShouldSync:   false,
        }
    }

    return &integration.IntegrationSyncState{
        EventID:          event.EventID,
        SyncStatus:       integration.SyncStatusBillable,
        ShouldSync:       true,
        BillableQuantity: event.QtyBillable,
        TotalQuantity:    event.QtyTotal,
        FreeQuantity:     event.QtyFreeApplied,
    }
}

Task 9: Create Temporal Models

File Path: /internal/temporal/models/stripe_types.go

Define input/output models for Temporal workflows and activities:

package models

import (
    "time"
    "github.com/shopspring/decimal"
)

type StripeSyncInput struct {
    WindowStart  time.Time
    WindowEnd    time.Time
    ProviderType string
}

type CronSyncInput struct {
    ProviderType string
}

type StripeSyncResult struct {
    EventsProcessed int
    EventsSynced    int
    BatchesSynced   int
}

type SyncDecisionResult struct {
    EventsProcessed int
}

type AggregationResult struct {
    AggregatedUsage []AggregatedUsage
    TotalEvents     int
}

type AggregatedUsage struct {
    CustomerID         string
    ProviderCustomerID string
    MeterID            string
    ProviderMeterID    string
    EventName          string
    TotalUsage         decimal.Decimal
    EventIDs           []string
}

type StripeSyncActivityResult struct {
    SyncedCount int
}

Stripe Client Implementation

Task 10: Create Stripe Client Interface

File Path: /internal/stripe/client.go

Create Stripe API client following FlexPrice's httpclient patterns:

package stripe

import (
    "context"
    "github.com/flexprice/flexprice/internal/httpclient"
)

type Client interface {
    CreateMeterEvent(ctx context.Context, req *MeterEventRequest) (string, error)
    GetMeter(ctx context.Context, meterID string) (*Meter, error)
    CreateMeter(ctx context.Context, req *CreateMeterRequest) (*Meter, error)
}

type client struct {
    httpClient httpclient.Client
    apiKey     string
    baseURL    string
}

type MeterEventRequest struct {
    EventName string                 `json:"event_name"`
    Payload   map[string]interface{} `json:"payload"`
    Timestamp int64                  `json:"timestamp,omitempty"`
}

type MeterEventResponse struct {
    ID        string `json:"id"`
    EventName string `json:"event_name"`
}

func NewClient(httpClient httpclient.Client, apiKey string) Client {
    return &client{
        httpClient: httpClient,
        apiKey:     apiKey,
        baseURL:    "https://api.stripe.com",
    }
}

func (c *client) CreateMeterEvent(ctx context.Context, req *MeterEventRequest) (string, error) {
    // Implementation using FlexPrice's httpclient patterns
    headers := map[string]string{
        "Authorization":   "Bearer " + c.apiKey,
        "Content-Type":    "application/json",
        "Stripe-Version":  "2025-04-30.preview",
    }
    
    response := &MeterEventResponse{}
    err := c.httpClient.Post(ctx, c.baseURL+"/v1/billing/meter_events", req, response, headers)
    if err != nil {
        return "", err
    }
    
    return response.ID, nil
}

Service Layer Implementation

Task 11: Create Integration Service

File Path: /internal/service/integration.go

Create service layer for integration operations following FlexPrice's service patterns:

package service

import (
    "context"
    "github.com/flexprice/flexprice/internal/domain/integration"
    "github.com/flexprice/flexprice/internal/temporal"
)

type IntegrationService struct {
    integrationRepo integration.Repository
    temporalClient  temporal.Client
}

func NewIntegrationService(
    integrationRepo integration.Repository,
    temporalClient temporal.Client,
) *IntegrationService {
    return &IntegrationService{
        integrationRepo: integrationRepo,
        temporalClient:  temporalClient,
    }
}

func (s *IntegrationService) CreateCustomerMapping(ctx context.Context, req *CreateCustomerMappingRequest) error {
    mapping := &integration.CustomerMapping{
        CustomerID:         req.CustomerID,
        ProviderType:       req.ProviderType,
        ProviderCustomerID: req.ProviderCustomerID,
        TenantID:           types.GetTenantID(ctx),
        EnvironmentID:      types.GetEnvironmentID(ctx),
    }
    
    return s.integrationRepo.CreateCustomerMapping(ctx, mapping)
}

func (s *IntegrationService) TriggerSync(ctx context.Context, req *TriggerSyncRequest) error {
    workflowInput := models.StripeSyncInput{
        WindowStart:  req.WindowStart,
        WindowEnd:    req.WindowEnd,
        ProviderType: req.ProviderType,
    }
    
    return s.temporalClient.ExecuteWorkflow(ctx, "stripe-sync-manual", workflowInput)
}

API Implementation

Task 12: Create Integration API Endpoints

File Path: /internal/api/v1/integration.go

Implement REST API endpoints for integration management:

package v1

import (
    "net/http"
    "github.com/gin-gonic/gin"
    "github.com/flexprice/flexprice/internal/service"
)

type IntegrationHandler struct {
    integrationService *service.IntegrationService
}

func NewIntegrationHandler(integrationService *service.IntegrationService) *IntegrationHandler {
    return &IntegrationHandler{integrationService: integrationService}
}

func (h *IntegrationHandler) CreateCustomerMapping(c *gin.Context) {
    var req service.CreateCustomerMappingRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }

    err := h.integrationService.CreateCustomerMapping(c.Request.Context(), &req)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }

    c.JSON(http.StatusCreated, gin.H{"status": "created"})
}

func (h *IntegrationHandler) TriggerSync(c *gin.Context) {
    var req service.TriggerSyncRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }

    err := h.integrationService.TriggerSync(c.Request.Context(), &req)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{"status": "triggered"})
}

Task 13: Register API Routes

File Path: /internal/api/router.go

Add integration routes to the existing router:

// Add to existing router.go RegisterRoutes function
func RegisterRoutes(r *gin.Engine, handlers *Handlers) {
    // ... existing routes
    
    // Integration routes
    integrationRoutes := v1Group.Group("/integrations")
    {
        integrationRoutes.POST("/customer-mappings", handlers.Integration.CreateCustomerMapping)
        integrationRoutes.POST("/sync/trigger", handlers.Integration.TriggerSync)
        integrationRoutes.GET("/sync/status", handlers.Integration.GetSyncStatus)
    }
}

Temporal Worker Registration

Task 14: Register Workflows and Activities

File Path: /internal/temporal/registration.go

Register new workflows and activities with Temporal worker:

// Add to existing registration.go
func RegisterWorkflowsAndActivities(w worker.Worker, activities *Activities) {
    // ... existing registrations
    
    // Stripe sync workflows
    w.RegisterWorkflow(workflows.StripeSyncWorkflow)
    w.RegisterWorkflow(workflows.CronStripeSyncWorkflow)
    
    // Stripe sync activities
    w.RegisterActivity(activities.StripeSyncActivities.ProcessSyncDecisionsActivity)
    w.RegisterActivity(activities.StripeSyncActivities.AggregateBillableEventsActivity)
    w.RegisterActivity(activities.StripeSyncActivities.SyncToStripeActivity)
}

Configuration

Task 15: Add Stripe Configuration

File Path: /internal/config/config.go

Add Stripe configuration to existing config structure:

// Add to existing Configuration struct
type Configuration struct {
    // ... existing fields
    
    Stripe StripeConfig `mapstructure:"stripe"`
}

type StripeConfig struct {
    Enabled           bool          `mapstructure:"enabled"`
    APIKey            string        `mapstructure:"api_key"`
    SyncFrequency     time.Duration `mapstructure:"sync_frequency"`
    AggregationWindow time.Duration `mapstructure:"aggregation_window"`
    RetryAttempts     int           `mapstructure:"retry_attempts"`
    RateLimit         int           `mapstructure:"rate_limit"`
}

Task 16: Add Configuration Validation

File Path: /internal/config/config.go

Add validation for Stripe configuration:

func (c *Configuration) Validate() error {
    // ... existing validation
    
    if c.Stripe.Enabled {
        if c.Stripe.APIKey == "" {
            return errors.New("stripe.api_key is required when stripe is enabled")
        }
        if c.Stripe.SyncFrequency <= 0 {
            return errors.New("stripe.sync_frequency must be greater than 0")
        }
    }
    
    return nil
}

Testing Implementation

Task 17: Create Integration Tests

File Path: /internal/service/integration_test.go

Create comprehensive tests for integration service using FlexPrice's testing patterns:

package service

import (
    "testing"
    "github.com/flexprice/flexprice/internal/testutil"
    "github.com/stretchr/testify/suite"
)

type IntegrationServiceTestSuite struct {
    testutil.BaseServiceSuite
    integrationService *IntegrationService
}

func (suite *IntegrationServiceTestSuite) SetupTest() {
    // Setup using FlexPrice's existing test patterns
    suite.BaseServiceSuite.SetupTest()
    suite.integrationService = NewIntegrationService(
        suite.Store.IntegrationStore,
        suite.TemporalClient,
    )
}

func (suite *IntegrationServiceTestSuite) TestCreateCustomerMapping() {
    // Test customer mapping creation
}

func (suite *IntegrationServiceTestSuite) TestSyncDecisionLogic() {
    // Test sync decision logic (billable vs skipped events)
}

func TestIntegrationServiceTestSuite(t *testing.T) {
    suite.Run(t, new(IntegrationServiceTestSuite))
}

Task 18: Create Temporal Workflow Tests

File Path: /internal/temporal/workflows/stripe_sync_workflow_test.go

Create tests for Temporal workflows using FlexPrice's testing patterns:

package workflows

import (
    "testing"
    "go.temporal.io/sdk/testsuite"
)

func TestStripeSyncWorkflow(t *testing.T) {
    testSuite := &testsuite.WorkflowTestSuite{}
    env := testSuite.NewTestWorkflowEnvironment()
    
    // Mock activities
    env.OnActivity(ProcessSyncDecisionsActivity, mock.Anything).Return(&models.SyncDecisionResult{EventsProcessed: 10}, nil)
    env.OnActivity(AggregateBillableEventsActivity, mock.Anything).Return(&models.AggregationResult{TotalEvents: 5}, nil)
    env.OnActivity(SyncToStripeActivity, mock.Anything).Return(&models.StripeSyncActivityResult{SyncedCount: 5}, nil)
    
    // Execute workflow
    env.ExecuteWorkflow(StripeSyncWorkflow, models.StripeSyncInput{})
    
    // Verify results
    require.True(t, env.IsWorkflowCompleted())
}

Deployment

Task 19: Add Docker Configuration

File Path: /docker-compose.yml

Add Stripe configuration to existing docker-compose:

# Add to existing docker-compose.yml services section
services:
  flexprice:
    environment:
      - STRIPE_ENABLED=true
      - STRIPE_API_KEY=${STRIPE_API_KEY}
      - STRIPE_SYNC_FREQUENCY=1h
      - STRIPE_AGGREGATION_WINDOW=1h

Task 20: Update Environment Configuration

File Path: /.env.example

Add Stripe environment variables to example configuration:

# Add to existing .env.example
# Stripe Configuration
STRIPE_ENABLED=false
STRIPE_API_KEY=sk_test_your_stripe_secret_key
STRIPE_SYNC_FREQUENCY=1h
STRIPE_AGGREGATION_WINDOW=1h
STRIPE_RETRY_ATTEMPTS=3
STRIPE_RATE_LIMIT=100

Task 21: Create Webhook Handler for Customer Sync

File Path: /internal/webhook/handler/customer_sync.go

Create webhook handler for customer synchronization using existing webhook infrastructure:

package handler

import (
    "context"
    "github.com/flexprice/flexprice/internal/service"
    "github.com/flexprice/flexprice/internal/webhook/dto"
)

type CustomerSyncHandler struct {
    integrationService *service.IntegrationService
}

func NewCustomerSyncHandler(integrationService *service.IntegrationService) *CustomerSyncHandler {
    return &CustomerSyncHandler{integrationService: integrationService}
}

func (h *CustomerSyncHandler) HandleStripeCustomerCreated(ctx context.Context, payload *dto.StripeCustomerWebhook) error {
    // Create customer mapping when Stripe customer is created
    req := &service.CreateCustomerMappingRequest{
        CustomerID:         payload.FlexPriceCustomerID,
        ProviderType:       "stripe",
        ProviderCustomerID: payload.StripeCustomerID,
    }
    
    return h.integrationService.CreateCustomerMapping(ctx, req)
}

Task 22: Create Bulk Customer Import Endpoint

File Path: /internal/api/v1/customer_import.go

Create endpoint for bulk customer import:

package v1

import (
    "net/http"
    "github.com/gin-gonic/gin"
    "github.com/flexprice/flexprice/internal/service"
)

type CustomerImportHandler struct {
    integrationService *service.IntegrationService
}

func NewCustomerImportHandler(integrationService *service.IntegrationService) *CustomerImportHandler {
    return &CustomerImportHandler{integrationService: integrationService}
}

func (h *CustomerImportHandler) BulkImportStripeCustomers(c *gin.Context) {
    var req service.BulkImportRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }

    result, err := h.integrationService.BulkImportStripeCustomers(c.Request.Context(), &req)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }

    c.JSON(http.StatusOK, result)
}

Task 23: Create Retry Workflow

File Path: /internal/temporal/workflows/stripe_retry_workflow.go

Create workflow for handling failed sync retries:

package workflows

import (
    "time"
    "go.temporal.io/sdk/workflow"
    "github.com/flexprice/flexprice/internal/temporal/models"
)

func StripeRetryWorkflow(ctx workflow.Context, input models.RetryInput) error {
    logger := workflow.GetLogger(ctx)
    
    // Wait for retry delay (exponential backoff)
    err := workflow.Sleep(ctx, input.RetryDelay)
    if err != nil {
        return err
    }
    
    // Execute retry activity
    activityOptions := workflow.ActivityOptions{
        StartToCloseTimeout: time.Minute * 3,
        RetryPolicy: &temporalsdk.RetryPolicy{
            InitialInterval: time.Second * 30,
            MaximumAttempts: 1, // Single retry attempt per workflow
        },
    }
    
    ctx = workflow.WithActivityOptions(ctx, activityOptions)
    
    var result models.RetryResult
    err = workflow.ExecuteActivity(ctx, RetryFailedEventsActivity, input).Get(ctx, &result)
    if err != nil {
        logger.Error("Retry failed", "error", err, "eventIDs", input.EventIDs)
        return err
    }
    
    logger.Info("Retry completed", "retriedCount", result.RetriedCount)
    return nil
}

Task 24: Create Admin API for Monitoring

File Path: /internal/api/v1/integration_admin.go

Create admin endpoints for monitoring sync status:

package v1

import (
    "net/http"
    "github.com/gin-gonic/gin"
    "github.com/flexprice/flexprice/internal/service"
)

type IntegrationAdminHandler struct {
    integrationService *service.IntegrationService
}

func NewIntegrationAdminHandler(integrationService *service.IntegrationService) *IntegrationAdminHandler {
    return &IntegrationAdminHandler{integrationService: integrationService}
}

func (h *IntegrationAdminHandler) GetSyncStatus(c *gin.Context) {
    status, err := h.integrationService.GetSyncStatus(c.Request.Context())
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }
    
    c.JSON(http.StatusOK, status)
}

func (h *IntegrationAdminHandler) GetFailedEvents(c *gin.Context) {
    failedEvents, err := h.integrationService.GetFailedEvents(c.Request.Context())
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }
    
    c.JSON(http.StatusOK, failedEvents)
}

func (h *IntegrationAdminHandler) RetryFailedEvents(c *gin.Context) {
    var req service.RetryFailedEventsRequest
    if err := c.ShouldBindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }

    err := h.integrationService.RetryFailedEvents(c.Request.Context(), &req)
    if err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{"status": "retry_triggered"})
}

Task 25: Create Backfill Workflow

File Path: /internal/temporal/workflows/stripe_backfill_workflow.go

Create workflow for historical data backfill:

package workflows

import (
    "time"
    "go.temporal.io/sdk/workflow"
    "github.com/flexprice/flexprice/internal/temporal/models"
)

func StripeBackfillWorkflow(ctx workflow.Context, input models.BackfillInput) (*models.BackfillResult, error) {
    logger := workflow.GetLogger(ctx)
    logger.Info("Starting Stripe backfill workflow", 
        "startDate", input.StartDate, 
        "endDate", input.EndDate,
        "chunkSize", input.ChunkSize)

    var totalProcessed int
    
    // Split time range into chunks to avoid overwhelming the system
    chunks := splitTimeRange(input.StartDate, input.EndDate, input.ChunkSize)
    
    for _, chunk := range chunks {
        childWorkflowOptions := workflow.ChildWorkflowOptions{
            WorkflowID: fmt.Sprintf("stripe-backfill-chunk-%d", chunk.Start.Unix()),
        }
        childCtx := workflow.WithChildOptions(ctx, childWorkflowOptions)
        
        syncInput := models.StripeSyncInput{
            WindowStart:  chunk.Start,
            WindowEnd:    chunk.End,
            ProviderType: "stripe",
            IsBackfill:   true,
        }
        
        var chunkResult models.StripeSyncResult
        err := workflow.ExecuteChildWorkflow(childCtx, StripeSyncWorkflow, syncInput).Get(ctx, &chunkResult)
        if err != nil {
            logger.Error("Backfill chunk failed", "error", err, "chunk", chunk)
            continue // Continue with next chunk
        }
        
        totalProcessed += chunkResult.EventsProcessed
        
        // Add delay between chunks to avoid overwhelming Stripe API
        if chunk != chunks[len(chunks)-1] { // Don't wait after last chunk
            workflow.Sleep(ctx, time.Minute)
        }
    }
    
    return &models.BackfillResult{
        TotalProcessed: totalProcessed,
        ChunksProcessed: len(chunks),
    }, nil
}

Task 26: Create Event Publisher for Integration Events

File Path: /internal/publisher/integration_publisher.go

Create publisher for integration-related events using existing publisher infrastructure:

package publisher

import (
    "context"
    "github.com/flexprice/flexprice/internal/pubsub"
    "github.com/flexprice/flexprice/internal/types"
)

type IntegrationPublisher struct {
    publisher pubsub.Publisher
}

func NewIntegrationPublisher(publisher pubsub.Publisher) *IntegrationPublisher {
    return &IntegrationPublisher{publisher: publisher}
}

func (p *IntegrationPublisher) PublishSyncCompleted(ctx context.Context, event *SyncCompletedEvent) error {
    return p.publisher.Publish(ctx, &pubsub.Message{
        Topic:   "integration.sync.completed",
        Payload: event,
        Headers: map[string]string{
            "tenant_id":     types.GetTenantID(ctx),
            "provider_type": event.ProviderType,
        },
    })
}

func (p *IntegrationPublisher) PublishSyncFailed(ctx context.Context, event *SyncFailedEvent) error {
    return p.publisher.Publish(ctx, &pubsub.Message{
        Topic:   "integration.sync.failed",
        Payload: event,
        Headers: map[string]string{
            "tenant_id":     types.GetTenantID(ctx),
            "provider_type": event.ProviderType,
        },
    })
}

type SyncCompletedEvent struct {
    ProviderType    string `json:"provider_type"`
    BatchID         string `json:"batch_id"`
    EventsProcessed int    `json:"events_processed"`
    EventsSynced    int    `json:"events_synced"`
}

type SyncFailedEvent struct {
    ProviderType string   `json:"provider_type"`
    EventIDs     []string `json:"event_ids"`
    ErrorMessage string   `json:"error_message"`
}

Task 27: Create CLI Commands for Management

File Path: /cmd/stripe.go

Create CLI commands for Stripe integration management:

package cmd

import (
    "context"
    "fmt"
    "time"
    "github.com/spf13/cobra"
    "github.com/flexprice/flexprice/internal/service"
)

var stripeCmd = &cobra.Command{
    Use:   "stripe",
    Short: "Stripe integration management commands",
}

var stripeSyncCmd = &cobra.Command{
    Use:   "sync",
    Short: "Trigger Stripe sync manually",
    RunE: func(cmd *cobra.Command, args []string) error {
        // Initialize services (using existing pattern from other commands)
        integrationService := initializeIntegrationService()
        
        windowStart, _ := cmd.Flags().GetString("start")
        windowEnd, _ := cmd.Flags().GetString("end")
        
        start, err := time.Parse("2006-01-02T15:04:05Z", windowStart)
        if err != nil {
            return fmt.Errorf("invalid start time format: %w", err)
        }
        
        end, err := time.Parse("2006-01-02T15:04:05Z", windowEnd)
        if err != nil {
            return fmt.Errorf("invalid end time format: %w", err)
        }
        
        req := &service.TriggerSyncRequest{
            WindowStart:  start,
            WindowEnd:    end,
            ProviderType: "stripe",
        }
        
        err = integrationService.TriggerSync(context.Background(), req)
        if err != nil {
            return fmt.Errorf("failed to trigger sync: %w", err)
        }
        
        fmt.Println("Stripe sync triggered successfully")
        return nil
    },
}

var stripeBackfillCmd = &cobra.Command{
    Use:   "backfill",
    Short: "Backfill historical data to Stripe",
    RunE: func(cmd *cobra.Command, args []string) error {
        // Implementation for backfill command
        return nil
    },
}

func init() {
    rootCmd.AddCommand(stripeCmd)
    stripeCmd.AddCommand(stripeSyncCmd)
    stripeCmd.AddCommand(stripeBackfillCmd)
    
    stripeSyncCmd.Flags().String("start", "", "Start time (RFC3339 format)")
    stripeSyncCmd.Flags().String("end", "", "End time (RFC3339 format)")
    stripeSyncCmd.MarkFlagRequired("start")
    stripeSyncCmd.MarkFlagRequired("end")
    
    stripeBackfillCmd.Flags().String("start-date", "", "Start date for backfill")
    stripeBackfillCmd.Flags().String("end-date", "", "End date for backfill")
    stripeBackfillCmd.Flags().Duration("chunk-size", time.Hour*24, "Chunk size for backfill")
}

Task 28: Create Documentation

File Path: /docs/stripe-integration.md

Create comprehensive documentation for the Stripe integration:

# Stripe Integration Guide

## Overview
FlexPrice's Stripe integration enables bidirectional usage synchronization while maintaining existing Stripe billing infrastructure.

## Setup

### 1. Environment Configuration
```env
STRIPE_ENABLED=true
STRIPE_API_KEY=sk_live_your_stripe_secret_key
STRIPE_SYNC_FREQUENCY=1h
STRIPE_AGGREGATION_WINDOW=1h

2. Customer Mapping

Map FlexPrice customers to Stripe customers:

curl -X POST /api/v1/integrations/customer-mappings \
  -H "Content-Type: application/json" \
  -d '{
    "customer_id": "cust_flexprice_123",
    "provider_type": "stripe",
    "provider_customer_id": "cus_stripe_456"
  }'

3. Meter Configuration

Configure which FlexPrice meters sync to Stripe:

curl -X POST /api/v1/integrations/provider-configs \
  -H "Content-Type: application/json" \
  -d '{
    "meter_id": "meter_api_calls",
    "provider_type": "stripe",
    "provider_meter_id": "stripe_meter_123",
    "sync_enabled": true
  }'

Usage

Manual Sync

flexprice stripe sync --start="2024-12-10T15:00:00Z" --end="2024-12-10T16:00:00Z"

Backfill Historical Data

flexprice stripe backfill --start-date="2024-01-01" --end-date="2024-12-10"

Monitoring

Sync Status

curl /api/v1/integrations/sync/status

Failed Events

curl /api/v1/admin/integrations/failed-events


### Task 29: Create Integration Tests
**File Path**: `/tests/integration/stripe_integration_test.go`

Create end-to-end integration tests:

```go
package integration

import (
    "testing"
    "time"
    "github.com/stretchr/testify/suite"
    "github.com/flexprice/flexprice/internal/testutil"
)

type StripeIntegrationTestSuite struct {
    testutil.BaseServiceSuite
    stripeMockServer *httptest.Server
}

func (suite *StripeIntegrationTestSuite) SetupTest() {
    suite.BaseServiceSuite.SetupTest()
    suite.setupStripeMockServer()
}

func (suite *StripeIntegrationTestSuite) TestEndToEndSyncFlow() {
    ctx := suite.GetContext()
    
    // 1. Create customer mapping
    mapping := suite.createCustomerMapping()
    
    // 2. Create events in FlexPrice
    events := suite.createTestEvents()
    
    // 3. Process events through FlexPrice billing
    suite.processEvents(events)
    
    // 4. Trigger sync workflow
    suite.triggerSyncWorkflow()
    
    // 5. Verify events were synced to Stripe
    suite.verifySyncedEvents()
    
    // 6. Verify sync state is tracked correctly
    suite.verifySyncState()
}

func (suite *StripeIntegrationTestSuite) TestFreeTierEventsNotSynced() {
    // Test that events covered by free tier are not synced to Stripe
}

func (suite *StripeIntegrationTestSuite) TestRetryLogic() {
    // Test retry logic for failed sync attempts
}

func TestStripeIntegrationTestSuite(t *testing.T) {
    suite.Run(t, new(StripeIntegrationTestSuite))
}

Task 30: Create Deployment Scripts

File Path: /scripts/deploy-stripe-integration.sh

Create deployment script for Stripe integration:

#!/bin/bash
set -e

echo "Deploying Stripe Integration..."

# 1. Run database migrations
echo "Running database migrations..."
make migrate-up

# 2. Generate Ent code
echo "Generating Ent code..."
make generate-ent

# 3. Build application
echo "Building application..."
make build

# 4. Deploy to production
echo "Deploying to production..."
make deploy

# 5. Verify deployment
echo "Verifying deployment..."
curl -f http://localhost:8080/health || exit 1

echo "Stripe Integration deployment completed successfully!"

1 reply

BRO3886 Jun 10, 2025
Author

-- Table: integration_sync_state
-- Purpose: Track sync decisions and status for each event
CREATE TABLE integration_sync_state (
    id VARCHAR(36) PRIMARY KEY,
    event_id VARCHAR(36) NOT NULL,
    tenant_id VARCHAR(36) NOT NULL,
    environment_id VARCHAR(36) NOT NULL,
    provider_type VARCHAR(50) NOT NULL, -- 'stripe', 'hubspot', etc.
    sync_status VARCHAR(50) NOT NULL,   -- 'billable', 'skipped_free', 'synced', 'failed'
    should_sync BOOLEAN DEFAULT FALSE,
    skip_reason VARCHAR(255),           -- 'free_tier_applied', 'credit_applied', etc.
    subscription_id VARCHAR(36),
    meter_id VARCHAR(36),
    
    -- Usage quantities
    billable_quantity DECIMAL(20,8),
    free_quantity DECIMAL(20,8),
    total_quantity DECIMAL(20,8),
    
    -- Provider sync details
    provider_event_id VARCHAR(255),     -- Stripe meter event ID
    provider_meter_id VARCHAR(255),     -- Stripe meter ID
    provider_customer_id VARCHAR(255),  -- Stripe customer ID
    sync_payload JSONB,                 -- Complete sync request payload
    
    -- Retry management
    sync_attempt_count INTEGER DEFAULT 0,
    last_sync_attempt TIMESTAMP,
    next_retry_at TIMESTAMP,
    error_message TEXT,
    error_code VARCHAR(100),
    
    -- Batch aggregation
    batch_id VARCHAR(36),               -- Groups events in same batch
    aggregation_window_start TIMESTAMP,
    aggregation_window_end TIMESTAMP,
    
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Indexes for efficient querying
CREATE UNIQUE INDEX idx_integration_sync_event_provider 
    ON integration_sync_state(event_id, provider_type, tenant_id, environment_id);

CREATE INDEX idx_integration_sync_status 
    ON integration_sync_state(tenant_id, environment_id, provider_type, sync_status);

CREATE INDEX idx_integration_sync_retry 
    ON integration_sync_state(next_retry_at, sync_status) 
    WHERE sync_status IN ('failed', 'retrying');

CREATE INDEX idx_integration_sync_batch 
    ON integration_sync_state(batch_id, provider_type) 
    WHERE batch_id IS NOT NULL;

CREATE INDEX idx_integration_sync_window 
    ON integration_sync_state(aggregation_window_start, aggregation_window_end, sync_status)
    WHERE sync_status = 'billable';

BRO3886 · 2025-06-11T10:53:20Z

BRO3886
Jun 11, 2025
Author

FlexPrice Stripe Integration - Requirements Document

Executive Summary

This document defines the requirements for FlexPrice's Stripe integration, enabling bidirectional synchronization between FlexPrice's usage-based billing system and Stripe's billing infrastructure. The integration serves as an enhancement layer for existing Stripe customers while providing a migration pathway to FlexPrice.

Timeline: 4-5 days for MVP implementation
Approach: Simplified webhook-driven customer sync + batch event aggregation
Trade-off: MVP speed over production sophistication (with clear evolution path)

Business Requirements

Primary Objectives

Customer Acquisition: Enable FlexPrice to compete in Stripe-dominated markets
Migration Pathway: Allow gradual adoption without disrupting existing Stripe workflows
Enhanced Billing: Provide advanced usage-based billing capabilities missing in Stripe
Demo Capability: Production-viable demo within 5 days for customer presentations

Success Criteria

Functional: Customers can use FlexPrice for usage tracking while maintaining Stripe subscriptions
Performance: Handle hourly sync batches without impacting FlexPrice core performance
Reliability: 99%+ success rate for customer sync and event sync operations
Extensibility: Architecture supports future provider integrations (HubSpot, RazorPay)

Technical Requirements

Core Requirements

CR-1: Customer Synchronization

Requirement: Bidirectional customer mapping between FlexPrice and Stripe
Input: Stripe customer.created webhook with metadata
Output: FlexPrice customer with integration mapping
SLA: <5 seconds webhook response time

CR-2: Usage Event Aggregation

Requirement: Batch sync of billable usage events to Stripe
Input: FlexPrice events_processed table data
Output: Stripe meter events with aggregated quantities
SLA: Hourly sync with <15 minute processing time

CR-3: Multi-tenant Support

Requirement: Isolated Stripe configurations per FlexPrice tenant
Support: Different Stripe API keys, meter mappings per tenant
Security: Encrypted credential storage

CR-4: Error Handling & Observability

Requirement: Comprehensive error tracking and retry mechanisms
Monitoring: Sync success rates, error categorization, performance metrics
Alerting: Failed sync detection and operator notifications

Functional Requirements

FR-1: Webhook Customer Management

Given: Stripe customer.created webhook received
When: Customer has valid metadata (external_id, flexprice_customer_id)
Then: Customer is created/mapped in FlexPrice with Stripe integration

FR-2: Batch Event Processing

Given: FlexPrice has processed billable events in the last hour
When: Temporal hourly workflow executes
Then: Events are aggregated by (customer_id, meter_id) and synced to Stripe

FR-3: Idempotent Operations

Given: Same sync operation executes multiple times
When: Using stripe_event_id or batch_id for deduplication
Then: No duplicate charges or data inconsistencies occur

FR-4: Configuration Management

Given: Tenant needs to configure Stripe integration
When: Admin sets up meter mappings and API credentials
Then: Sync operates according to tenant-specific configuration

Architecture Overview

System Architecture

graph TB
    subgraph "Client Ecosystem"
        App[Client Application]
        Stripe[Stripe Dashboard]
    end
    
    subgraph "FlexPrice Core"
        API[FlexPrice API]
        Billing[Billing Engine]
        CH[(ClickHouse)]
        PG[(PostgreSQL)]
    end
    
    subgraph "Stripe Integration Layer"
        Webhook[Webhook Handler]
        Temporal[Temporal Worker]
        StripeClient[Stripe Client]
        BatchTracker[Batch Tracker]
    end
    
    subgraph "External Services"
        StripeAPI[Stripe API]
        WebhookEndpoint[Stripe Webhooks]
    end
    
    %% Customer Flow
    App --> StripeAPI
    StripeAPI --> WebhookEndpoint
    WebhookEndpoint --> Webhook
    Webhook --> PG
    
    %% Event Flow
    App --> API
    API --> Billing
    Billing --> CH
    Temporal --> CH
    Temporal --> StripeClient
    StripeClient --> StripeAPI
    Temporal --> BatchTracker
    BatchTracker --> PG
    
    %% Monitoring
    Webhook -.-> BatchTracker
    Temporal -.-> BatchTracker

Data Architecture

erDiagram
    CUSTOMERS {
        uuid id PK
        string external_id "Client internal ID"
        string tenant_id
        string environment_id
        jsonb properties
        timestamp created_at
    }
    
    CUSTOMER_INTEGRATION_MAPPINGS {
        uuid customer_id PK,FK
        string provider_type PK "stripe"
        string provider_customer_id "cus_stripe_abc"
        string tenant_id PK
        string environment_id PK
        jsonb metadata
        timestamp created_at
        timestamp updated_at
    }
    
    STRIPE_SYNC_BATCHES {
        uuid id PK
        uuid tenant_id
        uuid environment_id
        uuid customer_id FK
        uuid meter_id FK
        string stripe_customer_id
        string stripe_meter_id
        decimal aggregated_quantity
        integer event_count
        string stripe_event_id "Idempotency key"
        string sync_status "success/failed/retrying"
        integer retry_count
        text error_message
        timestamp window_start
        timestamp window_end
        timestamp created_at
        timestamp synced_at
    }
    
    METER_PROVIDER_MAPPINGS {
        uuid meter_id PK,FK
        string provider_type PK "stripe"
        string provider_meter_id "meter_stripe_xyz"
        string tenant_id PK
        string environment_id PK
        boolean sync_enabled
        jsonb configuration
        timestamp created_at
    }
    
    STRIPE_TENANT_CONFIGS {
        uuid tenant_id PK
        uuid environment_id PK
        text api_key_encrypted
        boolean sync_enabled
        integer aggregation_window_minutes
        jsonb webhook_config
        timestamp created_at
        timestamp updated_at
    }
    
    CUSTOMERS ||--o{ CUSTOMER_INTEGRATION_MAPPINGS : maps_to
    CUSTOMERS ||--o{ STRIPE_SYNC_BATCHES : generates
    METERS ||--o{ METER_PROVIDER_MAPPINGS : configured_for
    METERS ||--o{ STRIPE_SYNC_BATCHES : measured_by

Flow Specifications

Flow 1: Customer Management

sequenceDiagram
    participant Client as Client App
    participant Stripe as Stripe API
    participant Webhook as FlexPrice Webhook
    participant DB as FlexPrice DB
    participant Mapping as Integration Mapping

    Note over Client,Mapping: Customer Creation & Sync Flow

    %% Customer creation
    Client->>Stripe: Create customer with metadata
    Note right of Stripe: metadata: {<br/>"external_id": "client_123",<br/>"flexprice_customer_id": "fp_xyz"<br/>}
    Stripe-->>Client: customer_id: "cus_stripe_abc"

    %% Webhook processing
    Stripe->>Webhook: customer.created webhook
    Webhook->>Webhook: Validate webhook signature
    Webhook->>Webhook: Extract metadata
    
    alt Valid metadata present
        Webhook->>DB: Find customer by external_id
        
        alt Customer exists
            DB-->>Webhook: Customer found
            Webhook->>Mapping: Update/create mapping
            Note right of Mapping: customer_id ↔ cus_stripe_abc
        else Customer doesn't exist
            DB-->>Webhook: Customer not found
            Webhook->>DB: Create new customer
            Note right of DB: Use metadata for customer data
            Webhook->>Mapping: Create integration mapping
        end
        
        Webhook-->>Stripe: 200 OK - Success
    else Invalid/missing metadata
        Webhook->>Webhook: Log error
        Webhook-->>Stripe: 400 Bad Request
    end

    Note over Client,Mapping: Customer ready for usage events

Flow 2: Event Synchronization

sequenceDiagram
    participant App as Client App
    participant FP as FlexPrice API
    participant CH as ClickHouse
    participant Temporal as Temporal Worker
    participant Batch as Batch Tracker
    participant Stripe as Stripe API

    Note over App,Stripe: Hourly Event Sync Flow

    %% Event ingestion (existing FlexPrice flow)
    loop Multiple usage events
        App->>FP: POST /events
        FP->>CH: Store raw event
        FP->>CH: Process → events_processed
        Note right of CH: Calculate qty_billable,<br/>qty_free, cost
    end

    %% Temporal cron workflow
    Note over Temporal: Cron Trigger: Every Hour
    Temporal->>CH: Query billable events
    Note right of CH: SELECT customer_id, meter_id,<br/>SUM(qty_billable), COUNT(*)<br/>FROM events_processed<br/>WHERE qty_billable > 0<br/>AND timestamp in last hour<br/>GROUP BY customer_id, meter_id

    CH-->>Temporal: Aggregated batches
    
    loop For each customer+meter batch
        Temporal->>Temporal: Lookup customer mapping
        
        alt Customer mapping exists
            Temporal->>Temporal: Lookup meter mapping
            
            alt Meter mapping exists
                Temporal->>Batch: Create batch record (pending)
                Temporal->>Stripe: Create meter event
                Note right of Stripe: Single event with<br/>aggregated quantity
                
                alt Stripe API success
                    Stripe-->>Temporal: stripe_event_id
                    Temporal->>Batch: Update batch (success)
                else Stripe API failure
                    Stripe-->>Temporal: Error response
                    Temporal->>Batch: Update batch (failed)
                    Temporal->>Temporal: Schedule retry
                end
            else Meter mapping missing
                Temporal->>Batch: Log skip (meter not configured)
            end
        else Customer mapping missing
            Temporal->>Batch: Log skip (customer not mapped)
        end
    end

    Temporal->>Temporal: Complete workflow

Error Handling Flow

flowchart TD
    A[Sync Operation] --> B{Operation Type}
    
    B -->|Customer Webhook| C[Webhook Error Handler]
    B -->|Event Sync| D[Temporal Error Handler]
    
    C --> E{Error Type}
    E -->|Invalid Payload| F[Return 400, Log Error]
    E -->|Database Error| G[Return 500, Stripe Retries]
    E -->|Mapping Conflict| H[Upsert Pattern, Return 200]
    
    D --> I{Error Type}
    I -->|Stripe Rate Limit| J[Exponential Backoff Retry]
    I -->|Network Timeout| K[Linear Retry with Circuit Breaker]
    I -->|Auth Error| L[Immediate Failure, Alert Ops]
    I -->|Invalid Data| M[Skip Batch, Log for Manual Review]
    
    J --> N{Retry Count < Max}
    N -->|Yes| O[Wait & Retry]
    N -->|No| P[Mark Failed, Alert]
    
    O --> D
    P --> Q[Manual Intervention Required]
    
    style F fill:#ffcccc
    style G fill:#ffcccc
    style L fill:#ffcccc
    style Q fill:#ffcccc
    style H fill:#ccffcc

Design Decisions & Trade-offs

Decision 1: Simplified vs Complex State Management

Original Design: Per-event state tracking with integration_sync_state table

-- Complex state per event
CREATE TABLE integration_sync_state (
    event_id UUID,
    sync_status VARCHAR, -- 7+ states
    skip_reason VARCHAR,
    retry_count INTEGER,
    -- ... many fields
);

Current Design: Batch-level tracking only

-- Simple batch tracking
CREATE TABLE stripe_sync_batches (
    id UUID,
    customer_id UUID,
    meter_id UUID,
    aggregated_quantity DECIMAL,
    sync_status VARCHAR -- 3 states
);

Trade-off Analysis:

✅ Pro: Faster MVP implementation (days vs weeks)
✅ Pro: Simpler debugging and operations
✅ Pro: Leverages existing FlexPrice billing calculations
❌ Con: Lost individual event traceability in Stripe
❌ Con: Harder to debug specific event processing issues
❌ Con: Less sophisticated error handling

Decision: Accept trade-off for MVP speed, add detailed tracking in Phase 2

Decision 2: Direct ClickHouse Integration

Alternative: Event streaming via Kafka/separate event store
Current: Direct queries to events_processed table

Trade-off Analysis:

✅ Pro: Reuses existing billing logic and data
✅ Pro: No additional infrastructure complexity
✅ Pro: Guaranteed consistency with FlexPrice billing
❌ Con: Tight coupling with billing engine schema
❌ Con: Harder to extend to other event sources
❌ Con: Performance impact on ClickHouse during sync

Decision: Accept coupling for MVP, consider event streaming for scale

Decision 3: Webhook-First Customer Sync

Alternative: API polling or JIT (Just-In-Time) creation
Current: Stripe webhook-driven synchronization

Trade-off Analysis:

✅ Pro: Eliminates race conditions between customer creation and events
✅ Pro: Real-time sync with deterministic ordering
✅ Pro: Leverages Stripe's reliable webhook infrastructure
❌ Con: Single point of failure (webhook endpoint)
❌ Con: Requires webhook endpoint reliability and security
❌ Con: No fallback mechanism for webhook failures

Decision: Webhook-first with monitoring, add reconciliation fallback in Phase 2

Decision 4: Hourly Aggregation Windows

Alternatives: Real-time, 15-minute, daily aggregation
Current: 1-hour fixed windows

Trade-off Analysis:

✅ Pro: Reduces Stripe API calls (rate limit friendly)
✅ Pro: Predictable batch sizes and processing time
✅ Pro: Balances real-time needs with system performance
❌ Con: Up to 1-hour delay in Stripe billing visibility
❌ Con: Fixed window may not align with customer billing periods
❌ Con: Batching makes debugging individual events harder

Decision: 1-hour windows for MVP, make configurable per tenant later

Implementation Plan

Phase 1: Core MVP (Days 1-3)

Goal: Basic customer sync + event sync working end-to-end

Day 1: Foundation

Database schema creation (integration mappings, batch tracking)
Customer webhook endpoint implementation
Basic Stripe client wrapper
Customer mapping logic (create/update)

Day 2: Event Sync

Temporal workflow for hourly aggregation
ClickHouse query for billable events aggregation
Stripe meter event creation
Basic batch tracking and status updates

Day 3: Integration & Testing

End-to-end flow testing
Error handling for common failures
Basic logging and monitoring
Configuration management (API keys, meter mappings)

Phase 2: Production Hardening (Days 4-5)

Goal: Production-ready reliability and observability

Day 4: Reliability

Comprehensive error handling and retry logic
Webhook signature validation
Idempotency guarantees (stripe_event_id usage)
Circuit breaker patterns for Stripe API

Day 5: Operations

Health check endpoints
Monitoring dashboard and alerts
Manual retry API endpoints
Documentation and runbooks

Phase 3: Future Enhancements (Post-MVP)

Scope: Advanced features for production scale

Individual event traceability within batches
Reconciliation jobs for data drift detection
Multi-provider extensibility framework
Advanced retry strategies with exponential backoff
Real-time sync options for premium customers
Billing period boundary handling
Currency conversion and multi-currency support

Risk Assessment

High-Risk Items

Webhook Reliability: Single point of failure for customer sync
- Mitigation: Monitoring, alerting, manual fallback procedures
Stripe API Rate Limits: Batch operations hitting limits
- Mitigation: Exponential backoff, circuit breakers, batch size tuning
Data Consistency: Drift between FlexPrice and Stripe billing
- Mitigation: Comprehensive logging, reconciliation alerts

Medium-Risk Items

Temporal Workflow Failures: Sync jobs failing silently
- Mitigation: Workflow monitoring, retry mechanisms, dead letter queues
ClickHouse Performance: Heavy aggregation queries impacting billing
- Mitigation: Query optimization, read replicas, monitoring
Configuration Drift: Meter mappings becoming inconsistent
- Mitigation: Configuration validation, audit trails

Low-Risk Items

Schema Evolution: Database changes breaking integration
- Mitigation: Backward compatibility practices, migration testing
Security: API key exposure or webhook vulnerabilities
- Mitigation: Encryption, signature validation, secret rotation

Success Metrics

Functional Metrics

Customer Sync Success Rate: >99% webhook processing success
Event Sync Accuracy: 100% billable quantity consistency (FlexPrice vs Stripe)
Sync Latency: <15 minutes from FlexPrice billing to Stripe sync
Error Recovery: <5% failed batches requiring manual intervention

Performance Metrics

Webhook Response Time: <5 seconds average
Batch Processing Time: <15 minutes for hourly aggregation
API Call Efficiency: <100 Stripe API calls per hour per tenant
Resource Usage: <10% impact on FlexPrice core performance

Business Metrics

Customer Adoption: Successful demo capability within 5 days
Integration Reliability: Zero customer-impacting incidents in first month
Scalability: Support for 10+ pilot customers without infrastructure changes

Conclusion

This requirements document defines a pragmatic approach to Stripe integration that prioritizes MVP delivery speed while maintaining a clear evolution path to production sophistication. The simplified architecture trades some advanced features for implementation speed, with explicit plans to add complexity as needed.

The design leverages FlexPrice's existing strengths (billing engine, event processing) while adding minimal new complexity. The webhook-driven customer sync and batch event aggregation provide a solid foundation that can be enhanced with more sophisticated patterns as the integration matures.

Key Success Factor: Disciplined scope management to deliver a working demo in 5 days while maintaining code quality for future enhancements.

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flexprice

Stripe Sync #283

Uh oh!

{{title}}

Uh oh!

Replies: 4 comments 1 reply

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

Flexprice

Stripe Sync #283

Uh oh!

BRO3886 Jun 10, 2025

Stripe Integration - High Level Design

System Overview

Key Objectives

Architecture Overview

Core Components

1. Event Processing Pipeline

2. Integration Sync Decision Engine

3. Temporal Workflow Architecture

Data Flow Architecture

1. Customer Onboarding Flow

2. Usage Event Sync Flow

3. Sync Decision Matrix

Customer Journey

Phase 1: Initial Setup

Phase 2: Usage Tracking

Phase 3: Automated Sync

Phase 4: Monitoring & Management

Integration Patterns

1. Three-Layer ID Mapping

2. Event Aggregation Strategy

3. Provider Extensibility

Error Handling & Resilience

1. Retry Strategy

2. Circuit Breaker Pattern

3. Failure Categories & Responses

Monitoring & Observability

1. Key Metrics Dashboard

2. Alert Hierarchy

Security & Compliance

1. Data Security

2. API Security Model

Performance & Scalability

1. Scalability Targets

2. Performance Optimization

Deployment Architecture

1. Infrastructure Layout

2. Deployment Strategy

Risk Assessment & Mitigation

1. Technical Risks

2. Business Risks

Success Criteria

Technical Success

Business Success

Operational Success

Replies: 4 comments · 1 reply

Uh oh!

BRO3886 Jun 10, 2025 Author

Stripe Integration - Low Level Design

Sequence Diagrams & State Machine Definitions

1. Core Event Flow Sequences

1.1 Complete Usage Event Sync Flow

1.2 Customer Onboarding Sequence

1.3 Sync Decision Processing Sequence

1.4 Event Aggregation Sequence

1.5 Stripe API Sync Sequence

1.6 Error Handling & Retry Sequence

2. State Machine Definitions

2.1 Integration Sync State Machine

2.2 Customer Mapping State Machine

2.3 Temporal Workflow State Machine

2.4 Provider Configuration State Machine

2.5 Batch Processing State Machine

3. State Transition Rules

3.1 Sync Status Transitions

3.2 Workflow Transition Rules

3.3 Error Recovery Rules

Uh oh!

BRO3886 Jun 10, 2025 Author

Customer Integration and Sync Design

Overview

Problem Statement

Use Case

Key Requirements

Database Design

Customer Integration Mapping Table

Ent Schema (Go)

BRO3886
Jun 10, 2025

Replies: 4 comments 1 reply

BRO3886
Jun 10, 2025
Author

BRO3886
Jun 10, 2025
Author

BRO3886
Jun 10, 2025
Author