Forem: Ugur Aslim

GitHub Actions for Parallel FastAPI + React Testing: Optimizing CI/CD Feedback Loops in Monorepos

Ugur Aslim — Sat, 23 May 2026 09:26:21 +0000

GitHub Actions for Parallel FastAPI + React Testing: Optimizing CI/CD Feedback Loops in Monorepos

I spent three months watching our CI pipeline consume 8 minutes per pull request. Eight minutes. That's long enough to switch contexts, check Slack, and forget what you were debugging. At CitizenApp, we have 9 AI features and 40+ API endpoints—skipping CI isn't an option, but neither is waiting forever for feedback.

The fix wasn't complicated, but it required thinking differently about how GitHub Actions executes jobs. Most teams either run everything sequentially (slow) or parallelize recklessly without understanding database isolation (flaky). I'm going to show you the exact pattern that cut our CI time to 90 seconds and kept our flake rate at zero.

The Problem: Sequential Testing Kills Productivity

Here's what our original workflow looked like:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run FastAPI tests
        run: pytest tests/
      - name: Setup Node
        uses: actions/setup-node@v4
        with:
          node-version: "20"
      - name: Install React dependencies
        run: npm install
      - name: Run React tests
        run: npm run test

This is a single job doing everything sequentially. Python dependencies install while Node sits idle. FastAPI tests run while React dependencies aren't even installed yet. It's like waiting for a single cashier when you have five registers.

The real cost: 8 minutes total, and you're blocked for any feedback.

The Solution: Matrix Strategy + Service Containers

I prefer running backend and frontend tests in separate parallel jobs with isolated PostgreSQL instances. Here's why:

Different environments: Backend needs PostgreSQL; frontend doesn't. Why install both in one job?
Independent scaling: If React tests take 20 seconds and FastAPI tests take 90, parallelize them instead of concatenating.
Service isolation: Each test job gets its own database instance on a unique port. No race conditions, no port conflicts.

Here's the workflow:

name: CI

on:
  pull_request:
    branches: [main]
  push:
    branches: [main]

jobs:
  backend-test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_USER: testuser
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: testdb
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
          cache: "pip"

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run FastAPI tests
        env:
          DATABASE_URL: postgresql://testuser:testpass@localhost:5432/testdb
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: pytest tests/ -v --cov=src --cov-report=xml

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml

  frontend-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Node
        uses: actions/setup-node@v4
        with:
          node-version: "20"
          cache: "npm"

      - name: Install dependencies
        run: npm ci

      - name: Run React tests
        run: npm run test:ci

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage/coverage-final.json

  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
          cache: "pip"
      - run: pip install ruff black
      - run: ruff check .
      - run: black --check .

This runs three jobs in parallel: backend-test, frontend-test, and lint. The PR is blocked only on the slowest job. In our case, that's backend tests at ~90 seconds. Frontend finishes in 25 seconds. Linting takes 8 seconds.

Why Service Containers Matter

The services key in GitHub Actions is magic. It spins up a PostgreSQL container for that job only, with health checks built in. Each test job gets its own isolated database—no test pollution, no port conflicts if you ever run multiple workflows simultaneously.

services:
  postgres:
    image: postgres:16-alpine
    env:
      POSTGRES_USER: testuser
      POSTGRES_PASSWORD: testpass
      POSTGRES_DB: testdb
    options: >-
      --health-cmd pg_isready
      --health-interval 10s
      --health-timeout 5s
      --health-retries 5
    ports:
      - 5432:5432

The --health-cmd pg_isready is critical. GitHub Actions waits for this check to pass before running your steps. I learned this the hard way—without it, your test job starts and immediately fails because the database isn't ready yet.

Intelligent Dependency Caching

Notice cache: "pip" and cache: "npm". This is GitHub Actions' built-in caching—it hashes your requirements.txt or package-lock.json and restores dependencies automatically. On subsequent runs, this saves 30–40 seconds alone.

For Python:

- uses: actions/setup-python@v4
  with:
    python-version: "3.11"
    cache: "pip"

For Node:

- uses: actions/setup-node@v4
  with:
    node-version: "20"
    cache: "npm"

This is not optional. Without caching, you're downloading and installing dependencies every single run. That's 3–4 minutes you don't need to waste.

Gotcha: Database Initialization Timing

Here's what burned me initially: I was running pytest immediately after installing dependencies, assuming the PostgreSQL container would be ready. It wasn't. The container was still starting up, and tests failed with "connection refused" errors.

The fix: The --health-cmd and --health-retries options handle this, but you can also add explicit waits in your test setup:

# tests/conftest.py
import time
from sqlalchemy import create_engine
from sqlalchemy.exc import OperationalError

def wait_for_db(max_retries=30):
    for attempt in range(max_retries):
        try:
            engine = create_engine(os.getenv("DATABASE_URL"))
            with engine.connect() as conn:
                conn.execute(text("SELECT 1"))
            return
        except OperationalError:
            if attempt == max_retries - 1:
                raise
            time.sleep(1)

@pytest.fixture(scope="session", autouse=True)
def setup_db():
    wait_for_db()
    # Run migrations
    subprocess.run(["alembic", "upgrade", "head"], check=True)

Without this explicit wait, you get flaky tests that pass 80% of the time. Not fun in production.

The Numbers

Before: 8 minutes, 2 seconds (sequential)
After: 1 minute, 35 seconds (parallel, cached)
Improvement: 80% faster

That's not premature optimization—that's respecting developers' time. Every developer on CitizenApp makes 10–15 PRs per week. Over a year, we're saving roughly 200 hours of waiting.

The Takeaway

Parallelization is a force multiplier, but only if you respect dependencies and isolation. Backend and frontend tests have different needs—acknowledge that in your workflow design. Use service containers for database-dependent tests. Cache aggressively. And always, always verify your database is actually ready before you run tests.

Building Dynamic RBAC in React 19: From Permission Strings to Component-Level Access Control

Ugur Aslim — Sat, 23 May 2026 09:22:46 +0000

Building Dynamic RBAC in React 19: From Permission Strings to Component-Level Access Control

String-based permission checks scattered across your React codebase are a maintenance nightmare. I know because I shipped CitizenApp with that anti-pattern, and it nearly bit me when we added our fifth AI feature.

The problem? Permissions were hardcoded in components. When the marketing team wanted to trial a feature with select customers, I had to grep through half the codebase, find every if (user.role === 'admin') check, and create some Frankenstein conditional. Worse, there was no single source of truth for what "feature_x_access" actually meant across our tenant hierarchy.

This post shows you how to build a type-safe, composable RBAC layer that lives outside your components. Your UI asks "can I do X?" and the permission engine answers. Clean separation. Testable. Scales.

The Core Philosophy: Permissions Are Data, Not Logic

Don't embed permissions in your component tree. Treat permissions as configuration that your components consume. This single shift unlocks everything.

Here's what bad looks like:

// ❌ Don't do this
export function AIFeatureCard() {
  const { user } = useAuth();

  if (user.role !== 'admin' && user.role !== 'premium_subscriber') {
    return null;
  }

  return <div>AI Feature</div>;
}

Problems:

Role names are magic strings
Permission logic is fragmented across 20 components
Adding a new role? Find and update every check
No way to test permissions without rendering components
No audit trail of what permission was checked where

Here's what good looks like:

// ✅ Do this
const canAccessAIFeature = await checkPermission('features:ai:access', {
  userId,
  tenantId,
});

if (canAccessAIFeature) {
  return <AIFeatureCard />;
}

Now permissions are data. You can log them, cache them, test them, audit them.

Type-Safe Permission Definitions

Start with TypeScript. Define every permission your app has as a const object:

// permissions.ts
export const PERMISSIONS = {
  // Organization management
  'org:create': 'Create organization',
  'org:update': 'Update organization settings',
  'org:delete': 'Delete organization',
  'org:invite_members': 'Invite team members',

  // AI features (your feature gates)
  'features:ai:access': 'Access any AI feature',
  'features:ai:documents': 'Use document analysis',
  'features:ai:workflows': 'Create automation workflows',
  'features:ai:exports': 'Export AI-generated content',

  // Admin
  'admin:billing': 'Manage billing',
  'admin:audit_logs': 'View audit logs',
} as const;

// Extract type: 'org:create' | 'org:update' | ...
export type PermissionKey = keyof typeof PERMISSIONS;

This gives you autocomplete and catches typos at compile time. No more permission strings as magic text.

Building the Permission Engine

Your permission engine lives on the backend and is queried from React. Here's the FastAPI service:

# permissions.py
from enum import Enum
from typing import Set
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

class RoleType(str, Enum):
    OWNER = "owner"
    ADMIN = "admin"
    MEMBER = "member"
    GUEST = "guest"

# Define role -> permissions mapping
ROLE_PERMISSIONS: dict[RoleType, Set[str]] = {
    RoleType.OWNER: {
        "org:create", "org:update", "org:delete", "org:invite_members",
        "features:ai:access", "features:ai:documents", "features:ai:workflows",
        "features:ai:exports", "admin:billing", "admin:audit_logs",
    },
    RoleType.ADMIN: {
        "org:update", "org:invite_members",
        "features:ai:access", "features:ai:documents", "features:ai:workflows",
        "features:ai:exports", "admin:audit_logs",
    },
    RoleType.MEMBER: {
        "features:ai:access", "features:ai:documents", "features:ai:workflows",
    },
    RoleType.GUEST: {
        "features:ai:documents",  # Read-only
    },
}

# Handle permission inheritance for multi-tenant
class PermissionResolver:
    def __init__(self, db: Session):
        self.db = db

    async def get_user_permissions(
        self, user_id: str, tenant_id: str
    ) -> Set[str]:
        """Resolve all permissions for a user in a tenant."""
        # Query user role in this tenant
        membership = self.db.query(TenantMembership).filter(
            TenantMembership.user_id == user_id,
            TenantMembership.tenant_id == tenant_id,
        ).first()

        if not membership:
            return set()

        # Start with their direct role permissions
        permissions = ROLE_PERMISSIONS.get(membership.role, set()).copy()

        # Add custom permissions (if you've granted individual perms)
        custom = self.db.query(UserPermission).filter(
            UserPermission.user_id == user_id,
            UserPermission.tenant_id == tenant_id,
        ).all()

        for perm in custom:
            permissions.add(perm.permission_key)

        return permissions

    async def can_perform(
        self, user_id: str, tenant_id: str, permission: str
    ) -> bool:
        """Check if user can perform an action."""
        permissions = await self.get_user_permissions(user_id, tenant_id)
        return permission in permissions

# Expose as endpoint
@app.post("/api/permissions/check")
async def check_permission(
    request: CheckPermissionRequest,  # {permission, tenant_id}
    user_id: str = Depends(get_current_user_id),
    db: Session = Depends(get_db),
) -> dict[str, bool]:
    resolver = PermissionResolver(db)
    allowed = await resolver.can_perform(
        user_id, request.tenant_id, request.permission
    )
    return {"allowed": allowed}

Notice: No logic in components. All rules live in ROLE_PERMISSIONS and the database. When you add a feature, you update ROLE_PERMISSIONS once. Done.

React Hook: usePermission

Now the React side. Create a hook that queries your permission endpoint:

// usePermission.ts
import { useAuth } from './useAuth';
import { PermissionKey, PERMISSIONS } from './permissions';

interface UsePermissionOptions {
  cacheSeconds?: number;
}

export function usePermission(
  permissionKey: PermissionKey,
  options: UsePermissionOptions = {}
) {
  const { user, tenantId } = useAuth();
  const [allowed, setAllowed] = React.useState<boolean | null>(null);
  const [loading, setLoading] = React.useState(true);
  const [error, setError] = React.useState<Error | null>(null);

  React.useEffect(() => {
    if (!user || !tenantId) {
      setLoading(false);
      setAllowed(false);
      return;
    }

    const checkPerm = async () => {
      try {
        const response = await fetch('/api/permissions/check', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            permission: permissionKey,
            tenant_id: tenantId,
          }),
        });

        if (!response.ok) throw new Error('Permission check failed');
        const data = await response.json();
        setAllowed(data.allowed);
      } catch (err) {
        setError(err as Error);
        setAllowed(false); // Fail closed
      } finally {
        setLoading(false);
      }
    };

    checkPerm();
  }, [user?.id, tenantId, permissionKey]);

  return { allowed, loading, error };
}

