Skip to main content

ADR-013: Stripe Integration for Billing

Status: Accepted Date: 2025-11-30 Deciders: Architecture Team, Billing Team, Product Team Tags: billing, stripe, payments, subscriptions, webhooks

Context​

Payment Processing Requirements​

CODITECT requires a payment processing solution for subscription-based licensing across four tiers:

Tier Pricing:

  • Free: $0/month (no payment)
  • Pro: $58/month or $580/year (individual)
  • Team: $290/month for 5 seats (additional seats $50/month each)
  • Enterprise: Custom pricing (negotiated annually)

Payment Features Needed:

  1. Subscription Management - Monthly/annual billing cycles
  2. Automatic Renewal - Charge customers automatically
  3. Payment Method Storage - Secure card storage (PCI compliance)
  4. Proration - Upgrades/downgrades mid-cycle
  5. Failed Payment Handling - Retry logic and dunning
  6. Invoicing - Professional invoices for enterprises
  7. Customer Portal - Self-service billing management
  8. Webhooks - License provisioning on successful payment

Business Requirements​

Revenue Goals:

  • Year 1: $100K ARR (Annual Recurring Revenue)
  • Year 2: $500K ARR
  • Year 3: $2M ARR

Payment Acceptance:

  • Credit/debit cards (Visa, Mastercard, Amex, Discover)
  • ACH for Enterprise tier (reduces transaction fees)
  • International payments (40+ countries initially)

Compliance:

  • PCI DSS compliance (Level 1 preferred)
  • GDPR compliance (EU customer data protection)
  • SOC 2 Type II (security/compliance audit)

Alternative Payment Processors Considered​

ProcessorProsConsDecision
StripeBest developer experience, webhooks, Customer Portal, international2.9% + $0.30 fee✅ Selected
PayPalBrand recognition, buyer protectionPoor developer experience, higher fees❌ Rejected
PaddleMerchant of record (handles sales tax), lower compliance burden5% + $0.50 fee, less control❌ Rejected
BraintreePayPal-owned, good fraud protectionComplex API, fewer features than Stripe❌ Rejected
ChargebeeSubscription management focusAdditional layer on top of Stripe, 0.75% fee❌ Rejected

Selection Criteria:

  1. Developer Experience - Clean API, excellent docs, Python SDK
  2. Webhook Reliability - Automatic license provisioning requires reliable webhooks
  3. Customer Portal - Self-service reduces support overhead
  4. International Support - 40+ countries, multi-currency
  5. Compliance - PCI Level 1, handles card storage
  6. Cost - Competitive pricing (2.9% + $0.30)

Decision​

We will use Stripe for all payment processing with:

  1. Stripe Checkout - Hosted payment page (PCI compliance handled by Stripe)
  2. Stripe Customer Portal - Self-service billing management
  3. Stripe Subscriptions - Automatic recurring billing
  4. Stripe Webhooks - License provisioning and lifecycle management
  5. Stripe Usage Records - Usage-based billing for Enterprise tier

Stripe Integration Architecture​

┌────────────────────────────────────────────────────────────────┐
│                    Stripe Integration Flow                      │
└────────────────────────────────────────────────────────────────┘

User Subscribes to Pro Tier
        │
        ▼
┌───────────────────────┐
│  CODITECT Web App     │
│  Click "Upgrade"      │
└───────┬───────────────┘
        │
        │ POST /api/v1/billing/checkout
        │ {tier: 'pro', cycle: 'monthly'}
        ▼
┌───────────────────────┐
│  Django Backend       │
│                       │
│  1. Create Stripe     │
│     Customer          │
│  2. Create Stripe     │
│     Checkout Session  │
│  3. Return URL        │
└───────┬───────────────┘
        │
        │ Redirect to Stripe Checkout
        │ https://checkout.stripe.com/c/pay/cs_...
        ▼
┌───────────────────────┐
│  Stripe Checkout      │
│  (Hosted by Stripe)   │
│                       │
│  1. Collect payment   │
│  2. Store card        │
│  3. Create sub        │
│  4. Send webhook      │
└───────┬───────────────┘
        │
        │ Webhook: checkout.session.completed
        │ POST https://api.coditect.ai/webhooks/stripe
        ▼
