Stripe usage-based billing, step by step

// Steps 1 and 2: a meter that sums usage, and the two prices that bill it.
// Run once per Stripe account: node setup.js
import { fileURLToPath } from 'node:url';
import Stripe from 'stripe';

export const METER_EVENT_NAME = 'api_requests';

// 100,000 requests included, then $0.50 per 1,000: 0.05 cents per request.
/** @type {Stripe.PriceCreateParams.Tier[]} */
export const USAGE_TIERS = [
  { up_to: 100000, unit_amount: 0 },
  { up_to: 'inf', unit_amount_decimal: Stripe.Decimal.from('0.05') },
];

export async function createBillingObjects(stripe) {
  const meter = await stripe.billing.meters.create({
    display_name: 'API requests',
    event_name: METER_EVENT_NAME,
    default_aggregation: { formula: 'sum' },
    customer_mapping: { type: 'by_id', event_payload_key: 'stripe_customer_id' },
    value_settings: { event_payload_key: 'value' },
  });

  // $29 a month, charged in advance.
  const basePrice = await stripe.prices.create({
    currency: 'usd',
    unit_amount: 2900,
    recurring: { interval: 'month' },
    product_data: { name: 'Search API plan' },
  });

  // Usage, charged in arrears from the meter.
  const usagePrice = await stripe.prices.create({
    currency: 'usd',
    billing_scheme: 'tiered',
    tiers_mode: 'graduated',
    tiers: USAGE_TIERS,
    recurring: { interval: 'month', usage_type: 'metered', meter: meter.id },
    product_data: { name: 'Search API requests', unit_label: 'request' },
  });

  return { meterId: meter.id, basePriceId: basePrice.id, usagePriceId: usagePrice.id };
}

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const ids = await createBillingObjects(new Stripe(process.env.STRIPE_SECRET_KEY));
  console.log(`export METER_ID=${ids.meterId} BASE_PRICE_ID=${ids.basePriceId} USAGE_PRICE_ID=${ids.usagePriceId}`);
}

"""Steps 1 and 2: a meter that sums usage, and the two prices that bill it.
Run once per Stripe account: uv run python setup_billing.py"""
import os
from decimal import Decimal

from stripe import StripeClient
from stripe.params import PriceCreateParamsTier

METER_EVENT_NAME = "api_requests"

# 100,000 requests included, then $0.50 per 1,000: 0.05 cents per request.
USAGE_TIERS: list[PriceCreateParamsTier] = [
    {"up_to": 100_000, "unit_amount": 0},
    {"up_to": "inf", "unit_amount_decimal": Decimal("0.05")},
]


def create_billing_objects(client):
    meter = client.v1.billing.meters.create({
        "display_name": "API requests",
        "event_name": METER_EVENT_NAME,
        "default_aggregation": {"formula": "sum"},
        "customer_mapping": {"type": "by_id", "event_payload_key": "stripe_customer_id"},
        "value_settings": {"event_payload_key": "value"},
    })

    # $29 a month, charged in advance.
    base_price = client.v1.prices.create({
        "currency": "usd",
        "unit_amount": 2900,
        "recurring": {"interval": "month"},
        "product_data": {"name": "Search API plan"},
    })

    # Usage, charged in arrears from the meter.
    usage_price = client.v1.prices.create({
        "currency": "usd",
        "billing_scheme": "tiered",
        "tiers_mode": "graduated",
        "tiers": USAGE_TIERS,
        "recurring": {"interval": "month", "usage_type": "metered", "meter": meter.id},
        "product_data": {"name": "Search API requests", "unit_label": "request"},
    })

    return {"meter_id": meter.id, "base_price_id": base_price.id, "usage_price_id": usage_price.id}


if __name__ == "__main__":
    ids = create_billing_objects(StripeClient(os.environ["STRIPE_SECRET_KEY"]))
    print(
        f"export METER_ID={ids['meter_id']} BASE_PRICE_ID={ids['base_price_id']} "
        f"USAGE_PRICE_ID={ids['usage_price_id']}"
    )

// Step 3: subscribe a customer to the flat fee and the metered price.

// Production: Stripe Checkout. The metered line item takes no quantity.
export function createCheckoutSession(stripe, { customerId, basePriceId, usagePriceId, successUrl }) {
  return stripe.checkout.sessions.create({
    mode: 'subscription',
    customer: customerId,
    line_items: [{ price: basePriceId, quantity: 1 }, { price: usagePriceId }],
    success_url: successUrl,
  });
}

// Tests and simulations: subscribe directly with Stripe's test card.
export async function subscribeWithTestCard(stripe, { customerId, basePriceId, usagePriceId }) {
  const card = await stripe.paymentMethods.attach('pm_card_visa', { customer: customerId });
  return stripe.subscriptions.create({
    customer: customerId,
    default_payment_method: card.id,
    items: [{ price: basePriceId }, { price: usagePriceId }],
  });
}

