Multi-tenancy real con FastAPI y PostgreSQL: planes, cuotas y aislamiento de datos

Implementé multi-tenancy real en un motor RAG SaaS: tenants con planes, enforcement de cuotas con contadores atómicos (UPSERT), rate limiting con Redis (sliding window), RBAC con 3 roles, y aislamiento de datos donde cada query se filtra por tenant_id. En este artículo comparto la arquitectura completa, las trampas que evité, y el código clave.

---

Por qué "agregar un tenant_id" no es multi-tenancy

La mayoría de los tutoriales de multi-tenancy se quedan en "agrega una columna tenant_id y filtra en cada query". Eso es el 10% del problema. En producción vas a necesitar:

• Planes con límites reales que no se puedan burlar con requests concurrentes
• Contadores atómicos que no pierdan incrementos bajo carga
• Rate limiting que sobreviva a reinicios del servidor
• Roles que restrinjan acciones, no solo visibilidad
• Aislamiento real donde un bug en un service no exponga datos de otro tenant
• Billing cycles que se reinicien correctamente cada mes

Este artículo muestra cómo resolví cada uno en un sistema real con FastAPI async y PostgreSQL.

---

El stack

Componente	Tecnología
Backend	Python 3.12 + FastAPI (100% async)
Base de datos	PostgreSQL 16
ORM	SQLAlchemy async + Alembic
Cache/Rate limit	Redis 7
Auth	JWT (access 30min, refresh 7d con rotación)
Passwords	bcrypt (via passlib)

---

Diseño de base de datos

Las 4 tablas fundamentales

┌──────────────┐     ┌──────────────┐
│    plans     │     │   tenants    │
├──────────────┤     ├──────────────┤
│ id (UUID)    │◄────│ plan_id (FK) │
│ name         │     │ id (UUID)    │
│ slug         │     │ name         │
│ max_users    │     │ slug (unique)│
│ max_kbs      │     │ is_active    │
│ max_documents│     │ plan_started │
│ max_storage  │     │ created_at   │
│ max_messages │     └──────┬───────┘
│ max_tokens   │            │
│ max_api_keys │     ┌──────┴───────┐
│ price_monthly│     │    users     │
│ max_concurrent│    ├──────────────┤
└──────────────┘     │ id (UUID)    │
                     │ tenant_id(FK)│
                     │ email        │
                     │ role (enum)  │
                     │ is_active    │
                     └──────────────┘

┌───────────────────────┐
│ tenant_monthly_usage  │
├───────────────────────┤
│ id (UUID)             │
│ tenant_id (FK)        │
│ billing_period (str)  │
│ messages_count (int)  │
│ tokens_used (bigint)  │
│ UNIQUE(tenant_id,     │
│   billing_period)     │
└───────────────────────┘

Modelo del Plan

Cada plan define todos los límites que el sistema va a enforcar:

class Plan(Base):
    __tablename__ = "plans"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    slug: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    max_users: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    max_concurrent: Mapped[int] = mapped_column(Integer, nullable=False, default=2)
    max_kbs: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    max_documents: Mapped[int] = mapped_column(Integer, nullable=False, default=20)
    max_storage_mb: Mapped[int] = mapped_column(Integer, nullable=False, default=200)
    max_messages_month: Mapped[int] = mapped_column(Integer, nullable=False, default=500)
    max_tokens_month: Mapped[int] = mapped_column(BigInteger, nullable=False, default=200000)
    max_api_keys: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    price_monthly: Mapped[float | None] = mapped_column(Numeric(10, 2), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

Decisión clave: -1 significa ilimitado. En vez de tener un booleano is_unlimited por cada recurso, uso -1 como convención. Simplifica todas las comparaciones:

if plan.max_users == -1:
    return  # Ilimitado, no verificar

Modelo del Tenant

class Tenant(Base):
    __tablename__ = "tenants"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    slug: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    plan_id: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True), ForeignKey("plans.id"), nullable=True
    )
    plan_started_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True, default=None
    )
    notification_preferences: Mapped[dict | None] = mapped_column(JSONB, nullable=True)

plan_started_at es crucial: define el ancla del ciclo de billing. Cuando un tenant pasa de Free a un plan pago, se resetea al momento del pago. Esto determina cuándo se reinician los contadores mensuales.