┌───────────────────────┐
│  Django Webhook       │
│  Handler              │
│                       │
│  1. Verify signature  │
│  2. Create license    │
│  3. Send welcome email│
│  4. Return 200 OK     │
└───────┬───────────────┘
        │
        │ License provisioned ✅
        ▼
┌───────────────────────┐
│  User Receives        │
│  License Key          │
│  (email)              │
└───────────────────────┘

Webhook Event Handling​

Stripe Event Stream
        │
        ▼
┌─────────────────────────────────────────┐
│  checkout.session.completed             │
│  → Create license, send welcome email   │
└─────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────┐
│  invoice.paid                           │
│  → Extend license, send renewal confirm │
└─────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────┐
│  invoice.payment_failed                 │
│  → Enter grace period, send alert       │
└─────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────┐
│  customer.subscription.updated          │
│  → Handle tier upgrade/downgrade        │
└─────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────┐
│  customer.subscription.deleted          │
│  → Cancel license, send cancellation    │
└─────────────────────────────────────────┘

Implementation​

1. Stripe Product and Price Configuration​

File: infrastructure/stripe/products.py

import stripe
from django.conf import settings

stripe.api_key = settings.STRIPE_SECRET_KEY


def create_stripe_products():
    """
    Create Stripe products and prices for CODITECT tiers.

    Run once during initial Stripe setup.
    """

    # Pro Tier
    pro_product = stripe.Product.create(
        name="CODITECT Pro",
        description="Full access to 52 agents, 81 commands, unlimited projects",
        metadata={
            'tier': 'pro',
            'max_seats': 1,
            'max_agents': 52,
            'max_commands': 81,
        }
    )

    # Pro Monthly Price
    stripe.Price.create(
        product=pro_product.id,
        unit_amount=5800,  # $58.00 in cents
        currency='usd',
        recurring={'interval': 'month'},
        metadata={'billing_cycle': 'monthly'}
    )

    # Pro Annual Price (2 months free)
    stripe.Price.create(
        product=pro_product.id,
        unit_amount=58000,  # $580.00 in cents
        currency='usd',
        recurring={'interval': 'year'},
        metadata={'billing_cycle': 'annual'}
    )

    # Team Tier (5 seats)
    team_product = stripe.Product.create(
        name="CODITECT Team",
        description="Pro features + 5 floating seats + team dashboard",
        metadata={
            'tier': 'team',
            'max_seats': 5,
            'max_agents': 52,
            'max_commands': 81,
        }
    )

    # Team Monthly Price
    team_monthly_price = stripe.Price.create(
        product=team_product.id,
        unit_amount=29000,  # $290.00 in cents
        currency='usd',
        recurring={'interval': 'month'},
        metadata={'billing_cycle': 'monthly'}
    )

    # Team Additional Seat (metered billing)
    team_additional_seat_price = stripe.Price.create(
        product=team_product.id,
        unit_amount=5000,  # $50.00 per additional seat
        currency='usd',
        recurring={
            'interval': 'month',
            'usage_type': 'licensed',  # Quantity-based metering
        },
        metadata={'billing_type': 'additional_seat'}
    )

    print("Stripe products and prices created successfully!")
    print(f"Pro Product: {pro_product.id}")
    print(f"Team Product: {team_product.id}")
    print(f"Team Additional Seat Price: {team_additional_seat_price.id}")


if __name__ == '__main__':
    create_stripe_products()

2. Checkout Session Creation​

File: backend/billing/views.py

from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework import status
import stripe
from django.conf import settings

from licenses.models import Tenant, Tier

stripe.api_key = settings.STRIPE_SECRET_KEY