Usage in components:

// AIFeatureCard.tsx
export function AIFeatureCard() {
  const { allowed, loading } = usePermission('features:ai:access');

  if (loading) return <Skeleton />;
  if (!allowed) return null;

  return <div className="p-4 bg-blue-50">AI Feature</div>;
}

Clean. Testable. Type-safe.

Caching: The Performance Multiplier

This burned me: I shipped this without caching, and every component requesting the same permission hammered the backend. Use React Query:

// usePermission.ts (improved)
import { useQuery } from '@tanstack/react-query';

export function usePermission(
  permissionKey: PermissionKey,
  options: UsePermissionOptions = {}
) {
  const { user, tenantId } = useAuth();
  const cacheSeconds = options.cacheSeconds ?? 300; // 5 min default

  return useQuery({
    queryKey: ['permission', tenantId, permissionKey],
    queryFn: async () => {
      const response = await fetch('/api/permissions/check', {
        method: 'POST',
        body: JSON.stringify({
          permission: permissionKey,
          tenant_id: tenantId,
        }),
      });
      const data = await response.json();
      return data.allowed;
    },
    staleTime: cacheSeconds * 1000,
    enabled: !!user && !!tenantId,
  });
}

Now the second component asking for features:ai:access hits the cache, not your backend.

Gotcha: Stale Permission State After Tenant Changes

Testing FastAPI + SQLAlchemy with Real PostgreSQL Fixtures: No More Mocking Misery

Ugur Aslim — Sat, 23 May 2026 09:12:23 +0000

Testing FastAPI + SQLAlchemy with Real PostgreSQL Fixtures: No More Mocking Misery

I spent three months debugging why a query worked in tests but failed in production. The culprit? I'd mocked the entire database layer.

Mocks are seductive. They're fast, isolated, and you control everything. But they're also liars. They hide migration bugs, concurrency issues, transaction edge cases, and the thousand small ways your ORM behaves differently under real database constraints. I'd rather catch those bugs in CI than at 2 AM in production.

This post shows you how to test FastAPI + SQLAlchemy against real PostgreSQL instances—cheaply, quickly, and with perfect isolation. No mocking the data layer. No false confidence.

Why Real Databases Beat Mocks

Before diving into code, let me be direct about why I changed my mind.

Mocks hide real problems. When you mock session.query(), you're not testing how SQLAlchemy constructs queries, handles N+1 problems, or manages transactions. You're testing your test setup.

Migrations are invisible to mocks. I once added a NOT NULL constraint in a migration. Tests passed (mocked). Production broke (real database). Now I run real migrations in every test database.

Concurrency and locks don't exist in mocks. Row-level locking, deadlock conditions, and transaction isolation levels are real problems in multi-tenant systems. Mocks can't catch them.

Your ORM behaves differently than you expect. SQLAlchemy's relationship loading, cascade deletes, and lazy vs. eager loading have edge cases. Real tests find them.

The trade-off isn't as bad as it sounds. Docker + pytest fixtures make spinning up isolated PostgreSQL databases fast—we're talking milliseconds for setup, and modern CI can parallelize tests across multiple database instances.

The Setup: Docker Compose + Pytest Fixtures

I use a minimal Docker Compose setup that spins up PostgreSQL and tears it down cleanly between test runs.

docker-compose.test.yml:

version: '3.8'
services:
  postgres-test:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: test_user
      POSTGRES_PASSWORD: test_password
      POSTGRES_DB: test_db
    ports:
      - "5433:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U test_user"]
      interval: 1s
      timeout: 5s
      retries: 10
    tmpfs:
      - /var/lib/postgresql/data

The tmpfs line is crucial—it stores the database in RAM, making writes blazingly fast. Perfect for tests.

conftest.py — this is where the magic happens:

import pytest
import subprocess
import time
from sqlalchemy import create_engine, text
from sqlalchemy.pool import NullPool
from sqlalchemy.orm import sessionmaker, Session

# Import your models
from app.models import Base
from app.database import get_db


@pytest.fixture(scope="session")
def docker_compose():
    """Spin up Docker Compose once per test session."""
    subprocess.run(
        ["docker-compose", "-f", "docker-compose.test.yml", "up", "-d"],
        check=True,
        cwd=".",
    )

    # Wait for postgres to be ready
    max_retries = 30
    for i in range(max_retries):
        try:
            engine = create_engine(
                "postgresql://test_user:test_password@localhost:5433/test_db"
            )
            with engine.connect() as conn:
                conn.execute(text("SELECT 1"))
            break
        except Exception as e:
            if i == max_retries - 1:
                raise
            time.sleep(0.5)

    yield

    # Tear down
    subprocess.run(
        ["docker-compose", "-f", "docker-compose.test.yml", "down"],
        check=True,
        cwd=".",
    )


@pytest.fixture
def db_session(docker_compose):
    """Create a fresh database session for each test."""
    # Create engine with NullPool—don't reuse connections between tests
    engine = create_engine(
        "postgresql://test_user:test_password@localhost:5433/test_db",
        poolclass=NullPool,
    )

    # Run migrations (I use Alembic; you could also use Base.metadata.create_all)
    subprocess.run(
        ["alembic", "upgrade", "head"],
        check=True,
        cwd=".",
    )

    SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
    session = SessionLocal()

    yield session

    # Cleanup: rollback any uncommitted transactions
    session.rollback()
    session.close()

    # Drop all tables between tests for true isolation
    Base.metadata.drop_all(engine)


@pytest.fixture
def client(db_session):
    """Provide a FastAPI test client with dependency injection."""
    from fastapi.testclient import TestClient
    from app.main import app

    def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    return TestClient(app)

Notice the key decisions:

NullPool: Each test gets a fresh connection. No connection reuse surprises.
drop_all() after each test: Complete isolation. No shared state.
Alembic migrations run in every test: You catch migration bugs immediately.

Real Test Example: Catching What Mocks Miss

Let's test a user creation endpoint with a unique email constraint:

from app.models import User
from datetime import datetime


def test_create_user_success(client, db_session):
    """Test successful user creation."""
    response = client.post(
        "/users",
        json={"email": "alice@example.com", "name": "Alice"},
    )

    assert response.status_code == 201
    assert response.json()["email"] == "alice@example.com"

    # Verify it's actually in the database
    user = db_session.query(User).filter_by(email="alice@example.com").first()
    assert user is not None
    assert user.name == "Alice"


def test_create_duplicate_email_fails(client, db_session):
    """Test that duplicate emails are rejected.

    This catches:
    - Whether your database constraint actually exists
    - Whether your error handling returns the right status code
    - Whether the transaction rolled back properly
    """
    # Insert first user
    client.post(
        "/users",
        json={"email": "bob@example.com", "name": "Bob"},
    )

    # Try to insert duplicate
    response = client.post(
        "/users",
        json={"email": "bob@example.com", "name": "Bob 2"},
    )

    assert response.status_code == 409  # Conflict
    assert "already exists" in response.json()["detail"].lower()

    # Verify only one user exists
    count = db_session.query(User).filter_by(email="bob@example.com").count()
    assert count == 1


def test_n_plus_one_query_detection(client, db_session):
    """Catch N+1 queries against real PostgreSQL."""
    from sqlalchemy import event

    # Track queries
    queries = []

    @event.listens_for(engine, "before_cursor_execute")
    def receive_before_cursor_execute(conn, cursor, statement, params, context, executemany):
        queries.append(statement)

    # Create users with posts
    for i in range(5):
        user = User(email=f"user{i}@example.com", name=f"User {i}")
        db_session.add(user)
    db_session.commit()

    # Fetch users and their posts efficiently
    users = db_session.query(User).options(joinedload(User.posts)).all()

    # Against a real database, you can verify query patterns
    # Mocks can't catch these subtle performance bugs
    assert len(queries) < 10  # Rough check; adjust for your schema

That last test is impossible to write meaningfully against mocks. With real PostgreSQL, you actually catch N+1 queries, missing indexes, and transaction issues.

Gotcha: Test Database Concurrency

Here's what burned me: I created a test for concurrent writes, but because each test got its own db_session, I wasn't actually testing multiple concurrent connections.

Solution: When you need true concurrency tests, use separate database sessions:

def test_concurrent_writes(db_session):
    """Test actual concurrent database writes."""
    from concurrent.futures import ThreadPoolExecutor
    from app.database import SessionLocal

    def create_user(email: str):
        session = SessionLocal()
        try:
            user = User(email=email, name=email.split("@")[0])
            session.add(user)
            session.commit()
            return user.id
        finally:
            session.close()

    # Run 10 writes concurrently
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [
            executor.submit(create_user, f"user{i}@example.com")
            for i in range(10)
        ]
        results = [f.result() for f in futures]

    # Verify all succeeded
    assert len(set(results)) == 10  # All IDs unique

This actually stress-tests your database and connection pool. Mocks wouldn't catch connection pool exhaustion or deadlocks.

Speed: It's Faster Than You Think

People worry real database tests are slow. They're not—especially with tmpfs and good fixture design:

Database spin-up: ~2s (once per session)
Per-test setup: ~50ms
Typical test execution: ~100ms

A 100-test suite runs in under 15 seconds locally. In CI with parallelization

I Built a Multi-Tenant SaaS for 50+ Tenants — Here's the Complete Architecture

Ugur Aslim — Fri, 22 May 2026 18:44:07 +0000

I Built a Multi-Tenant SaaS for 50+ Tenants — Here's the Complete Architecture

Six months into building CitizenApp — a GDPR-compliant citizen management SaaS — a customer asked: "Are you sure my data is completely separate from other organisations using this?"

I said yes. Then I went and checked the code. I found three endpoints that could theoretically return cross-tenant data if a specific race condition hit. Nothing had leaked. But it could have.

That conversation triggered a complete rearchitecture of our tenant isolation stack. This post documents every significant decision I made — and what I'd do differently if I started over today.

The First Decision: Choosing Your Isolation Model

Everything downstream depends on how you isolate tenants at the data layer. Three options:

Strategy	Isolation	Complexity	Migration cost	Cross-tenant queries
`tenant_id` column	Logical	Low	Low	Easy
Schema per tenant	Physical	Medium	High	Hard
Database per tenant	Full	High	Very high	Impossible

For CitizenApp, I chose tenant_id column on every table, enforced at two layers: the application repository and PostgreSQL Row-Level Security. Here's why:

Scale: Under 500 tenants at launch. Schema-per-tenant complexity pays off at 10,000+ tenants when you need schema-level customisation per client.
GDPR compliance: Logical isolation with RLS is sufficient for GDPR. Physical isolation adds ops complexity without a compliance benefit at this scale.
Query patterns: I need cross-tenant analytics in the admin dashboard. Schema-per-tenant makes this significantly harder.
Migration tooling: Alembic works cleanly with a single schema. Per-schema migrations require custom tooling.

Every data model inherits from a base mixin:

class TenantMixin:
    tenant_id: Mapped[int] = mapped_column(
        ForeignKey("tenants.id"), nullable=False, index=True
    )

class VatandasModel(TenantMixin, Base):
    __tablename__ = "vatandas"
    id: Mapped[int] = mapped_column(primary_key=True)
    ad_soyad: Mapped[str] = mapped_column(String(200))
    tc_no: Mapped[str] = mapped_column(String(11))  # encrypted at rest

The Lesson I Learned the Hard Way: Application-Layer Isolation Isn't Enough

Most teams stop here:

@router.get("/users")
async def list_users(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
    return db.query(User).filter(User.tenant_id == current_user.tenant_id).all()

This looks safe. It isn't — not as a guarantee:

Forgotten filters: A new endpoint queries User without the tenant check. Easy mistake, invisible in tests.
Scope creep: An admin panel needs to see all users, so you bypass the filter. Now that code path exists, and someone copies it.
N+1 relationships: You load users, loop through, load their audit logs. The second query forgets the tenant filter.
Background jobs: A Celery task runs as a "system user" with tenant_id = None. It can see everything.

These bugs are invisible until they're exploited. Your tests pass because they run within a single tenant context.

The Fix: PostgreSQL Row-Level Security as the Second Enforcement Layer

RLS lives in the database. PostgreSQL evaluates it before returning any row. You cannot read data you're not allowed to read — the database won't let you, regardless of what application code runs.

Here's the pattern:

-- Enable RLS on every tenant-scoped table
ALTER TABLE vatandas ENABLE ROW LEVEL SECURITY;
ALTER TABLE vatandas FORCE ROW LEVEL SECURITY;

-- Policy: only see rows matching the session's tenant
CREATE POLICY vatandas_tenant_isolation ON vatandas
  FOR ALL
  USING (tenant_id = current_setting('app.current_tenant_id')::int);

-- Superuser/admin bypass
CREATE POLICY vatandas_admin_all ON vatandas
  FOR ALL
  USING (current_setting('app.is_admin', true)::boolean = true);

The key is current_setting('app.current_tenant_id'). This is a PostgreSQL session variable your application sets after authentication. The database reads it on every query — automatically, everywhere.

Cascade this to every table. Once you set the session context once, every query across all tenant-scoped tables is filtered — JOINs, subqueries, transactions touching three tables:

ALTER TABLE audit_logs ENABLE ROW LEVEL SECURITY;
CREATE POLICY audit_tenant_isolation ON audit_logs
  FOR ALL USING (tenant_id = current_setting('app.current_tenant_id')::int);

ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
CREATE POLICY documents_tenant_isolation ON documents
  FOR ALL USING (tenant_id = current_setting('app.current_tenant_id')::int);

Wiring RLS into FastAPI + SQLAlchemy

Set the context once per request, immediately after JWT verification:

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session

engine = create_engine("postgresql://...", echo=False)
SessionLocal = sessionmaker(bind=engine)

def set_rls_context(session: Session, tenant_id: int, is_admin: bool = False):
    """Set PostgreSQL session variables for RLS enforcement."""
    session.execute(text("SET LOCAL app.current_tenant_id = :tid"), {"tid": tenant_id})
    session.execute(text("SET LOCAL app.is_admin = :admin"), {"admin": str(is_admin).lower()})

async def get_db(current_user: User = Depends(get_current_user)) -> Session:
    session = SessionLocal()
    try:
        set_rls_context(
            session,
            tenant_id=current_user.tenant_id,
            is_admin=current_user.role == "admin"
        )
        yield session
    finally:
        session.close()

Now your endpoints are clean — no manual tenant filtering:

@router.get("/citizens")
async def list_citizens(db: Session = Depends(get_db)):
    # No .filter(tenant_id == ...) needed — RLS handles it
    return db.query(VatandasModel).all()

The Connection Pooling Gotcha

Session variables don't persist across physical connections in a connection pool. If you use PgBouncer or similar, a subsequent request may get a different connection with no context set.

Solution: Use SET LOCAL (not SET) — it scopes the variable to the current transaction. Alternatively, set it at the start of every request via middleware:

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        token = request.headers.get("authorization", "").replace("Bearer ", "")
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        request.state.tenant_id = payload["tenant_id"]
        response = await call_next(request)
        return response

The Migration Gotcha

Alembic migrations run as a privileged role. If FORCE ROW LEVEL SECURITY is set and that role has RLS applied, migrations can fail. Handle it:

# alembic/env.py
def run_migrations_online():
    with connectable.connect() as connection:
        # Temporarily bypass RLS for migration scripts
        connection.execute(text("SET app.current_tenant_id = '0'"))
        connection.execute(text("SET app.is_admin = 'true'"))
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

Local Development Gotcha

current_setting() raises an error if not set (unless you use the two-argument form current_setting('app.current_tenant_id', true) which returns NULL). A NULL tenant_id means zero rows returned — which is the safe default, but confusing in dev. Always set a test context in fixtures:

@pytest.fixture
def db_with_rls():
    session = SessionLocal()
    set_rls_context(session, tenant_id=1)
    yield session
    session.close()

JWT Design: What's in the Token and Why

Token design has permanent consequences. Tokens issued today will be used until they expire — you can't retroactively change their structure without forcing re-login.

CitizenApp's access token payload:

{
    "sub": "42",           # user_id
    "tenant_id": 7,        # embedded — no DB lookup per request
    "role": "operator",    # embedded — no DB lookup per request
    "email": "u@co.com",   # for audit logging without a DB hit
    "exp": 1716307200,     # 15-minute expiry
    "type": "access"
}

Why embed tenant_id and role directly: Every request needs these. If they're not in the token, every API call requires a database lookup. At 100 requests/second, that's 100 unnecessary DB queries/second. Token-embedded claims mean zero DB round-trips for auth.

The tradeoff: A role change takes up to 15 minutes to propagate. Acceptable for CitizenApp. If you need immediate revocation, use short-lived tokens (5 minutes) plus a Redis blocklist.

Refresh token rotation with theft detection:

async def rotate_refresh_token(db: AsyncSession, old_token: str) -> tuple[str, str]:
    stored = await get_refresh_token(db, old_token)

    if not stored or stored.is_revoked:
        # Token reuse detected — possible theft
        await revoke_token_family(db, stored.family_id)
        raise HTTPException(status_code=401, detail="Token reuse detected")

    await revoke_token(db, old_token)
    new_access  = create_access_token(stored.user_id, stored.tenant_id)
    new_refresh = await create_refresh_token(db, stored.user_id, stored.family_id)
    return new_access, new_refresh

If a refresh token is used twice, the entire token family is revoked. This stops an attacker even if they intercepted the token first.

RBAC: Three Roles, One Decision Point

CitizenApp has three roles: admin, operator, viewer. I deliberately did not implement fine-grained permissions.

Why not: Premature flexibility. At three roles with clear semantics, a permission matrix adds indirection without value. The hierarchy:

viewer → read-only access to citizen data
operator → read + write citizen data, run imports
admin → everything + user management + billing + tenant settings

def require_role(*roles: str):
    def dependency(current_user: UserModel = Depends(get_current_user)):
        if current_user.role not in roles:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return dependency

@router.delete("/citizens/{id}")
async def delete_citizen(
    id: int,
    user: UserModel = Depends(require_role("admin", "operator")),
    repo: CitizenRepository = Depends(get_citizen_repo),
):
    ...

Clean, readable, testable. When the permission model needs to evolve, it's one function.

Encryption at Rest: Fernet Over pgcrypto

PII like the TC identity number (Turkish national ID) must be encrypted at rest. Three options:

Application-level (Fernet/AES) — encrypt in Python before INSERT
pgcrypto — encrypt at the database level
Transparent Data Encryption — filesystem/disk level

I chose application-level Fernet:

from cryptography.fernet import Fernet

class EncryptionService:
    def __init__(self, key: bytes):
        self.fernet = Fernet(key)

    def encrypt(self, value: str) -> str:
        return self.fernet.encrypt(value.encode()).decode()

    def decrypt(self, value: str) -> str:
        return self.fernet.decrypt(value.encode()).decode()

# SQLAlchemy event listener — transparent to the rest of the code
@event.listens_for(VatandasModel.tc_no, "set")
def encrypt_tc_no(target, value, oldvalue, initiator):
    if value and not value.startswith("gAAAAA"):  # not already encrypted
        target.tc_no = encryption_service.encrypt(value)

Why not pgcrypto: The encryption key must be available to PostgreSQL, meaning it's accessible to anyone with database access. Application-level keeps the key in the application environment (Render secrets), completely separate from the database.

Key rotation is zero-downtime with MultiFernet:

# During rotation: old ciphertext decrypts with old key, new data uses new key
fernet = MultiFernet([new_key_fernet, old_key_fernet])

Async SQLAlchemy: The Patterns That Bit Me

CitizenApp uses FastAPI + asyncpg + async SQLAlchemy. Correct choice, but with sharp edges.

Session scoping — auto-commit/rollback:

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

The N+1 trap — async doesn't protect you from it:

# BAD: N queries inside a loop
citizens = await repo.get_all()
for c in citizens:
    c.audit_count = await audit_repo.count_for(c.id)  # N extra queries

# GOOD: single query with aggregation
stmt = (
    select(VatandasModel, func.count(AuditModel.id).label("audit_count"))
    .outerjoin(AuditModel, AuditModel.citizen_id == VatandasModel.id)
    .where(VatandasModel.tenant_id == tenant_id)
    .group_by(VatandasModel.id)
)

The lazy loading trap — async SQLAlchemy doesn't support lazy loading. Accessing a relationship outside a transaction raises MissingGreenlet. Declare what you need upfront:

stmt = (
    select(UserModel)
    .options(selectinload(UserModel.tenant))  # explicit eager load
    .where(UserModel.id == user_id)
)

Alembic Migrations: Zero-Downtime Rules

Every schema change is an Alembic migration. No exceptions.

Three rules I follow strictly:

1. Never rename a column directly. Add the new column → backfill → update application code → drop old column in a separate migration.

2. New NOT NULL columns need a DEFAULT or two-step migration. Adding NOT NULL without a default takes a table lock.

3. Always create indexes CONCURRENTLY:

# In Alembic migration — standard CREATE INDEX locks the table
op.execute(
    "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_vatandas_tenant_created "
    "ON vatandas(tenant_id, created_at DESC)"
)

Alembic doesn't generate CONCURRENTLY by default. Write it manually.

Frontend: React 19 + TanStack Query

State management: No Redux, no Zustand. TanStack Query for server state, useState for local UI state. The vast majority of React state in a SaaS is server data — citizen lists, audit logs, user lists. TanStack Query handles all of it.

React 19 useOptimistic for instant UX:

const [optimisticCitizens, addOptimistic] = useOptimistic(citizens);

async function handleDelete(id: number) {
    addOptimistic(prev => prev.filter(c => c.id !== id));  // instant UI update

    try {
        await deleteCitizen(id);
        await queryClient.invalidateQueries(["citizens"]);
    } catch {
        // useOptimistic rolls back automatically on error
        toast.error("Delete failed");
    }
}

The UI updates immediately. The API catches up. If the call fails, the optimistic update rolls back automatically — no manual state cleanup.

Tenant context from JWT (frontend-side):

function useTenant() {
    const token = localStorage.getItem("access_token");
    if (!token) return null;
    const payload = JSON.parse(atob(token.split(".")[1]));
    return { id: payload.tenant_id, plan: payload.plan };
}

The frontend decodes the JWT without verification (signature verification is the server's job) to personalise the UI. The backend always validates the token independently — showing wrong UI data is a UX problem, returning wrong API data is a security problem.

What I'd Change Starting Over

1. RLS from day one.

Retrofitting RLS on an existing production schema with live tenants is painful. It should be in the first migration.

2. Separate read and write models.

List endpoints do too much: filter, sort, paginate, aggregate audit counts. Dedicated read models or database views optimised for list queries would be cleaner than the combined write models handling both.

3. Formal API versioning from day one.

CitizenApp is on v1 with no formal versioning. For multi-tenant SaaS where different tenants may be on different frontend versions, /api/v1/ from the start with a migration plan matters.

4. Event sourcing for audit.

The current audit log records events but not full before/after state. Reconstructing "what did this record look like at this point in time" requires replaying events. Full snapshots in the audit table would make compliance reporting significantly easier.

The Result

After implementing RLS alongside the repository pattern:

Zero cross-tenant data incidents in 6 months of production
Application bugs that would have been security holes became no-ops — the database simply returned nothing
New developers on the team can't accidentally bypass isolation — the DB layer doesn't let them

Multi-tenant architecture is about defaults that are safe by construction. The goal isn't to write perfect code — it's to make dangerous mistakes structurally impossible.

Building a multi-tenant SaaS and want a second opinion on the architecture? I'm available for consulting.

GDPR-Ready SaaS Architecture: What You Actually Need to Build

Ugur Aslim — Fri, 22 May 2026 15:17:01 +0000

Every SaaS founder I talk to says "we'll add GDPR later." This is almost always a mistake — not because GDPR is hard, but because retrofitting compliance onto an existing architecture is far more expensive than building it in from the start.

Here's what "GDPR-ready architecture" actually means at the code level.

What GDPR Actually Requires (Technically)

Lawyers will give you a long list. Engineers need a shorter one:

Lawful basis for processing — you know why you're storing each piece of data
Data minimisation — you only store what you actually need
Right to access — users can request their data in a portable format
Right to erasure — users can request deletion ("right to be forgotten")
Breach notification — you know when data was accessed and by whom (audit trail)
Data at rest encryption — sensitive fields are encrypted in the database
Data residency — you know where the data is stored geographically

Most of these are architectural decisions, not legal ones.

The Database Layer: Encryption at Rest

Don't encrypt the whole database — encrypt the sensitive fields. This is cheaper, faster, and more targeted.

# utils/encryption.py
from cryptography.fernet import Fernet, MultiFernet
import os

def get_fernet() -> MultiFernet:
    """
    MultiFernet supports key rotation without re-encrypting existing data.
    PRIMARY_KEY is active for encryption.
    SECONDARY_KEY (optional) is used for decrypting data encrypted with the old key.
    """
    primary = Fernet(os.environ["ENCRYPTION_KEY_PRIMARY"])
    keys = [primary]

    secondary_raw = os.environ.get("ENCRYPTION_KEY_SECONDARY")
    if secondary_raw:
        keys.append(Fernet(secondary_raw))

    return MultiFernet(keys)

def encrypt(value: str) -> str:
    return get_fernet().encrypt(value.encode()).decode()

def decrypt(value: str) -> str:
    return get_fernet().decrypt(value.encode()).decode()

In the model:

# models/user.py
from sqlalchemy import event
from sqlalchemy.orm import reconstructor

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    _national_id_encrypted: Mapped[Optional[str]] = mapped_column("national_id", String(500))

    @property
    def national_id(self) -> Optional[str]:
        if self._national_id_encrypted is None:
            return None
        return decrypt(self._national_id_encrypted)

    @national_id.setter
    def national_id(self, value: Optional[str]) -> None:
        self._national_id_encrypted = encrypt(value) if value else None

The application never touches raw PII. It goes in encrypted, comes out decrypted — transparently.

The Audit Trail

Every access to sensitive data must be logged. Not for your benefit — for the user's. They have the right to know who accessed their data and when.

# models/audit_log.py
class AuditEventType(str, enum.Enum):
    # Auth events
    USER_LOGIN = "user.login"
    USER_LOGIN_FAILED = "user.login_failed"
    USER_LOGOUT = "user.logout"
    USER_2FA_ENABLED = "user.2fa_enabled"

    # Data access
    USER_DATA_ACCESSED = "user.data_accessed"
    USER_DATA_EXPORTED = "user.data_exported"
    USER_DATA_DELETED = "user.data_deleted"

    # Admin actions
    ADMIN_USER_VIEWED = "admin.user_viewed"
    ADMIN_ROLE_CHANGED = "admin.role_changed"

class AuditLog(Base):
    __tablename__ = "audit_logs"

    id: Mapped[int] = mapped_column(primary_key=True)
    event_type: Mapped[AuditEventType] = mapped_column(nullable=False)
    actor_user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
    target_user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id"))
    ip_address: Mapped[Optional[str]] = mapped_column(String(45))  # IPv6 max length
    user_agent: Mapped[Optional[str]] = mapped_column(String(500))
    metadata: Mapped[Optional[dict]] = mapped_column(JSONB)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow, index=True)

Critical: audit logs are write-only for application code. No UPDATE, no DELETE. If a log record needs to be "removed", that itself is a logged event.

The Right to Erasure

"Delete my account" is a GDPR requirement, but it's not always a DELETE FROM users WHERE id = ?.

You need to think through your data relationships:

# services/gdpr.py
async def erase_user_data(db: AsyncSession, user_id: int, actor_id: int) -> None:
    """
    GDPR right to erasure implementation.

    We anonymise rather than delete where foreign keys require it
    (e.g., audit logs referencing this user must be kept for compliance,
    but the personal data within them can be removed).
    """
    user = await db.get(User, user_id)
    if not user:
        raise ValueError(f"User {user_id} not found")

    # Anonymise PII fields (keep the record for referential integrity)
    user.email = f"deleted_{user_id}@anonymised.invalid"
    user.full_name = "Deleted User"
    user.national_id = None
    user.phone = None
    user.is_deleted = True
    user.deleted_at = datetime.utcnow()

    # Revoke all active sessions
    await db.execute(
        update(RefreshToken)
        .where(RefreshToken.user_id == user_id)
        .values(used=True)
    )

    # Log the erasure (this itself is a compliance record)
    db.add(AuditLog(
        event_type=AuditEventType.USER_DATA_DELETED,
        actor_user_id=actor_id,
        target_user_id=user_id,
        metadata={"reason": "gdpr_erasure_request"},
    ))

    await db.commit()

The Right to Data Portability

Users can request their data in a machine-readable format. Build this early:

@router.get("/me/export")
async def export_my_data(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_async_db),
):
    """
    GDPR Article 20 — right to data portability.
    Returns all personal data in a structured JSON format.
    """
    await audit_log(db, AuditEventType.USER_DATA_EXPORTED, actor=current_user)

    return {
        "export_date": datetime.utcnow().isoformat(),
        "data_controller": "uguraslim.com",
        "user": {
            "id": current_user.id,
            "email": current_user.email,
            "full_name": current_user.full_name,
            "created_at": current_user.created_at.isoformat(),
        },
        "activity_summary": await get_user_activity_summary(db, current_user.id),
    }

