integration-stripe-sdd-software-design-document

---

title: Stripe Integration - Software Design Document type: reference component_type: reference version: 1.0.0 created: '2025-12-27' updated: '2025-12-27' status: active tags:

• ai-ml
• authentication
• security
• testing
• api
• architecture
• automation
• backend summary: 'Stripe Integration - Software Design Document Document Version: 1.0.0 Status: Planning Last Updated: December 17, 2025 Author: CODITECT Engineering Team --- Executive Summary This document describes the software design for CODITECT''s Stripe...' moe_confidence: 0.950 moe_classified: 2025-12-31

---

Stripe Integration - Software Design Document

Document Version: 1.0.0 Status: Planning Last Updated: December 17, 2025 Author: CODITECT Engineering Team

---

1. Executive Summary

This document describes the software design for CODITECT's Stripe integration, enabling subscription management, payment processing, usage-based billing, and revenue analytics for the CODITECT platform.

1.1 Purpose

Provide a comprehensive payment and billing infrastructure that:

• Handles subscription lifecycle management
• Processes one-time and recurring payments
• Supports usage-based billing (metered billing)
• Enables customer self-service portal
• Provides revenue analytics and reporting
• Maintains PCI DSS compliance through Stripe

1.2 Scope

In Scope	Out of Scope
Subscription CRUD	Custom payment gateway
Payment processing	Direct credit card handling
Usage-based billing	Tax calculation (use Stripe Tax)
Customer portal	Invoicing (use Stripe Invoicing)
Webhook handling	Fraud detection (use Stripe Radar)
Revenue metrics	Multi-currency conversion logic

1.3 Key Objectives

1. Zero PCI Burden - Stripe handles all card data
2. Subscription Flexibility - Multiple plans, trials, upgrades
3. Usage Metering - Track and bill API/feature usage
4. Self-Service - Customer billing portal
5. Revenue Visibility - Real-time metrics and reporting

---

2. System Architecture

2.1 High-Level Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                           CODITECT PLATFORM                              │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐     │
│  │   Frontend      │    │   Backend API   │    │   Background    │     │
│  │   (React)       │    │   (Rust/Python) │    │   Workers       │     │
│  └────────┬────────┘    └────────┬────────┘    └────────┬────────┘     │
│           │                      │                      │               │
│           │    ┌─────────────────┴─────────────────┐   │               │
│           │    │                                   │   │               │
│           ▼    ▼                                   ▼   ▼               │
│  ┌─────────────────────────────────────────────────────────────┐       │
│  │                   STRIPE INTEGRATION LAYER                   │       │
│  ├─────────────────────────────────────────────────────────────┤       │
│  │                                                             │       │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │       │
│  │  │ Subscription│  │  Payment    │  │   Usage     │         │       │
│  │  │  Manager    │  │  Processor  │  │   Meter     │         │       │
│  │  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘         │       │
│  │         │                │                │                 │       │
│  │  ┌──────┴────────────────┴────────────────┴──────┐         │       │
│  │  │           StripeClient (API Wrapper)          │         │       │
│  │  └──────────────────────┬────────────────────────┘         │       │
│  │                         │                                   │       │
│  │  ┌──────────────────────┴────────────────────────┐         │       │
│  │  │         Webhook Handler (Event Processing)    │         │       │
│  │  └───────────────────────────────────────────────┘         │       │
│  │                                                             │       │
│  └─────────────────────────────────────────────────────────────┘       │
│                                    │                                    │
└────────────────────────────────────┼────────────────────────────────────┘
                                     │
                                     ▼
                    ┌────────────────────────────────┐
                    │         STRIPE API             │
                    ├────────────────────────────────┤
                    │  • Customers                   │
                    │  • Subscriptions               │
                    │  • Payment Intents             │
                    │  • Invoices                    │
                    │  • Usage Records               │
                    │  • Checkout Sessions           │
                    │  • Customer Portal             │
                    │  • Webhooks                    │
                    └────────────────────────────────┘

2.2 Component Architecture

