Add finance app with Stripe Checkout subscriptions (#21) (#23)

## Summary - Closes #21 — new Django `finance` app with Stripe as payment provider - Subscription price defaults to **$10 USD / month** via `SUBSCRIPTION_PRICE_AMOUNT_CENTS = 1000` in `settings.py` (env-overridable) - Persists **Invoice** and **Payment** rows; both registered in Django admin (with payment inline on invoices) - Checkout Session API redirects users to Stripe hosted payment; webhook verifies signatures and upserts ledger idempotently ## API - `POST /api/finance/checkout/` — JWT auth → `{ checkout_url, session_id }` - `GET /api/finance/invoices/` / `GET /api/finance/payments/` — own records - `POST /api/finance/webhooks/stripe/` — Stripe signature-verified webhook ## Config Documented in `.env.example` / `.env.prod.example`: `STRIPE_SECRET_KEY`, `STRIPE_PUBLISHABLE_KEY`, `STRIPE_WEBHOOK_SECRET`, optional `STRIPE_PRICE_ID`, `FRONTEND_BASE_URL` ## Test plan - [x] `uv run python manage.py test finance` (17 tests) - [ ] Set Stripe test keys locally; create checkout session; complete payment in Stripe test mode - [ ] Confirm Invoice/Payment appear in `/admin/` - [ ] Point Stripe webhook to `/api/finance/webhooks/stripe/` and verify `checkout.session.completed` / `invoice.paid`Reviewed-on: #23
2026-07-26 17:35:06 -07:00
parent 9984d1c340
commit ad44359804
23 changed files with 1540 additions and 0 deletions
@@ -25,6 +25,21 @@ EMAIL_USE_TLS=true
 # Captcha (optional local)
 CAPTCHA_SECRET_KEY=

+# Stripe / finance (optional local — required for checkout + webhooks)
+STRIPE_SECRET_KEY=
+STRIPE_PUBLISHABLE_KEY=
+STRIPE_WEBHOOK_SECRET=
+# Optional: pre-created Stripe Price ID. When empty, Checkout uses
+# SUBSCRIPTION_PRICE_* from settings.py ($10 USD / month by default).
+STRIPE_PRICE_ID=
+# SUBSCRIPTION_PRICE_AMOUNT_CENTS=1000
+# SUBSCRIPTION_PRICE_CURRENCY=usd
+# SUBSCRIPTION_PRICE_INTERVAL=month
+# SUBSCRIPTION_PRODUCT_NAME=Chat Subscription
+FRONTEND_BASE_URL=http://localhost:3000
+# STRIPE_CHECKOUT_SUCCESS_URL=http://localhost:3000/billing/success?session_id={CHECKOUT_SESSION_ID}
+# STRIPE_CHECKOUT_CANCEL_URL=http://localhost:3000/billing/cancel
+
 # Gunicorn / ASGI
 GUNICORN_WORKERS=2
 GUNICORN_BIND=0.0.0.0:8000
@@ -45,6 +45,17 @@ EMAIL_USE_TLS=true
 # Captcha
 CAPTCHA_SECRET_KEY=replace-with-captcha-secret

+# Stripe / finance
+STRIPE_SECRET_KEY=replace-with-stripe-secret-key
+STRIPE_PUBLISHABLE_KEY=replace-with-stripe-publishable-key
+STRIPE_WEBHOOK_SECRET=replace-with-stripe-webhook-secret
+# Optional: pre-created Stripe Price ID. When empty, Checkout uses
+# SUBSCRIPTION_PRICE_* from settings.py ($10 USD / month by default).
+STRIPE_PRICE_ID=
+FRONTEND_BASE_URL=https://chat.aimloperations.com
+# STRIPE_CHECKOUT_SUCCESS_URL=https://chat.aimloperations.com/billing/success?session_id={CHECKOUT_SESSION_ID}
+# STRIPE_CHECKOUT_CANCEL_URL=https://chat.aimloperations.com/billing/cancel
+
 # Gunicorn / ASGI (UvicornWorker for WebSockets)
 GUNICORN_WORKERS=2
 GUNICORN_BIND=0.0.0.0:8000
@@ -90,6 +90,9 @@ with `COMPOSE_DATABASE_URL` if needed.
 | `OLLAMA_MODEL` / `OLLAMA_EMBED_MODEL` | from `DEBUG` | optional | Override model names |
 | `EMAIL_HOST_*` | empty | yes (prod/beta) | SMTP2GO |
 | `CAPTCHA_SECRET_KEY` | empty | recommended | |
+| `STRIPE_SECRET_KEY` / `STRIPE_PUBLISHABLE_KEY` / `STRIPE_WEBHOOK_SECRET` | empty | yes for billing | Stripe API + webhook |
+| `STRIPE_PRICE_ID` | empty | optional | Pre-created Price; else `$10/mo` from settings |
+| `FRONTEND_BASE_URL` | `http://localhost:3000` | set in prod | Checkout success/cancel base |
 | `CORS_ALLOWED_ORIGINS` | local + chat FE | set in prod | Frontend origin |
 | `USE_TLS_PROXY` | false (dev) | true behind NPM | Sets `SECURE_PROXY_SSL_HEADER` |
 | `GUNICORN_WORKERS` / `GUNICORN_BIND` | 2 / `0.0.0.0:8000` | optional | Entrypoint |
@@ -0,0 +1,86 @@
+from django.contrib import admin
+
+from finance.models import Invoice, Payment
+
+
+class PaymentInline(admin.TabularInline):
+    model = Payment
+    extra = 0
+    readonly_fields = (
+        "provider",
+        "status",
+        "amount",
+        "currency",
+        "stripe_payment_intent_id",
+        "stripe_charge_id",
+        "paid_at",
+        "failure_message",
+        "created",
+        "last_modified",
+    )
+    can_delete = False
+    show_change_link = True
+
+
+@admin.register(Invoice)
+class InvoiceAdmin(admin.ModelAdmin):
+    list_display = (
+        "id",
+        "user",
+        "company",
+        "provider",
+        "status",
+        "amount_due",
+        "amount_paid",
+        "currency",
+        "period_start",
+        "period_end",
+        "stripe_invoice_id",
+        "stripe_checkout_session_id",
+        "created",
+    )
+    list_filter = ("provider", "status", "currency")
+    search_fields = (
+        "user__email",
+        "user__username",
+        "company__name",
+        "stripe_invoice_id",
+        "stripe_checkout_session_id",
+        "stripe_subscription_id",
+        "stripe_customer_id",
+        "description",
+    )
+    readonly_fields = ("created", "last_modified")
+    raw_id_fields = ("user", "company")
+    inlines = [PaymentInline]
+    date_hierarchy = "created"
+
+
+@admin.register(Payment)
+class PaymentAdmin(admin.ModelAdmin):
+    list_display = (
+        "id",
+        "user",
+        "company",
+        "invoice",
+        "provider",
+        "status",
+        "amount",
+        "currency",
+        "stripe_payment_intent_id",
+        "stripe_charge_id",
+        "paid_at",
+        "created",
+    )
+    list_filter = ("provider", "status", "currency")
+    search_fields = (
+        "user__email",
+        "user__username",
+        "company__name",
+        "stripe_payment_intent_id",
+        "stripe_charge_id",
+        "invoice__stripe_invoice_id",
+    )
+    readonly_fields = ("created", "last_modified")
+    raw_id_fields = ("user", "company", "invoice")
+    date_hierarchy = "created"
@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class FinanceConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "finance"
+    verbose_name = "Finance"
@@ -0,0 +1,237 @@
+# Generated by Django 6.0 on 2026-07-27 00:16
+
+import django.db.models.deletion
+import django.utils.timezone
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("chat_backend", "0023_promptmetric_tokens_in_promptmetric_tokens_out"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="Invoice",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created", models.DateTimeField(default=django.utils.timezone.now)),
+                (
+                    "last_modified",
+                    models.DateTimeField(default=django.utils.timezone.now),
+                ),
+                (
+                    "provider",
+                    models.CharField(
+                        choices=[("stripe", "Stripe")], default="stripe", max_length=32
+                    ),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("draft", "Draft"),
+                            ("open", "Open"),
+                            ("paid", "Paid"),
+                            ("void", "Void"),
+                            ("uncollectible", "Uncollectible"),
+                            ("payment_failed", "Payment failed"),
+                        ],
+                        db_index=True,
+                        default="open",
+                        max_length=32,
+                    ),
+                ),
+                ("currency", models.CharField(default="usd", max_length=8)),
+                (
+                    "amount_due",
+                    models.PositiveIntegerField(
+                        default=0,
+                        help_text="Amount due in the smallest currency unit (e.g. cents).",
+                    ),
+                ),
+                (
+                    "amount_paid",
+                    models.PositiveIntegerField(
+                        default=0,
+                        help_text="Amount paid in the smallest currency unit (e.g. cents).",
+                    ),
+                ),
+                ("period_start", models.DateTimeField(blank=True, null=True)),
+                ("period_end", models.DateTimeField(blank=True, null=True)),
+                (
+                    "stripe_invoice_id",
+                    models.CharField(
+                        blank=True,
+                        db_index=True,
+                        max_length=255,
+                        null=True,
+                        unique=True,
+                    ),
+                ),
+                (
+                    "stripe_checkout_session_id",
+                    models.CharField(
+                        blank=True,
+                        db_index=True,
+                        max_length=255,
+                        null=True,
+                        unique=True,
+                    ),
+                ),
+                (
+                    "stripe_subscription_id",
+                    models.CharField(
+                        blank=True, db_index=True, max_length=255, null=True
+                    ),
+                ),
+                (
+                    "stripe_customer_id",
+                    models.CharField(blank=True, default="", max_length=255),
+                ),
+                ("hosted_invoice_url", models.URLField(blank=True, default="")),
+                (
+                    "description",
+                    models.CharField(blank=True, default="", max_length=512),
+                ),
+                (
+                    "company",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="invoices",
+                        to="chat_backend.company",
+                    ),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="invoices",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["-created"],
+            },
+        ),
+        migrations.CreateModel(
+            name="Payment",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created", models.DateTimeField(default=django.utils.timezone.now)),
+                (
+                    "last_modified",
+                    models.DateTimeField(default=django.utils.timezone.now),
+                ),
+                (
+                    "provider",
+                    models.CharField(
+                        choices=[("stripe", "Stripe")], default="stripe", max_length=32
+                    ),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("pending", "Pending"),
+                            ("succeeded", "Succeeded"),
+                            ("failed", "Failed"),
+                            ("canceled", "Canceled"),
+                            ("requires_action", "Requires action"),
+                        ],
+                        db_index=True,
+                        default="pending",
+                        max_length=32,
+                    ),
+                ),
+                ("currency", models.CharField(default="usd", max_length=8)),
+                (
+                    "amount",
+                    models.PositiveIntegerField(
+                        default=0,
+                        help_text="Amount in the smallest currency unit (e.g. cents).",
+                    ),
+                ),
+                (
+                    "stripe_payment_intent_id",
+                    models.CharField(
+                        blank=True,
+                        db_index=True,
+                        max_length=255,
+                        null=True,
+                        unique=True,
+                    ),
+                ),
+                (
+                    "stripe_charge_id",
+                    models.CharField(
+                        blank=True,
+                        db_index=True,
+                        max_length=255,
+                        null=True,
+                        unique=True,
+                    ),
+                ),
+                ("paid_at", models.DateTimeField(blank=True, null=True)),
+                (
+                    "failure_message",
+                    models.CharField(blank=True, default="", max_length=512),
+                ),
+                (
+                    "company",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="payments",
+                        to="chat_backend.company",
+                    ),
+                ),
+                (
+                    "invoice",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="payments",
+                        to="finance.invoice",
+                    ),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="payments",
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "ordering": ["-created"],
+            },
+        ),
+    ]
@@ -0,0 +1,161 @@
+from django.conf import settings
+from django.db import models
+from django.utils import timezone
+
+from chat_backend.models import Company, TimeInfoBase
+
+
+class Invoice(TimeInfoBase):
+    """Local ledger row for a billed period / Stripe invoice or checkout session."""
+
+    class Provider(models.TextChoices):
+        STRIPE = "stripe", "Stripe"
+
+    class Status(models.TextChoices):
+        DRAFT = "draft", "Draft"
+        OPEN = "open", "Open"
+        PAID = "paid", "Paid"
+        VOID = "void", "Void"
+        UNCOLLECTIBLE = "uncollectible", "Uncollectible"
+        PAYMENT_FAILED = "payment_failed", "Payment failed"
+
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.CASCADE,
+        related_name="invoices",
+    )
+    company = models.ForeignKey(
+        Company,
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="invoices",
+    )
+    provider = models.CharField(
+        max_length=32,
+        choices=Provider.choices,
+        default=Provider.STRIPE,
+    )
+    status = models.CharField(
+        max_length=32,
+        choices=Status.choices,
+        default=Status.OPEN,
+        db_index=True,
+    )
+    currency = models.CharField(max_length=8, default="usd")
+    amount_due = models.PositiveIntegerField(
+        default=0,
+        help_text="Amount due in the smallest currency unit (e.g. cents).",
+    )
+    amount_paid = models.PositiveIntegerField(
+        default=0,
+        help_text="Amount paid in the smallest currency unit (e.g. cents).",
+    )
+    period_start = models.DateTimeField(null=True, blank=True)
+    period_end = models.DateTimeField(null=True, blank=True)
+    stripe_invoice_id = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        unique=True,
+        db_index=True,
+    )
+    stripe_checkout_session_id = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        unique=True,
+        db_index=True,
+    )
+    stripe_subscription_id = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        db_index=True,
+    )
+    stripe_customer_id = models.CharField(max_length=255, blank=True, default="")
+    hosted_invoice_url = models.URLField(blank=True, default="")
+    description = models.CharField(max_length=512, blank=True, default="")
+
+    class Meta:
+        ordering = ["-created"]
+
+    def __str__(self) -> str:
+        return f"Invoice {self.pk} ({self.status}) user={self.user_id}"
+
+
+class Payment(TimeInfoBase):
+    """Local ledger row for a payment attempt / Stripe PaymentIntent or charge."""
+
+    class Provider(models.TextChoices):
+        STRIPE = "stripe", "Stripe"
+
+    class Status(models.TextChoices):
+        PENDING = "pending", "Pending"
+        SUCCEEDED = "succeeded", "Succeeded"
+        FAILED = "failed", "Failed"
+        CANCELED = "canceled", "Canceled"
+        REQUIRES_ACTION = "requires_action", "Requires action"
+
+    invoice = models.ForeignKey(
+        Invoice,
+        on_delete=models.CASCADE,
+        related_name="payments",
+        null=True,
+        blank=True,
+    )
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.CASCADE,
+        related_name="payments",
+    )
+    company = models.ForeignKey(
+        Company,
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="payments",
+    )
+    provider = models.CharField(
+        max_length=32,
+        choices=Provider.choices,
+        default=Provider.STRIPE,
+    )
+    status = models.CharField(
+        max_length=32,
+        choices=Status.choices,
+        default=Status.PENDING,
+        db_index=True,
+    )
+    currency = models.CharField(max_length=8, default="usd")
+    amount = models.PositiveIntegerField(
+        default=0,
+        help_text="Amount in the smallest currency unit (e.g. cents).",
+    )
+    stripe_payment_intent_id = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        unique=True,
+        db_index=True,
+    )
+    stripe_charge_id = models.CharField(
+        max_length=255,
+        blank=True,
+        null=True,
+        unique=True,
+        db_index=True,
+    )
+    paid_at = models.DateTimeField(null=True, blank=True)
+    failure_message = models.CharField(max_length=512, blank=True, default="")
+
+    class Meta:
+        ordering = ["-created"]
+
+    def __str__(self) -> str:
+        return f"Payment {self.pk} ({self.status}) user={self.user_id}"
+
+    def mark_succeeded(self, *, paid_at=None):
+        self.status = self.Status.SUCCEEDED
+        self.paid_at = paid_at or timezone.now()
+        self.save(update_fields=["status", "paid_at", "last_modified"])
@@ -0,0 +1,51 @@
+from rest_framework import serializers
+
+from finance.models import Invoice, Payment
+
+
+class InvoiceSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Invoice
+        fields = [
+            "id",
+            "provider",
+            "status",
+            "currency",
+            "amount_due",
+            "amount_paid",
+            "period_start",
+            "period_end",
+            "stripe_invoice_id",
+            "stripe_checkout_session_id",
+            "stripe_subscription_id",
+            "hosted_invoice_url",
+            "description",
+            "created",
+            "last_modified",
+        ]
+        read_only_fields = fields
+
+
+class PaymentSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Payment
+        fields = [
+            "id",
+            "invoice",
+            "provider",
+            "status",
+            "currency",
+            "amount",
+            "stripe_payment_intent_id",
+            "stripe_charge_id",
+            "paid_at",
+            "failure_message",
+            "created",
+            "last_modified",
+        ]
+        read_only_fields = fields
+
+
+class CheckoutSessionSerializer(serializers.Serializer):
+    success_url = serializers.URLField(required=False, allow_blank=False)
+    cancel_url = serializers.URLField(required=False, allow_blank=False)
@@ -0,0 +1,71 @@
+"""Stripe Checkout session helpers."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import stripe
+from django.conf import settings
+
+
+class StripeNotConfiguredError(RuntimeError):
+    """Raised when Stripe secret key is missing."""
+
+
+def configure_stripe() -> str:
+    secret = settings.STRIPE_SECRET_KEY
+    if not secret:
+        raise StripeNotConfiguredError(
+            "STRIPE_SECRET_KEY is not configured. Set it in the environment."
+        )
+    stripe.api_key = secret
+    return secret
+
+
+def subscription_line_items() -> list[dict[str, Any]]:
+    """Build Checkout line_items from settings-backed subscription pricing."""
+    price_id = settings.STRIPE_PRICE_ID
+    if price_id:
+        return [{"price": price_id, "quantity": 1}]
+
+    return [
+        {
+            "price_data": {
+                "currency": settings.SUBSCRIPTION_PRICE_CURRENCY,
+                "unit_amount": settings.SUBSCRIPTION_PRICE_AMOUNT_CENTS,
+                "recurring": {"interval": settings.SUBSCRIPTION_PRICE_INTERVAL},
+                "product_data": {
+                    "name": settings.SUBSCRIPTION_PRODUCT_NAME,
+                },
+            },
+            "quantity": 1,
+        }
+    ]
+
+
+def create_checkout_session(
+    *,
+    user,
+    success_url: str | None = None,
+    cancel_url: str | None = None,
+):
+    """Create a Stripe Checkout Session for the subscription plan."""
+    configure_stripe()
+
+    metadata = {
+        "user_id": str(user.pk),
+        "company_id": str(user.company_id) if user.company_id else "",
+    }
+    customer_email = getattr(user, "email", None) or None
+
+    session = stripe.checkout.Session.create(
+        mode="subscription",
+        line_items=subscription_line_items(),
+        success_url=success_url or settings.STRIPE_CHECKOUT_SUCCESS_URL,
+        cancel_url=cancel_url or settings.STRIPE_CHECKOUT_CANCEL_URL,
+        customer_email=customer_email,
+        client_reference_id=str(user.pk),
+        metadata=metadata,
+        subscription_data={"metadata": metadata},
+    )
+    return session
@@ -0,0 +1,332 @@
+"""Idempotent Stripe webhook handlers that upsert Invoice / Payment rows."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone as dt_timezone
+from typing import Any
+
+from django.contrib.auth import get_user_model
+from django.db import transaction
+from django.utils import timezone
+
+from finance.models import Invoice, Payment
+
+logger = logging.getLogger(__name__)
+User = get_user_model()
+
+
+def _ts_to_dt(value: int | None):
+    if not value:
+        return None
+    return datetime.fromtimestamp(value, tz=dt_timezone.utc)
+
+
+def _resolve_user(*, user_id: str | None = None, customer_email: str | None = None):
+    if user_id:
+        try:
+            return User.objects.get(pk=int(user_id))
+        except (User.DoesNotExist, TypeError, ValueError):
+            logger.warning("Webhook: user_id=%s not found", user_id)
+    if customer_email:
+        user = User.objects.filter(email__iexact=customer_email).first()
+        if user:
+            return user
+        logger.warning("Webhook: email=%s not found", customer_email)
+    return None
+
+
+def _user_from_metadata(metadata: dict | None, *, email: str | None = None):
+    metadata = metadata or {}
+    return _resolve_user(
+        user_id=metadata.get("user_id") or metadata.get("client_reference_id"),
+        customer_email=email,
+    )
+
+
+@transaction.atomic
+def upsert_invoice_from_stripe(
+    *,
+    user,
+    stripe_invoice: dict[str, Any] | None = None,
+    stripe_checkout_session_id: str | None = None,
+    stripe_subscription_id: str | None = None,
+    stripe_customer_id: str | None = None,
+    status: str,
+    amount_due: int = 0,
+    amount_paid: int = 0,
+    currency: str = "usd",
+    period_start=None,
+    period_end=None,
+    hosted_invoice_url: str = "",
+    description: str = "",
+) -> Invoice:
+    stripe_invoice_id = None
+    if stripe_invoice:
+        stripe_invoice_id = stripe_invoice.get("id")
+        stripe_subscription_id = (
+            stripe_subscription_id or stripe_invoice.get("subscription") or None
+        )
+        stripe_customer_id = (
+            stripe_customer_id or stripe_invoice.get("customer") or None
+        )
+        amount_due = int(stripe_invoice.get("amount_due") or amount_due or 0)
+        amount_paid = int(stripe_invoice.get("amount_paid") or amount_paid or 0)
+        currency = (stripe_invoice.get("currency") or currency or "usd").lower()
+        period_start = period_start or _ts_to_dt(
+            (stripe_invoice.get("period_start") or stripe_invoice.get("created"))
+        )
+        period_end = period_end or _ts_to_dt(stripe_invoice.get("period_end"))
+        hosted_invoice_url = (
+            hosted_invoice_url or stripe_invoice.get("hosted_invoice_url") or ""
+        )
+        description = description or stripe_invoice.get("description") or ""
+
+    lookup: dict[str, Any] = {}
+    if stripe_invoice_id:
+        lookup["stripe_invoice_id"] = stripe_invoice_id
+    elif stripe_checkout_session_id:
+        lookup["stripe_checkout_session_id"] = stripe_checkout_session_id
+    else:
+        raise ValueError("Need stripe_invoice_id or stripe_checkout_session_id")
+
+    defaults = {
+        "user": user,
+        "company": getattr(user, "company", None),
+        "provider": Invoice.Provider.STRIPE,
+        "status": status,
+        "currency": currency,
+        "amount_due": amount_due,
+        "amount_paid": amount_paid,
+        "period_start": period_start,
+        "period_end": period_end,
+        "stripe_subscription_id": stripe_subscription_id or None,
+        "stripe_customer_id": stripe_customer_id or "",
+        "hosted_invoice_url": hosted_invoice_url or "",
+        "description": description or "",
+    }
+    if stripe_invoice_id:
+        defaults["stripe_invoice_id"] = stripe_invoice_id
+    if stripe_checkout_session_id:
+        defaults["stripe_checkout_session_id"] = stripe_checkout_session_id
+
+    invoice, _created = Invoice.objects.update_or_create(
+        **lookup,
+        defaults=defaults,
+    )
+    if stripe_invoice_id and invoice.stripe_invoice_id != stripe_invoice_id:
+        invoice.stripe_invoice_id = stripe_invoice_id
+        invoice.save(update_fields=["stripe_invoice_id", "last_modified"])
+    return invoice
+
+
+@transaction.atomic
+def upsert_payment_from_stripe(
+    *,
+    user,
+    invoice: Invoice | None,
+    amount: int,
+    currency: str = "usd",
+    status: str,
+    stripe_payment_intent_id: str | None = None,
+    stripe_charge_id: str | None = None,
+    paid_at=None,
+    failure_message: str = "",
+) -> Payment:
+    if not stripe_payment_intent_id and not stripe_charge_id:
+        raise ValueError("Need stripe_payment_intent_id or stripe_charge_id")
+
+    lookup: dict[str, Any] = {}
+    if stripe_payment_intent_id:
+        lookup["stripe_payment_intent_id"] = stripe_payment_intent_id
+    else:
+        lookup["stripe_charge_id"] = stripe_charge_id
+
+    defaults = {
+        "user": user,
+        "company": getattr(user, "company", None),
+        "invoice": invoice,
+        "provider": Payment.Provider.STRIPE,
+        "status": status,
+        "currency": (currency or "usd").lower(),
+        "amount": int(amount or 0),
+        "paid_at": paid_at,
+        "failure_message": failure_message or "",
+    }
+    if stripe_payment_intent_id:
+        defaults["stripe_payment_intent_id"] = stripe_payment_intent_id
+    if stripe_charge_id:
+        defaults["stripe_charge_id"] = stripe_charge_id
+
+    payment, _created = Payment.objects.update_or_create(
+        **lookup,
+        defaults=defaults,
+    )
+    return payment
+
+
+def handle_checkout_session_completed(session: dict[str, Any]) -> Invoice | None:
+    metadata = session.get("metadata") or {}
+    customer_details = session.get("customer_details") or {}
+    user = _user_from_metadata(
+        metadata,
+        email=customer_details.get("email") or session.get("customer_email"),
+    )
+    if user is None and session.get("client_reference_id"):
+        user = _resolve_user(user_id=session.get("client_reference_id"))
+    if user is None:
+        logger.error(
+            "checkout.session.completed: cannot resolve user for session %s",
+            session.get("id"),
+        )
+        return None
+
+    amount_total = int(session.get("amount_total") or 0)
+    invoice = upsert_invoice_from_stripe(
+        user=user,
+        stripe_checkout_session_id=session.get("id"),
+        stripe_subscription_id=session.get("subscription") or None,
+        stripe_customer_id=session.get("customer") or None,
+        status=(
+            Invoice.Status.PAID
+            if session.get("payment_status") == "paid"
+            else Invoice.Status.OPEN
+        ),
+        amount_due=amount_total,
+        amount_paid=amount_total if session.get("payment_status") == "paid" else 0,
+        currency=(session.get("currency") or "usd").lower(),
+        description="Subscription checkout",
+    )
+
+    payment_intent = session.get("payment_intent")
+    if payment_intent and session.get("payment_status") == "paid":
+        upsert_payment_from_stripe(
+            user=user,
+            invoice=invoice,
+            amount=amount_total,
+            currency=(session.get("currency") or "usd").lower(),
+            status=Payment.Status.SUCCEEDED,
+            stripe_payment_intent_id=(
+                payment_intent if isinstance(payment_intent, str) else None
+            ),
+            paid_at=timezone.now(),
+        )
+    return invoice
+
+
+def handle_invoice_paid(stripe_invoice: dict[str, Any]) -> Invoice | None:
+    metadata = stripe_invoice.get("metadata") or {}
+    user = _user_from_metadata(
+        metadata,
+        email=stripe_invoice.get("customer_email"),
+    )
+    if user is None:
+        existing = None
+        if stripe_invoice.get("id"):
+            existing = (
+                Invoice.objects.filter(stripe_invoice_id=stripe_invoice["id"])
+                .select_related("user")
+                .first()
+            )
+        if existing is None and stripe_invoice.get("subscription"):
+            existing = (
+                Invoice.objects.filter(
+                    stripe_subscription_id=stripe_invoice["subscription"]
+                )
+                .select_related("user")
+                .order_by("-created")
+                .first()
+            )
+        if existing:
+            user = existing.user
+    if user is None:
+        logger.error(
+            "invoice.paid: cannot resolve user for invoice %s",
+            stripe_invoice.get("id"),
+        )
+        return None
+
+    invoice = upsert_invoice_from_stripe(
+        user=user,
+        stripe_invoice=stripe_invoice,
+        status=Invoice.Status.PAID,
+    )
+
+    payment_intent = stripe_invoice.get("payment_intent")
+    charge = stripe_invoice.get("charge")
+    if payment_intent or charge:
+        paid_at = _ts_to_dt(
+            (stripe_invoice.get("status_transitions") or {}).get("paid_at")
+        ) or timezone.now()
+        upsert_payment_from_stripe(
+            user=user,
+            invoice=invoice,
+            amount=int(stripe_invoice.get("amount_paid") or 0),
+            currency=(stripe_invoice.get("currency") or "usd").lower(),
+            status=Payment.Status.SUCCEEDED,
+            stripe_payment_intent_id=(
+                payment_intent if isinstance(payment_intent, str) else None
+            ),
+            stripe_charge_id=charge if isinstance(charge, str) else None,
+            paid_at=paid_at,
+        )
+    return invoice
+
+
+def handle_invoice_payment_failed(stripe_invoice: dict[str, Any]) -> Invoice | None:
+    metadata = stripe_invoice.get("metadata") or {}
+    user = _user_from_metadata(
+        metadata,
+        email=stripe_invoice.get("customer_email"),
+    )
+    if user is None:
+        existing = (
+            Invoice.objects.filter(stripe_invoice_id=stripe_invoice.get("id"))
+            .select_related("user")
+            .first()
+        )
+        if existing:
+            user = existing.user
+    if user is None:
+        logger.error(
+            "invoice.payment_failed: cannot resolve user for invoice %s",
+            stripe_invoice.get("id"),
+        )
+        return None
+
+    invoice = upsert_invoice_from_stripe(
+        user=user,
+        stripe_invoice=stripe_invoice,
+        status=Invoice.Status.PAYMENT_FAILED,
+    )
+
+    payment_intent = stripe_invoice.get("payment_intent")
+    if payment_intent:
+        upsert_payment_from_stripe(
+            user=user,
+            invoice=invoice,
+            amount=int(stripe_invoice.get("amount_due") or 0),
+            currency=(stripe_invoice.get("currency") or "usd").lower(),
+            status=Payment.Status.FAILED,
+            stripe_payment_intent_id=(
+                payment_intent if isinstance(payment_intent, str) else None
+            ),
+            failure_message="Stripe invoice payment failed",
+        )
+    return invoice
+
+
+def dispatch_stripe_event(event: dict[str, Any]):
+    """Route a verified Stripe event to the appropriate handler."""
+    event_type = event.get("type")
+    data_object = (event.get("data") or {}).get("object") or {}
+
+    if event_type == "checkout.session.completed":
+        return handle_checkout_session_completed(data_object)
+    if event_type == "invoice.paid":
+        return handle_invoice_paid(data_object)
+    if event_type == "invoice.payment_failed":
+        return handle_invoice_payment_failed(data_object)
+
+    logger.info("Ignoring unhandled Stripe event type: %s", event_type)
+    return None
@@ -0,0 +1,122 @@
+"""Tests for Stripe Checkout Session API (mocked Stripe SDK)."""
+
+from unittest.mock import MagicMock, patch
+
+from django.test import override_settings
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from chat_backend.tests.factories import make_company, make_user
+from finance.models import Invoice
+
+
+class CreateCheckoutSessionViewTestCase(APITestCase):
+    def setUp(self):
+        self.company = make_company()
+        self.user = make_user(company=self.company)
+        self.client.force_authenticate(user=self.user)
+        self.url = reverse("finance_checkout")
+
+    @override_settings(
+        STRIPE_SECRET_KEY="sk_test_fake",
+        SUBSCRIPTION_PRICE_AMOUNT_CENTS=1000,
+        SUBSCRIPTION_PRICE_CURRENCY="usd",
+        SUBSCRIPTION_PRICE_INTERVAL="month",
+        SUBSCRIPTION_PRODUCT_NAME="Chat Subscription",
+        STRIPE_PRICE_ID="",
+        STRIPE_CHECKOUT_SUCCESS_URL="http://localhost:3000/ok",
+        STRIPE_CHECKOUT_CANCEL_URL="http://localhost:3000/cancel",
+    )
+    @patch("finance.services.stripe_service.stripe.checkout.Session.create")
+    def test_creates_checkout_session_and_draft_invoice(self, mock_create):
+        mock_session = MagicMock()
+        mock_session.id = "cs_test_abc"
+        mock_session.url = "https://checkout.stripe.com/c/pay/cs_test_abc"
+        mock_session.customer = None
+        mock_create.return_value = mock_session
+
+        response = self.client.post(self.url, {}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(
+            response.data["checkout_url"],
+            "https://checkout.stripe.com/c/pay/cs_test_abc",
+        )
+        self.assertEqual(response.data["session_id"], "cs_test_abc")
+
+        mock_create.assert_called_once()
+        kwargs = mock_create.call_args.kwargs
+        self.assertEqual(kwargs["mode"], "subscription")
+        line_item = kwargs["line_items"][0]
+        self.assertEqual(line_item["price_data"]["unit_amount"], 1000)
+        self.assertEqual(line_item["price_data"]["currency"], "usd")
+        self.assertEqual(
+            line_item["price_data"]["recurring"]["interval"], "month"
+        )
+        self.assertEqual(kwargs["metadata"]["user_id"], str(self.user.pk))
+
+        invoice = Invoice.objects.get(stripe_checkout_session_id="cs_test_abc")
+        self.assertEqual(invoice.user, self.user)
+        self.assertEqual(invoice.company, self.company)
+        self.assertEqual(invoice.amount_due, 1000)
+        self.assertEqual(invoice.status, Invoice.Status.OPEN)
+
+    @override_settings(STRIPE_SECRET_KEY="", STRIPE_PRICE_ID="")
+    def test_missing_stripe_key_returns_503(self):
+        response = self.client.post(self.url, {}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE)
+
+    def test_unauthenticated_rejected(self):
+        self.client.force_authenticate(user=None)
+        response = self.client.post(self.url, {}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+    @override_settings(
+        STRIPE_SECRET_KEY="sk_test_fake",
+        STRIPE_PRICE_ID="price_abc123",
+        STRIPE_CHECKOUT_SUCCESS_URL="http://localhost:3000/ok",
+        STRIPE_CHECKOUT_CANCEL_URL="http://localhost:3000/cancel",
+    )
+    @patch("finance.services.stripe_service.stripe.checkout.Session.create")
+    def test_uses_stripe_price_id_when_set(self, mock_create):
+        mock_session = MagicMock()
+        mock_session.id = "cs_test_price"
+        mock_session.url = "https://checkout.stripe.com/c/pay/cs_test_price"
+        mock_session.customer = None
+        mock_create.return_value = mock_session
+
+        response = self.client.post(self.url, {}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        line_item = mock_create.call_args.kwargs["line_items"][0]
+        self.assertEqual(line_item, {"price": "price_abc123", "quantity": 1})
+
+
+class InvoicePaymentListViewTestCase(APITestCase):
+    def setUp(self):
+        self.company = make_company()
+        self.user = make_user(company=self.company)
+        self.other = make_user(
+            email="other@test.com",
+            username="other@test.com",
+            company=self.company,
+        )
+        self.client.force_authenticate(user=self.user)
+        Invoice.objects.create(
+            user=self.user,
+            company=self.company,
+            amount_due=1000,
+            stripe_checkout_session_id="cs_mine",
+        )
+        Invoice.objects.create(
+            user=self.other,
+            company=self.company,
+            amount_due=1000,
+            stripe_checkout_session_id="cs_other",
+        )
+
+    def test_list_own_invoices_only(self):
+        response = self.client.get(reverse("finance_invoices"))
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]["stripe_checkout_session_id"], "cs_mine")
@@ -0,0 +1,66 @@
+"""Tests for finance Invoice / Payment models and admin registration."""
+
+from django.contrib import admin
+from django.test import TestCase
+
+from chat_backend.tests.factories import make_company, make_user
+from finance.models import Invoice, Payment
+
+
+class InvoicePaymentModelTestCase(TestCase):
+    def setUp(self):
+        self.company = make_company()
+        self.user = make_user(company=self.company)
+
+    def test_create_invoice_and_payment(self):
+        invoice = Invoice.objects.create(
+            user=self.user,
+            company=self.company,
+            status=Invoice.Status.OPEN,
+            amount_due=1000,
+            currency="usd",
+            stripe_checkout_session_id="cs_test_1",
+        )
+        payment = Payment.objects.create(
+            user=self.user,
+            company=self.company,
+            invoice=invoice,
+            amount=1000,
+            currency="usd",
+            status=Payment.Status.PENDING,
+            stripe_payment_intent_id="pi_test_1",
+        )
+        self.assertEqual(invoice.provider, Invoice.Provider.STRIPE)
+        self.assertEqual(payment.invoice_id, invoice.pk)
+        self.assertEqual(Invoice.objects.count(), 1)
+        self.assertEqual(Payment.objects.count(), 1)
+
+    def test_mark_payment_succeeded(self):
+        payment = Payment.objects.create(
+            user=self.user,
+            amount=1000,
+            stripe_payment_intent_id="pi_test_2",
+        )
+        payment.mark_succeeded()
+        payment.refresh_from_db()
+        self.assertEqual(payment.status, Payment.Status.SUCCEEDED)
+        self.assertIsNotNone(payment.paid_at)
+
+    def test_unique_stripe_checkout_session_id(self):
+        Invoice.objects.create(
+            user=self.user,
+            stripe_checkout_session_id="cs_unique",
+            amount_due=1000,
+        )
+        with self.assertRaises(Exception):
+            Invoice.objects.create(
+                user=self.user,
+                stripe_checkout_session_id="cs_unique",
+                amount_due=1000,
+            )
+
+
+class FinanceAdminRegistrationTestCase(TestCase):
+    def test_invoice_and_payment_registered(self):
+        self.assertIn(Invoice, admin.site._registry)
+        self.assertIn(Payment, admin.site._registry)
@@ -0,0 +1,159 @@
+"""Tests for Stripe webhook verification and ledger upserts."""
+
+from unittest.mock import patch
+
+from django.test import override_settings
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from chat_backend.tests.factories import make_company, make_user
+from finance.models import Invoice, Payment
+from finance.services.webhooks import (
+    dispatch_stripe_event,
+    handle_checkout_session_completed,
+    handle_invoice_paid,
+    handle_invoice_payment_failed,
+)
+
+
+class WebhookHandlerUnitTestCase(APITestCase):
+    def setUp(self):
+        self.company = make_company()
+        self.user = make_user(company=self.company)
+
+    def test_checkout_session_completed_creates_invoice_and_payment(self):
+        session = {
+            "id": "cs_test_completed",
+            "metadata": {"user_id": str(self.user.pk)},
+            "customer": "cus_123",
+            "subscription": "sub_123",
+            "payment_intent": "pi_123",
+            "payment_status": "paid",
+            "amount_total": 1000,
+            "currency": "usd",
+            "customer_email": self.user.email,
+        }
+        invoice = handle_checkout_session_completed(session)
+        self.assertIsNotNone(invoice)
+        self.assertEqual(invoice.status, Invoice.Status.PAID)
+        self.assertEqual(invoice.amount_paid, 1000)
+        self.assertEqual(invoice.stripe_subscription_id, "sub_123")
+        payment = Payment.objects.get(stripe_payment_intent_id="pi_123")
+        self.assertEqual(payment.status, Payment.Status.SUCCEEDED)
+        self.assertEqual(payment.invoice_id, invoice.pk)
+
+    def test_checkout_session_completed_is_idempotent(self):
+        session = {
+            "id": "cs_test_idem",
+            "metadata": {"user_id": str(self.user.pk)},
+            "payment_status": "paid",
+            "amount_total": 1000,
+            "currency": "usd",
+            "payment_intent": "pi_idem",
+        }
+        handle_checkout_session_completed(session)
+        handle_checkout_session_completed(session)
+        self.assertEqual(
+            Invoice.objects.filter(stripe_checkout_session_id="cs_test_idem").count(),
+            1,
+        )
+        self.assertEqual(
+            Payment.objects.filter(stripe_payment_intent_id="pi_idem").count(),
+            1,
+        )
+
+    def test_invoice_paid_upserts(self):
+        stripe_invoice = {
+            "id": "in_paid_1",
+            "metadata": {"user_id": str(self.user.pk)},
+            "customer": "cus_1",
+            "subscription": "sub_1",
+            "amount_due": 1000,
+            "amount_paid": 1000,
+            "currency": "usd",
+            "status": "paid",
+            "payment_intent": "pi_paid_1",
+            "charge": "ch_paid_1",
+            "period_start": 1_700_000_000,
+            "period_end": 1_700_259_200,
+            "hosted_invoice_url": "https://invoice.stripe.com/i/test",
+            "status_transitions": {"paid_at": 1_700_000_100},
+        }
+        invoice = handle_invoice_paid(stripe_invoice)
+        self.assertEqual(invoice.status, Invoice.Status.PAID)
+        self.assertEqual(invoice.stripe_invoice_id, "in_paid_1")
+        payment = Payment.objects.get(stripe_payment_intent_id="pi_paid_1")
+        self.assertEqual(payment.stripe_charge_id, "ch_paid_1")
+        self.assertEqual(payment.status, Payment.Status.SUCCEEDED)
+
+    def test_invoice_payment_failed(self):
+        stripe_invoice = {
+            "id": "in_fail_1",
+            "metadata": {"user_id": str(self.user.pk)},
+            "amount_due": 1000,
+            "amount_paid": 0,
+            "currency": "usd",
+            "payment_intent": "pi_fail_1",
+        }
+        invoice = handle_invoice_payment_failed(stripe_invoice)
+        self.assertEqual(invoice.status, Invoice.Status.PAYMENT_FAILED)
+        payment = Payment.objects.get(stripe_payment_intent_id="pi_fail_1")
+        self.assertEqual(payment.status, Payment.Status.FAILED)
+
+    def test_dispatch_ignores_unknown_events(self):
+        result = dispatch_stripe_event(
+            {"type": "customer.created", "data": {"object": {}}}
+        )
+        self.assertIsNone(result)
+
+
+class StripeWebhookViewTestCase(APITestCase):
+    def setUp(self):
+        self.url = reverse("finance_stripe_webhook")
+        self.company = make_company()
+        self.user = make_user(company=self.company)
+
+    @override_settings(STRIPE_WEBHOOK_SECRET="")
+    def test_missing_webhook_secret_returns_503(self):
+        response = self.client.post(
+            self.url,
+            data=b"{}",
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE)
+
+    @override_settings(STRIPE_WEBHOOK_SECRET="whsec_test")
+    @patch("finance.views.stripe.Webhook.construct_event")
+    def test_invalid_signature_returns_400(self, mock_construct):
+        import stripe
+
+        mock_construct.side_effect = stripe.SignatureVerificationError(
+            "bad sig", "sig_header"
+        )
+        response = self.client.post(
+            self.url,
+            data=b"{}",
+            content_type="application/json",
+            HTTP_STRIPE_SIGNATURE="t=1,v1=bad",
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    @override_settings(STRIPE_WEBHOOK_SECRET="whsec_test")
+    @patch("finance.views.dispatch_stripe_event")
+    @patch("finance.views.stripe.Webhook.construct_event")
+    def test_valid_event_dispatched(self, mock_construct, mock_dispatch):
+        mock_construct.return_value = {
+            "id": "evt_1",
+            "type": "checkout.session.completed",
+            "data": {"object": {"id": "cs_x"}},
+        }
+        response = self.client.post(
+            self.url,
+            data=b'{"id":"evt_1"}',
+            content_type="application/json",
+            HTTP_STRIPE_SIGNATURE="t=1,v1=good",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertTrue(response.data["received"])
+        mock_dispatch.assert_called_once()
@@ -0,0 +1,31 @@
+from django.urls import path
+
+from finance.views import (
+    CreateCheckoutSessionView,
+    InvoiceListView,
+    PaymentListView,
+    StripeWebhookView,
+)
+
+urlpatterns = [
+    path(
+        "checkout/",
+        CreateCheckoutSessionView.as_view(),
+        name="finance_checkout",
+    ),
+    path(
+        "invoices/",
+        InvoiceListView.as_view(),
+        name="finance_invoices",
+    ),
+    path(
+        "payments/",
+        PaymentListView.as_view(),
+        name="finance_payments",
+    ),
+    path(
+        "webhooks/stripe/",
+        StripeWebhookView.as_view(),
+        name="finance_stripe_webhook",
+    ),
+]
@@ -0,0 +1,134 @@
+import logging
+
+import stripe
+from django.conf import settings
+from rest_framework import permissions, status
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from finance.models import Invoice, Payment
+from finance.serializers import (
+    CheckoutSessionSerializer,
+    InvoiceSerializer,
+    PaymentSerializer,
+)
+from finance.services.stripe_service import (
+    StripeNotConfiguredError,
+    create_checkout_session,
+)
+from finance.services.webhooks import dispatch_stripe_event
+
+logger = logging.getLogger(__name__)
+
+
+class CreateCheckoutSessionView(APIView):
+    """Create a Stripe Checkout Session and return the hosted redirect URL."""
+
+    def post(self, request):
+        serializer = CheckoutSessionSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        try:
+            session = create_checkout_session(
+                user=request.user,
+                success_url=serializer.validated_data.get("success_url"),
+                cancel_url=serializer.validated_data.get("cancel_url"),
+            )
+        except StripeNotConfiguredError as exc:
+            return Response(
+                {"detail": str(exc)},
+                status=status.HTTP_503_SERVICE_UNAVAILABLE,
+            )
+        except stripe.StripeError as exc:
+            logger.exception("Stripe Checkout Session creation failed")
+            return Response(
+                {"detail": str(getattr(exc, "user_message", None) or exc)},
+                status=status.HTTP_502_BAD_GATEWAY,
+            )
+
+        # Persist a draft invoice keyed by checkout session for admin visibility
+        # before the webhook fires.
+        Invoice.objects.update_or_create(
+            stripe_checkout_session_id=session.id,
+            defaults={
+                "user": request.user,
+                "company": request.user.company,
+                "provider": Invoice.Provider.STRIPE,
+                "status": Invoice.Status.OPEN,
+                "currency": settings.SUBSCRIPTION_PRICE_CURRENCY,
+                "amount_due": settings.SUBSCRIPTION_PRICE_AMOUNT_CENTS,
+                "amount_paid": 0,
+                "stripe_customer_id": getattr(session, "customer", None) or "",
+                "description": settings.SUBSCRIPTION_PRODUCT_NAME,
+            },
+        )
+
+        return Response(
+            {
+                "checkout_url": session.url,
+                "session_id": session.id,
+            },
+            status=status.HTTP_201_CREATED,
+        )
+
+
+class InvoiceListView(APIView):
+    def get(self, request):
+        invoices = Invoice.objects.filter(user=request.user)
+        return Response(InvoiceSerializer(invoices, many=True).data)
+
+
+class PaymentListView(APIView):
+    def get(self, request):
+        payments = Payment.objects.filter(user=request.user)
+        return Response(PaymentSerializer(payments, many=True).data)
+
+
+class StripeWebhookView(APIView):
+    """Verify Stripe signatures and upsert local invoice/payment rows."""
+
+    permission_classes = (permissions.AllowAny,)
+    authentication_classes = ()
+
+    def post(self, request):
+        payload = request.body
+        sig_header = request.META.get("HTTP_STRIPE_SIGNATURE", "")
+        webhook_secret = settings.STRIPE_WEBHOOK_SECRET
+
+        if not webhook_secret:
+            logger.error("STRIPE_WEBHOOK_SECRET is not configured")
+            return Response(
+                {"detail": "Webhook secret not configured"},
+                status=status.HTTP_503_SERVICE_UNAVAILABLE,
+            )
+
+        try:
+            event = stripe.Webhook.construct_event(
+                payload=payload,
+                sig_header=sig_header,
+                secret=webhook_secret,
+            )
+        except ValueError:
+            return Response(
+                {"detail": "Invalid payload"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        except stripe.SignatureVerificationError:
+            return Response(
+                {"detail": "Invalid signature"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        if hasattr(event, "to_dict"):
+            event = event.to_dict()
+
+        try:
+            dispatch_stripe_event(event)
+        except Exception:
+            logger.exception("Error handling Stripe event %s", event.get("id"))
+            return Response(
+                {"detail": "Webhook handler error"},
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+
+        return Response({"received": True}, status=status.HTTP_200_OK)
@@ -169,6 +169,7 @@ INSTALLED_APPS = [
    "whitenoise.runserver_nostatic",
    "django.contrib.staticfiles",
    "chat_backend",
+    "finance",
    "rest_framework",
    "corsheaders",
    "rest_framework_simplejwt.token_blacklist",
@@ -296,6 +297,42 @@ os.makedirs(directory_path, exist_ok=True)
 ALLOW_IMAGE_GENERATION = env_bool("ALLOW_IMAGE_GENERATION", False)
 ALLOW_INTERNET_ACCESS = env_bool("ALLOW_INTERNET_ACCESS", True)

+# ---------------------------------------------------------------------------
+# Finance / Stripe (subscription billing)
+# ---------------------------------------------------------------------------
+STRIPE_SECRET_KEY = env("STRIPE_SECRET_KEY", "") or ""
+STRIPE_PUBLISHABLE_KEY = env("STRIPE_PUBLISHABLE_KEY", "") or ""
+STRIPE_WEBHOOK_SECRET = env("STRIPE_WEBHOOK_SECRET", "") or ""
+# Optional: use a pre-created Stripe Price. When empty, Checkout uses
+# price_data built from SUBSCRIPTION_PRICE_* below.
+STRIPE_PRICE_ID = env("STRIPE_PRICE_ID", "") or ""
+
+# Subscription list price — $10.00 USD / month (amount in cents).
+SUBSCRIPTION_PRICE_AMOUNT_CENTS = int(
+    env("SUBSCRIPTION_PRICE_AMOUNT_CENTS", "1000") or "1000"
+)
+SUBSCRIPTION_PRICE_CURRENCY = (
+    env("SUBSCRIPTION_PRICE_CURRENCY", "usd") or "usd"
+).lower()
+SUBSCRIPTION_PRICE_INTERVAL = (
+    env("SUBSCRIPTION_PRICE_INTERVAL", "month") or "month"
+).lower()
+SUBSCRIPTION_PRODUCT_NAME = (
+    env("SUBSCRIPTION_PRODUCT_NAME", "Chat Subscription") or "Chat Subscription"
+)
+
+FRONTEND_BASE_URL = (
+    env("FRONTEND_BASE_URL", "http://localhost:3000") or "http://localhost:3000"
+).rstrip("/")
+STRIPE_CHECKOUT_SUCCESS_URL = env(
+    "STRIPE_CHECKOUT_SUCCESS_URL",
+    f"{FRONTEND_BASE_URL}/billing/success?session_id={{CHECKOUT_SESSION_ID}}",
+) or f"{FRONTEND_BASE_URL}/billing/success?session_id={{CHECKOUT_SESSION_ID}}"
+STRIPE_CHECKOUT_CANCEL_URL = env(
+    "STRIPE_CHECKOUT_CANCEL_URL",
+    f"{FRONTEND_BASE_URL}/billing/cancel",
+) or f"{FRONTEND_BASE_URL}/billing/cancel"
+
 if DJANGO_ENV in {"prod", "beta"}:
    # Compose treats $ in .env as variable expansion — escape each $ as $$.
    if not SECRET_KEY:
@@ -23,6 +23,7 @@ urlpatterns = (
    [
        path("admin/", admin.site.urls),
        path("api/", include("chat_backend.urls")),
+        path("api/finance/", include("finance.urls")),
    ]
    + static(settings.STATIC_URL, document_root=settings.STATIC_ROOT)
    + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)
@@ -45,6 +45,7 @@ dependencies = [
    "httpx==0.28.1",
    "python-dateutil==2.9.0.post0",
    "pytz==2025.2",
+    "stripe>=12.0.0,<14.0.0",
 ]

 [dependency-groups]
@@ -661,6 +661,7 @@ dependencies = [
    { name = "python-dateutil" },
    { name = "python-docx" },
    { name = "pytz" },
+    { name = "stripe" },
    { name = "unstructured" },
    { name = "uvicorn" },
    { name = "whitenoise" },
@@ -707,6 +708,7 @@ requires-dist = [
    { name = "python-dateutil", specifier = "==2.9.0.post0" },
    { name = "python-docx", specifier = "==1.2.0" },
    { name = "pytz", specifier = "==2025.2" },
+    { name = "stripe", specifier = ">=12.0.0,<14.0.0" },
    { name = "unstructured", specifier = "==0.18.21" },
    { name = "uvicorn", specifier = "==0.38.0" },
    { name = "whitenoise", specifier = "==6.9.0" },
@@ -3976,6 +3978,19 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/49/4b/359f28a903c13438ef59ebeee215fb25da53066db67b305c125f1c6d2a25/sqlparse-0.5.5-py3-none-any.whl", hash = "sha256:12a08b3bf3eec877c519589833aed092e2444e68240a3577e8e26148acc7b1ba", size = 46138, upload-time = "2025-12-19T07:17:46.573Z" },
 ]

+[[package]]
+name = "stripe"
+version = "13.2.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "requests" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/4f/c1/e687ba904a78eeb9ac8e0845630c625a03d380b7569647530c4bafd12677/stripe-13.2.0.tar.gz", hash = "sha256:ed6ad1c27725e1e32a336fa9d835cfa8f0bd6dafcc98b3ab7bc7e8791390453b", size = 1357785, upload-time = "2025-11-05T23:04:23.043Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/a0/39/5a7a7ac27372e8bd1ec45f52b884d039f504900a1a41885745c5a7976519/stripe-13.2.0-py3-none-any.whl", hash = "sha256:e7d18bd44bab1812bc4e9e75da4ceacedd587f71737bb9193955edf420605f88", size = 1962001, upload-time = "2025-11-05T23:04:21.056Z" },
+]
+
 [[package]]
 name = "tenacity"
 version = "9.1.4"