Modelo del Usuario

class UserRole(str, PyEnum):
    SUPER_ADMIN = "super_admin"
    TENANT_ADMIN = "tenant_admin"
    MEMBER = "member"


class User(Base):
    __tablename__ = "users"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    full_name: Mapped[str] = mapped_column(String(255), default="")
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole, name="user_role", values_callable=lambda e: [x.value for x in e]),
        default=UserRole.MEMBER,
    )
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    email_verified: Mapped[bool] = mapped_column(Boolean, default=False)

Un usuario pertenece a exactamente un tenant. No hay usuarios multi-tenant. Si necesitás que alguien acceda a dos organizaciones, crea dos cuentas. Simplifica enormemente el modelo de permisos.

---

RBAC: 3 roles, 0 ambiguedad

Los roles

Rol	Alcance	Puede hacer
super_admin	Plataforma	Ver todos los tenants, cambiar planes, ver métricas globales
tenant_admin	Su tenant	CRUD de KBs, documentos, FAQs, invitar usuarios, ver analytics
member	Su tenant (restringido)	Chat, ver KBs asignadas, feedback

Dependency injection para autorización

FastAPI tiene un sistema de dependencias que es perfecto para RBAC. Creé un factory de dependencias:

from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    payload = decode_access_token(token)
    user_id = payload.get("sub")
    if not user_id:
        raise UnauthorizedException("Token payload invalido")

    user = await get_user_by_id(db, UUID(user_id))
    if not user:
        raise UnauthorizedException("Usuario no encontrado")
    if not user.is_active:
        raise UnauthorizedException("Cuenta desactivada")

    # Verificar que el tenant esté activo (super_admin bypasses)
    if user.role != UserRole.SUPER_ADMIN:
        tenant = await db.get(Tenant, user.tenant_id)
        if not tenant or not tenant.is_active:
            raise UnauthorizedException("Tenant desactivado")

    return user

Detalle sutil: el super_admin bypasea la verificación de tenant activo. Si desactivamos un tenant malicioso, el super_admin todavía puede entrar a investigar.

Factory para restricción por rol

def require_role(*roles):
    """Dependency factory: restringe endpoint a roles específicos."""
    async def _check(current_user=Depends(get_current_user)):
        if current_user.role not in roles:
            raise ForbiddenException("No tienes permisos para esta accion")
        return current_user
    return _check


# Shortcuts
def require_super_admin():
    return require_role(UserRole.SUPER_ADMIN)

def require_tenant_admin_or_above():
    return require_role(UserRole.SUPER_ADMIN, UserRole.TENANT_ADMIN)

Uso en los endpoints:

@router.get("/tenants", response_model=list[TenantResponse])
async def list_tenants(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if current_user.role != UserRole.SUPER_ADMIN:
        raise ForbiddenException("Solo super_admin puede listar todos los tenants")
    return await tenant_service.list_tenants(db)


@router.put("/{kb_id}/access")
async def update_kb_access(
    kb_id: UUID,
    data: KBAccessConfig,
    db: AsyncSession = Depends(require_tenant_admin_or_above()),
):
    # Solo tenant_admin o super_admin llegan acá
    ...

---

Aislamiento de datos: tenant_id en todo

El patrón

Cada tabla que contiene datos de usuario tiene una columna tenant_id (directa o transitiva). No hay excepciones.

class KnowledgeBase(Base):
    __tablename__ = "knowledge_bases"
    # ...
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )

class ApiKey(Base):
    __tablename__ = "api_keys"
    # ...
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )

Filtrado en cada service

Cada función que lista o busca datos filtra por tenant_id del usuario autenticado:

async def list_knowledge_bases(db: AsyncSession, user: User) -> list[dict]:
    stmt = (
        select(KnowledgeBase, func.count(Document.id).label("document_count"))
        .outerjoin(Document, Document.knowledge_base_id == KnowledgeBase.id)
    )

    if user.role == UserRole.SUPER_ADMIN:
        # Super admin ve solo sus propias KBs + system KBs
        stmt = stmt.where(or_(
            KnowledgeBase.tenant_id == user.tenant_id,
            KnowledgeBase.is_system == True,
        ))
    else:
        stmt = stmt.where(or_(
            KnowledgeBase.tenant_id == user.tenant_id,
            KnowledgeBase.is_system == True,
        ))
        # Para members, aplicar filtro de acceso
        access_filter = await get_accessible_kb_filter(user)
        if access_filter is not None:
            stmt = stmt.where(or_(KnowledgeBase.is_system == True, access_filter))

    stmt = stmt.group_by(KnowledgeBase.id).order_by(KnowledgeBase.created_at.desc())
    result = await db.execute(stmt)
    # ...