stripe/
├── src/
│   ├── __init__.py
│   ├── client.py              # Stripe API client wrapper
│   ├── config.py              # Configuration management
│   │
│   ├── customers/
│   │   ├── __init__.py
│   │   ├── manager.py         # Customer CRUD operations
│   │   └── sync.py            # Customer sync with CODITECT users
│   │
│   ├── subscriptions/
│   │   ├── __init__.py
│   │   ├── manager.py         # Subscription lifecycle
│   │   ├── plans.py           # Plan/Price management
│   │   └── trials.py          # Trial handling
│   │
│   ├── payments/
│   │   ├── __init__.py
│   │   ├── processor.py       # Payment processing
│   │   ├── intents.py         # Payment Intent handling
│   │   └── methods.py         # Payment method management
│   │
│   ├── billing/
│   │   ├── __init__.py
│   │   ├── metering.py        # Usage metering
│   │   ├── invoices.py        # Invoice operations
│   │   └── credits.py         # Credit/balance management
│   │
│   ├── webhooks/
│   │   ├── __init__.py
│   │   ├── handler.py         # Webhook endpoint handler
│   │   ├── signature.py       # Signature verification
│   │   └── events/            # Event-specific handlers
│   │       ├── __init__.py
│   │       ├── subscription.py
│   │       ├── invoice.py
│   │       ├── payment.py
│   │       └── customer.py
│   │
│   ├── portal/
│   │   ├── __init__.py
│   │   └── session.py         # Customer portal sessions
│   │
│   └── analytics/
│       ├── __init__.py
│       ├── revenue.py         # Revenue metrics
│       └── churn.py           # Churn analysis
│
├── tests/
│   ├── __init__.py
│   ├── test_client.py
│   ├── test_subscriptions.py
│   ├── test_payments.py
│   ├── test_webhooks.py
│   └── fixtures/              # Test fixtures
│
├── config/
│   ├── plans.yaml             # Plan definitions
│   └── webhook_events.yaml    # Webhook event configs
│
└── docs/
    ├── sdd-software-design-document.md
    ├── tdd-technical-design-document.md
    ├── project-plan.md
    ├── tasklist-with-checkboxes.md
    └── adrs/

---

3. Core Components

3.1 StripeClient

Central API client wrapping Stripe SDK with error handling and retries.

from dataclasses import dataclass
from typing import Any, Dict, Optional
import stripe
from stripe.error import StripeError

@dataclass
class StripeConfig:
    """Stripe configuration."""
    api_key: str
    webhook_secret: str
    api_version: str = "2023-10-16"
    max_retries: int = 3
    timeout: int = 30

class StripeClient:
    """Stripe API client wrapper."""

    def __init__(self, config: StripeConfig):
        self.config = config
        stripe.api_key = config.api_key
        stripe.api_version = config.api_version
        stripe.max_network_retries = config.max_retries

    async def create_customer(
        self,
        email: str,
        name: Optional[str] = None,
        metadata: Optional[Dict[str, str]] = None,
    ) -> stripe.Customer:
        """Create a Stripe customer."""
        return stripe.Customer.create(
            email=email,
            name=name,
            metadata=metadata or {},
        )

    async def create_subscription(
        self,
        customer_id: str,
        price_id: str,
        trial_days: Optional[int] = None,
        metadata: Optional[Dict[str, str]] = None,
    ) -> stripe.Subscription:
        """Create a subscription."""
        params = {
            "customer": customer_id,
            "items": [{"price": price_id}],
            "metadata": metadata or {},
        }
        if trial_days:
            params["trial_period_days"] = trial_days
        return stripe.Subscription.create(**params)

    async def create_checkout_session(
        self,
        customer_id: str,
        price_id: str,
        success_url: str,
        cancel_url: str,
        mode: str = "subscription",
    ) -> stripe.checkout.Session:
        """Create a Checkout Session."""
        return stripe.checkout.Session.create(
            customer=customer_id,
            line_items=[{"price": price_id, "quantity": 1}],
            mode=mode,
            success_url=success_url,
            cancel_url=cancel_url,
        )

    async def create_portal_session(
        self,
        customer_id: str,
        return_url: str,
    ) -> stripe.billing_portal.Session:
        """Create a Customer Portal session."""
        return stripe.billing_portal.Session.create(
            customer=customer_id,
            return_url=return_url,
        )