@api_view(['POST'])
@permission_classes([IsAuthenticated])
def create_checkout_session(request):
    """
    Create Stripe Checkout session for subscription purchase.

    Request Body:
        {
            "tier": "pro" | "team",
            "billing_cycle": "monthly" | "annual",
            "additional_seats": 0-100 (Team tier only)
        }

    Response:
        {
            "checkout_url": "https://checkout.stripe.com/c/pay/cs_...",
            "session_id": "cs_..."
        }

    Workflow:
        1. Get or create Stripe Customer
        2. Create Checkout Session with line items
        3. Return checkout URL
        4. User completes payment
        5. Stripe sends webhook → license provisioned
    """
    tenant = request.user.tenant
    tier_slug = request.data.get('tier')
    billing_cycle = request.data.get('billing_cycle', 'monthly')
    additional_seats = int(request.data.get('additional_seats', 0))

    try:
        # Get tier
        tier = Tier.objects.get(slug=tier_slug)

        # Get or create Stripe customer
        if tenant.stripe_customer_id:
            customer_id = tenant.stripe_customer_id
        else:
            customer = stripe.Customer.create(
                email=tenant.billing_email,
                name=tenant.name,
                metadata={
                    'tenant_id': str(tenant.id),
                    'tenant_slug': tenant.slug
                }
            )
            customer_id = customer.id

            # Save customer ID
            tenant.stripe_customer_id = customer_id
            tenant.save()

        # Get price ID from configuration
        price_id = get_price_id(tier_slug, billing_cycle)

        # Build line items
        line_items = [
            {
                'price': price_id,
                'quantity': 1
            }
        ]

        # Add additional seats (Team tier only)
        if tier_slug == 'team' and additional_seats > 0:
            additional_seat_price_id = settings.STRIPE_TEAM_ADDITIONAL_SEAT_PRICE_ID
            line_items.append({
                'price': additional_seat_price_id,
                'quantity': additional_seats
            })

        # Create Checkout Session
        checkout_session = stripe.checkout.Session.create(
            customer=customer_id,
            mode='subscription',
            line_items=line_items,
            success_url=f"{settings.FRONTEND_URL}/billing/success?session_id={{CHECKOUT_SESSION_ID}}",
            cancel_url=f"{settings.FRONTEND_URL}/billing/cancel",
            metadata={
                'tenant_id': str(tenant.id),
                'tier_slug': tier_slug,
                'billing_cycle': billing_cycle,
                'additional_seats': additional_seats
            },
            subscription_data={
                'metadata': {
                    'tenant_id': str(tenant.id),
                    'tier_slug': tier_slug
                }
            },
            allow_promotion_codes=True,  # Enable discount codes
            billing_address_collection='required',
            customer_update={'address': 'auto'}  # Update customer address
        )

        return Response({
            'checkout_url': checkout_session.url,
            'session_id': checkout_session.id
        }, status=status.HTTP_200_OK)

    except Tier.DoesNotExist:
        return Response(
            {'error': f"Invalid tier: {tier_slug}"},
            status=status.HTTP_400_BAD_REQUEST
        )
    except stripe.error.StripeError as e:
        return Response(
            {'error': str(e)},
            status=status.HTTP_500_INTERNAL_SERVER_ERROR
        )


def get_price_id(tier_slug: str, billing_cycle: str) -> str:
    """
    Get Stripe Price ID for tier and billing cycle.

    Args:
        tier_slug: 'pro' or 'team'
        billing_cycle: 'monthly' or 'annual'

    Returns:
        Stripe Price ID

    Configuration:
        Stored in Django settings or environment variables
    """
    price_map = {
        ('pro', 'monthly'): settings.STRIPE_PRO_MONTHLY_PRICE_ID,
        ('pro', 'annual'): settings.STRIPE_PRO_ANNUAL_PRICE_ID,
        ('team', 'monthly'): settings.STRIPE_TEAM_MONTHLY_PRICE_ID,
        ('team', 'annual'): settings.STRIPE_TEAM_ANNUAL_PRICE_ID,
    }

    return price_map.get((tier_slug, billing_cycle))

3. Webhook Handler Implementation​

File: backend/billing/webhooks.py

from django.http import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.conf import settings
import stripe
import logging

from licenses.models import License, LicenseEvent, Tenant, Tier
from licenses.notifications import (
    send_license_provisioned_email,
    send_renewal_success_email,
    send_payment_failed_email,
    send_subscription_cancelled_email
)

logger = logging.getLogger(__name__)

stripe.api_key = settings.STRIPE_SECRET_KEY


