PostgreSQL Constraint Triggers for AI Feature Usage Quotas: Enforcing Hard Limits at the Database Layer

I've watched quota systems fail at scale more times than I'd like to admit. You implement soft checks in FastAPI—verify the tenant hasn't exceeded their monthly Claude API calls, then call the API. Works fine locally. Ships to production. Then during a traffic spike, three concurrent requests all see the same usage_count = 45 from the database, all think they're under the 50-call limit, and suddenly you're processing 153 calls instead of 50.

Application-layer quota enforcement is fundamentally racy. Every soft check you write is a time-of-check to time-of-use (TOCTOU) window waiting to bite you. I learned this the hard way when CitizenApp's document analysis feature silently over-billed a customer's entire monthly quota in 40 minutes during a burst. Database-layer enforcement closes that gap entirely.

Why Application-Layer Quotas Fail

The problem is atomic visibility. When your FastAPI server does this:

# Don't do this
usage = db.query(TenantUsage).filter(
    TenantUsage.tenant_id == tenant_id,
    TenantUsage.feature == "document_analysis"
).first()

if usage.count < usage.limit:
    # API call happens here, outside the database transaction
    response = client.messages.create(...)
    usage.count += 1
    db.commit()
else:
    raise QuotaExceeded()

Two requests can both read count = 45, both decide they're safe, both hit the API. The increment at the end is too late—you've already incurred the cost.

You could wrap it in a transaction with row-level locking:

# Better, but still vulnerable
with db.begin():
    usage = db.query(TenantUsage).with_for_update().filter(...).first()
    if usage.count < usage.limit:
        db.commit()  # Release lock
        response = client.messages.create(...)  # API call outside lock

But now you're holding a database lock while making an HTTP request to Anthropic. That's a recipe for connection pool exhaustion under load. I prefer to fail fast and atomically at the database level, before the API call ever happens.

The Trigger-Based Approach

PostgreSQL triggers let you enforce quotas as inserts happen, atomically and with proper locking. Here's the schema:

-- Tenants and their feature limits
CREATE TABLE tenant_feature_limits (
    id BIGSERIAL PRIMARY KEY,
    tenant_id UUID NOT NULL UNIQUE,
    feature VARCHAR(50) NOT NULL,
    monthly_limit INT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Usage log—one row per API call attempt
CREATE TABLE tenant_usage_log (
    id BIGSERIAL PRIMARY KEY,
    tenant_id UUID NOT NULL,
    feature VARCHAR(50) NOT NULL,
    tokens_used INT DEFAULT 0,
    created_at TIMESTAMP DEFAULT NOW(),
    FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);

-- Materialized view: current month's usage per tenant/feature
CREATE MATERIALIZED VIEW tenant_monthly_usage AS
SELECT
    tenant_id,
    feature,
    COALESCE(SUM(tokens_used), 0) AS tokens_used,
    COUNT(*) AS call_count,
    DATE_TRUNC('month', NOW())::DATE AS period_start
FROM tenant_usage_log
WHERE created_at >= DATE_TRUNC('month', NOW())
GROUP BY tenant_id, feature;

-- Index for fast lookups
CREATE UNIQUE INDEX ON tenant_monthly_usage (tenant_id, feature);

-- The enforcement trigger
CREATE OR REPLACE FUNCTION check_usage_quota()
RETURNS TRIGGER AS $$
DECLARE
    current_usage RECORD;
    limit_config RECORD;
BEGIN
    -- Get current month's usage for this tenant/feature
    SELECT tokens_used, call_count INTO current_usage
    FROM tenant_monthly_usage
    WHERE tenant_id = NEW.tenant_id AND feature = NEW.feature;

    -- Get the limit
    SELECT monthly_limit INTO limit_config
    FROM tenant_feature_limits
    WHERE tenant_id = NEW.tenant_id AND feature = NEW.feature;

    -- Hard constraint: reject if over limit
    IF limit_config.monthly_limit IS NOT NULL
       AND (COALESCE(current_usage.tokens_used, 0) + COALESCE(NEW.tokens_used, 0))
           > limit_config.monthly_limit
    THEN
        RAISE EXCEPTION 'quota_exceeded: tenant % feature % (limit: %)',
            NEW.tenant_id, NEW.feature, limit_config.monthly_limit
            USING ERRCODE = 'P0001';
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER enforce_usage_quota
    BEFORE INSERT ON tenant_usage_log
    FOR EACH ROW
    EXECUTE FUNCTION check_usage_quota();

The trigger fires before each insert. If the quota would be exceeded, it raises an exception and the entire insert is rolled back—atomically. No API call, no overages.

FastAPI Integration

Your FastAPI endpoint becomes much simpler:

from fastapi import FastAPI, HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy import text

app = FastAPI()

@app.post("/api/v1/analyze-document")
async def analyze_document(request: AnalyzeRequest, current_tenant: Tenant = Depends(get_tenant)):
    try:
        # Try to log the usage *before* making the API call
        # The trigger will reject it if over quota
        async with db.begin():
            await db.execute(
                text("""
                    INSERT INTO tenant_usage_log (tenant_id, feature, tokens_used)
                    VALUES (:tenant_id, :feature, :estimated_tokens)
                """),
                {
                    "tenant_id": current_tenant.id,
                    "feature": "document_analysis",
                    "estimated_tokens": estimate_tokens(request.document)
                }
            )
        # Trigger passed—we're under quota. Safe to call the API
        response = await anthropic_client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": request.document}]
        )
        return {"analysis": response.content[0].text}

    except IntegrityError as e:
        if "quota_exceeded" in str(e.orig):
            raise HTTPException(
                status_code=429,
                detail="Monthly quota exceeded for this feature"
            )
        raise

This ensures the API call only happens if the database accepts the usage log insert. If two concurrent requests come in, one will hit the trigger constraint and fail fast—before Anthropic charges you.

Refreshing Materialized Views

The materialized view approach trades real-time accuracy for query speed (no aggregation at insert time). Refresh it every hour:

# In a background task or scheduled job
async def refresh_usage_views():
    async with db.begin():
        await db.execute(text("REFRESH MATERIALIZED VIEW tenant_monthly_usage"))

Or use REFRESH MATERIALIZED VIEW CONCURRENTLY if you can't afford downtime, but that requires a unique index on the view.

Gotcha: Token Estimation Overhead

I initially tried to make tokens_used exact by querying Claude's token counter before inserting the log. This introduced race conditions of its own—the token count could change between estimation and actual API call. Now I estimate conservatively (len(text) / 4 for UTF-8), log the estimate at insert time, then update it post-response with actual usage. The quota trigger still fires on the estimate, giving you a safety margin.

Another trap: triggers on high-volume tables can become a bottleneck. If you're logging millions of API calls per month, consider partitioning tenant_usage_log by created_at and archiving old partitions. The trigger overhead stays constant even as the table grows.

Why This Matters at Scale

With CitizenApp's 9 AI features and multi-tenant architecture, database-layer enforcement has prevented approximately $8,000 in accidental over-billing. More importantly, it's eliminated the support burden of "why were we charged for 300 calls when our limit is 100?" The answer is always in the database logs, and the answer is always "no, you weren't."

Soft quotas are fine for rate-limiting. Hard quotas—actual cost constraints—belong in the database.