3.2 SubscriptionManager

Handles subscription lifecycle operations.

from enum import Enum
from datetime import datetime
from typing import List, Optional
from dataclasses import dataclass

class SubscriptionStatus(str, Enum):
    """Subscription status."""
    ACTIVE = "active"
    PAST_DUE = "past_due"
    CANCELED = "canceled"
    INCOMPLETE = "incomplete"
    TRIALING = "trialing"
    UNPAID = "unpaid"
    PAUSED = "paused"

@dataclass
class Subscription:
    """CODITECT subscription model."""
    id: str
    stripe_subscription_id: str
    customer_id: str
    plan_id: str
    status: SubscriptionStatus
    current_period_start: datetime
    current_period_end: datetime
    trial_end: Optional[datetime] = None
    canceled_at: Optional[datetime] = None
    cancel_at_period_end: bool = False
    metadata: dict = None

class SubscriptionManager:
    """Manage subscription lifecycle."""

    def __init__(self, stripe_client: StripeClient, db):
        self.stripe = stripe_client
        self.db = db

    async def create_subscription(
        self,
        user_id: str,
        plan_id: str,
        trial_days: int = 14,
    ) -> Subscription:
        """Create a new subscription for user."""
        # Get or create Stripe customer
        customer = await self._ensure_customer(user_id)

        # Get price ID for plan
        price_id = await self._get_price_id(plan_id)

        # Create Stripe subscription
        stripe_sub = await self.stripe.create_subscription(
            customer_id=customer.stripe_id,
            price_id=price_id,
            trial_days=trial_days,
            metadata={"user_id": user_id, "plan_id": plan_id},
        )

        # Store in database
        subscription = Subscription(
            id=generate_id(),
            stripe_subscription_id=stripe_sub.id,
            customer_id=customer.id,
            plan_id=plan_id,
            status=SubscriptionStatus(stripe_sub.status),
            current_period_start=datetime.fromtimestamp(stripe_sub.current_period_start),
            current_period_end=datetime.fromtimestamp(stripe_sub.current_period_end),
            trial_end=datetime.fromtimestamp(stripe_sub.trial_end) if stripe_sub.trial_end else None,
        )

        await self.db.subscriptions.insert(subscription)
        return subscription

    async def upgrade_subscription(
        self,
        subscription_id: str,
        new_plan_id: str,
        prorate: bool = True,
    ) -> Subscription:
        """Upgrade subscription to new plan."""
        subscription = await self.db.subscriptions.get(subscription_id)
        new_price_id = await self._get_price_id(new_plan_id)

        # Update in Stripe
        stripe_sub = stripe.Subscription.retrieve(subscription.stripe_subscription_id)
        stripe.Subscription.modify(
            subscription.stripe_subscription_id,
            items=[{
                "id": stripe_sub["items"]["data"][0].id,
                "price": new_price_id,
            }],
            proration_behavior="create_prorations" if prorate else "none",
        )

        # Update local record
        subscription.plan_id = new_plan_id
        await self.db.subscriptions.update(subscription)
        return subscription

    async def cancel_subscription(
        self,
        subscription_id: str,
        immediate: bool = False,
    ) -> Subscription:
        """Cancel subscription."""
        subscription = await self.db.subscriptions.get(subscription_id)

        if immediate:
            stripe.Subscription.delete(subscription.stripe_subscription_id)
            subscription.status = SubscriptionStatus.CANCELED
            subscription.canceled_at = datetime.utcnow()
        else:
            stripe.Subscription.modify(
                subscription.stripe_subscription_id,
                cancel_at_period_end=True,
            )
            subscription.cancel_at_period_end = True

        await self.db.subscriptions.update(subscription)
        return subscription

    async def resume_subscription(self, subscription_id: str) -> Subscription:
        """Resume a subscription scheduled for cancellation."""
        subscription = await self.db.subscriptions.get(subscription_id)

        stripe.Subscription.modify(
            subscription.stripe_subscription_id,
            cancel_at_period_end=False,
        )

        subscription.cancel_at_period_end = False
        await self.db.subscriptions.update(subscription)
        return subscription

3.3 UsageMeter

Track and report usage for metered billing.