Leccion importante: el super_admin NO ve los datos de otros tenants. Es el operador de la plataforma. Ve estadisticas agregadas (total tenants, revenue, etc.) pero no puede leer las conversaciones ni KBs de otros. Esto fue una decision deliberada de privacidad.

Acceso granular a KBs

Dentro de un mismo tenant, no todos los miembros ven todas las KBs. Hay un sistema de control de acceso:

async def _check_kb_access(db: AsyncSession, kb: KnowledgeBase, user: User) -> None:
    has_access = await user_can_access_kb(db, user, kb)
    if not has_access:
        raise ForbiddenException("No tienes acceso a esta base de conocimiento")

Las KBs pueden ser:

• Publicas dentro del tenant: todos los miembros las ven
• Privadas: solo miembros asignados via grupos las ven
• System: KBs del sistema, visibles para todos, solo modificables por super_admin

CASCADE: borrado limpio

Todas las foreign keys usan ondelete="CASCADE". Cuando se borra un tenant, todo se borra en cascada: usuarios, KBs, documentos, chunks, conversaciones, API keys, usage logs. Una sola operacion:

async def delete_tenant(db: AsyncSession, tenant_id: UUID, current_user: User) -> dict:
    tenant = await get_tenant(db, tenant_id)

    # Safety checks
    if current_user.tenant_id == tenant_id:
        raise BadRequestException("No puedes eliminar tu propio tenant")
    if tenant.slug == "system-tenant":
        raise ForbiddenException("No se puede eliminar el tenant del sistema")

    # 1. Cancelar suscripcion activa
    try:
        await admin_cancel_subscription(db, tenant_id, immediate=True)
    except NotFoundException:
        pass

    # 2. Limpiar cache Redis
    redis = await get_redis()
    if redis:
        keys = await redis.keys(f"*:{tenant_id}:*")
        if keys:
            await redis.delete(*keys)

    # 3. DELETE → CASCADE hace el resto
    await db.delete(tenant)
    await db.commit()

---

Plan enforcement: contadores atomicos con UPSERT

El problema

Imaginate que un tenant en plan Free tiene limite de 50 mensajes/mes. Dos usuarios mandan un mensaje al mismo tiempo. Si hacés SELECT count → check → INSERT, tenés una race condition: ambos leen 49, ambos pasan el check, ambos insertan. Ahora hay 51.

La solucion: UPSERT atomico

La tabla tenant_monthly_usage usa un constraint UNIQUE en (tenant_id, billing_period):

class TenantMonthlyUsage(Base):
    __tablename__ = "tenant_monthly_usage"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )
    billing_period: Mapped[str] = mapped_column(String(10), nullable=False)
    messages_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    tokens_used: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)

    __table_args__ = (
        UniqueConstraint("tenant_id", "billing_period", name="uq_tenant_billing_usage"),
    )

El incremento usa INSERT ... ON CONFLICT DO UPDATE:

async def increment_message_count(db: AsyncSession, tenant_id: UUID) -> None:
    anchor = await _get_billing_anchor(db, tenant_id)
    bp = _get_billing_period_key(anchor)
    await db.execute(text("""
        INSERT INTO tenant_monthly_usage (id, tenant_id, billing_period, messages_count, tokens_used)
        VALUES (gen_random_uuid(), :tid, :bp, 1, 0)
        ON CONFLICT (tenant_id, billing_period)
        DO UPDATE SET messages_count = tenant_monthly_usage.messages_count + 1,
                      updated_at = now()
    """), {"tid": tenant_id, "bp": bp})
    await db.commit()

¿Por qué esto funciona? PostgreSQL ejecuta el INSERT ... ON CONFLICT DO UPDATE de forma atómica. No hay ventana de race condition. Si dos requests llegan al mismo tiempo, uno hara INSERT y el otro hara UPDATE, pero el contador siempre será correcto.