Data Residency: Where Are Your Servers?

GDPR doesn't ban non-EU servers, but it restricts data transfers. The easiest path: keep EU data in the EU.

For the CitizenApp stack:

Neon Postgres → Frankfurt (eu-central-1) ✓
Render Docker → Frankfurt region ✓
Vercel Edge → edge network (content only, no PII) ✓

Document this. When a client asks "where is our data stored?", you should have a one-line answer.

The GDPR Checklist at Architecture Level

Requirement	Implementation
Lawful basis	Consent field on User model + timestamp
Data minimisation	Explicit schema review — delete unused columns
Encryption at rest	Fernet on PII fields
Access logging	AuditLog table, write-only
Right to access	`/me/export` endpoint
Right to erasure	Anonymisation + session revocation
Breach detection	Suspicious login alerts + IP rate limiting
Data residency	All infra in Frankfurt

The Cost of Doing It Late

I've seen codebases where GDPR was added as an afterthought. The typical cost:

2-4 weeks of schema migration work
Careful data backfill of encrypted fields (risky)
Retrofitting audit logging on hundreds of endpoints
Finding all the places PII leaks into logs, caches, error messages

Building it in from day one: maybe 3-5 days of extra upfront work.

It's not a close comparison.

Building a GDPR-compliant SaaS and want a second opinion on your architecture? Get in touch.

JWT Refresh Token Rotation in FastAPI — The Right Way

Ugur Aslim — Fri, 22 May 2026 15:13:40 +0000

Most FastAPI tutorials show you how to generate a JWT. Almost none of them show you what happens when that token gets stolen — and how proper refresh token rotation prevents the damage.

This is the pattern I use in production. It's what ships in CitizenApp. Here's the full implementation.

The Problem With Naive JWT Auth

A typical JWT setup looks like this:

# ⚠️ This is incomplete — don't ship this
@app.post("/login")
def login(credentials: LoginSchema, db: Session = Depends(get_db)):
    user = authenticate_user(db, credentials.username, credentials.password)
    access_token = create_access_token({"sub": user.id})
    refresh_token = create_refresh_token({"sub": user.id})
    return {"access_token": access_token, "refresh_token": refresh_token}

The access token expires in 15 minutes. The refresh token expires in... 30 days? Never?

The silent flaw: if an attacker steals the refresh token (XSS, network intercept, log exposure), they have unlimited access for the entire token lifetime. Your 15-minute access token is now meaningless.

Rotation + Reuse Detection

The fix is refresh token rotation with reuse detection:

Every /refresh call issues a new refresh token and invalidates the old one
If an old refresh token is presented again → all sessions for that user are revoked

Reuse of a rotated-away token is a strong signal that the token was stolen.

# models.py
class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    family: Mapped[uuid.UUID] = mapped_column(nullable=False)  # rotation chain
    used: Mapped[bool] = mapped_column(default=False)
    expires_at: Mapped[datetime] = mapped_column(nullable=False)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

The family column links a rotation chain. When reuse is detected, we revoke the entire family.

The Rotation Endpoint

# auth/routes.py
@router.post("/refresh")
async def refresh_tokens(
    request: Request,
    payload: RefreshRequest,
    db: AsyncSession = Depends(get_async_db),
):
    token_hash = hash_token(payload.refresh_token)

    # Load the token record
    result = await db.execute(
        select(RefreshToken)
        .where(RefreshToken.token_hash == token_hash)
        .with_for_update()  # pessimistic lock — prevents race conditions
    )
    record = result.scalar_one_or_none()

    if not record:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    # Reuse detection — token was already rotated
    if record.used:
        # Revoke entire rotation family
        await db.execute(
            update(RefreshToken)
            .where(RefreshToken.family == record.family)
            .values(used=True)
        )
        await db.commit()
        raise HTTPException(
            status_code=401,
            detail="Refresh token reuse detected — all sessions revoked"
        )

    if record.expires_at < datetime.utcnow():
        raise HTTPException(status_code=401, detail="Refresh token expired")

    # Mark old token as used (rotated)
    record.used = True

    # Issue new token in the same family
    new_raw = secrets.token_urlsafe(48)
    new_record = RefreshToken(
        user_id=record.user_id,
        token_hash=hash_token(new_raw),
        family=record.family,
        expires_at=datetime.utcnow() + timedelta(days=30),
    )
    db.add(new_record)
    await db.commit()

    # Issue new access token
    access_token = create_access_token({"sub": str(record.user_id)})

    return {
        "access_token": access_token,
        "refresh_token": new_raw,
    }

Hashing Refresh Tokens at Rest

Never store raw refresh tokens in the database. If your DB is compromised, hashed tokens are useless to an attacker.

import hashlib
import secrets

def hash_token(raw: str) -> str:
    """SHA-256 hash — fast, one-way, collision-resistant."""
    return hashlib.sha256(raw.encode()).hexdigest()

def generate_refresh_token() -> tuple[str, str]:
    """Returns (raw_token, hashed_token)."""
    raw = secrets.token_urlsafe(48)
    return raw, hash_token(raw)

The raw token goes to the client (HTTP-only cookie ideally). The hash goes in the DB.

Storing in HTTP-Only Cookies

XSS can't steal what JavaScript can't read:

@router.post("/login")
async def login(response: Response, ...):
    ...
    raw_refresh, _ = generate_refresh_token()
    response.set_cookie(
        key="refresh_token",
        value=raw_refresh,
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 30,  # 30 days
        path="/auth/refresh",         # scoped — not sent on every request
    )
    return {"access_token": access_token}

Test Coverage

Every auth path needs tests. Here's the rotation test:

# tests/test_auth.py
async def test_refresh_token_rotation(client, user_factory):
    user = await user_factory()
    login_resp = await client.post("/auth/login", json={"username": user.email, "password": "testpass"})
    original_refresh = login_resp.cookies["refresh_token"]

    # First refresh — should succeed
    r1 = await client.post("/auth/refresh", cookies={"refresh_token": original_refresh})
    assert r1.status_code == 200
    new_refresh = r1.cookies["refresh_token"]
    assert new_refresh != original_refresh

    # Reuse old token — should trigger reuse detection
    r2 = await client.post("/auth/refresh", cookies={"refresh_token": original_refresh})
    assert r2.status_code == 401
    assert "reuse detected" in r2.json()["detail"]

    # New token should also be revoked after reuse detection
    r3 = await client.post("/auth/refresh", cookies={"refresh_token": new_refresh})
    assert r3.status_code == 401

Summary