from datetime import datetime
from typing import Optional
from dataclasses import dataclass

@dataclass
class UsageRecord:
    """Usage record for metered billing."""
    subscription_item_id: str
    quantity: int
    timestamp: datetime
    action: str = "increment"  # increment or set
    idempotency_key: Optional[str] = None

class UsageMeter:
    """Track and report usage for metered billing."""

    def __init__(self, stripe_client: StripeClient, db):
        self.stripe = stripe_client
        self.db = db

    async def record_usage(
        self,
        subscription_id: str,
        meter_id: str,
        quantity: int,
        timestamp: Optional[datetime] = None,
        idempotency_key: Optional[str] = None,
    ) -> UsageRecord:
        """Record usage for a metered subscription item."""
        subscription = await self.db.subscriptions.get(subscription_id)

        # Find the subscription item for this meter
        stripe_sub = stripe.Subscription.retrieve(
            subscription.stripe_subscription_id,
            expand=["items.data.price"],
        )

        subscription_item_id = None
        for item in stripe_sub["items"]["data"]:
            if item["price"]["lookup_key"] == meter_id:
                subscription_item_id = item.id
                break

        if not subscription_item_id:
            raise ValueError(f"Meter {meter_id} not found in subscription")

        # Report to Stripe
        usage_record = stripe.SubscriptionItem.create_usage_record(
            subscription_item_id,
            quantity=quantity,
            timestamp=int((timestamp or datetime.utcnow()).timestamp()),
            action="increment",
            idempotency_key=idempotency_key,
        )

        # Store locally for analytics
        record = UsageRecord(
            subscription_item_id=subscription_item_id,
            quantity=quantity,
            timestamp=timestamp or datetime.utcnow(),
            idempotency_key=idempotency_key,
        )
        await self.db.usage_records.insert(record)

        return record

    async def get_usage_summary(
        self,
        subscription_id: str,
        meter_id: str,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None,
    ) -> dict:
        """Get usage summary for billing period."""
        # Query from local database for speed
        records = await self.db.usage_records.query(
            subscription_id=subscription_id,
            meter_id=meter_id,
            start_date=start_date,
            end_date=end_date,
        )

        total_quantity = sum(r.quantity for r in records)
        return {
            "meter_id": meter_id,
            "total_quantity": total_quantity,
            "record_count": len(records),
            "start_date": start_date,
            "end_date": end_date,
        }

3.4 WebhookHandler

Process Stripe webhook events securely.

import hmac
import hashlib
from typing import Callable, Dict
from dataclasses import dataclass

@dataclass
class WebhookEvent:
    """Parsed webhook event."""
    id: str
    type: str
    data: dict
    created: int
    livemode: bool