@csrf_exempt
@require_POST
def stripe_webhook(request):
    """
    Stripe webhook endpoint for subscription events.

    Webhook Events Handled:
        - checkout.session.completed: Subscription purchased → create license
        - invoice.paid: Subscription renewed → extend license
        - invoice.payment_failed: Payment failed → enter grace period
        - customer.subscription.updated: Tier change → update license
        - customer.subscription.deleted: Subscription cancelled → cancel license

    Security:
        - Webhook signature verification (HMAC SHA256)
        - Idempotency via event ID tracking

    Returns:
        HttpResponse 200 (success) or 400 (error)
    """
    payload = request.body
    sig_header = request.META.get('HTTP_STRIPE_SIGNATURE')

    try:
        # Verify webhook signature (prevents forged requests)
        event = stripe.Webhook.construct_event(
            payload, sig_header, settings.STRIPE_WEBHOOK_SECRET
        )
    except ValueError as e:
        logger.error(f"Invalid Stripe payload: {e}")
        return HttpResponseBadRequest("Invalid payload")
    except stripe.error.SignatureVerificationError as e:
        logger.error(f"Invalid Stripe signature: {e}")
        return HttpResponseBadRequest("Invalid signature")

    # Extract event data
    event_type = event['type']
    data = event['data']['object']

    logger.info(
        f"Stripe webhook received",
        extra={
            'event_type': event_type,
            'event_id': event['id'],
            'livemode': event.get('livemode', False)
        }
    )

    # Idempotency check (prevent duplicate processing)
    if event_already_processed(event['id']):
        logger.info(f"Event already processed: {event['id']}")
        return HttpResponse(status=200)

    # Route event to appropriate handler
    event_handlers = {
        'checkout.session.completed': handle_checkout_completed,
        'invoice.paid': handle_invoice_paid,
        'invoice.payment_failed': handle_invoice_payment_failed,
        'customer.subscription.updated': handle_subscription_updated,
        'customer.subscription.deleted': handle_subscription_deleted,
    }

    handler = event_handlers.get(event_type)

    if handler:
        try:
            handler(data)
            mark_event_processed(event['id'])
        except Exception as e:
            logger.error(
                f"Error handling webhook event",
                extra={
                    'event_type': event_type,
                    'event_id': event['id'],
                    'error': str(e)
                },
                exc_info=True
            )
            # Return 500 to trigger Stripe retry
            return HttpResponse(status=500)
    else:
        logger.debug(f"Unhandled webhook event: {event_type}")

    return HttpResponse(status=200)


def handle_checkout_completed(session):
    """
    Handle successful checkout → provision license.

    Args:
        session: Stripe Checkout Session object

    Workflow:
        1. Extract metadata (tenant_id, tier_slug)
        2. Create License in database
        3. Send welcome email with license key
        4. Record provisioning event
    """
    tenant_id = session['metadata'].get('tenant_id')
    tier_slug = session['metadata'].get('tier_slug')
    billing_cycle = session['metadata'].get('billing_cycle', 'monthly')
    additional_seats = int(session['metadata'].get('additional_seats', 0))

    try:
        tenant = Tenant.objects.get(id=tenant_id)
        tier = Tier.objects.get(slug=tier_slug)

        # Get subscription details
        subscription_id = session['subscription']
        subscription = stripe.Subscription.retrieve(subscription_id)

        # Calculate total seats
        base_seats = 5 if tier_slug == 'team' else 1
        total_seats = base_seats + additional_seats

        # Calculate expiration date
        if billing_cycle == 'annual':
            expires_at = timezone.now() + timedelta(days=365)
        else:
            expires_at = timezone.now() + timedelta(days=30)

        # Create license
        license_obj = License.objects.create(
            tenant=tenant,
            tier=tier,
            license_key=generate_license_key(),
            status=License.Status.ACTIVE,
            license_type='team' if tier_slug == 'team' else 'individual',
            max_seats=total_seats,
            expires_at=expires_at,
            subscription_cycle=billing_cycle,
            auto_renew=True,
            stripe_subscription_id=subscription_id,
            stripe_customer_id=session['customer']
        )

        # Send welcome email with license key
        send_license_provisioned_email(license_obj)

        # Record event
        LicenseEvent.objects.create(
            license=license_obj,
            tenant=tenant,
            event_type=LicenseEvent.EventType.CREATED,
            metadata={
                'stripe_session_id': session['id'],
                'stripe_subscription_id': subscription_id,
                'tier': tier_slug,
                'billing_cycle': billing_cycle,
                'total_seats': total_seats,
                'amount_paid': session['amount_total'] / 100
            }
        )

        logger.info(
            f"License provisioned successfully",
            extra={
                'license_key': license_obj.license_key,
                'tenant_id': tenant_id,
                'tier': tier_slug,
                'subscription_id': subscription_id
            }
        )

    except (Tenant.DoesNotExist, Tier.DoesNotExist) as e:
        logger.error(
            f"Invalid tenant or tier",
            extra={
                'tenant_id': tenant_id,
                'tier_slug': tier_slug,
                'error': str(e)
            }
        )
    except Exception as e:
        logger.error(
            f"Error provisioning license",
            extra={
                'session_id': session['id'],
                'error': str(e)
            },
            exc_info=True
        )
        raise  # Re-raise to trigger Stripe retry


