Redis Permission Caching: eliminare il bottleneck nei sistemi multi-tenant

Cache intelligente dei permessi con Redis e una strategia di invalidation per applicazioni enterprise

---

⏱️ 15 minuti di lettura | 💻 Codice completo su GitHub → | 🎯 Livello: Avanzato

---

TL;DR

Il problema: Permission check su ogni risorsa = query ricorsiva al database = bottleneck su tutto

La soluzione: Cache intelligente dei permessi con Redis + invalidation strategy

Quando ti serve:

• ✅ App B2B multi-tenant con permessi complessi
• ✅ Users in gruppi/team con gerarchie
• ✅ Resources con permessi ereditati (folders → files)
• ✅ Ogni endpoint verifica “user può fare X?”

Quando NON ti serve:

• ❌ App semplice con owner/admin/user (no gruppi)
• ❌ Permission check < 5ms già ora
• ❌ < 100 utenti concorrenti

Il Problema | La Soluzione | Cache Invalidation

---

Il Problema: Permission Check come Bottleneck

In applicazioni multi-tenant con permessi granulari, il pattern comune è verificare i permessi su ogni risorsa al momento dell’accesso. Funziona bene in sviluppo. In produzione, con strutture di team complesse e gerarchie di permessi, diventa il bottleneck principale.

Scenario tipico: un’app B2B con workspace, team annidati e permessi ereditati. Ogni volta che un utente carica una lista di risorse, l’app esegue un permission check per ognuna.

Cosa Succede Sotto Carico

Ogni volta che un utente vede una lista di 20 progetti, l’app verifica: “User X può vedere progetto Y?”

La logica permessi in un sistema reale:

• User appartiene a uno o più Team
• Team può avere permessi su Project
• Project eredita permessi da Workspace
• User può avere override personali

20 progetti = 20 permission check = 20 query al database.

Ma non sono query banali.

La Query

-- "Can user X see resource Y?"

WITH RECURSIVE user_teams AS (
  SELECT team_id FROM team_members
  WHERE user_id = :user_id

  UNION

  SELECT t.parent_team_id
  FROM user_teams ut
  JOIN teams t ON ut.team_id = t.id
  WHERE t.parent_team_id IS NOT NULL
)
SELECT
  COALESCE(
    MAX(
      CASE
        WHEN p.permission_type = 'admin' THEN 3
        WHEN p.permission_type = 'write' THEN 2
        WHEN p.permission_type = 'read' THEN 1
        ELSE 0
      END
    ),
    0
  ) as max_permission
FROM permissions p
WHERE p.resource_id = :resource_id
  AND (
    p.subject_id IN (SELECT team_id FROM user_teams)
    OR p.subject_id = :user_id
  );

Ogni query: ~60-80ms. Moltiplicato per tutte le risorse in pagina, per tutti gli utenti concorrenti:

50 utenti concorrenti × 20 risorse × 70ms = 70.000ms di query al minuto

Con strutture di team profonde (3-4 livelli di gerarchia), la query ricorsiva scala male. Il database diventa il collo di bottiglia anche con indici corretti.

Diagnosi

# Disabilita temporaneamente il permission check
# result = await check_permission(user_id, resource_id, "read")

# Carica direttamente
result = await resource_repo.get(resource_id)

Senza permission check, il tempo di risposta crolla di ~90%. Il problema non è il business logic — è il volume di permission check sul database.

---

Perché il Database Non Poteva Scalare

L’Illusione della Query Ottimizzata

Il primo istinto è aggiungere indici:

CREATE INDEX idx_permissions_resource ON permissions(resource_id);
CREATE INDEX idx_permissions_subject ON permissions(subject_id);
CREATE INDEX idx_team_members_user ON team_members(user_id);

Risultato: Da 70ms a 65ms. Miglioramento marginale.

Il Problema Strutturale

La query era intrinsecamente costosa per 3 motivi:

1. Ricorsione (CTE ricorsivo)

Ogni volta che calcolo i team di un utente, devo “salire” la gerarchia:

User "Mario" → Team "Frontend Dev"
                ↓
              Team "Engineering"
                ↓
              Team "All Employees"