Detalle anti-fraude: los contadores solo incrementan. No hay endpoint para decrementar. No hay forma de que un usuario manipule su usage. La unica forma de "resetear" es que cambie el billing period (es decir, que pase un mes).

Calculo del billing period

def _get_billing_period_key(anchor: datetime) -> str:
    now = datetime.now(timezone.utc)
    billing_day = anchor.day

    # Clamping: si el anchor es dia 31 y estamos en febrero (28 dias),
    # el billing day efectivo es 28
    max_day = calendar.monthrange(now.year, now.month)[1]
    effective_day = min(billing_day, max_day)

    if now.day >= effective_day:
        return f"{now.year}-{now.month:02d}-{effective_day:02d}"
    else:
        # El periodo actual empezo el mes pasado
        if now.month == 1:
            prev_year, prev_month = now.year - 1, 12
        else:
            prev_year, prev_month = now.year, now.month - 1
        max_day_prev = calendar.monthrange(prev_year, prev_month)[1]
        effective_day_prev = min(billing_day, max_day_prev)
        return f"{prev_year}-{prev_month:02d}-{effective_day_prev:02d}"

¿Por qué es complicado? Porque los meses no tienen la misma cantidad de dias. Si un tenant pagó el 31 de enero, su siguiente ciclo no empieza el 31 de febrero (no existe). El clamping maneja esto.

Verificacion de limites

Antes de ejecutar el pipeline RAG, verifico los limites:

async def check_message_limit(db: AsyncSession, tenant_id: UUID) -> None:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return
    if plan.max_messages_month == -1:
        return  # Ilimitado
    usage = await get_monthly_usage(db, tenant_id)
    if usage["messages_month"] >= plan.max_messages_month:
        raise PlanLimitException(
            f"Limite de mensajes mensuales alcanzado "
            f"({plan.max_messages_month} max para plan {plan.name})"
        )

PlanLimitException devuelve HTTP 403, no 429. Es una restriccion de plan, no de rate limiting. El frontend distingue entre "pagá mas" (403) y "esperá un rato" (429).

Limites en todos los recursos

No solo mensajes. Cada recurso tiene su check:

# Antes de crear un usuario
await check_user_limit(db, tenant_id)

# Antes de crear una KB
await check_kb_limit(db, tenant_id)

# Antes de subir un documento
await check_document_limit(db, tenant_id)
await check_storage_limit(db, tenant_id, new_file_size_bytes=file_size)

# Antes de crear una API key
await check_api_key_limit(db, tenant_id)

# Antes de enviar un mensaje al LLM
await check_message_limit(db, tenant_id)
await check_concurrent_limit(db, tenant_id)

El patron es siempre el mismo:

async def check_kb_limit(db: AsyncSession, tenant_id: UUID) -> None:
    tenant = await get_tenant(db, tenant_id)
    if not tenant.plan_id:
        return
    plan = await db.get(Plan, tenant.plan_id)
    if not plan or plan.max_kbs == -1:
        return
    stmt = select(func.count(KnowledgeBase.id)).where(KnowledgeBase.tenant_id == tenant_id)
    result = await db.execute(stmt)
    current_count = result.scalar() or 0
    if current_count >= plan.max_kbs:
        raise PlanLimitException(
            f"Limite de bases de conocimiento alcanzado ({plan.max_kbs} max)"
        )

---

Rate limiting con Redis: sliding window

Por que Redis y no PostgreSQL

Los contadores de billing period estan en PostgreSQL porque necesitan durabilidad. El rate limiting esta en Redis porque necesita velocidad y TTL automatico.

Si el servidor se reinicia a mitad de un stream, no quiero un contador "fantasma" en PostgreSQL que bloquee al tenant. Redis con TTL se limpia solo.

Sliding window con sorted sets

import time
import redis.asyncio as redis

async def is_allowed(key: str, max_requests: int, window_seconds: int = 60) -> bool:
    client = await get_redis()
    if client is None:
        return True  # Fail-open

    redis_key = f"ratelimit:{key}"
    now = time.time()
    cutoff = now - window_seconds

    try:
        pipe = client.pipeline()
        pipe.zremrangebyscore(redis_key, 0, cutoff)   # Limpiar expirados
        pipe.zcard(redis_key)                          # Contar actuales
        pipe.zadd(redis_key, {str(now): now})          # Agregar este request
        pipe.expire(redis_key, window_seconds + 1)     # TTL safety net
        results = await pipe.execute()

        current_count = results[1]
        if current_count >= max_requests:
            await client.zrem(redis_key, str(now))     # Rollback
            return False
        return True
    except Exception:
        return True  # Fail-open