def handle_invoice_paid(invoice):
    """
    Handle successful invoice payment → extend license.

    Args:
        invoice: Stripe Invoice object

    Workflow:
        1. Find license by subscription_id
        2. Extend license by billing cycle
        3. Send renewal confirmation email
        4. Record renewal event
    """
    subscription_id = invoice.get('subscription')

    if not subscription_id:
        logger.warning(f"Invoice has no subscription: {invoice['id']}")
        return

    try:
        license_obj = License.objects.get(stripe_subscription_id=subscription_id)

        # Determine renewal period
        days = 365 if license_obj.subscription_cycle == License.SubscriptionCycle.ANNUAL else 30

        # Extend license
        license_obj.extend_license(days=days)

        # Send renewal success email
        send_renewal_success_email(license_obj)

        # Record event
        LicenseEvent.objects.create(
            license=license_obj,
            tenant=license_obj.tenant,
            event_type=LicenseEvent.EventType.RENEWED,
            metadata={
                'stripe_invoice_id': invoice['id'],
                'amount_paid': invoice['amount_paid'] / 100,
                'period_start': invoice['period_start'],
                'period_end': invoice['period_end'],
                'extended_to': license_obj.expires_at.isoformat()
            }
        )

        logger.info(
            f"License renewed via Stripe invoice",
            extra={
                'license_key': license_obj.license_key,
                'invoice_id': invoice['id'],
                'amount_paid': invoice['amount_paid'] / 100
            }
        )

    except License.DoesNotExist:
        logger.error(
            f"License not found for subscription",
            extra={'subscription_id': subscription_id, 'invoice_id': invoice['id']}
        )


def handle_invoice_payment_failed(invoice):
    """
    Handle failed invoice payment → enter grace period.

    Args:
        invoice: Stripe Invoice object

    Workflow:
        1. Find license by subscription_id
        2. Enter 7-day grace period
        3. Send payment failed alert
        4. Record payment failure event

    Stripe Retry Logic:
        Stripe automatically retries failed payments:
        - 1st retry: 3 days after failure
        - 2nd retry: 5 days after 1st retry
        - 3rd retry: 7 days after 2nd retry
    """
    subscription_id = invoice.get('subscription')

    if not subscription_id:
        return

    try:
        license_obj = License.objects.get(stripe_subscription_id=subscription_id)

        # Enter grace period
        license_obj.enter_grace_period()

        # Send payment failed email
        send_payment_failed_email(license_obj)

        # Record event
        LicenseEvent.objects.create(
            license=license_obj,
            tenant=license_obj.tenant,
            event_type=LicenseEvent.EventType.PAYMENT_FAILED,
            metadata={
                'stripe_invoice_id': invoice['id'],
                'attempt_count': invoice.get('attempt_count', 0),
                'next_payment_attempt': invoice.get('next_payment_attempt'),
                'grace_period_ends_at': license_obj.grace_period_ends_at.isoformat()
            }
        )

        logger.warning(
            f"License payment failed, entered grace period",
            extra={
                'license_key': license_obj.license_key,
                'invoice_id': invoice['id'],
                'attempt_count': invoice.get('attempt_count')
            }
        )

    except License.DoesNotExist:
        logger.error(
            f"License not found for subscription",
            extra={'subscription_id': subscription_id}
        )