Pattern	What it prevents
Token rotation	Stolen token reuse after expiry
Reuse detection	Attacker using a rotated-away token
Family revocation	Entire session chain compromised
Hash at rest	DB breach → token exposure
HTTP-only cookie	XSS token theft

This combination is what I ship by default on every project. Security isn't a feature you bolt on — it's the foundation you build on.

Questions or found an edge case I missed? Reach me at uaslim@me.com

React 19 + TanStack Query: Patterns That Actually Work in Production

Ugur Aslim — Fri, 22 May 2026 15:11:07 +0000

React 19 shipped with a set of new primitives that overlap with what TanStack Query has been doing for years. That overlap is intentional — but it doesn't make TanStack Query obsolete. It changes where the boundary sits.

Here's what I've settled on after shipping a production app with React 19 + TanStack Query v5.

What React 19 Actually Adds

Three things that matter for data fetching:

1. use() for promise unwrapping

// Unwrap a promise in a component — but it suspends!
function UserName({ userPromise }: { userPromise: Promise<User> }) {
  const user = use(userPromise);   // suspends until resolved
  return <span>{user.name}</span>;
}

2. Actions and useActionState

// Form mutations with built-in pending/error state
const [state, submitAction, isPending] = useActionState(
  async (prevState, formData) => {
    const result = await updateProfile(formData);
    return result;
  },
  null
);

3. useOptimistic

// Optimistic UI without external state manager
const [optimisticItems, addOptimistic] = useOptimistic(items);

These are great primitives. They don't replace TanStack Query. Here's why.

Where TanStack Query Still Wins

React 19's primitives handle a single request well. TanStack Query handles the lifecycle of requests across your entire app:

Deduplication — 10 components request the same user, 1 network call
Background refetching — stale-while-revalidate, focus refetch, interval polling
Cache coordination — invalidate after mutation, optimistic update with rollback
DevTools — see every query, its state, when it last fetched
Infinite queries — pagination with automatic page management

Once your app has more than a handful of async calls, you want TanStack Query managing the cache.

The Pattern I Use: Server for Mutations, Query for Reads

The clean split:

// ✅ Reads → TanStack Query
function useUser(userId: string) {
  return useQuery({
    queryKey: ['users', userId],
    queryFn: () => api.get<User>(`/users/${userId}`),
    staleTime: 5 * 60 * 1000,   // 5 minutes
  });
}

// ✅ Mutations → useMutation + React 19 useOptimistic
function useUpdateUser() {
  const queryClient = useQueryClient();

  return useMutation({
    mutationFn: (data: UpdateUserPayload) => api.patch('/users/me', data),
    onSuccess: () => {
      queryClient.invalidateQueries({ queryKey: ['users'] });
    },
  });
}

Optimistic Updates: The Right Way

React 19 has useOptimistic, but TanStack Query's onMutate + onError gives you rollback for free:

function useToggleFavourite() {
  const queryClient = useQueryClient();

  return useMutation({
    mutationFn: (itemId: string) => api.post(`/items/${itemId}/favourite`),

    onMutate: async (itemId) => {
      // Cancel any outgoing refetches (avoid overwriting optimistic update)
      await queryClient.cancelQueries({ queryKey: ['items'] });

      // Snapshot the previous value
      const previousItems = queryClient.getQueryData<Item[]>(['items']);

      // Optimistically update
      queryClient.setQueryData<Item[]>(['items'], (old = []) =>
        old.map(item =>
          item.id === itemId
            ? { ...item, isFavourite: !item.isFavourite }
            : item
        )
      );

      // Return snapshot for rollback
      return { previousItems };
    },

    onError: (_err, _itemId, context) => {
      // Roll back on error
      if (context?.previousItems) {
        queryClient.setQueryData(['items'], context.previousItems);
      }
    },

    onSettled: () => {
      // Always refetch after mutation
      queryClient.invalidateQueries({ queryKey: ['items'] });
    },
  });
}

This pattern handles every failure mode: network error, server error, optimistic state gets rolled back automatically.

Forms with React 19 Actions + Query Invalidation

Where React 19 Actions genuinely shine: form submissions that need pending state and validation feedback.

// components/ProfileForm.tsx
function ProfileForm() {
  const queryClient = useQueryClient();
  const { data: user } = useUser('me');

  const [state, submitAction, isPending] = useActionState(
    async (_prevState: FormState, formData: FormData) => {
      const result = await updateProfileAction(formData);   // server action

      if (result.success) {
        // Invalidate the query cache after successful mutation
        await queryClient.invalidateQueries({ queryKey: ['users', 'me'] });
        return { success: true, errors: null };
      }

      return { success: false, errors: result.errors };
    },
    { success: false, errors: null }
  );

  return (
    <form action={submitAction}>
      <input name="name" defaultValue={user?.name} />
      {state.errors?.name && (
        <p className="text-red-500 text-sm">{state.errors.name}</p>
      )}
      <button disabled={isPending}>
        {isPending ? 'Saving...' : 'Save changes'}
      </button>
    </form>
  );
}

The action handles submission + pending state. TanStack Query handles cache invalidation. Clean separation.

Query Key Factories: Avoid Key Spaghetti

As your app grows, query keys become a maintenance problem. Key factories fix this:

// lib/query-keys.ts
export const queryKeys = {
  users: {
    all: ['users'] as const,
    lists: () => [...queryKeys.users.all, 'list'] as const,
    list: (filters: UserFilters) => [...queryKeys.users.lists(), filters] as const,
    details: () => [...queryKeys.users.all, 'detail'] as const,
    detail: (id: string) => [...queryKeys.users.details(), id] as const,
  },
  items: {
    all: ['items'] as const,
    byUser: (userId: string) => [...queryKeys.items.all, 'byUser', userId] as const,
  },
} as const;

// Usage — no magic strings anywhere
const { data } = useQuery({
  queryKey: queryKeys.users.detail(userId),
  queryFn: () => api.get<User>(`/users/${userId}`),
});

// Invalidation is precise
queryClient.invalidateQueries({ queryKey: queryKeys.users.details() });
// ^ invalidates ALL detail queries, not the list

This makes refactoring safe. Change the key structure in one place, TypeScript catches every consumer.

Suspense Integration

React 19 makes Suspense more ergonomic. TanStack Query v5 has first-class Suspense support:

// The query — note useSuspenseQuery
function UserProfile({ userId }: { userId: string }) {
  // No loading/error state to handle — Suspense/ErrorBoundary do it
  const { data: user } = useSuspenseQuery({
    queryKey: queryKeys.users.detail(userId),
    queryFn: () => api.get<User>(`/users/${userId}`),
  });

  return <div>{user.name}</div>;
}

// The parent
function ProfilePage({ userId }: { userId: string }) {
  return (
    <ErrorBoundary fallback={<ErrorCard />}>
      <Suspense fallback={<ProfileSkeleton />}>
        <UserProfile userId={userId} />
      </Suspense>
    </ErrorBoundary>
  );
}

useSuspenseQuery — not useQuery — triggers Suspense. The component itself is clean: no if (isLoading), no if (error).

Prefetching for Perceived Performance

The biggest performance win in TanStack Query that most apps don't use:

// Prefetch on hover — data is ready before user clicks
function UserCard({ userId }: { userId: string }) {
  const queryClient = useQueryClient();

  return (
    <div
      onMouseEnter={() => {
        queryClient.prefetchQuery({
          queryKey: queryKeys.users.detail(userId),
          queryFn: () => api.get<User>(`/users/${userId}`),
          staleTime: 5 * 60 * 1000,
        });
      }}
    >
      {/* ... */}
    </div>
  );
}

When the user hovers on a card, the data fetches in the background. By the time they click, it's already in cache. The navigation feels instant.

What I Reach For and When

Scenario	Tool
Read data, cache it, background refresh	`useQuery`
Create / update / delete with side effects	`useMutation` + `invalidateQueries`
Form with pending state + validation	`useActionState` (React 19)
Optimistic update with rollback	`useMutation.onMutate`
Loading/error boundaries	`useSuspenseQuery` + `ErrorBoundary`
Dependent requests	`enabled` option on `useQuery`
Infinite scroll / pagination	`useInfiniteQuery`
One-off promise in component	`use()` (React 19)

React 19 fills the gaps at the edges. TanStack Query owns the cache layer. Together they cover every async pattern cleanly.

Building a React 19 app and want feedback on your data fetching architecture? Let's talk.

How I Use AI to Ship 3 Faster Without Cutting Corners

Ugur Aslim — Fri, 22 May 2026 15:04:30 +0000

Every week someone asks me: "Do you actually use AI, or is that just marketing copy?"

The honest answer: yes, heavily — but not the way most people imagine. AI doesn't replace engineering judgment. It amplifies it. Here's exactly what that looks like on a real project.

The Stack

Three tools, three distinct roles:

Claude — architecture, code review, documentation, complex reasoning
Cursor — in-editor generation, refactoring, inline explanation
GitHub Copilot — fast autocomplete for repetitive patterns

None of them replace each other. Each has a context where it's genuinely faster.

Phase 1: Architecture Decisions

Before writing a single line of code, I describe the problem to Claude:

I'm building a multi-tenant SaaS where each organisation's data 
must be logically isolated. PostgreSQL, FastAPI. Options:

1. Row-level security with tenant_id FK everywhere
2. Separate schemas per tenant  
3. Separate databases

Current scale: <500 tenants, <10k rows per tenant.
GDPR compliance required. Prioritise maintainability over performance.
Which pattern and why?

Claude's response is usually better than any Stack Overflow thread — because it reasons through my constraints, not a generic scenario. I use this for:

Schema design tradeoffs
Auth architecture decisions
Caching strategy
Deployment topology

Time saved: 1-2 hours per architecture decision. The thinking I do is genuinely deeper, not shallower.

Phase 2: Scaffolding the Boring Parts

Once the architecture is decided, there's a class of code that's important but not interesting: Pydantic schemas, CRUD endpoints, test fixtures, migration files.

In Cursor, I describe what I need:

Create a FastAPI router for the User resource.
Endpoints: GET /users, GET /users/{id}, POST /users, PATCH /users/{id}, DELETE /users/{id}
Use async SQLAlchemy. Inject AsyncSession via Depends.
Return 404 for missing users, 409 for email conflicts.
Follow the pattern in routers/organisation.py.

The output is ~90% correct and needs 10% adjustment. That's fine — the adjustment is the interesting part.

What I never do: accept generated code without reading every line. AI generates plausible-looking code that can have subtle bugs. Plausible ≠ correct.

Phase 3: Test Generation

This is where I get the biggest leverage. Writing tests is important but tedious. AI is excellent at it.

# I write the function
async def transfer_credits(
    db: AsyncSession,
    from_user_id: int,
    to_user_id: int,
    amount: Decimal,
) -> Transaction:
    ...

# I ask Claude: "Write pytest tests for transfer_credits.
# Cover: happy path, insufficient balance, invalid users,
# concurrent transfers (race condition), decimal precision."

Claude generates tests I would have written but faster. More importantly, it sometimes suggests edge cases I hadn't thought of — the concurrent transfer scenario above was a Claude suggestion.

Result: 107 tests in CitizenApp. Probably 40+ of those were AI-drafted, then reviewed and adjusted by me.

Phase 4: Code Review

Before any PR, I paste the diff into Claude:

Review this authentication middleware. I'm particularly worried about:
1. The JWT validation edge cases
2. Whether the rate limiting is applied before or after auth
3. Any timing attack vectors in the token comparison

[paste diff]

Claude caught a subtle issue in my Fernet key rotation logic that my own review missed — the old key wasn't being kept available for decrypting existing data during rotation.

Where AI Fails (And Human Judgment Is Irreplaceable)

After 18 months of heavy AI-assisted development, here's where it consistently falls short:

1. Business logic tradeoffs

AI doesn't know that your SaaS has a "freemium" tier that needs different rate limiting. It doesn't know your biggest client needs an audit log export in a non-standard format. Domain knowledge is yours.

2. Long-range architectural consequences

AI gives excellent advice for today's problem. It often misses how today's decision will constrain you in 6 months.