¿Por qué sorted sets y no un simple INCR? Porque necesito una ventana deslizante, no una ventana fija. Con INCR, si un usuario manda 60 requests en el segundo 59 del minuto, y 60 mas en el segundo 1 del siguiente minuto, pasaría 120 requests en 2 segundos. Con sorted sets, cada request tiene su timestamp y la ventana se desliza continuamente.

¿Por qué fail-open? Si Redis se cae, prefiero dejar pasar requests (degradacion graciosa) que bloquear a todos los usuarios. Los limites de billing en PostgreSQL siguen activos como red de seguridad.

Concurrencia de streams

Para limitar conexiones SSE simultaneas por tenant, uso un patron diferente: INCR/DECR con TTL de seguridad.

_CONCURRENT_TTL = 300  # 5 minutos

async def increment_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    try:
        pipe = client.pipeline()
        pipe.incr(redis_key)
        pipe.expire(redis_key, _CONCURRENT_TTL)
        results = await pipe.execute()
        return results[0]
    except Exception:
        return 0


async def decrement_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    try:
        count = await client.decr(redis_key)
        if count <= 0:
            await client.delete(redis_key)
            return 0
        await client.expire(redis_key, _CONCURRENT_TTL)
        return count
    except Exception:
        return 0

¿Por qué TTL de 5 minutos? Si el servidor crashea durante un stream, el DECR nunca se ejecuta y el contador queda inflado. El TTL auto-cura esta situacion: despues de 5 minutos sin actividad, el contador se borra solo.

El uso en el endpoint de chat:

@router.post("/conversations/{conv_id}/messages/stream")
async def stream_chat(conv_id: UUID, data: MessageCreate, ...):
    # Verificar presupuesto LLM (soft check)
    faq_only_mode = await is_llm_budget_exhausted(db, current_user.tenant_id)
    if not faq_only_mode:
        await check_concurrent_limit(db, current_user.tenant_id)

    # ... preparar pipeline ...

    async def event_generator():
        await increment_concurrent(str(current_user.tenant_id))
        try:
            async for event in _event_generator_inner():
                yield event
        finally:
            # SIEMPRE decrementa, incluso si hay error
            await decrement_concurrent(str(current_user.tenant_id))

    return EventSourceResponse(event_generator())

El finally es critico. Sin el, cualquier excepcion durante el streaming dejaria el contador inflado.

---

Registro: un tenant por usuario, automatico

Cuando alguien se registra, se crea automaticamente un tenant con plan Free:

async def register(db: AsyncSession, data: UserRegister) -> User:
    existing = await get_user_by_email(db, data.email)
    if existing:
        raise BadRequestException("El email ya esta registrado")

    # Crear tenant automaticamente
    tenant = await create_tenant(db, data.tenant_name, plan_slug="free")

    user = User(
        email=data.email,
        hashed_password=hash_password(data.password),
        full_name=data.full_name,
        role=UserRole.TENANT_ADMIN,  # El creador es admin
        tenant_id=tenant.id,
        email_verified=False,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

El slug del tenant se genera automaticamente con deduplicacion:

def _generate_slug(name: str) -> str:
    slug = re.sub(r'[^a-z0-9]+', '-', name.lower()).strip('-')
    return slug[:80] if slug else 'tenant'


async def create_tenant(db: AsyncSession, name: str, plan_slug: str = "free") -> Tenant:
    # Buscar plan
    stmt = select(Plan).where(Plan.slug == plan_slug, Plan.is_active == True)
    result = await db.execute(stmt)
    plan = result.scalar_one_or_none()

    base_slug = _generate_slug(name)
    slug = base_slug

    # Si ya existe, agregar sufijo aleatorio
    existing = await db.execute(select(Tenant).where(Tenant.slug == slug))
    if existing.scalar_one_or_none():
        slug = f"{base_slug}-{uuid.uuid4().hex[:6]}"

    tenant = Tenant(name=name, slug=slug, plan_id=plan.id if plan else None)
    db.add(tenant)
    await db.flush()
    return tenant

---

JWT: tenant_id en el token

El access token incluye tenant_id y role directamente en el payload:

def create_access_token(user_id: str, tenant_id: str, role: str,
                        expire_minutes: int | None = None) -> str:
    minutes = expire_minutes or settings.access_token_expire_minutes
    expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
    payload = {
        "sub": user_id,
        "tenant_id": tenant_id,
        "role": role,
        "exp": expire,
        "type": "access",
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)

¿Por qué incluir tenant_id en el JWT? Para no tener que hacer una query a la DB en cada request solo para saber de qué tenant es el usuario. En get_current_user igual verifico contra la DB (por si el usuario fue desactivado), pero el tenant_id del token sirve como hint rápido.

Refresh token rotation

Cada vez que se usa un refresh token, se revoca y se emite uno nuevo:

async def refresh_access_token(db: AsyncSession, refresh_token_value: str):
    stmt = select(RefreshToken).where(
        RefreshToken.token == refresh_token_value,
        RefreshToken.revoked == False,
    )
    result = await db.execute(stmt)
    token_record = result.scalar_one_or_none()

    if not token_record:
        raise UnauthorizedException("Refresh token invalido")

    # Revocar el viejo (rotacion)
    token_record.revoked = True

    user = await get_user_by_id(db, token_record.user_id)
    if not user or not user.is_active:
        await db.commit()
        raise UnauthorizedException("Usuario no encontrado o desactivado")

    # Emitir nuevos tokens
    new_access = create_access_token(str(user.id), str(user.tenant_id), user.role.value)
    new_refresh_value = create_refresh_token_value()
    new_refresh = RefreshToken(
        user_id=user.id,
        token=new_refresh_value,
        expires_at=datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expire_days),
    )
    db.add(new_refresh)
    await db.commit()

    return new_access, new_refresh_value

Si alguien intercepta un refresh token y lo usa, el token original queda revocado. Cuando el usuario legítimo intente refrescar, va a fallar y sabrá que hubo un compromiso.

---

Threshold warnings: avisar antes de que explote

No espero a que el tenant llegue al 100% del límite para avisarle. Implementé warnings en 80% y 95%:

async def get_usage_warnings(db: AsyncSession, tenant_id: UUID) -> list[dict]:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return []

    usage = await get_monthly_usage(db, tenant_id)
    warnings = []

    resources = [
        ("messages", usage["messages_month"], plan.max_messages_month),
        ("tokens", usage["tokens_month"], plan.max_tokens_month),
    ]

    for resource, used, max_val in resources:
        if max_val == -1 or max_val <= 0:
            continue
        pct = round((used / max_val) * 100, 1)
        if pct >= 95:
            warnings.append({
                "resource": resource, "used": used, "max": max_val,
                "pct": pct, "level": "critical"
            })
        elif pct >= 80:
            warnings.append({
                "resource": resource, "used": used, "max": max_val,
                "pct": pct, "level": "warning"
            })

    return warnings

Los warnings se envian por email a los admins del tenant, con deduplicacion para no spammear:

# Solo enviar si:
# 1. No se envió para este billing period, O
# 2. Se está escalando de warning → critical
if last_sent == bp and last_level == "critical":
    return  # Ya envié critical para este periodo

# Guardar marker de dedup
updated_prefs["_last_warning_bp"] = bp
updated_prefs["_last_warning_level"] = most_severe["level"]

---

FAQ-only fallback: degradacion graciosa

Cuando un tenant agota su presupuesto de LLM, no bloqueo todo. Las FAQs siguen funcionando:

async def is_llm_budget_exhausted(db: AsyncSession, tenant_id: UUID) -> bool:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return False

    usage = await get_monthly_usage(db, tenant_id)

    if plan.max_messages_month != -1 and usage["messages_month"] >= plan.max_messages_month:
        return True
    if plan.max_tokens_month != -1 and usage["tokens_month"] >= plan.max_tokens_month:
        return True
    return False

En el chat, si is_llm_budget_exhausted devuelve True, se intenta FAQ match primero. Si matchea, respuesta gratuita. Si no, se envia un evento SSE upgrade_required y el frontend muestra un card de upgrade.

---

Custom exceptions: errores semánticos