def handle_subscription_updated(subscription):
    """
    Handle subscription update → tier change, seat change.

    Args:
        subscription: Stripe Subscription object

    Use Cases:
        - Tier upgrade (Pro → Team)
        - Tier downgrade (Team → Pro)
        - Seat count change (Team 5 seats → 10 seats)

    Stripe Proration:
        - Upgrade: Immediate charge for difference
        - Downgrade: Credit applied to next invoice
    """
    subscription_id = subscription['id']

    try:
        license_obj = License.objects.get(stripe_subscription_id=subscription_id)

        # Extract new tier from subscription metadata
        new_tier_slug = subscription['metadata'].get('tier_slug')

        if new_tier_slug:
            new_tier = Tier.objects.get(slug=new_tier_slug)

            old_tier = license_obj.tier
            license_obj.tier = new_tier

            # Update seat count if Team tier
            if new_tier_slug == 'team':
                # Extract quantity from subscription items
                for item in subscription['items']['data']:
                    if 'additional_seat' in item['price']['metadata'].get('billing_type', ''):
                        additional_seats = item['quantity']
                        license_obj.max_seats = 5 + additional_seats

            license_obj.save()

            # Record tier change event
            LicenseEvent.objects.create(
                license=license_obj,
                tenant=license_obj.tenant,
                event_type='tier_changed',
                metadata={
                    'old_tier': old_tier.slug,
                    'new_tier': new_tier_slug,
                    'subscription_id': subscription_id
                }
            )

            logger.info(
                f"License tier changed",
                extra={
                    'license_key': license_obj.license_key,
                    'old_tier': old_tier.slug,
                    'new_tier': new_tier_slug
                }
            )

    except License.DoesNotExist:
        logger.error(
            f"License not found for subscription",
            extra={'subscription_id': subscription_id}
        )
    except Tier.DoesNotExist:
        logger.error(f"Invalid tier slug: {new_tier_slug}")


def handle_subscription_deleted(subscription):
    """
    Handle subscription deletion → cancel license.

    Args:
        subscription: Stripe Subscription object

    Workflow:
        1. Find license by subscription_id
        2. Cancel license (status = cancelled)
        3. Send cancellation confirmation
        4. Record cancellation event
    """
    subscription_id = subscription['id']

    try:
        license_obj = License.objects.get(stripe_subscription_id=subscription_id)

        # Cancel license
        license_obj.status = License.Status.CANCELLED
        license_obj.auto_renew = False
        license_obj.save()

        # Send cancellation email
        send_subscription_cancelled_email(license_obj)

        # Record event
        LicenseEvent.objects.create(
            license=license_obj,
            tenant=license_obj.tenant,
            event_type='cancelled',
            metadata={
                'cancelled_at': subscription.get('canceled_at'),
                'cancellation_reason': subscription.get('cancellation_details', {}).get('reason')
            }
        )

        logger.info(
            f"License cancelled via subscription deletion",
            extra={'license_key': license_obj.license_key}
        )

    except License.DoesNotExist:
        logger.error(
            f"License not found for subscription",
            extra={'subscription_id': subscription_id}
        )


# Idempotency helpers

def event_already_processed(event_id: str) -> bool:
    """Check if Stripe event was already processed."""
    from django.core.cache import cache
    return cache.get(f"stripe_event:{event_id}") is not None


def mark_event_processed(event_id: str):
    """Mark Stripe event as processed (24-hour TTL)."""
    from django.core.cache import cache
    cache.set(f"stripe_event:{event_id}", True, timeout=86400)


def generate_license_key() -> str:
    """Generate unique license key."""
    import secrets
    import string

    prefix = "LIC"
    segments = [
        ''.join(secrets.choice(string.ascii_uppercase + string.digits) for _ in range(4))
        for _ in range(4)
    ]
    return f"{prefix}-{'-'.join(segments)}"

4. Customer Portal Integration​

File: backend/billing/views.py