3. Security audit of its own output

AI generates code with security vulnerabilities. It's also good at finding them — but only when you ask explicitly. Never assume AI-generated code is secure without a dedicated security review pass.

4. The judgment call

"Should we add this feature now or defer it?" AI will give you both sides of the argument. The decision is still yours.

The 3× Number

Where does "3× faster" come from?

Rough breakdown on a typical feature:

Architecture decisions: 50% faster (Claude eliminates most of the research)
Boilerplate/scaffolding: 80% faster (Cursor generates the structure)
Test writing: 60% faster (AI drafts, I review/adjust)
Documentation: 70% faster (Claude explains, I verify)
Code review: 30% faster (catches more issues, faster)

Weighted across a project, ~3× is a conservative estimate. Some tasks are 5×. Some (judgment calls, architecture) are 1.1×.

The Mental Model That Works

Think of AI as a very fast, very well-read junior engineer who:

Knows every library, every pattern, every Stack Overflow answer
Never gets tired or distracted
Has no domain knowledge about your business
Occasionally makes confident, plausible mistakes
Needs clear briefs and explicit review

Treat it that way, and it's transformative. Treat it as a magic code generator, and you'll ship subtle bugs at 3× the speed.

Using AI tools on a project I can help with? Let's talk.

Vercel + Render Hybrid Deployment: Why I Split My Stack (and How)

Ugur Aslim — Fri, 22 May 2026 14:50:16 +0000

Most tutorials deploy everything to one platform. Most production apps shouldn't.

Here's the hybrid stack I use for every project — React on Vercel Edge, FastAPI on Render Docker — and exactly why each service is on the platform it's on.

The Stack at a Glance

Frontend  →  Vercel Edge CDN      (React 19 + Vite)
Backend   →  Render Docker        (FastAPI + Uvicorn)
Database  →  Neon Postgres         (Frankfurt, eu-central-1)
Cache     →  Valkey (Redis-compat) (Render internal network)

Each choice is deliberate. Let me explain the reasoning.

Why Vercel for the Frontend

Vercel does one thing better than anyone else: deploying JavaScript frontends at the edge.

What you get:

Global CDN with ~50 PoPs — sub-50ms TTFB worldwide
Automatic branch previews on every PR
Edge middleware for rewrites, redirects, A/B testing
Zero-config HTTPS, HTTP/2, Brotli compression
Image optimisation via next/image (or custom transforms)

For a React SPA built with Vite, Vercel's CDN just serves static files. The edge network means the HTML/JS/CSS is physically close to your user — before their first API call.

The free tier is genuinely production-ready for personal projects and small SaaS. Bandwidth limits only matter at scale.

Why Render for the Backend

FastAPI needs a real server — persistent process, filesystem access, long-running connections for WebSockets. Serverless doesn't work well for this.

Render Docker gives you:

# The Dockerfile I use for FastAPI on Render
FROM python:3.14-slim

WORKDIR /app

# Install deps in a separate layer for cache efficiency
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Non-root user for security
RUN adduser --disabled-password --gecos '' appuser
USER appuser

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

Render auto-deploys on push to main, handles HTTPS termination, and gives you proper logs. The --workers 2 gives you two Uvicorn processes per instance — enough for most early-stage SaaS traffic.

Render's internal network is important: your Valkey/Redis instance talks to your FastAPI instance on a private network. No public exposure, no auth tokens needed, ~0.1ms latency.

The CORS Configuration

Split deployment means your API and frontend are on different domains. CORS must be explicit:

# main.py
from fastapi.middleware.cors import CORSMiddleware

ALLOWED_ORIGINS = [
    "https://your-app.vercel.app",         # Vercel preview domain
    "https://yourdomain.com",              # Production custom domain
    "https://*.vercel.app",                # All branch previews
]

# Add localhost in dev
if settings.ENVIRONMENT == "development":
    ALLOWED_ORIGINS.extend([
        "http://localhost:5173",           # Vite dev server
        "http://localhost:3000",
    ])

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,               # Required for cookies (refresh tokens)
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
)

allow_credentials=True is required if you're using HTTP-only cookies for refresh tokens (which you should be).

Environment Variables: The Right Pattern

Never hardcode the API URL. Use Vite environment variables:

// src/lib/api.ts
const BASE_URL = import.meta.env.VITE_API_URL;

if (!BASE_URL) {
  throw new Error('VITE_API_URL is not set');
}

export const apiClient = axios.create({
  baseURL: BASE_URL,
  withCredentials: true,    // Send cookies cross-origin
  timeout: 10_000,
});

In Vercel, set per-environment:

Production: VITE_API_URL = https://api.yourdomain.com
Preview: VITE_API_URL = https://api-staging.onrender.com

In Render:

ENVIRONMENT = production
DATABASE_URL = postgresql+asyncpg://... (from Neon)
REDIS_URL = redis://... (Render internal)
ENCRYPTION_KEY_PRIMARY = ...

Database: Neon Postgres in Frankfurt

For GDPR compliance, EU data stays in the EU. Neon's eu-central-1 (Frankfurt) region puts Postgres physically close to both the Render instance (also Frankfurt) and the user base.

Neon's connection pooling via PgBouncer is essential for serverless/edge workloads:

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Use the pooled connection string from Neon dashboard
# It routes through PgBouncer — handles connection limits properly
DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_async_engine(
    DATABASE_URL,
    pool_size=5,           # Per-worker pool
    max_overflow=10,
    pool_pre_ping=True,    # Detect stale connections
    pool_recycle=300,      # Recycle connections every 5 min
)

AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

The pool_pre_ping=True is important on Neon — serverless databases can pause and the connection will appear valid but fail on first use without it.

CI/CD: GitHub Actions → Both Platforms

One push to main triggers both deploys in parallel:

# .github/workflows/deploy.yml
name: Deploy

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.14' }
      - run: pip install -r requirements.txt
      - run: pytest --tb=short -q
        env:
          DATABASE_URL: ${{ secrets.TEST_DATABASE_URL }}
          ENCRYPTION_KEY_PRIMARY: ${{ secrets.TEST_ENCRYPTION_KEY }}

  deploy-backend:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - name: Trigger Render deploy
        run: |
          curl -X POST "${{ secrets.RENDER_DEPLOY_HOOK }}"

  deploy-frontend:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: '22' }
      - run: npm ci && npm run build
        env:
          VITE_API_URL: ${{ secrets.VITE_API_URL }}
      - uses: amondnet/vercel-action@v25
        with:
          vercel-token: ${{ secrets.VERCEL_TOKEN }}
          vercel-org-id: ${{ secrets.VERCEL_ORG_ID }}
          vercel-project-id: ${{ secrets.VERCEL_PROJECT_ID }}
          vercel-args: '--prod'

Tests gate both deploys. Neither platform gets a broken build.

Health Checks and Monitoring

Render can restart unhealthy instances automatically:

# routes/health.py
from fastapi import APIRouter
from sqlalchemy import text

router = APIRouter()

@router.get("/health")
async def health_check(db: AsyncSession = Depends(get_async_db)):
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ok", "db": "connected"}
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "unhealthy", "error": str(e)}
        )

In Render settings: set health check path to /health, threshold to 3 failures. Render will replace the instance automatically.

The Latency Profile

With Frankfurt for both Render and Neon:

Hop	Latency
Browser → Vercel Edge (nearest PoP)	~10–30ms
JS/CSS/assets served from edge	~5ms
API call: Browser → Render (Frankfurt)	~20–60ms (EU users)
FastAPI → Neon Postgres	~3–8ms (same region)
FastAPI → Valkey Redis	~0.5–2ms (internal network)

For EU users, total page load feels fast. For users outside EU, the API calls will be slower — if that matters, you'd add a Render instance in another region or put a CDN in front of the API.

Cost at Small Scale

Service	Tier	Monthly cost
Vercel	Hobby	Free
Render	Starter (512MB)	$7
Neon	Free tier (0.5 CU)	Free
Valkey on Render	Starter	$7
Total		~$14/month

At this price point, you have a fully production-grade stack with proper separation of concerns, GDPR-compliant EU data residency, CI/CD, health checks, and auto-deploy.

Scale to the next tier when you need it — the architecture doesn't change.

Running this stack on a project? I'd be happy to review your setup. Get in touch.

My Cursor + Claude Engineering Workflow: A Real Project Walkthrough

Ugur Aslim — Fri, 22 May 2026 14:31:35 +0000

Every "AI productivity" article I've read is either vague advice ("use AI for boilerplate!") or cherry-picked demos that don't look like real work. This is neither.

This is a concrete walkthrough of a single feature — adding TOTP 2FA to CitizenApp — showing exactly where Cursor and Claude helped, where they didn't, and what the actual time savings looked like.

The Feature: TOTP 2FA with Recovery Codes

The requirement: users can enable TOTP-based two-factor authentication (Google Authenticator-compatible). If they lose their device, they can use one of 10 one-time recovery codes.

This is a non-trivial feature: it touches the auth flow, the user model, the session model, email sending, and the frontend in at least 4 places.

Estimated time without AI: 2–3 days.
Actual time: ~6 hours.

Here's how.

Step 1: Architecture Decision (Claude)

Before opening my editor, I opened Claude with context:

I'm adding TOTP 2FA to a FastAPI app. Current auth:
- bcrypt password hashing
- JWT access tokens (15min)
- HTTP-only refresh token cookies (30 days)
- PostgreSQL + SQLAlchemy async

Requirements:
- Users can enable/disable 2FA
- TOTP compatible with Google Authenticator, Authy
- 10 recovery codes on setup, each single-use
- Login flow: password → if 2FA enabled → TOTP prompt

Questions:
1. Where to store TOTP secret? Encrypted at rest?
2. How to handle the "pending 2FA" state during login — 
   issue a limited token or use session?
3. Recovery code format — hash them or store plain?

Libraries: pyotp seems obvious. Anything better?

Claude's response clarified three things I hadn't fully thought through:

Yes, encrypt the TOTP secret at rest — same Fernet key as other PII. If your DB is breached, unencrypted TOTP secrets let an attacker bypass 2FA permanently.
Use a short-lived "pending 2FA" JWT rather than a session. Stateless — no DB lookup during the TOTP verification step. Include {"sub": user_id, "2fa_pending": true} in the payload with a 5-minute expiry.
Hash recovery codes — same as password hashing. Bcrypt is overkill (slow), SHA-256 is fine since recovery codes are long random strings.

Total time: 20 minutes. I had a clear, validated architecture before writing any code.

Step 2: Database Schema (Cursor)

In Cursor, I described the schema to Agent mode:

Add to models/user.py:
- totp_secret field (encrypted string, nullable)
- totp_enabled bool (default false)
- totp_verified_at datetime (nullable)

Create new file models/recovery_code.py:
- id, user_id FK, code_hash (sha256, unique), used bool, used_at datetime
- 10 codes per user, generated on 2FA setup

Cursor generated both files. I reviewed:

✅ Correct SQLAlchemy 2.0 mapped_column syntax
✅ Proper ForeignKey with cascade delete
✅ Index on code_hash for lookup performance
⚠️ Missing: used_at timezone handling — I added timezone=True to the datetime column

One manual fix. Two files I didn't have to write from scratch.

Then Alembic migration:

alembic revision --autogenerate -m "add_totp_2fa"

I reviewed the generated migration, confirmed the column types were correct, ran it.

Step 3: The TOTP Service (Cursor + Manual)

The TOTP logic is self-contained enough for Cursor to handle most of it:

Create services/totp.py with:
- generate_totp_secret() → encrypted secret
- get_totp_uri(user_email, secret) → otpauth:// URI for QR code
- verify_totp(encrypted_secret, code, window=1) → bool
- generate_recovery_codes(user_id, db) → list of raw codes (store hashes)
- verify_recovery_code(user_id, code, db) → bool (mark used)
Use pyotp. Decrypt secret before passing to pyotp.

Generated code was ~90% correct. What I fixed manually:

# Cursor generated this:
totp = pyotp.TOTP(secret)
return totp.verify(code)