class WebhookHandler:
    """Handle Stripe webhook events."""

    def __init__(self, webhook_secret: str, db):
        self.webhook_secret = webhook_secret
        self.db = db
        self._handlers: Dict[str, Callable] = {}

        # Register default handlers
        self._register_default_handlers()

    def _register_default_handlers(self):
        """Register handlers for common events."""
        self.register("customer.subscription.created", self._handle_subscription_created)
        self.register("customer.subscription.updated", self._handle_subscription_updated)
        self.register("customer.subscription.deleted", self._handle_subscription_deleted)
        self.register("invoice.paid", self._handle_invoice_paid)
        self.register("invoice.payment_failed", self._handle_payment_failed)
        self.register("customer.created", self._handle_customer_created)
        self.register("checkout.session.completed", self._handle_checkout_completed)

    def register(self, event_type: str, handler: Callable):
        """Register a handler for an event type."""
        self._handlers[event_type] = handler

    def verify_signature(self, payload: bytes, signature: str) -> bool:
        """Verify webhook signature."""
        try:
            stripe.Webhook.construct_event(
                payload, signature, self.webhook_secret
            )
            return True
        except stripe.error.SignatureVerificationError:
            return False

    async def handle(self, payload: bytes, signature: str) -> dict:
        """Handle incoming webhook."""
        # Verify signature
        if not self.verify_signature(payload, signature):
            raise ValueError("Invalid webhook signature")

        # Parse event
        event = stripe.Webhook.construct_event(
            payload, signature, self.webhook_secret
        )

        # Check for duplicate (idempotency)
        if await self._is_duplicate(event.id):
            return {"status": "duplicate", "event_id": event.id}

        # Get handler
        handler = self._handlers.get(event.type)
        if not handler:
            return {"status": "unhandled", "event_type": event.type}

        # Process event
        await handler(event)

        # Mark as processed
        await self._mark_processed(event.id)

        return {"status": "processed", "event_id": event.id}

    async def _handle_subscription_created(self, event):
        """Handle new subscription."""
        subscription = event.data.object
        # Sync to local database
        await self.db.subscriptions.sync_from_stripe(subscription)

    async def _handle_subscription_updated(self, event):
        """Handle subscription update."""
        subscription = event.data.object
        await self.db.subscriptions.sync_from_stripe(subscription)

        # Check for status changes
        previous = event.data.previous_attributes
        if "status" in previous:
            await self._notify_status_change(
                subscription.id,
                previous["status"],
                subscription.status,
            )

    async def _handle_subscription_deleted(self, event):
        """Handle subscription cancellation."""
        subscription = event.data.object
        await self.db.subscriptions.update_status(
            subscription.id,
            SubscriptionStatus.CANCELED,
        )

    async def _handle_invoice_paid(self, event):
        """Handle successful payment."""
        invoice = event.data.object
        await self.db.invoices.mark_paid(invoice.id)

    async def _handle_payment_failed(self, event):
        """Handle failed payment."""
        invoice = event.data.object
        await self._notify_payment_failure(invoice)
        # Stripe will retry automatically based on settings

    async def _handle_customer_created(self, event):
        """Handle new customer."""
        customer = event.data.object
        await self.db.customers.sync_from_stripe(customer)

    async def _handle_checkout_completed(self, event):
        """Handle completed checkout session."""
        session = event.data.object
        # Provision access based on checkout
        await self._provision_access(session)

---

4. Data Models

4.1 Core Models

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional

class PlanTier(str, Enum):
    """Subscription plan tiers."""
    FREE = "free"
    STARTER = "starter"
    PROFESSIONAL = "professional"
    ENTERPRISE = "enterprise"

class BillingInterval(str, Enum):
    """Billing interval."""
    MONTHLY = "month"
    YEARLY = "year"

@dataclass
class Plan:
    """Subscription plan definition."""
    id: str
    name: str
    tier: PlanTier
    stripe_product_id: str
    stripe_price_id_monthly: Optional[str] = None
    stripe_price_id_yearly: Optional[str] = None
    features: List[str] = field(default_factory=list)
    limits: Dict[str, int] = field(default_factory=dict)
    price_monthly: int = 0  # in cents
    price_yearly: int = 0   # in cents
    is_active: bool = True

@dataclass
class Customer:
    """CODITECT customer linked to Stripe."""
    id: str
    user_id: str
    stripe_customer_id: str
    email: str
    name: Optional[str] = None
    default_payment_method_id: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.utcnow)
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class Invoice:
    """Invoice record."""
    id: str
    stripe_invoice_id: str
    customer_id: str
    subscription_id: Optional[str] = None
    amount_due: int = 0
    amount_paid: int = 0
    currency: str = "usd"
    status: str = "draft"
    period_start: Optional[datetime] = None
    period_end: Optional[datetime] = None
    paid_at: Optional[datetime] = None
    hosted_invoice_url: Optional[str] = None
    invoice_pdf: Optional[str] = None

@dataclass
class PaymentMethod:
    """Stored payment method."""
    id: str
    stripe_payment_method_id: str
    customer_id: str
    type: str  # card, bank_account, etc.
    card_brand: Optional[str] = None
    card_last4: Optional[str] = None
    card_exp_month: Optional[int] = None
    card_exp_year: Optional[int] = None
    is_default: bool = False

4.2 Usage Models

@dataclass
class UsageLimit:
    """Usage limit for a feature."""
    feature: str
    limit: int
    period: str = "month"  # month, day, hour

