Stripe metered billing til SaaS — forbrug-baseret fakturering

Stripe metered billing fakturerer kunder baseret på faktisk forbrug — perfekt til SaaS der skalerer med antal aktive enheder. Her er den komplette implementation.

---

Metered billing vs. flat-rate

# Flat-rate: Fast beløb pr. periode uanset forbrug
# Metered: Beløb baseret på rapporteret forbrug

# M-Bus Gateway model:
# - Fast: 99 DKK/md (platform-adgang)
# - Metered: 19 DKK pr. aktiv lejeenhed/md

import stripe

stripe.api_key = settings.stripe_secret_key.get_secret_value()

# Opret Stripe Price for metered billing:
metered_price = stripe.Price.create(
    product="prod_mbus_platform",
    currency="dkk",
    unit_amount=1900,          # 19 DKK i øre
    recurring={
        "interval": "month",
        "usage_type": "metered",           # Metered (ikke licensed)
        "aggregate_usage": "max",          # Brug højeste usage-record pr. periode
        # Alternativer: "sum" (total), "last_during_period", "last_ever"
    },
    nickname="Aktiv lejeenhed/md",
)

# Flat-rate platform-adgang:
flat_price = stripe.Price.create(
    product="prod_mbus_platform",
    currency="dkk",
    unit_amount=9900,          # 99 DKK
    recurring={"interval": "month"},
    nickname="Platform-adgang",
)

---

Opret subscription med metered item

# server/src/stripe/service.py

import stripe
from datetime import datetime

async def create_subscription_with_metered(
    customer_id: str,
    flat_price_id: str,
    metered_price_id: str,
) -> stripe.Subscription:
    """
    Opret subscription med to items:
    1. Flat rate (platform-adgang)
    2. Metered (aktive enheder)
    """
    subscription = stripe.Subscription.create(
        customer=customer_id,
        items=[
            {"price": flat_price_id},       # Fast månedlig pris
            {"price": metered_price_id},     # Metered — ingen quantity her
        ],
        payment_behavior="default_incomplete",
        expand=["latest_invoice.payment_intent"],
    )

    # Gem subscription_item_id for metered item (bruges til usage records):
    metered_item = next(
        item for item in subscription["items"]["data"]
        if item["price"]["id"] == metered_price_id
    )
    return subscription, metered_item["id"]


async def report_usage(
    subscription_item_id: str,
    quantity: int,
    timestamp: datetime | None = None,
) -> stripe.UsageRecord:
    """
    Rapportér forbrug til Stripe.
    Kaldse én gang pr. dag/uge med aktuelle enhedsantal.
    """
    return stripe.SubscriptionItem.create_usage_record(
        subscription_item_id,
        quantity=quantity,
        timestamp=int((timestamp or datetime.utcnow()).timestamp()),
        action="set",  # "set" erstatter (ved max aggregation), "increment" summerer
    )

---

Celery task: Månedlig enhedssynk

# server/src/workers/tasks/subscription_sync.py

from celery import shared_task
from sqlalchemy import select, func
from server.src.db.models import Occupancy, Tenant
import stripe
import structlog

log = structlog.get_logger()


@shared_task(name="billing.sync_unit_counts", acks_late=True)
def sync_active_unit_counts():
    """
    Kør månedligt: Tæl aktive lejemål pr. tenant og rapport til Stripe.
    Kørsel: 1. i måneden kl. 05:00 UTC (inden Stripe fakturerer).
    """
    from server.src.db import get_sync_session
    from server.src.config import get_settings

    settings = get_settings()
    stripe.api_key = settings.stripe_secret_key.get_secret_value()

    with get_sync_session() as session:
        # Tæl aktive occupancies pr. tenant
        active_counts = session.execute(
            select(Tenant.id, Tenant.stripe_subscription_item_id, func.count(Occupancy.id))
            .join(Occupancy, Occupancy.tenant_id == Tenant.id)
            .where(
                Occupancy.deleted_at.is_(None),
                Occupancy.end_date.is_(None),  # Aktiv (ingen slutdato)
            )
            .group_by(Tenant.id, Tenant.stripe_subscription_item_id)
        ).all()

        for tenant_id, stripe_item_id, count in active_counts:
            if not stripe_item_id:
                log.warning("Tenant mangler Stripe subscription item", tenant_id=str(tenant_id))
                continue

            try:
                stripe.SubscriptionItem.create_usage_record(
                    stripe_item_id,
                    quantity=count,
                    action="set",
                )
                log.info("Usage reported", tenant_id=str(tenant_id), units=count)
            except stripe.StripeError as e:
                log.error("Stripe usage report failed",
                          tenant_id=str(tenant_id), error=str(e))