class PlanLimitException(HTTPException):
    def __init__(self, detail: str = "Plan limit reached"):
        super().__init__(status_code=status.HTTP_403_FORBIDDEN, detail=detail)

class TooManyRequestsException(HTTPException):
    def __init__(self, detail: str = "Too many requests"):
        super().__init__(status_code=429, detail=detail)

El frontend distingue:

• 403 PlanLimit: "Llegaste al limite de tu plan. Upgrade disponible."
• 429 RateLimit: "Demasiados requests. Reintentá en unos segundos."
• 401 Unauthorized: "Sesion expirada. Volvé a iniciar sesion."

---

Numeros que importan

Recurso	Free	Starter	Pro
Usuarios	3	10	50
Knowledge Bases	3	10	-1 (ilimitado)
Documentos	20	100	-1
Storage	200 MB	1 GB	10 GB
Mensajes IA/mes	50	500	5000
API Keys	1	5	20
Concurrencia	2	5	20

---

Lecciones aprendidas

1. UPSERT > SELECT + INSERT

Nunca hagas "read count → check → insert" para contadores de uso. La race condition es inevitable bajo carga. El UPSERT atomico de PostgreSQL es la unica forma correcta. Me ahorré horas de debugging de "por qué el tenant tiene 51 mensajes si el limite es 50".

2. Fail-open para rate limiting, fail-closed para billing

Si Redis se cae, los rate limits se relajan (fail-open) pero los limites de plan en PostgreSQL siguen activos (fail-closed). Nunca dejes que una dependencia caida bloquee a todos tus usuarios.

3. El super_admin no es un dios omnisciente

Mi primer diseño daba acceso total al super_admin sobre los datos de todos los tenants. Mal. El super_admin es el operador de la plataforma, no el dueño de los datos. Ve metricas agregadas, no conversaciones individuales. Esto simplifica compliance y genera confianza.

4. -1 para ilimitado es mejor que un booleano

Tener max_kbs: int con -1 = ilimitado es mas limpio que max_kbs: int + is_kbs_unlimited: bool. Una sola columna, un solo check. Lo apliqué a todos los limites de plan.

5. Billing period basado en fecha de pago, no en mes calendario

Si un tenant pagó el 15 de marzo, su ciclo va del 15/3 al 14/4, no del 1/3 al 31/3. Parece un detalle menor hasta que un usuario se queja de que "pagué ayer y ya usé la mitad de mi cuota".

6. TTL como safety net para contadores en Redis

Los contadores de concurrencia en Redis necesitan TTL. Sin el, un crash del servidor deja contadores "zombie" que bloquean al tenant. Con TTL de 5 minutos, el sistema se auto-cura.

7. CASCADE simplifica el borrado pero hay que limpiar Redis tambien

El ON DELETE CASCADE de PostgreSQL es maravilloso para borrar todo lo relacionado a un tenant. Pero no limpia Redis. Hay que hacerlo explicitamente. Me enteré cuando un tenant borrado seguia contando rate limits.

---

Lo que sigue

• Audit log: registrar quién hizo qué, cuándo, sobre qué recurso
• Fine-grained permissions: permisos por KB y por accion (no solo por rol global)
• Multi-region: tenants asignados a regiones especificas para compliance

---

Conclusion

Multi-tenancy real no es una columna extra en la base de datos. Es un sistema de enforcement que abarca desde el modelo de datos hasta el rate limiting, pasando por la autorizacion y el billing. Los puntos mas criticos:

1. Contadores atomicos (UPSERT) para que los limites se respeten bajo concurrencia
2. Redis para lo efimero (rate limits, concurrencia) y PostgreSQL para lo durable (billing, usage)
3. Roles claros con dependency injection que haga imposible saltear la autorizacion
4. Fail-open donde duele poco, fail-closed donde duele mucho
5. Aislamiento por defecto — cada query filtra por tenant_id, sin excepciones

Si estas construyendo un SaaS multi-tenant, invertí en el enforcement desde el dia 1. Refactorear esto despues es ordenes de magnitud mas doloroso que hacerlo bien desde el principio.

---

Si estas construyendo algo multi-tenant y te chocaste con un problema que no cubrí, dejá un comentario. Y si este articulo te ahorro tiempo, un like ayuda a que llegue a mas personas.