# I changed to:
totp = pyotp.TOTP(secret)
# window=1 allows 30s clock drift — important for mobile TOTP apps
return totp.verify(code, valid_window=1)

Small but important: valid_window=1 means ±1 time step (±30 seconds). Without it, users with slightly drifted clocks fail verification constantly.

Step 4: The Auth Flow Changes (Manual + Claude Review)

This was the part I wrote mostly by hand — the login flow change is architectural, not boilerplate:

# routes/auth.py — modified login endpoint
@router.post("/login")
async def login(credentials: LoginSchema, db: AsyncSession = Depends(get_async_db)):
    user = await authenticate_user(db, credentials.username, credentials.password)

    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # If 2FA is enabled, issue a limited pending token instead of full auth
    if user.totp_enabled:
        pending_token = create_access_token(
            {"sub": str(user.id), "2fa_pending": True},
            expires_delta=timedelta(minutes=5)
        )
        return {"requires_2fa": True, "pending_token": pending_token}

    # Normal flow — no 2FA
    return await issue_full_tokens(user)


@router.post("/verify-2fa")
async def verify_2fa(
    payload: Verify2FASchema,
    db: AsyncSession = Depends(get_async_db),
):
    # Validate the pending token
    token_data = decode_pending_token(payload.pending_token)
    if not token_data or not token_data.get("2fa_pending"):
        raise HTTPException(status_code=401, detail="Invalid pending token")

    user = await get_user(db, int(token_data["sub"]))

    # Try TOTP first, then recovery code
    if not (
        verify_totp(user.totp_secret, payload.code)
        or await verify_recovery_code(user.id, payload.code, db)
    ):
        raise HTTPException(status_code=401, detail="Invalid 2FA code")

    return await issue_full_tokens(user)

After writing this, I pasted it to Claude with: "Review for security issues, particularly around the pending token and the TOTP/recovery code branch."

Claude flagged one issue: the or short-circuit means if TOTP passes, the recovery code check is never run — which is correct behaviour, but I should log the authentication method used in the audit trail.

I added it. Three lines. Worth catching.

Step 5: Tests (Cursor + Review)

I wrote a brief test spec in a comment, then let Cursor generate the test cases:

# Tests needed:
# - login without 2FA → full tokens
# - login with 2FA → pending token + requires_2fa flag
# - verify 2FA with valid TOTP → full tokens
# - verify 2FA with invalid TOTP → 401
# - verify 2FA with valid recovery code → full tokens, code marked used
# - verify 2FA with used recovery code → 401
# - verify 2FA with expired pending token → 401

Cursor generated 7 tests. I reviewed each one, fixed 2 issues:

Mock for pyotp.TOTP.verify wasn't isolated — tests were dependent on each other
Recovery code test wasn't checking used=True after verification

Fixed, ran pytest. All 7 passed. Added to the existing 107-test suite.

Step 6: Frontend (Cursor, mostly autonomous)

The React side: Cursor handled it with minimal guidance.

Add to pages/Login.tsx:
- If login response has requires_2fa: true, show 2FA input form
- Submit pending_token + code to /auth/verify-2fa
- Handle recovery code toggle (same input, different UX label)
- Show loading state during verification

Add to pages/AccountSettings.tsx:
- Enable 2FA flow: show QR code + secret + "verify to confirm" step
- Disable 2FA flow: require password confirmation
- Show recovery codes after setup (copy-to-clipboard, one-time display)
- Regenerate recovery codes option

Frontend took ~90 minutes. Cursor did most of the repetitive React + TypeScript. I reviewed for accessibility (focusable inputs, keyboard navigation on the code entry) and UX edge cases (what happens if QR scan fails → show manual entry option).

The Honest Time Breakdown

Task	Without AI	With AI	Saved
Architecture decision	2h (research + thinking)	20min (Claude review)	~1.5h
DB schema + migration	30min	10min	20min
TOTP service	1h	25min	35min
Auth flow changes	1.5h	1.5h (manual)	0
Tests	1h	30min	30min
Frontend	3h	1.5h	1.5h
Total	~9h	~4.25h	~4.75h

That's roughly 2× for this feature — not 3×. The architectural work (login flow changes, security review) wasn't AI-accelerated much.

The 3× number shows up on features with more boilerplate relative to architecture: CRUD endpoints, schema migrations, test fixtures, UI forms.

What I'd Never Let AI Do Unreviewed

Any security-sensitive logic — the valid_window issue is a perfect example of plausible-but-wrong AI output
DB migrations — generated migrations can silently drop columns if you're not careful
Auth flow — I wrote the pending token flow entirely by hand
Error handling — AI usually handles happy path, misses edge cases

The rule I follow: AI drafts, I ship. Everything AI generates gets read line by line before it goes into main.

Cursor Config That Helps

Two settings that materially improve Cursor's output quality:

// .cursor/rules (project-level)
{
  "rules": [
    "Always use SQLAlchemy 2.0 mapped_column syntax, not Column()",
    "Use async/await throughout — no sync database calls",
    "Follow existing patterns in the codebase before inventing new ones",
    "Always include error handling for database operations",
    "Use Pydantic v2 model_validator not root_validator"
  ]
}

These prevent the most common errors in AI-generated FastAPI code.

Want to walk through your project's architecture with me? 30 minutes, no pitch.

Astro and Islands Architecture: Why Your Portfolio Doesn't Need React for Everything

Ugur Aslim — Fri, 22 May 2026 14:23:55 +0000

Most portfolio sites are over-engineered. You reach for React because you know React, wire up a router, bundle 300 KB of JavaScript, and ship it to a user who just wants to read your about page. Astro was built to break that cycle.

This site — uaslim.com — is an Astro project. Zero client-side JavaScript on most pages. Sub-second loads globally. Full support for React, TypeScript, and Tailwind where needed. Here's how Astro achieves that without compromise.

The Core Philosophy: HTML-First

Astro starts from a different premise than React, Next.js, or SvelteKit. Those frameworks assume you're building an application — state, routing, hydration, the whole runtime. Astro assumes you're building a document. HTML is the primary output; JavaScript is optional.

An Astro component looks familiar:

---
// Component script — runs at build time (server-side), never in browser
const posts = await getCollection('blog');
const sorted = posts.sort((a, b) =>
  new Date(b.data.date).getTime() - new Date(a.data.date).getTime()
);
---

<!-- Template — compiled to pure HTML -->
<ul>
  {sorted.map(post => (
    <li>
      <a href={`/blog/${post.id}/`}>{post.data.title}</a>
      <time>{post.data.date}</time>
    </li>
  ))}
</ul>

The frontmatter (between ---) runs at build time. The template compiles to static HTML. The browser receives markup, not a JavaScript bundle that generates markup. For a blog post list, that's the correct model.

Islands Architecture

Islands Architecture is Astro's answer to the question: what if I need interactivity in one part of the page?

The metaphor: a static HTML page is an ocean. Interactive components are islands. Each island is an independent, self-contained piece of UI that hydrates independently — without hydrating the entire page.

---
import StaticHeader from './StaticHeader.astro';   // 0 JS
import BlogList from './BlogList.astro';            // 0 JS
import SearchWidget from './SearchWidget.tsx';      // React — hydrated!
import Footer from './Footer.astro';               // 0 JS
---

<StaticHeader />
<BlogList posts={posts} />

<!-- client:load → hydrate immediately when page loads -->
<SearchWidget client:load />

<Footer />

The client: directive is how you opt a component into client-side JavaScript. Without it, even a React component renders to static HTML at build time and ships no runtime JavaScript.

The Five Hydration Directives

Directive	When it hydrates	Use case
`client:load`	Immediately on page load	Critical UI, above the fold
`client:idle`	When browser is idle (`requestIdleCallback`)	Non-critical widgets
`client:visible`	When the element enters viewport	Below-the-fold features
`client:media="(max-width: 768px)"`	When media query matches	Mobile-only components
`client:only="react"`	Client-side only, never SSR'd	Components that depend on browser APIs

The defaults are deliberately conservative. client:idle and client:visible are the most useful — they defer JavaScript parsing until the browser has nothing better to do, or until the user actually scrolls to the component.

<!-- Chat widget — loads only when user scrolls to it -->
<AiChatWidget client:visible />

<!-- Cookie banner — only on mobile -->
<MobileCookieBanner client:media="(max-width: 768px)" />

<!-- Analytics dashboard — needs browser APIs, never SSR -->
<AnalyticsDashboard client:only="react" />

This granularity is what makes Islands Architecture powerful. You're not choosing between "fully static" and "fully hydrated" — you're choosing per component.

Framework Agnosticism

Astro doesn't pick a frontend framework for you. It supports React, Vue, Svelte, Solid, Preact, and Lit in the same project. You import the integration, and the client: directives work the same way.

npx astro add react      # @astrojs/react
npx astro add vue        # @astrojs/vue
npx astro add svelte     # @astrojs/svelte

In practice this means: use React for the interactive search widget you already have. Use Svelte for the lightweight toggle that doesn't need a full React runtime. Use plain .astro components for everything static. The bundles are separate — React's 45 KB runtime only loads on pages that need it.

---
import ReactSearchWidget from './SearchWidget.tsx';  // loads React runtime
import SvelteCounter from './Counter.svelte';         // loads Svelte runtime (~2 KB)
---

<ReactSearchWidget client:load />
<SvelteCounter client:idle />

Content Collections: Typed Markdown at Scale

Content Collections are Astro's built-in system for managing structured content like blog posts. Instead of reading Markdown files manually, you define a schema with Zod and get full TypeScript validation:

// src/content.config.ts
import { defineCollection, z } from 'astro:content';
import { glob } from 'astro/loaders';

const blog = defineCollection({
  loader: glob({ pattern: '**/*.md', base: './src/content/blog' }),
  schema: z.object({
    title: z.string(),
    date: z.string(),
    description: z.string(),
    tags: z.array(z.string()).optional(),
    readingTime: z.number().optional(),
  }),
});

export const collections = { blog };

Now every Markdown file in src/content/blog/ is validated against that schema at build time. Missing a required field? Build fails. Typo in a tag? TypeScript catches it. The data you get from getCollection('blog') is fully typed:

---
import { getCollection } from 'astro:content';

// posts: CollectionEntry<'blog'>[] — fully typed
const posts = await getCollection('blog');

// post.data.title — string, guaranteed
// post.data.date — string, guaranteed
// post.data.readingTime — number | undefined, correct
---

For a blog, this replaces a CMS, a database, and a content API. Your Markdown files are the source of truth, Git is version control, and TypeScript is validation.

The Build Output: What Actually Ships

Running astro build produces a dist/ folder of pure HTML, CSS, and minimal JavaScript. For a typical content site:

dist/
  index.html          — 12 KB
  blog/
    index.html        — 8 KB
    my-post/
      index.html      — 15 KB
  _astro/
    main.css          — 18 KB (Tailwind, purged)
    SearchWidget.js   — 48 KB (React runtime + component, only on pages that use it)

Compare that to a Next.js app with the same content: 200+ KB of JavaScript on the initial page load, even for pages with no interactivity.

The Lighthouse scores reflect this. A static Astro page with no interactive islands routinely scores 98-100 on performance. The JavaScript budget only grows when you explicitly spend it.

View Transitions: SPA Feel, Static Output

Astro's View Transitions API gives you smooth page-to-page animations without a client-side router. One import, one attribute:

---
// layouts/Layout.astro
import { ViewTransitions } from 'astro:transitions';
---

<html>
  <head>
    <ViewTransitions />
  </head>
  <body>
    <slot />
  </body>
</html>

Under the hood this uses the browser's native View Transitions API with a polyfill for unsupported browsers. Navigation feels like an SPA — no full page flash, animated transitions — but the HTML is still served statically per page. No client-side routing JavaScript, no prefetch bundle bloat.

You can annotate specific elements for custom transitions:

<!-- Blog card that "expands" into the article on click -->
<img
  src={post.data.coverImage}
  transition:name={`cover-${post.id}`}
  transition:animate="slide"
/>

Astro matches the transition:name between the list and the detail page and animates between them using the browser's native API. The effect is striking; the cost is near-zero.

