Implementing Free Tier With Redis

To add a Free Tier, we need to move from just checking "is the user subscribed" to tracking usage.

Since AI inference is expensive, your free tier usually needs two things:

1. A Rate Limit: (e.g., 5 messages per minute).
2. A Daily Cap: (e.g., 20 messages per day).

1. The Strategy

• Database (Redis): We'll use Redis because it's lightning-fast for counting hits.
• Webhook: When a user pays, Stripe sends a "ping" to your server. We'll update their status in the database from free to pro.

2. The Integrated Middleware (Free + Paid)

Python

import stripe
import requests
from fastapi import FastAPI, HTTPException, Header, Request
from redis import Redis
import time

app = FastAPI()
redis = Redis(host='localhost', port=6379, decode_responses=True)
stripe.api_key = "sk_test_..."
WEBHOOK_SECRET = "whsec_..."

# Tier Configuration
TIERS = {
"free": {"day_limit": 20, "rpm": 3}, # 20 per day, 3 per minute
"pro": {"day_limit": 5000, "rpm": 60} # Effectively unlimited for a human
}

@app.post("/chat")
async def chat(prompt: str, user_id: str = Header(...)):
# 1. Get user status from Redis (Default to free)
user_status = redis.get(f"user:{user_id}:tier") or "free"
limits = TIERS[user_status]

# 2. Check Minute Rate Limit (RPM)
minute_key = f"usage:{user_id}:minute"
rpm_count = redis.incr(minute_key)
if rpm_count == 1: redis.expire(minute_key, 60)
if rpm_count > limits["rpm"]:
raise HTTPException(status_code=429, detail="Rate limit exceeded. Slow down!")

# 3. Check Daily Cap
day_key = f"usage:{user_id}:day"
daily_count = redis.incr(day_key)
if daily_count == 1: redis.expire(day_key, 86400)
if daily_count > limits["day_limit"]:
raise HTTPException(status_code=403, detail="Daily limit reached. Upgrade to Pro!")

# 4. Call Ollama
resp = requests.post("http://localhost:11434/api/generate",
json={"model": "llama3", "prompt": prompt, "stream": False})
return resp.json()

3. The Stripe Webhook (The "Automatic Upgrader")

This endpoint listens for Stripe events. When someone pays, it instantly switches their tier in Redis.

Python

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

try:
event = stripe.Webhook.construct_event(payload, sig_header, WEBHOOK_SECRET)
except Exception:
raise HTTPException(status_code=400)

# When a subscription is successful
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
user_id = session["client_reference_id"] # Pass this when creating checkout

# Upgrade them in Redis
redis.set(f"user:{user_id}:tier", "pro")
print(f"User {user_id} upgraded to PRO!")

# When a subscription is canceled
if event["type"] == "customer.subscription.deleted":
customer_id = event["data"]["object"]["customer"]
# Logic to find your internal user_id from stripe customer_id
# redis.set(f"user:{user_id}:tier", "free")

return {"status": "success"}

4. How to test this locally

1. Install Redis: docker run -d -p 6379:6379 redis
2. Stripe CLI: To test webhooks without a public URL, use the Stripe CLI:
   Bash
   stripe listen --forward-to localhost:8000/webhook
   
3. The Trigger: In a second terminal, simulate a purchase:
   Bash
   stripe trigger checkout.session.completed --override checkout_session:client_reference_id=user_123
   

Why this works for a business:

• Low Friction: Users can try your "ChatGPT" for free immediately.
• Zero Waste: If a free user spams you, Redis cuts them off before a single GPU cycle is wasted.
• Scalable: You can add an "Enterprise" tier just by adding a new dictionary entry in TIERS.

Would you like me to show you how to wrap this into a Docker Compose file so you can deploy the whole thing (Ollama + Redis + API) to a server with one command?