@dataclass
class UsageRecord:
    """Recorded usage event."""
    id: str
    subscription_id: str
    feature: str
    quantity: int
    timestamp: datetime
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class UsageSummary:
    """Usage summary for billing period."""
    subscription_id: str
    feature: str
    total_usage: int
    limit: int
    period_start: datetime
    period_end: datetime
    overage: int = 0
    overage_cost: int = 0  # in cents

---

5. API Endpoints

5.1 Subscription Endpoints

Method	Endpoint	Description
GET	/api/v1/billing/plans	List available plans
GET	/api/v1/billing/subscription	Get current subscription
POST	/api/v1/billing/subscription	Create subscription
PATCH	/api/v1/billing/subscription	Update subscription
DELETE	/api/v1/billing/subscription	Cancel subscription
POST	/api/v1/billing/subscription/resume	Resume cancelled subscription

5.2 Payment Endpoints

Method	Endpoint	Description
GET	/api/v1/billing/payment-methods	List payment methods
POST	/api/v1/billing/payment-methods	Add payment method
DELETE	/api/v1/billing/payment-methods/{id}	Remove payment method
POST	/api/v1/billing/payment-methods/{id}/default	Set default

5.3 Checkout & Portal

Method	Endpoint	Description
POST	/api/v1/billing/checkout	Create checkout session
POST	/api/v1/billing/portal	Create portal session

5.4 Usage & Invoices

Method	Endpoint	Description
GET	/api/v1/billing/usage	Get usage summary
GET	/api/v1/billing/invoices	List invoices
GET	/api/v1/billing/invoices/{id}	Get invoice details

5.5 Webhooks

Method	Endpoint	Description
POST	/webhooks/stripe	Stripe webhook endpoint

---

6. CODITECT Plan Structure

6.1 Plan Definitions

plans:
  free:
    name: "Free"
    tier: free
    price_monthly: 0
    price_yearly: 0
    features:
      - "3 projects"
      - "Basic AI assistance"
      - "Community support"
    limits:
      projects: 3
      ai_requests_per_month: 100
      storage_mb: 500

  starter:
    name: "Starter"
    tier: starter
    price_monthly: 1900  # $19/month
    price_yearly: 19000  # $190/year (save $38)
    features:
      - "10 projects"
      - "Standard AI assistance"
      - "Email support"
      - "Git integrations"
    limits:
      projects: 10
      ai_requests_per_month: 1000
      storage_mb: 5000
      team_members: 3

  professional:
    name: "Professional"
    tier: professional
    price_monthly: 4900  # $49/month
    price_yearly: 49000  # $490/year (save $98)
    features:
      - "Unlimited projects"
      - "Advanced AI assistance"
      - "Priority support"
      - "All integrations"
      - "Custom workflows"
    limits:
      projects: -1  # unlimited
      ai_requests_per_month: 10000
      storage_mb: 50000
      team_members: 10

  enterprise:
    name: "Enterprise"
    tier: enterprise
    price_monthly: 0  # custom pricing
    price_yearly: 0
    features:
      - "Everything in Professional"
      - "Dedicated support"
      - "SSO/SAML"
      - "Custom AI models"
      - "On-premise option"
      - "SLA guarantee"
    limits:
      projects: -1
      ai_requests_per_month: -1
      storage_mb: -1
      team_members: -1

6.2 Metered Features

metered_features:
  ai_requests:
    name: "AI Requests"
    lookup_key: "ai_requests"
    unit_amount: 1  # $0.01 per request overage
    included_in_plans:
      free: 100
      starter: 1000
      professional: 10000
      enterprise: unlimited

  storage:
    name: "Storage"
    lookup_key: "storage_mb"
    unit_amount: 5  # $0.05 per MB overage
    included_in_plans:
      free: 500
      starter: 5000
      professional: 50000
      enterprise: unlimited

---

7. Security Considerations

7.1 PCI Compliance

• No card data handling - All card data goes directly to Stripe
• Stripe.js/Elements - Frontend uses Stripe's secure components
• Tokenization - Only tokens stored, never card numbers
• Webhook verification - All webhooks verified via signature

7.2 API Security

# Always use environment variables
STRIPE_SECRET_KEY = os.environ["STRIPE_SECRET_KEY"]
STRIPE_WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]