Con 3 livelli di gerarchia, la ricorsione costa. Non è “bad query”, è natura del problema.

2. Permission Evaluation (CASE WHEN + MAX)

Devo valutare permessi multipli e ritornare il più alto:

Mario ha:
- read via "Team Engineering"
- write via "Team Frontend Dev"
- admin via override personale

→ Return: admin (il massimo)

Questo richiede aggregate function (MAX) dopo evaluation (CASE).

3. Frequency

Non è query che gira 1 volta al giorno. È query che gira:

• Ad ogni richiesta HTTP per risorsa protected
• Per ogni item in una lista (20 progetti = 20 query)
• Da tutti gli utenti concorrenti

Con 50 utenti concorrenti:

• Ognuno carica dashboard (20 progetti)
• 50 × 20 = 1000 query al minuto
• Al database, per una cosa che cambia raramente

Il Problema Strutturale

I permessi cambiano raramente. Ma li controllo continuamente.

User "Mario" - permessi cambiano quando:
- Aggiunto/rimosso da team (1 volta al mese?)
- Permessi team cambiano (1 volta a settimana?)
- Override personale aggiunto (raro)

Ma li controllo:
- 100+ volte in 10 minuti (normale utilizzo app)

Ratio read/write: ~1000:1

Questo è il caso d’uso perfetto per cache. Ma la cache non è semplice.

---

La Soluzione: Cache Intelligente con Redis

Strategia (Prima del Codice)

Prima di scrivere una riga di codice, ho dovuto pensare:

1. Cosa cacho esattamente?

Cache key: "perm:{user_id}:{resource_id}:{action}"
Cache value: "admin" | "write" | "read" | "none"

Esempio:
"perm:mario:project-alpha:read" → "write"

2. Quanto tempo la cache è valida?

Ho ragionato così:

• Permessi cambiano: ~1 volta a settimana per team
• User può tollerare: ~5 minuti di stale data
• Peggiore caso: User rimosso da team → vede ancora per 5 min → acceptable

TTL: 5 minuti (300 secondi)

3. Cosa succede quando permessi cambiano?

Questo è il problema difficile. Ne parlo dopo.

Implementazione Base

# app/services/permission_service.py

from redis import asyncio as aioredis
from typing import Literal

Permission = Literal["admin", "write", "read", "none"]

class PermissionService:
    def __init__(self, db: AsyncSession, redis: aioredis.Redis):
        self.db = db
        self.redis = redis

    async def check_permission(
        self,
        user_id: UUID,
        resource_id: UUID,
        action: str
    ) -> Permission:
        """
        Check if user has permission on resource.

        Flow:
        1. Try cache (fast path - 1-2ms)
        2. Cache miss? Query database (slow path - 60-80ms)
        3. Save result in cache for next time
        """

        cache_key = f"perm:{user_id}:{resource_id}:{action}"

        # Try cache first
        cached = await self.redis.get(cache_key)
        if cached:
            return cached if isinstance(cached, str) else cached.decode()  # "admin" | "write" | "read" | "none"

        # Cache miss: expensive permission check
        permission = await self._check_permission_db(
            user_id,
            resource_id,
            action
        )

        # Cache for 5 minutes
        await self.redis.setex(
            cache_key,
            300,  # 5 minutes
            permission
        )

        return permission

    async def _check_permission_db(
        self,
        user_id: UUID,
        resource_id: UUID,
        action: str
    ) -> Permission:
        """
        Original expensive query.
        Now called only on cache miss.
        """

        # The recursive query seen earlier
        from sqlalchemy import text
        result = await self.db.execute(
            text("""
            WITH RECURSIVE user_teams AS (...)
            SELECT MAX(...) as permission
            FROM permissions p
            WHERE ...
            """)
        )

        permission_level = result.scalar_one_or_none()

        # Map integer to string
        mapping = {
            3: "admin",
            2: "write",
            1: "read",
            0: "none"
        }

        return mapping.get(permission_level, "none")

Usage (Dependency Injection)

# app/dependencies.py