When Astro Is the Right Choice

Astro is well-suited for:

Marketing sites and landing pages — mostly content, minimal state, SEO critical
Documentation sites — large amounts of Markdown, fast navigation, search as an island
Portfolios and blogs — exactly what this site is
E-commerce product pages — static product listing, React cart widget as an island
Hybrid apps — static shell with interactive sections

Astro is a poor fit for:

Highly interactive single-page applications — a dashboard with complex state, real-time updates, frequent client-side navigation. React + Vite or Next.js App Router serves you better.
Apps where every route is behind auth — Astro can do SSR, but if every page is personalized and dynamic, the static-first model loses its advantage.
Real-time features — live data, WebSockets, collaborative editing. These need a proper SPA runtime.

The honest test: count the interactive components on your page. If it's one or two (a search bar, a contact form, a theme toggle), Astro is probably the right choice. If every component manages state and communicates with siblings, you want a full SPA framework.

SSR Mode: Opting Into Dynamic Rendering

Astro isn't locked to static output. Setting output: 'server' in astro.config.mjs switches to server-side rendering on every request, just like Next.js:

// astro.config.mjs
export default defineConfig({
  output: 'server',
  adapter: vercel(),   // or node(), cloudflare(), etc.
});

Or you can use output: 'hybrid' — static by default, with opt-in dynamic routes:

---
// pages/api/contact.ts — dynamic API endpoint
export const prerender = false;

export async function POST({ request }) {
  const data = await request.json();
  await sendEmail(data);
  return new Response(JSON.stringify({ ok: true }), { status: 200 });
}
---

---
// pages/blog/[slug].astro — stays static
// (prerender defaults to true in hybrid mode)
---

This hybrid model is genuinely powerful: static pages for content (fast, cacheable, CDN-friendly), dynamic endpoints for forms and APIs. No separate backend needed for simple interactions.

Content Security Without a Build Step

One thing Astro gets right by design: there's no client-side HTML injection surface by default. .astro component templates use JSX-like syntax, and values are escaped automatically:

---
const userInput = '<script>alert("xss")</script>';
---

<!-- Renders as escaped text — safe -->
<p>{userInput}</p>

<!-- Explicit opt-in required for raw HTML -->
<p set:html={sanitizedHtml} />

The set:html directive is the only way to inject raw markup, and its name is a visible signal in code review. In a CMS-driven site, this matters: you know exactly where unescaped content can flow.

The Developer Experience

A few things that make Astro pleasant to work with day-to-day:

Hot module replacement is instant. Astro's dev server only re-processes the changed component, not the whole page. On large sites this is noticeable.

TypeScript is on by default. No config required. .astro files support TypeScript in the frontmatter, and getCollection() returns typed data automatically.

Zero-config image optimization. The <Image /> component from astro:assets generates WebP/AVIF variants, adds correct width and height attributes (preventing layout shift), and lazy-loads by default. One line replaces a webpack image pipeline.

---
import { Image } from 'astro:assets';
import heroImage from '../images/hero.jpg';
---

<!-- Optimized, correctly sized, lazy-loaded, WebP format -->
<Image src={heroImage} alt="Hero" width={1200} height={630} />

Production Checklist

Before shipping an Astro site:

# Run a full build — catches type errors and missing content
npm run build

# Preview the static output locally — catches routing issues
npm run preview

# Check bundle size per page
npx astro build --verbose

# Validate HTML output
npx html-validate dist/**/*.html

The most common production issue: client:only components that reference window or document during SSR. Astro warns on these — fix them before the build, not after.

This site uses Astro 5 with the hybrid output mode — static pages for everything, a small server function for the contact form. The result is a Lighthouse performance score of 98 and a first contentful paint under 400ms globally, with zero JavaScript shipped to blog post pages. That's the Islands Architecture working exactly as designed.

Have questions about Astro's architecture or thinking about migrating a site? Drop me a line.

JWT Token Refresh Patterns in React 19: Avoiding the Silent Auth Death Spiral

Ugur Aslim — Fri, 22 May 2026 14:12:41 +0000

JWT Token Refresh Patterns in React 19: Avoiding the Silent Auth Death Spiral

I've watched authentication break in production more times than I want to admit. Usually it's silent—users get logged out mid-action, requests fail with 401s, and nobody notices until support tickets pile up. The culprit? Naive token refresh logic that doesn't handle concurrent requests.

Most solutions I see are either bloated (Redux middleware with retry queues) or fragile (localStorage checks that fail when two requests hit simultaneously). I'm going to show you the approach I use in CitizenApp: minimal, correct, and battle-tested with concurrent traffic.

The Problem: Why Simple Token Refresh Fails

Let me paint a scenario. User opens your app. Token expires. They click a button that triggers three API calls simultaneously. Here's what happens with naive refresh:

// ❌ Bad: Each request independently tries to refresh
async function apiCall(endpoint: string) {
  let token = localStorage.getItem('token');

  const res = await fetch(endpoint, {
    headers: { Authorization: `Bearer ${token}` }
  });

  if (res.status === 401) {
    // All three requests do this at the same time
    const refreshRes = await fetch('/api/auth/refresh', {
      method: 'POST',
      body: JSON.stringify({ refreshToken: localStorage.getItem('refreshToken') })
    });
    // ...
  }
}

You just sent three simultaneous refresh requests. Your backend probably revoked all but one token for security reasons. Congrats, now all three requests fail with 401 again, and you're in a retry loop. This is the "silent auth death spiral."

The Solution: React Context + AbortController

I prefer React Context for this over Redux because:

It's built into React—no external dependency
Token refresh is genuinely app-level state, not domain state
It's easier to reason about with AbortControllers for request queuing

Here's the implementation I use:

// auth.context.tsx
import React, { createContext, useCallback, useRef, useEffect } from 'react';

interface AuthContextType {
  token: string | null;
  refreshToken: string | null;
  setTokens: (token: string, refreshToken: string) => void;
  clearTokens: () => void;
  getValidToken: () => Promise<string>;
}

export const AuthContext = createContext<AuthContextType | null>(null);

export function AuthProvider({ children }: { children: React.ReactNode }) {
  const [token, setToken] = React.useState<string | null>(null);
  const [refreshToken, setRefreshToken] = React.useState<string | null>(null);
  const refreshPromiseRef = useRef<Promise<string> | null>(null);
  const abortControllerRef = useRef<AbortController | null>(null);

  // Load tokens from secure storage on mount
  useEffect(() => {
    const token = sessionStorage.getItem('token');
    const refreshToken = sessionStorage.getItem('refreshToken');
    if (token && refreshToken) {
      setToken(token);
      setRefreshToken(refreshToken);
    }
  }, []);

  const performRefresh = useCallback(
    async (refreshTokenValue: string): Promise<string> => {
      // Cancel previous refresh attempt if still in flight
      if (abortControllerRef.current) {
        abortControllerRef.current.abort();
      }

      abortControllerRef.current = new AbortController();

      try {
        const res = await fetch('/api/auth/refresh', {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ refreshToken: refreshTokenValue }),
          signal: abortControllerRef.current.signal
        });

        if (res.status === 401 || res.status === 403) {
          // Refresh token expired—user must re-login
          clearTokens();
          throw new Error('Refresh token expired');
        }

        if (!res.ok) {
          throw new Error(`Refresh failed: ${res.statusText}`);
        }

        const { token: newToken, refreshToken: newRefreshToken } = await res.json();
        setToken(newToken);
        setRefreshToken(newRefreshToken);
        sessionStorage.setItem('token', newToken);
        sessionStorage.setItem('refreshToken', newRefreshToken);

        return newToken;
      } catch (error) {
        if (error instanceof Error && error.name === 'AbortError') {
          throw new Error('Refresh cancelled');
        }
        throw error;
      }
    },
    []
  );

  const getValidToken = useCallback(async (): Promise<string> => {
    // If we already have a refresh in flight, wait for it
    if (refreshPromiseRef.current) {
      return refreshPromiseRef.current;
    }

    // If token exists and isn't expired, return it
    if (token && !isTokenExpired(token)) {
      return token;
    }

    // Otherwise, start a refresh
    if (!refreshToken) {
      throw new Error('No refresh token available');
    }

    refreshPromiseRef.current = performRefresh(refreshToken)
      .finally(() => {
        refreshPromiseRef.current = null;
      });

    return refreshPromiseRef.current;
  }, [token, refreshToken, performRefresh]);

  return (
    <AuthContext.Provider
      value={{
        token,
        refreshToken,
        setTokens: (t, rt) => {
          setToken(t);
          setRefreshToken(rt);
          sessionStorage.setItem('token', t);
          sessionStorage.setItem('refreshToken', rt);
        },
        clearTokens: () => {
          setToken(null);
          setRefreshToken(null);
          sessionStorage.clear();
        },
        getValidToken
      }}
    >
      {children}
    </AuthContext.Provider>
  );
}

function isTokenExpired(token: string): boolean {
  try {
    const payload = JSON.parse(atob(token.split('.')[1]));
    return payload.exp * 1000 < Date.now();
  } catch {
    return true;
  }
}

Now your fetch interceptor:

// api.ts
import { AuthContext } from './auth.context';

export function useApi() {
  const authContext = React.useContext(AuthContext);

  return useCallback(
    async (endpoint: string, options: RequestInit = {}) => {
      if (!authContext) throw new Error('AuthProvider not found');

      try {
        const token = await authContext.getValidToken();
        const headers = new Headers(options.headers || {});
        headers.set('Authorization', `Bearer ${token}`);

        let res = await fetch(endpoint, { ...options, headers });

        // If still 401 after refresh attempt, we're truly unauthorized
        if (res.status === 401) {
          authContext.clearTokens();
          window.location.href = '/login';
          throw new Error('Session expired');
        }

        return res;
      } catch (error) {
        if (error instanceof Error && error.message === 'No refresh token available') {
          window.location.href = '/login';
        }
        throw error;
      }
    },
    [authContext]
  );
}

Why This Works

Concurrency handling: The refreshPromiseRef ensures only one refresh happens at a time. Multiple concurrent requests wait for the same promise.

// Three requests hit simultaneously
await Promise.all([
  useApi()('/api/users'),
  useApi()('/api/posts'),
  useApi()('/api/notifications')
]);
// All three call getValidToken() → same refresh promise → single refresh request

AbortController prevents stale refreshes: If a user navigates away or a component unmounts mid-refresh, we abort the request instead of updating state on an unmounted component.

Session storage, not localStorage: Tokens should die with the browser tab. I never use localStorage for auth—it survives XSS attacks longer than it should.

The FastAPI Backend

Keep it simple:

# auth.py
from fastapi import HTTPException, status
from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRET = "your-secret"

def create_tokens(user_id: str):
    access_payload = {
        'sub': user_id,
        'exp': datetime.utcnow() + timedelta(minutes=15),
        'type': 'access'
    }
    refresh_payload = {
        'sub': user_id,
        'exp': datetime.utcnow() + timedelta(days=7),
        'type': 'refresh'
    }
    return {
        'token': jwt.encode(access_payload, SECRET),
        'refreshToken': jwt.encode(refresh_payload, SECRET)
    }

@app.post('/api/auth/refresh')
async def refresh(data: dict):
    try:
        payload = jwt.decode(data['refreshToken'], SECRET, algorithms=['HS256'])
        if payload.get('type') != 'refresh':
            raise HTTPException(status_code=401)
        return create_tokens(payload['sub'])
    except JWTError:
        raise HTTPException(status_code=403)

Gotcha: Token Expiration Skew

This burned me: I assumed isTokenExpired() would perfectly predict when a token fails. It doesn't. Clock skew between client and server, plus network latency, means a token can appear valid client-side but fail server-side by milliseconds.

Fix: Add a 30-second buffer:

function isTokenExpired(token: string, bufferSeconds = 30): boolean {
  try {
    const payload = JSON.parse(atob(token.split('.')[1]));
    return payload.exp * 1000 < Date.now() + bufferSeconds * 1000;
  } catch {
    return true;
  }
}

This makes the client refresh slightly early, avoiding the 401 entirely on most requests.