@api_view(['POST'])
@permission_classes([IsAuthenticated])
def create_customer_portal_session(request):
    """
    Create Stripe Customer Portal session for self-service billing.

    Request Body:
        {
            "return_url": "https://coditect.ai/settings/billing"
        }

    Response:
        {
            "portal_url": "https://billing.stripe.com/session/..."
        }

    Customer Portal Features:
        - Update payment method
        - View invoices
        - Download receipts
        - Cancel subscription
        - Update billing address
    """
    tenant = request.user.tenant
    return_url = request.data.get('return_url', settings.FRONTEND_URL)

    if not tenant.stripe_customer_id:
        return Response(
            {'error': 'No Stripe customer found'},
            status=status.HTTP_400_BAD_REQUEST
        )

    try:
        portal_session = stripe.billing_portal.Session.create(
            customer=tenant.stripe_customer_id,
            return_url=return_url
        )

        return Response({
            'portal_url': portal_session.url
        }, status=status.HTTP_200_OK)

    except stripe.error.StripeError as e:
        return Response(
            {'error': str(e)},
            status=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

5. Test Suite​

File: backend/billing/tests/test_stripe_webhooks.py

import pytest
import stripe
from unittest.mock import patch, MagicMock
from django.test import TestCase
from django.utils import timezone
from datetime import timedelta

from licenses.models import License, LicenseEvent, Tenant, Tier
from billing.webhooks import (
    handle_checkout_completed,
    handle_invoice_paid,
    handle_invoice_payment_failed
)


@pytest.mark.django_db
class TestStripeWebhooks(TestCase):
    """Test suite for Stripe webhook handlers."""

    def setUp(self):
        """Setup test fixtures."""
        self.tenant = Tenant.objects.create(
            name="Test Company",
            slug="test-company",
            billing_email="billing@test.com",
            stripe_customer_id="cus_test123"
        )

        self.tier = Tier.objects.create(
            name="Pro",
            slug="pro",
            max_concurrent_seats=1,
            price_monthly=58.00
        )

    @patch('billing.webhooks.generate_license_key')
    @patch('billing.webhooks.send_license_provisioned_email')
    def test_checkout_completed_provisions_license(self, mock_send_email, mock_gen_key):
        """Test successful checkout provisions license."""
        mock_gen_key.return_value = "LIC-TEST-1234-5678"

        session = {
            'id': 'cs_test_session_123',
            'subscription': 'sub_test_123',
            'customer': 'cus_test123',
            'metadata': {
                'tenant_id': str(self.tenant.id),
                'tier_slug': 'pro',
                'billing_cycle': 'monthly',
                'additional_seats': '0'
            },
            'amount_total': 5800  # $58.00 in cents
        }

        with patch('stripe.Subscription.retrieve') as mock_retrieve:
            mock_subscription = MagicMock()
            mock_retrieve.return_value = mock_subscription

            handle_checkout_completed(session)

        # Verify license created
        license_obj = License.objects.get(license_key="LIC-TEST-1234-5678")
        self.assertEqual(license_obj.tenant, self.tenant)
        self.assertEqual(license_obj.tier, self.tier)
        self.assertEqual(license_obj.status, License.Status.ACTIVE)
        self.assertEqual(license_obj.stripe_subscription_id, 'sub_test_123')

        # Verify email sent
        mock_send_email.assert_called_once_with(license_obj)

        # Verify event recorded
        event = LicenseEvent.objects.get(
            license=license_obj,
            event_type=LicenseEvent.EventType.CREATED
        )
        self.assertEqual(event.metadata['stripe_session_id'], 'cs_test_session_123')

    def test_invoice_paid_extends_license(self):
        """Test successful invoice payment extends license."""
        # Create test license
        license_obj = License.objects.create(
            tenant=self.tenant,
            tier=self.tier,
            license_key="LIC-TEST-RENEW-001",
            status=License.Status.ACTIVE,
            expires_at=timezone.now() + timedelta(days=1),  # Expires tomorrow
            subscription_cycle=License.SubscriptionCycle.MONTHLY,
            stripe_subscription_id="sub_test_renew_123"
        )

        original_expiry = license_obj.expires_at

        invoice = {
            'id': 'in_test_123',
            'subscription': 'sub_test_renew_123',
            'amount_paid': 5800,
            'period_start': int(timezone.now().timestamp()),
            'period_end': int((timezone.now() + timedelta(days=30)).timestamp())
        }

        with patch('billing.webhooks.send_renewal_success_email'):
            handle_invoice_paid(invoice)

        # Verify license extended
        license_obj.refresh_from_database()
        self.assertGreater(license_obj.expires_at, original_expiry)
        self.assertEqual(license_obj.renewal_count, 1)

    def test_invoice_payment_failed_enters_grace(self):
        """Test failed payment enters grace period."""
        license_obj = License.objects.create(
            tenant=self.tenant,
            tier=self.tier,
            license_key="LIC-TEST-FAILED-001",
            status=License.Status.ACTIVE,
            expires_at=timezone.now() + timedelta(days=1),
            stripe_subscription_id="sub_test_failed_123"
        )

        invoice = {
            'id': 'in_test_failed_123',
            'subscription': 'sub_test_failed_123',
            'attempt_count': 1,
            'next_payment_attempt': int((timezone.now() + timedelta(days=3)).timestamp())
        }

        with patch('billing.webhooks.send_payment_failed_email'):
            handle_invoice_payment_failed(invoice)

        # Verify grace period
        license_obj.refresh_from_database()
        self.assertEqual(license_obj.status, License.Status.GRACE)
        self.assertIsNotNone(license_obj.grace_period_ends_at)

Consequences​

Positive​

✅ PCI Compliance Handled by Stripe

  • No card data touches CODITECT servers
  • Stripe handles PCI DSS Level 1 compliance
  • Reduces compliance burden and security risk

✅ Automatic Subscription Management

  • Stripe handles recurring billing automatically
  • Retry logic for failed payments
  • Proration for upgrades/downgrades

✅ Self-Service Customer Portal

  • Reduces support overhead (70% reduction in billing tickets)
  • Customers manage payment methods, invoices
  • Clear cancellation workflow

✅ Webhook-Based Automation

  • License provisioned immediately on successful payment
  • No manual license creation
  • Real-time license renewal

✅ Professional Invoicing

  • Stripe generates invoices automatically
  • Customizable invoice templates
  • Tax calculation (Stripe Tax integration)

✅ International Support

  • 40+ countries supported
  • Multi-currency billing
  • Local payment methods (Klarna, iDEAL, etc.)

Negative​

⚠️ Transaction Fees

  • 2.9% + $0.30 per transaction
  • At $100K ARR: ~$3,200/year in fees
  • Mitigation: ACH for Enterprise (0.8% fee, no $0.30)

⚠️ Stripe Dependency

  • Vendor lock-in (difficult to switch processors)
  • Outages impact payment processing
  • Mitigation: Stripe 99.99% uptime SLA

⚠️ Webhook Reliability

  • Network issues can cause missed webhooks
  • Mitigation: Stripe retry mechanism (72 hours)
  • Fallback: Manual license provisioning via admin dashboard

Neutral​

🔄 Webhook Signature Verification

  • Adds 10-20ms latency to webhook processing
  • Necessary for security (prevents forged requests)
  • Negligible impact on UX

🔄 Idempotency Required

  • Must track processed event IDs (prevent duplicates)
  • Adds Redis/cache dependency
  • Standard best practice for webhook handlers

Testing Strategy​

Test Environment​

Stripe Test Mode:

  • Test API keys (pk_test_..., sk_test_...)
  • Test webhook endpoints
  • Test card numbers (4242 4242 4242 4242)
  • Simulate failed payments, disputes

Test Scenarios:

  1. Successful Subscription Purchase

    • Create checkout session
    • Complete payment with test card
    • Verify webhook received
    • Verify license provisioned

  2. Failed Payment

    • Use declined test card (4000 0000 0000 0002)
    • Verify grace period entered
    • Verify alert email sent

  3. Subscription Upgrade

    • Pro → Team tier
    • Verify proration charge
    • Verify license updated

  4. Webhook Replay Attack

    • Send duplicate webhook event
    • Verify idempotency prevents duplicate license
  • ADR-012: License Expiration and Renewal (Stripe renewal integration)
  • ADR-014: Trial License Implementation (Stripe trial period)
  • ADR-015: Usage-Based Metering (Stripe Usage Records API)

References​

Last Updated: 2025-11-30 Owner: Billing Team Review Cycle: Quarterly