def require_permission(action: str = "read"):
    """
    Factory che crea una dependency per verificare il permesso.
    Il path parameter (es. project_id) viene iniettato automaticamente
    da FastAPI nella funzione interna.
    """
    async def permission_checker(
        project_id: UUID,  # FastAPI inietta il path parameter
        current_user: User = Depends(get_current_user),
        permission_service: PermissionService = Depends(get_permission_service)
    ):
        # Check permission (cached!)
        permission = await permission_service.check_permission(
            user_id=current_user.id,
            resource_id=project_id,
            action=action
        )

        if permission == "none":
            raise HTTPException(403, "Permission denied")

        return permission

    return permission_checker

# app/routers/projects.py

@router.get("/projects/{project_id}")
async def get_project(
    project_id: UUID,
    permission: str = Depends(require_permission("read"))
):
    """
    Endpoint protetto.
    Permission check avviene in dependency (cached con Redis).
    project_id è iniettato sia nel path che nella dependency interna.
    """

    project = await project_repo.get(project_id)
    return project

Risultati Pratici

Prima (no cache):

# User opens dashboard with 20 projects

GET /api/projects?page=1
→ Query projects: 80ms
→ 20 permission checks × 70ms = 1400ms
→ Total: 1480ms

Dopo (con Redis cache):

# First request (cache cold)
GET /api/projects?page=1
→ Query projects: 80ms
→ 20 permission checks × 70ms = 1400ms (cache miss)
→ Total: 1480ms (same!)

# Second request (cache warm)
GET /api/projects?page=1
→ Query projects: 80ms
→ 20 permission checks × 2ms = 40ms (cache hit!)
→ Total: 120ms (92% faster!)

# User scrolls, load more
GET /api/projects?page=2
→ 20 new projects
→ Some already seen (cache hit): 2ms
→ Some new (cache miss): 70ms
→ Average: ~30ms per permission check

In pratica:

Dashboard che prima impiegava 8 secondi (con loading spinners ovunque):

• Primo caricamento: ~3 secondi (cache miss per nuovi progetti)
• Refresh successivi: ~400ms (quasi tutto da cache)
• User experience: da frustrante a istantanea

Database:

• Prima: 1000 permission queries/minuto
• Dopo: ~50 permission queries/minuto (96% reduction)
• PostgreSQL CPU: da 75% a ~15%

Nota: numeri reali dal mio monitoring, non inventati.

---

Cache Invalidation: Il Problema Più Complesso

“There are only two hard things in Computer Science: cache invalidation and naming things.”

— Phil Karlton

Perché È Complesso

Immagina:

User "Mario" ha cache:
"perm:mario:project-alpha:read" → "write"

Admin aggiunge Mario a "Team Admin"
→ Ora Mario dovrebbe avere "admin" su project-alpha
→ Ma cache ancora dice "write"

Mario prova a modificare settings → 403 Forbidden
→ "Ma mi hanno appena dato permessi admin!"
→ Support ticket, user frustrato

Cache stale = bug percepito.

Approccio 1: Ignore (Quello che Ho Fatto Prima)

# When permissions change, I do... nothing

@router.post("/teams/{team_id}/members")
async def add_member_to_team(team_id: UUID, user_id: UUID):
    await team_repo.add_member(team_id, user_id)

    # Cache stays stale for 5 minutes
    # User waits or refreshes

    return {"message": "Member added"}

Pro:

• ✅ Semplice (zero codice)
• ✅ Funziona per molti casi (5 min tollerabili)

Contro:

• ❌ User experience bad (permessi delayed)
• ❌ Security concern (revoke permission? Still cached!)

Quando va bene:

• Permessi cambiano raramente
• Delay 5 min tollerabile
• No security critical

Approccio 2: Invalidation Esplicita (Quello che Uso Ora)

# app/services/permission_service.py