"""Step 3: subscribe a customer to the flat fee and the metered price."""


def create_checkout_session(client, customer_id, base_price_id, usage_price_id, success_url):
    """Production: Stripe Checkout. The metered line item takes no quantity."""
    return client.v1.checkout.sessions.create({
        "mode": "subscription",
        "customer": customer_id,
        "line_items": [{"price": base_price_id, "quantity": 1}, {"price": usage_price_id}],
        "success_url": success_url,
    })


def subscribe_with_test_card(client, customer_id, base_price_id, usage_price_id):
    """Tests and simulations: subscribe directly with Stripe's test card."""
    card = client.v1.payment_methods.attach("pm_card_visa", {"customer": customer_id})
    return client.v1.subscriptions.create({
        "customer": customer_id,
        "default_payment_method": card.id,
        "items": [{"price": base_price_id}, {"price": usage_price_id}],
    })

// Step 4: report usage to the meter.
import { METER_EVENT_NAME } from './setup.js';

export function reportUsage(stripe, { customerId, quantity, identifier, timestamp }) {
  return stripe.billing.meterEvents.create({
    event_name: METER_EVENT_NAME,
    // Payload values are strings, the number included.
    payload: { stripe_customer_id: customerId, value: String(quantity) },
    identifier,
    timestamp,
  });
}

// One event per customer per period, sent after the period ends and
// before its invoice finalizes.
export function reportPeriodTotal(stripe, { customerId, quantity, periodStart, periodEnd }) {
  return reportUsage(stripe, {
    customerId,
    quantity,
    identifier: `${customerId}_${periodStart}`, // same on every retry
    timestamp: periodEnd - 1, // last second of the period
  });
}

"""Step 4: report usage to the meter."""
from setup_billing import METER_EVENT_NAME


def report_usage(client, customer_id, quantity, identifier, timestamp):
    return client.v1.billing.meter_events.create({
        "event_name": METER_EVENT_NAME,
        # Payload values are strings, the number included.
        "payload": {"stripe_customer_id": customer_id, "value": str(quantity)},
        "identifier": identifier,
        "timestamp": timestamp,
    })


def report_period_total(client, customer_id, quantity, period_start, period_end):
    """One event per customer per period, sent after the period ends and
    before its invoice finalizes."""
    return report_usage(
        client,
        customer_id,
        quantity,
        identifier=f"{customer_id}_{period_start}",  # same on every retry
        timestamp=period_end - 1,  # last second of the period
    )

// Steps 5 and 6: read the meter back, then check the finalized invoice.
import Stripe from 'stripe';

// Summary windows must start and end on whole minutes. Rounding both ends up
// keeps this period's event (periodEnd - 1) in and the last one out.
export const ceilToMinute = (seconds) => Math.ceil(seconds / 60) * 60;

export async function meterTotal(stripe, { meterId, customerId, periodStart, periodEnd }) {
  const summaries = await stripe.billing.meters.listEventSummaries(meterId, {
    customer: customerId,
    start_time: ceilToMinute(periodStart),
    end_time: ceilToMinute(periodEnd),
  });
  return summaries.data.reduce((total, summary) => total + summary.aggregated_value, 0);
}

// Match on the line's period: the invoice's period_start and period_end
// are not the service period.
export async function findUsageLine(stripe, { subscriptionId, usagePriceId, periodEnd }) {
  const lastSecond = periodEnd - 1;
  const invoices = await stripe.invoices.list({ subscription: subscriptionId, limit: 3 });
  for (const invoice of invoices.data) {
    const lines = await stripe.invoices.listLineItems(invoice.id, { limit: 100 });
    const line = lines.data.find(
      (candidate) =>
        candidate.pricing?.price_details?.price === usagePriceId &&
        candidate.period.start <= lastSecond &&
        lastSecond <= candidate.period.end,
    );
    if (line) return { invoice, line };
  }
  return null;
}

// Exact charge in cents for graduated per-unit tiers, fractions included
// (128,401 requests cost 1420.05).
export function expectedAmount(quantity, tiers) {
  let total = Stripe.Decimal.from(0);
  let previousUpTo = 0;
  for (const tier of tiers) {
    const upTo = typeof tier.up_to === 'number' ? Math.min(tier.up_to, quantity) : quantity;
    const unitAmount = Stripe.Decimal.from(String(tier.unit_amount_decimal ?? tier.unit_amount));
    total = total.add(unitAmount.mul(Stripe.Decimal.from(Math.max(0, upTo - previousUpTo))));
    previousUpTo = upTo;
  }
  return total;
}