# Celery beat schedule (kørsel 1. i måneden kl. 05:00 UTC):
CELERY_BEAT_SCHEDULE = {
    "sync-unit-counts": {
        "task": "billing.sync_active_unit_counts",
        "schedule": crontab(day_of_month="1", hour="5", minute="0"),
    },
}

---

Webhooks: Metered billing events

# server/src/stripe/router.py

@router.post("/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    event = stripe.Webhook.construct_event(
        payload, sig_header, settings.stripe_webhook_secret.get_secret_value()
    )

    handlers = {
        "invoice.created": handle_invoice_created,
        "invoice.upcoming": handle_invoice_upcoming,
        "invoice.paid": handle_invoice_paid,
        "invoice.payment_failed": handle_payment_failed,
        "customer.subscription.updated": handle_subscription_updated,
    }

    handler = handlers.get(event["type"])
    if handler:
        await handler(event["data"]["object"])

    return {"received": True}


async def handle_invoice_created(invoice: dict) -> None:
    """
    Invoice oprettet (typisk 1 time inden faktureringsperiode slutter).
    Mulighed for at korrigere usage records FØR faktura lukkes.
    """
    if invoice.get("billing_reason") == "subscription_cycle":
        # Verificér at usage records er korrekte
        tenant = await get_tenant_by_stripe_customer(invoice["customer"])
        if tenant:
            actual_count = await count_active_occupancies(tenant.id)
            # Opdatér usage record med præcist antal
            await report_usage(
                tenant.stripe_subscription_item_id,
                actual_count,
            )
            log.info("Usage corrected at invoice creation",
                     tenant_id=str(tenant.id), units=actual_count)

---

Proration ved plan-skift

# Opgrader fra Starter til Pro midt i faktureringsperiode:

async def upgrade_subscription(
    subscription_id: str,
    new_price_id: str,
    prorate: bool = True,
) -> stripe.Subscription:
    subscription = stripe.Subscription.retrieve(subscription_id)
    item_id = subscription["items"]["data"][0]["id"]

    updated = stripe.Subscription.modify(
        subscription_id,
        items=[{
            "id": item_id,
            "price": new_price_id,
        }],
        proration_behavior="create_prorations" if prorate else "none",
        # "create_prorations": Kreditér resterende periode, fakturér ny
        # "always_invoice": Opret straks faktura for difference
        # "none": Ingen proration — skift ved næste periode
    )
    return updated


# Forhåndsvis proration (til UI):
async def preview_proration(subscription_id: str, new_price_id: str) -> dict:
    upcoming = stripe.Invoice.upcoming(
        subscription=subscription_id,
        subscription_items=[{
            "id": subscription["items"]["data"][0]["id"],
            "price": new_price_id,
        }],
        subscription_proration_behavior="create_prorations",
        subscription_proration_date=int(datetime.utcnow().timestamp()),
    )
    return {
        "amount_due": upcoming["amount_due"] / 100,
        "currency": upcoming["currency"].upper(),
        "period_end": datetime.fromtimestamp(upcoming["period_end"]).isoformat(),
    }

---

Test af metered billing

# server/tests/test_stripe_billing.py

import pytest
from unittest.mock import patch, MagicMock


async def test_usage_report_on_invoice_created(client, test_tenant, mock_stripe):
    """Usage records korrigeres ved invoice.created webhook."""
    mock_stripe.SubscriptionItem.create_usage_record = MagicMock(return_value={})

    # Simulér webhook:
    event = {
        "type": "invoice.created",
        "data": {"object": {
            "customer": test_tenant.stripe_customer_id,
            "billing_reason": "subscription_cycle",
        }}
    }

    with patch("stripe.Webhook.construct_event", return_value=event):
        resp = await client.post(
            "/api/v1/stripe/webhook",
            content=b"{}",
            headers={"stripe-signature": "dummy"},
        )

    assert resp.status_code == 200
    mock_stripe.SubscriptionItem.create_usage_record.assert_called_once()
    call_args = mock_stripe.SubscriptionItem.create_usage_record.call_args
    assert call_args.kwargs["action"] == "set"

---

Konklusion

Stripe metered billing med aggregate_usage="max" er ideel til SaaS der fakturerer efter aktive lejeenheder: Quantity sættes dagligt/månedligt og Stripe beregner automatisk fakturabeløbet. invoice.created webhook giver mulighed for at korrigere usage records inden faktura lukkes. Proration-preview i UI giver transparens ved plan-skift.

Se Stripe abonnement SaaS guide eller Stripe webhook events guide.