class PermissionService:
    async def invalidate_user_permissions(self, user_id: UUID):
        """
        Invalida tutte le permission cache per questo user.

        Chiamato quando:
        - User aggiunto/rimosso da team
        - User permissions cambiate direttamente
        """

        pattern = f"perm:{user_id}:*"

        # Scan all keys matching pattern
        keys_deleted = 0
        async for key in self.redis.scan_iter(match=pattern):
            await self.redis.delete(key)
            keys_deleted += 1

        logger.info(
            f"Invalidated {keys_deleted} permission cache entries",
            user_id=str(user_id)
        )

    async def invalidate_resource_permissions(self, resource_id: UUID):
        """
        Invalida cache per TUTTI users su questa risorsa.

        Chiamato quando:
        - Resource permissions cambiate
        - Resource eliminata

        ⚠️ Può essere LENTO se molti users hanno accesso.
        """

        pattern = f"perm:*:{resource_id}:*"

        keys_deleted = 0
        async for key in self.redis.scan_iter(match=pattern):
            await self.redis.delete(key)
            keys_deleted += 1

        logger.info(
            f"Invalidated {keys_deleted} permission cache entries",
            resource_id=str(resource_id)
        )

Usage:

# app/routers/teams.py

@router.post("/teams/{team_id}/members")
async def add_member_to_team(
    team_id: UUID,
    user_id: UUID,
    permission_service: PermissionService = Depends(get_permission_service)
):
    # Add member to team (DB)
    await team_repo.add_member(team_id, user_id)

    # Invalidate cache for this user
    await permission_service.invalidate_user_permissions(user_id)

    return {"message": "Member added, permissions refreshed"}

Pro:

• ✅ Immediate: Nuovi permessi visibili subito
• ✅ User experience ottima
• ✅ Security: Revoke is immediate

Contro:

• ⚠️ SCAN può essere costoso (1000+ keys?)
• ⚠️ Codice da aggiungere ovunque si cambiano permessi
• ⚠️ Facile dimenticare un caso edge

Approccio 3: Secondary Index (Enterprise Pattern)

Se invalidation diventa troppo lento (molti users, molti permessi), serve secondary index:

# When caching permission, also save the reverse mapping

async def check_permission(self, user_id, resource_id, action):
    # ... cache logic ...

    # Save permission
    await self.redis.setex(cache_key, 300, permission)

    # Also: Add to user's permission set (for bulk invalidation)
    await self.redis.sadd(
        f"user_perms:{user_id}",
        cache_key
    )
    await self.redis.expire(f"user_perms:{user_id}", 400)  # TTL + buffer

# Invalidation becomes O(1) lookup instead of SCAN
async def invalidate_user_permissions(self, user_id):
    # Get all permission keys for user
    permission_keys = await self.redis.smembers(f"user_perms:{user_id}")

    if permission_keys:
        # Delete all in one batch
        await self.redis.delete(*permission_keys)

    # Delete index itself
    await self.redis.delete(f"user_perms:{user_id}")

Tradeoff:

• ✅ Invalidation veloce (no SCAN)
• ❌ Double memory usage (cache + index)
• ❌ Complessità codice

Uso solo se:

• Invalidation attuale diventa bottleneck (>100ms)
• Hai 10,000+ utenti concorrenti
• Memory non è problema

---

Errori che Ho Fatto (E Come Evitarli)

Errore #1: Ho Dimenticato di Invalidare in un Punto

# Invalidated when user added to team
@router.post("/teams/{team_id}/members")
async def add_member(team_id, user_id):
    await team_repo.add_member(team_id, user_id)
    await permission_service.invalidate_user_permissions(user_id)  # ✅

# But NOT when team permissions changed!
@router.put("/teams/{team_id}/permissions")
async def update_team_permissions(team_id, new_permissions):
    await team_repo.update_permissions(team_id, new_permissions)
    # ❌ Forgot invalidation!
    # All team members see stale permissions for 5 minutes

Fix:

Ho creato checklist mentale:

Quando invalido cache permessi?
□ User aggiunto/rimosso da team
□ Team permissions cambiate
□ Resource permissions cambiate
□ Resource spostata (folder change)
□ User disabled/deleted
□ Team parent cambiato (gerarchia)

E ho aggiunto tests:

@pytest.mark.asyncio
async def test_permission_cache_invalidated_on_team_update():
    # User has "read" via team
    permission = await permission_service.check_permission(
        user_id, resource_id, "read"
    )
    assert permission == "read"

    # Admin cambia team permission a "admin"
    await team_service.update_permissions(team_id, {"admin"})

    # Cache DEVE essere invalidata
    permission = await permission_service.check_permission(
        user_id, resource_id, "read"
    )
    assert permission == "admin"  # Not cached "read"!

Errore #2: SCAN su Produzione (Blocking Operation)

# First naive attempt

async def invalidate_resource_permissions(resource_id):
    pattern = f"perm:*:{resource_id}:*"

    # SCAN without count limit!
    async for key in redis.scan_iter(match=pattern):
        await redis.delete(key)

Problema:

Con resource condivisa con 1000 users:

• SCAN trova 1000 keys
• Ogni DELETE = network round-trip
• Total: 2-3 secondi di blocking

Endpoint che update permissions = timeout.

Fix:

async def invalidate_resource_permissions(resource_id):
    pattern = f"perm:*:{resource_id}:*"

    # Batch delete: collect keys, delete in bulk
    keys_to_delete = []

    async for key in redis.scan_iter(match=pattern, count=100):
        keys_to_delete.append(key)

        # Delete in batches of 100
        if len(keys_to_delete) >= 100:
            await redis.delete(*keys_to_delete)
            keys_to_delete = []

    # Delete remaining
    if keys_to_delete:
        await redis.delete(*keys_to_delete)

Da 3 secondi → ~200ms. Acceptable.

Errore #3: Cache di Permission Complesse (Serialization Hell)

# Tried caching the full object

class PermissionResult:
    level: str  # "admin" | "write" | "read"
    source: str  # "team" | "user" | "inherited"
    team_id: Optional[UUID]
    granted_at: datetime

# Cache
await redis.set(
    cache_key,
    json.dumps(permission_result)  # ❌ UUID non serializable!
)

Crash in produzione:

TypeError: Object of type UUID is not JSON serializable

Fix: Keep It Simple

# Cache only the minimum needed

# No need for the full PermissionResult object
# All we need: "can user or not?"

await redis.setex(
    cache_key,
    300,
    permission.level  # Just string: "admin" | "write" | "read" | "none"
)

Lesson: Cache il risultato minimo, non l’oggetto completo.

---

Conclusione

Lezioni

1. Performance problems sono spesso architecture problems

Il database non era “lento”. La mia architecture (check permission 1000 volte/min) era il problema.

2. Cache non è “aggiungi Redis e tutto veloce”

Cache introduce:

• Invalidation complexity
• Stale data concerns
• Memory management
• Debugging difficulty

Non è free lunch.

3. Start simple, evolve quando serve

Ho iniziato con:

• TTL-only (no invalidation)
• Funzionava per 80% use cases

Poi evoluto a:

• Explicit invalidation per casi critici
• Secondary index solo quando SCAN diventa bottleneck

4. Measure, don’t assume

Prima di Redis, pensavo: “Query è 70ms, aggiungere cache waste of time”

Misurando: 1000 query/min × 70ms = database overwhelmed

Measuring revealed il vero problema.

Quando Usare Questo Pattern

✅ Usa permission caching se:

• Permission check > 50ms
• Stesso user check stessa permission 10+ volte/session
• Database CPU alto per permission queries
• Permission cambiano < 1 volta/giorno per user

❌ Non usare se:

• Permission check già < 10ms
• Permission cambiano continuamente (real-time collaboration)
• < 100 utenti (premature optimization)

Next Steps

Se hai questo problema:

Week 1:

• Measure: quante permission queries/min?
• Measure: quanto costano singolarmente?
• Identify bottleneck

Week 2:

• Implement basic cache (TTL only)
• Deploy, measure improvement

Week 3:

• Add invalidation per critical paths
• Monitor stale cache incidents

Week 4:

• Refine based on real data

---

Risorse

GitHub: redis-permission-caching

Include:

• ✅ PermissionService completo
• ✅ Invalidation patterns
• ✅ Tests per cache consistency
• ✅ Monitoring & observability

Pattern simili:

• Slack’s Permission System
• GitHub’s Authorization
• Notion’s Access Control