# Never log sensitive data
logger.info(f"Processing payment for customer {customer_id}")
# NOT: logger.info(f"Card: {card_number}")

# Verify webhook signatures
if not webhook_handler.verify_signature(payload, signature):
    raise HTTPException(status_code=400, detail="Invalid signature")

7.3 Idempotency

# Use idempotency keys for mutations
stripe.PaymentIntent.create(
    amount=1000,
    currency="usd",
    idempotency_key=f"payment_{order_id}",
)

# Deduplicate webhook events
if await is_event_processed(event.id):
    return {"status": "duplicate"}

---

8. Integration with CODITECT

8.1 User Provisioning Flow

User Signs Up → Create Stripe Customer → Free Plan Activated
       │
       ▼
User Upgrades → Checkout Session → Payment → Subscription Created
       │
       ▼
Webhook Received → Update Database → Provision Features

8.2 Feature Gating

class FeatureGate:
    """Gate features based on subscription."""

    async def can_access(self, user_id: str, feature: str) -> bool:
        """Check if user can access feature."""
        subscription = await self.get_subscription(user_id)

        if not subscription:
            # Fall back to free tier limits
            plan = self.plans["free"]
        else:
            plan = self.plans[subscription.plan_id]

        return feature in plan.features

    async def check_limit(self, user_id: str, feature: str) -> tuple[bool, int]:
        """Check if user is within limit."""
        subscription = await self.get_subscription(user_id)
        plan = self.plans.get(subscription.plan_id, self.plans["free"])

        limit = plan.limits.get(feature, 0)
        if limit == -1:
            return True, float("inf")

        current_usage = await self.get_usage(user_id, feature)
        return current_usage < limit, limit - current_usage

8.3 Usage Tracking Integration

# Track AI request usage
@app.post("/api/v1/ai/complete")
async def ai_complete(request: AIRequest, user: User = Depends(get_current_user)):
    # Check limit before processing
    allowed, remaining = await feature_gate.check_limit(user.id, "ai_requests")
    if not allowed:
        raise HTTPException(status_code=429, detail="AI request limit exceeded")

    # Process request
    result = await ai_service.complete(request)

    # Record usage
    await usage_meter.record_usage(
        subscription_id=user.subscription_id,
        meter_id="ai_requests",
        quantity=1,
    )

    return result

---

9. Error Handling

9.1 Stripe Error Types

from stripe.error import (
    CardError,
    RateLimitError,
    InvalidRequestError,
    AuthenticationError,
    APIConnectionError,
    StripeError,
)

async def handle_stripe_error(e: StripeError) -> dict:
    """Map Stripe errors to user-friendly responses."""
    if isinstance(e, CardError):
        return {
            "error": "card_error",
            "message": e.user_message,
            "code": e.code,
        }
    elif isinstance(e, RateLimitError):
        return {
            "error": "rate_limit",
            "message": "Too many requests. Please try again.",
            "retry_after": 60,
        }
    elif isinstance(e, InvalidRequestError):
        return {
            "error": "invalid_request",
            "message": "Invalid request. Please check your input.",
        }
    else:
        # Log for debugging, return generic message
        logger.error(f"Stripe error: {e}")
        return {
            "error": "payment_error",
            "message": "Payment processing failed. Please try again.",
        }

---

10. Testing Strategy

10.1 Test Mode

# Use Stripe test mode
STRIPE_SECRET_KEY = "sk_test_..."

# Test card numbers
TEST_CARDS = {
    "success": "4242424242424242",
    "decline": "4000000000000002",
    "insufficient_funds": "4000000000009995",
    "3d_secure": "4000002760003184",
}

10.2 Webhook Testing

# Install Stripe CLI
brew install stripe/stripe-cli/stripe

# Forward webhooks to local
stripe listen --forward-to localhost:8000/webhooks/stripe

# Trigger test events
stripe trigger invoice.paid
stripe trigger customer.subscription.created

---

11. Related Documents

• tdd-technical-design-document.md
• project-plan.md
• adr-001-stripe-as-payment-provider.md
• adr-002-subscription-model.md

---

Document Control:

• Created: December 17, 2025
• Author: CODITECT Engineering Team
• Review Cycle: Monthly