export function compareUsage({ invoice, line }, { quantity, tiers }) {
  const problems = [];
  if (invoice.status === 'draft') problems.push(`${invoice.id} is still a draft`);
  // quantity_decimal is a Stripe.Decimal in stripe-node 23; Number() reads it.
  const billedQuantity = Number(line.quantity_decimal ?? line.quantity);
  if (billedQuantity !== quantity) problems.push(`Quantity: billed ${billedQuantity}, counted ${quantity}`);
  // subtotal excludes discounts and taxes. Stripe rounds it to a whole cent
  // but its docs leave the direction unclear, so either whole cent next to
  // the exact amount is a match.
  const exact = expectedAmount(quantity, tiers);
  const allowed = [Number(exact.toFixed(0, 'floor')), Number(exact.toFixed(0, 'ceil'))];
  if (!allowed.includes(line.subtotal)) problems.push(`Subtotal: billed ${line.subtotal} cents, expected ${exact}`);
  return problems;
}

"""Steps 5 and 6: read the meter back, then check the finalized invoice."""
import math
from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal


def ceil_to_minute(seconds):
    """Summary windows must start and end on whole minutes. Rounding both ends up
    keeps this period's event (period_end - 1) in and the last one out."""
    return math.ceil(seconds / 60) * 60


def meter_total(client, meter_id, customer_id, period_start, period_end):
    summaries = client.v1.billing.meters.event_summaries.list(meter_id, {
        "customer": customer_id,
        "start_time": ceil_to_minute(period_start),
        "end_time": ceil_to_minute(period_end),
    })
    return sum(summary.aggregated_value for summary in summaries.data)


def find_usage_line(client, subscription_id, usage_price_id, period_end):
    """Match on the line's period: the invoice's period_start and period_end
    are not the service period."""
    last_second = period_end - 1
    invoices = client.v1.invoices.list({"subscription": subscription_id, "limit": 3})
    for invoice in invoices.data:
        lines = client.v1.invoices.line_items.list(invoice.id, {"limit": 100})
        for line in lines.data:
            details = getattr(line.pricing, "price_details", None)
            if (
                details
                and details.price == usage_price_id
                and line.period.start <= last_second <= line.period.end
            ):
                return invoice, line
    return None


def expected_amount(quantity, tiers):
    """Exact charge in cents for graduated per-unit tiers, fractions included
    (128,401 requests cost 1420.05)."""
    total, previous_up_to = Decimal(0), 0
    for tier in tiers:
        up_to = min(tier["up_to"], quantity) if isinstance(tier["up_to"], int) else quantity
        unit_amount = Decimal(tier.get("unit_amount_decimal", tier.get("unit_amount")))
        total += max(0, up_to - previous_up_to) * unit_amount
        previous_up_to = up_to
    return total


def compare_usage(invoice, line, quantity, tiers):
    problems = []
    if invoice.status == "draft":
        problems.append(f"{invoice.id} is still a draft")
    billed_quantity = line.quantity_decimal if line.quantity_decimal is not None else line.quantity
    if billed_quantity != quantity:
        problems.append(f"Quantity: billed {billed_quantity}, counted {quantity}")
    # subtotal excludes discounts and taxes. Stripe rounds it to a whole cent
    # but its docs leave the direction unclear, so either whole cent next to
    # the exact amount is a match.
    exact = expected_amount(quantity, tiers)
    if line.subtotal not in (exact.to_integral_value(ROUND_FLOOR), exact.to_integral_value(ROUND_CEILING)):
        problems.append(f"Subtotal: billed {line.subtotal} cents, expected {exact.normalize():f}")
    return problems

node --env-file=.env setup.js   # prints an export line; run it
node --env-file=.env simulate-cycle.js
# Python: uv run --env-file .env python setup_billing.py, then simulate_cycle.py

// Step 7: run one billing cycle in a sandbox with a test clock.
// Uses the IDs printed by setup.js: node simulate-cycle.js
import { fileURLToPath } from 'node:url';
import Stripe from 'stripe';
import { USAGE_TIERS } from './setup.js';
import { subscribeWithTestCard } from './subscribe.js';
import { reportUsage } from './report-usage.js';
import { meterTotal, findUsageLine, compareUsage } from './check-invoice.js';

const GRACE_PERIOD_SECONDS = 60 * 60; // Stripe's default; match your own setting
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export async function advanceClock(stripe, clockId, frozenTime) {
  let clock = await stripe.testHelpers.testClocks.advance(clockId, { frozen_time: frozenTime });
  while (clock.status !== 'ready') {
    if (clock.status === 'internal_failure') throw new Error(`Test clock ${clockId} failed to advance.`);
    await sleep(2000);
    clock = await stripe.testHelpers.testClocks.retrieve(clockId);
  }
  return clock;
}

