Billing Infrastructure for Screenshot APIs: Metered Usage, Invoices, and Dunning
A pricing page is a promise. Billing infrastructure is what keeps that promise — counting usage accurately, charging at the right time, handling failed payments without losing customers. Most early-stage API products get the pricing page right and the billing infrastructure wrong: usage is miscounted, invoices are confusing, and failed payments are handled with a single email before the account gets suspended.
This post covers the three systems that make billing work: accurate metering, automated invoicing, and dunning.
Accurate Usage Metering
Metering is the foundation. Every other billing decision depends on knowing, precisely, how many screenshots a user has taken in a billing period.
import time
import redis
import json
from datetime import datetime, timezone
from dataclasses import dataclass
r = redis.Redis(host='localhost', port=6379, db=2)
@dataclass
class UsageEvent:
user_id: str
key_id: str
endpoint: str # "screenshot", "batch", "pdf"
timestamp: float
billable_units: int # 1 for screenshot, N for batch of N
metadata: dict # url_domain, response_time_ms, cached
def record_usage(event: UsageEvent):
"""
Record a billable usage event atomically.
Two writes: current-period counter + event log for audit.
"""
period = _billing_period_key(event.user_id)
pipe = r.pipeline()
# Increment period counter (for billing)
pipe.hincrby(f"usage:{period}", event.endpoint, event.billable_units)
pipe.hincrby(f"usage:{period}", "total", event.billable_units)
# Set TTL on counter (90 days — covers billing period + dispute window)
pipe.expire(f"usage:{period}", 90 * 24 * 3600)
# Append to audit log (for invoice line items and disputes)
log_entry = json.dumps({
"ts": event.timestamp,
"key": event.key_id,
"ep": event.endpoint,
"n": event.billable_units,
"meta": event.metadata,
})
pipe.rpush(f"usage_log:{event.user_id}:{_today()}", log_entry)
pipe.expire(f"usage_log:{event.user_id}:{_today()}", 90 * 24 * 3600)
pipe.execute()
def get_period_usage(user_id: str, period: str | None = None) -> dict:
"""
Get usage counts for a billing period.
period format: "2026-08" (year-month)
"""
if period is None:
period = _current_billing_period()
key = f"usage:{user_id}:{period}"
raw = r.hgetall(key)
return {k.decode(): int(v) for k, v in raw.items()}
def _billing_period_key(user_id: str) -> str:
period = _current_billing_period()
return f"{user_id}:{period}"
def _current_billing_period() -> str:
now = datetime.now(timezone.utc)
return f"{now.year}-{now.month:02d}"
def _today() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
The atomic pipeline write is important: counter and audit log must update together or not at all. A usage event that increments the counter but misses the audit log is invisible in disputes. A usage event in the audit log that missed the counter creates underbilling. The Redis pipeline makes both writes atomic from the application's perspective.
The audit log is the insurance policy. When a user disputes a charge ("I only made 200 calls, not 500"), you open the audit log for that period, filter by their key IDs, and show them the exact requests with timestamps and URLs. Most disputes evaporate at this point; the ones that don't reveal real bugs in your counting logic.
Invoice Generation
An invoice is a document. It needs to be accurate, readable, and permanent. Generated invoices should never change after issue — if a correction is needed, issue a credit note.
from dataclasses import dataclass, field
from typing import Optional
import hashlib
@dataclass
class InvoiceLineItem:
description: str
quantity: int
unit_price_usd: float
total_usd: float
@dataclass
class Invoice:
invoice_id: str
user_id: str
period: str # "2026-08"
issued_at: datetime
due_at: datetime
line_items: list[InvoiceLineItem] = field(default_factory=list)
subtotal_usd: float = 0.0
tax_usd: float = 0.0
total_usd: float = 0.0
status: str = "draft" # draft | issued | paid | void
payment_intent_id: Optional[str] = None
def generate_invoice(user_id: str, period: str, plan: dict) -> Invoice:
"""
Generate invoice for a completed billing period.
Called by monthly cron at period end.
"""
usage = get_period_usage(user_id, period)
total_screenshots = usage.get("total", 0)
line_items = []
# Base subscription
if plan["price_usd"] > 0:
line_items.append(InvoiceLineItem(
description=f"{plan['name']} Plan — {period}",
quantity=1,
unit_price_usd=plan["price_usd"],
total_usd=plan["price_usd"],
))
# Overage (screenshots beyond included quota)
included = plan["included_screenshots"]
if total_screenshots > included and plan.get("overage_per_1000"):
overage_count = total_screenshots - included
overage_amount = (overage_count / 1000) * plan["overage_per_1000"]
line_items.append(InvoiceLineItem(
description=f"Overage: {overage_count:,} screenshots beyond {included:,} included",
quantity=overage_count,
unit_price_usd=plan["overage_per_1000"] / 1000,
total_usd=round(overage_amount, 2),
))
subtotal = sum(item.total_usd for item in line_items)
# Tax calculation is jurisdiction-dependent — simplified here
tax = round(subtotal * _get_tax_rate(user_id), 2)
now = datetime.now(timezone.utc)
invoice = Invoice(
invoice_id=f"inv_{hashlib.sha256(f'{user_id}{period}'.encode()).hexdigest()[:12]}",
user_id=user_id,
period=period,
issued_at=now,
due_at=now, # Net-0 for subscription billing
line_items=line_items,
subtotal_usd=subtotal,
tax_usd=tax,
total_usd=round(subtotal + tax, 2),
status="draft",
)
db_save_invoice(invoice)
return invoice
The deterministic invoice_id based on user_id + period prevents duplicate invoices from retry logic. Running the invoice generation cron twice in a period produces the same invoice_id both times — the second run is a no-op that updates the existing draft rather than creating a second invoice.
Dunning: Recovering Failed Payments
Dunning is the process of retrying failed payments and communicating about them. Most failed payments are not fraud — they are expired cards, insufficient funds on a specific date, or temporary bank holds. A dunning sequence that communicates clearly and retries intelligently recovers 60-80% of initial failures.
# Days after initial failure to retry and send email
DUNNING_SCHEDULE = [
{"day": 3, "retry": True, "email": "payment_failed_day3"},
{"day": 7, "retry": True, "email": "payment_failed_day7"},
{"day": 14, "retry": True, "email": "payment_failed_day14_final"},
{"day": 21, "retry": False, "email": "account_suspended"},
]
def process_dunning(user_id: str, invoice_id: str, failure_date: datetime):
"""
Execute the appropriate dunning step based on days since failure.
Called by daily cron.
"""
days_since_failure = (datetime.now(timezone.utc) - failure_date).days
for step in DUNNING_SCHEDULE:
if days_since_failure >= step["day"]:
last_step = step
if not last_step:
return
if last_step["retry"]:
success = retry_payment(invoice_id)
if success:
_send_email(user_id, "payment_recovered")
_restore_account_access(user_id)
return
_send_email(user_id, last_step["email"])
if last_step["email"] == "account_suspended":
_suspend_account(user_id)
def _suspend_account(user_id: str):
"""
Suspend: block new API requests, preserve data.
Do NOT delete data — that's irreversible and burns goodwill.
"""
db_set_account_status(user_id, "suspended")
# Rate limit to 0 at the API layer, not by deleting the key
def _restore_account_access(user_id: str):
db_set_account_status(user_id, "active")
The email sequence matters as much as the retry schedule. Day-3 email tone: neutral and helpful ("Your payment didn't go through — here's how to update your card"). Day-7 email tone: slightly more urgent ("We tried again — please update your payment method to avoid interruption"). Day-14 email tone: clear consequence ("This is our final reminder. Your account will be suspended on [date+7]"). Day-21: account suspended with clear reactivation path.
Never delete data on suspension. A user who comes back to reactivate their account three months later should find their usage history, their API keys (disabled but restorable), and their invoices intact. Data deletion is permanent; the cost of storing a suspended account's data for 12 months is negligible.
The Minimum Viable Billing Stack
You don't need to build all of this from scratch. The minimum viable billing implementation for a new API:
- Stripe for payment processing, card storage, and payment intents — don't build this yourself
- Redis for real-time usage counters — fast enough for per-request metering
- PostgreSQL for invoice records and audit logs — needs durability
- A monthly cron that generates invoices and attempts charges
- A daily cron that runs the dunning schedule
The billing code above is the glue between these. Stripe handles the payment complexity; your code handles the usage counting and invoice logic that Stripe can't know about.
Part of the pricing and monetization series. Previous: Pricing Strategy. Next: Building a Usage Dashboard.