export async function simulateCycle(stripe, { meterId, basePriceId, usagePriceId, quantity }) {
  // Start at the current minute so the event timestamp is valid and in period one.
  const start = Math.floor(Date.now() / 60_000) * 60;
  const clock = await stripe.testHelpers.testClocks.create({ frozen_time: start });
  const customer = await stripe.customers.create({ test_clock: clock.id });
  const subscription = await subscribeWithTestCard(stripe, { customerId: customer.id, basePriceId, usagePriceId });
  const item = subscription.items.data.find((each) => each.price.id === usagePriceId);
  const period = { periodStart: item.current_period_start, periodEnd: item.current_period_end };

  const identifier = `${customer.id}_${period.periodStart}`;
  await reportUsage(stripe, { customerId: customer.id, quantity, identifier, timestamp: start });

  // Stripe processes meter events asynchronously: wait until the meter shows them.
  for (let attempt = 1; ; attempt++) {
    if ((await meterTotal(stripe, { meterId, customerId: customer.id, ...period })) >= quantity) break;
    if (attempt === 24) throw new Error('Usage did not reach the meter within two minutes.');
    await sleep(5000);
  }

  // Past the period end and the grace period, the invoice is final.
  await advanceClock(stripe, clock.id, period.periodEnd + GRACE_PERIOD_SECONDS + 60);
  const found = await findUsageLine(stripe, { subscriptionId: subscription.id, usagePriceId, periodEnd: period.periodEnd });
  const problems = found ? compareUsage(found, { quantity, tiers: USAGE_TIERS }) : ['No usage line found.'];
  return { clockId: clock.id, problems };
}

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const { STRIPE_SECRET_KEY, METER_ID, BASE_PRICE_ID, USAGE_PRICE_ID } = process.env;
  const result = await simulateCycle(new Stripe(STRIPE_SECRET_KEY), {
    meterId: METER_ID, basePriceId: BASE_PRICE_ID, usagePriceId: USAGE_PRICE_ID, quantity: 128400,
  });
  console.log(result.problems.join('\n') || 'Invoice matches.', `(test clock ${result.clockId})`);
}

"""Step 7: run one billing cycle in a sandbox with a test clock.
Uses the IDs printed by setup_billing.py: uv run python simulate_cycle.py"""
import os
import time

from stripe import StripeClient

from check_invoice import compare_usage, find_usage_line, meter_total
from report_usage import report_usage
from setup_billing import USAGE_TIERS
from subscribe import subscribe_with_test_card

GRACE_PERIOD_SECONDS = 60 * 60  # Stripe's default; match your own setting


def advance_clock(client, clock_id, frozen_time):
    clock = client.v1.test_helpers.test_clocks.advance(clock_id, {"frozen_time": frozen_time})
    while clock.status != "ready":
        if clock.status == "internal_failure":
            raise RuntimeError(f"Test clock {clock_id} failed to advance.")
        time.sleep(2)
        clock = client.v1.test_helpers.test_clocks.retrieve(clock_id)
    return clock


def simulate_cycle(client, meter_id, base_price_id, usage_price_id, quantity):
    # Start at the current minute so the event timestamp is valid and in period one.
    start = int(time.time()) // 60 * 60
    clock = client.v1.test_helpers.test_clocks.create({"frozen_time": start})
    customer = client.v1.customers.create({"test_clock": clock.id})
    subscription = subscribe_with_test_card(client, customer.id, base_price_id, usage_price_id)
    item = next(each for each in subscription.items.data if each.price.id == usage_price_id)
    period_start, period_end = item.current_period_start, item.current_period_end

    identifier = f"{customer.id}_{period_start}"
    report_usage(client, customer.id, quantity, identifier, timestamp=start)

    # Stripe processes meter events asynchronously: wait until the meter shows them.
    for _ in range(24):
        if meter_total(client, meter_id, customer.id, period_start, period_end) >= quantity:
            break
        time.sleep(5)
    else:
        raise RuntimeError("Usage did not reach the meter within two minutes.")

    # Past the period end and the grace period, the invoice is final.
    advance_clock(client, clock.id, period_end + GRACE_PERIOD_SECONDS + 60)
    found = find_usage_line(client, subscription.id, usage_price_id, period_end)
    if found is None:
        return clock.id, ["No usage line found."]
    invoice, line = found
    return clock.id, compare_usage(invoice, line, quantity, USAGE_TIERS)


if __name__ == "__main__":
    clock_id, problems = simulate_cycle(
        StripeClient(os.environ["STRIPE_SECRET_KEY"]),
        os.environ["METER_ID"], os.environ["BASE_PRICE_ID"], os.environ["USAGE_PRICE_ID"],
        quantity=128_400,
    )
    print("\n".join(problems) or "Invoice matches.", f"(test clock {clock_id})")