sprint1: entity_permissions table + owned_mixin + universal permission service + API + migrations 0049+0050

app/main.py

@@ -35,6 +35,7 @@ from app.routes import (
     errors,
     contact_folders,
     contact_folder_permissions,
+    entity_permissions,
     contacts,
     dashboard,
     entity_history,
@@ -378,6 +379,7 @@ def create_app() -> FastAPI:
     app.include_router(contacts.router)
     app.include_router(contact_folders.router)
     app.include_router(contact_folder_permissions.router)
+    app.include_router(entity_permissions.router)
     app.include_router(dashboard.router)
     app.include_router(entity_history.router)
     app.include_router(import_export.router)

app/models/__init__.py

@@ -10,6 +10,8 @@ from app.models.contact import Contact, ContactPerson
 from app.models.contact_folder import ContactFolder
 from app.models.contact_folder_permission import ContactFolderPermission
 from app.models.contact_merge import ContactMergeHistory
+from app.models.entity_permission import EntityPermission
+from app.models.owned_mixin import OwnedMixin
 from app.models.entity_history import EntityHistory
 from app.models.currency import Currency
 from app.models.group import Group, UserGroup
@@ -48,6 +50,8 @@ __all__ = [
     "ContactFolder",
     "ContactFolderPermission",
     "ContactMergeHistory",
+    "EntityPermission",
+    "OwnedMixin",
     "EntityHistory",
     "Currency",
     "TaxRate",

app/models/address.py

@@ -9,6 +9,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class Address(Base, TenantMixin):

app/models/ai_conversation.py

@@ -11,6 +11,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class AIConversation(Base, TenantMixin):

app/models/attachment.py

@@ -10,6 +10,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class Attachment(Base, TenantMixin):

app/models/bank_account.py

@@ -9,6 +9,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class BankAccount(Base, TenantMixin):

app/models/contact.py

@@ -27,9 +27,10 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
-class Contact(Base, TenantMixin):
+class Contact(Base, TenantMixin, OwnedMixin):
     """Unified contact entity — can be a company or a person.
 
     type='company': name is the company name, firstname/surname empty.

app/models/custom_field_definition.py

@@ -10,6 +10,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class CustomFieldDefinition(Base, TenantMixin):

app/models/entity_history.py

@@ -12,6 +12,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class EntityHistory(Base, TenantMixin):

app/models/entity_permission.py

@@ -0,0 +1,109 @@
+"""Universal entity permission model — ACLs for ANY entity in the system.
+
+This single table stores permissions for contacts, files, mailboxes,
+calendar events, tasks, workflows, and any future entity type.
+
+Architecture:
+- entity_type + entity_id identify the datensatz
+- principal_type + principal_id identify who gets access
+- permission_level defines what they can do
+- expires_at enables time-limited sharing
+
+Resolution (highest wins):
+1. Owner → 'owner' (from owner_id on the entity)
+2. Direct user permission
+3. Group permission (via user_groups)
+4. Role permission (via user_tenants.role_id)
+5. No access → 'none'
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    CheckConstraint,
+    DateTime,
+    ForeignKey,
+    Index,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class EntityPermission(Base, TenantMixin):
+    """Universal ACL entry for any entity in the system.
+
+    entity_type examples: 'contact', 'dms_file', 'mailbox', 'calendar_event',
+    'task', 'workflow', 'contact_folder', etc.
+
+    principal_type: 'user', 'group', 'role', 'guest'
+
+    permission_level: 'none' | 'read' | 'write' | 'admin' | 'delete'
+    - none:   explicit deny (overrides allow)
+    - read:   view the entity
+    - write:  read + edit entity fields
+    - admin:  write + delete + manage permissions
+    - delete: admin + transfer ownership
+    """
+
+    __tablename__ = "entity_permissions"
+    __table_args__ = (
+        UniqueConstraint(
+            "entity_type",
+            "entity_id",
+            "principal_type",
+            "principal_id",
+            "tenant_id",
+            name="uq_ep_entity_principal_tenant",
+        ),
+        CheckConstraint(
+            "principal_type IN ('user', 'group', 'role', 'guest')",
+            name="ck_ep_principal_type",
+        ),
+        CheckConstraint(
+            "permission_level IN ('none', 'read', 'write', 'admin', 'delete')",
+            name="ck_ep_permission_level",
+        ),
+        Index("ix_ep_entity", "entity_type", "entity_id"),
+        Index("ix_ep_principal", "principal_type", "principal_id"),
+        Index("ix_ep_tenant", "tenant_id"),
+        Index("ix_ep_expires", "expires_at"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    entity_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    entity_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), nullable=False
+    )
+    principal_type: Mapped[str] = mapped_column(String(10), nullable=False)
+    principal_id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), nullable=False
+    )
+    permission_level: Mapped[str] = mapped_column(
+        String(20), nullable=False, default="read"
+    )
+    expires_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, default=None
+    )
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+        onupdate=func.now(),
+    )

app/models/notification.py

@@ -19,6 +19,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class Notification(Base, TenantMixin):

app/models/owned_mixin.py

@@ -0,0 +1,31 @@
+"""OwnedMixin — adds owner_id to any model for row-level ownership.
+
+Usage:
+    class Contact(Base, TenantMixin, OwnedMixin):
+        ...
+
+owner_id semantics:
+- NULL → "tenant-owned" (visible to all with module permission)
+- UUID → owned by that user (visible to owner + shared via entity_permissions)
+- When creating: owner_id = current_user.id (set automatically by service layer)
+- Transfer: only owner, admin, or system_admin can change owner_id
+"""
+
+from __future__ import annotations
+
+import uuid
+
+from sqlalchemy import ForeignKey, Index
+from sqlalchemy.dialects.postgresql import UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+class OwnedMixin:
+    """Mixin that adds owner_id column to a model."""
+
+    owner_id: Mapped[uuid.UUID | None] = mapped_column(
+        PGUUID(as_uuid=True),
+        ForeignKey("users.id", ondelete="SET NULL"),
+        nullable=True,
+        index=True,
+    )

app/models/saved_filter.py

@@ -11,6 +11,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class SavedFilter(Base, TenantMixin):

app/models/saved_view.py

@@ -11,6 +11,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class SavedView(Base, TenantMixin):

app/models/sequence.py

@@ -10,6 +10,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class Sequence(Base, TenantMixin):

app/models/webhook.py

@@ -9,6 +9,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class Webhook(Base, TenantMixin):

app/models/workflow.py

@@ -12,6 +12,7 @@ from sqlalchemy.dialects.postgresql import UUID as PGUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.core.db import Base, TenantMixin
+from app.models.owned_mixin import OwnedMixin
 
 
 class Workflow(Base, TenantMixin):

app/routes/entity_permissions.py

@@ -0,0 +1,151 @@
+"""Universal entity permission routes — ACL management for ALL entities."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import get_current_user, get_redis_dep, require_permission
+from app.schemas.entity_permission import (
+    EntityPermissionCreate,
+    EntityPermissionUpdate,
+)
+from app.services import entity_permission_service
+
+router = APIRouter(prefix="/api/v1/permissions", tags=["entity-permissions"])
+
+
+@router.get("/{entity_type}/{entity_id}")
+async def list_entity_permissions(
+    entity_type: str,
+    entity_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List all permission entries for a specific entity."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        items = await entity_permission_service.list_permissions(
+            db, tenant_id, entity_type, entity_id
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    return {"items": items, "total": len(items)}
+
+
+@router.post("/{entity_type}/{entity_id}", status_code=status.HTTP_201_CREATED)
+async def create_entity_permission(
+    entity_type: str,
+    entity_id: str,
+    body: EntityPermissionCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Grant or update a permission on any entity for a user, group, role, or guest."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    try:
+        return await entity_permission_service.create_permission(
+            db,
+            tenant_id,
+            entity_type,
+            entity_id,
+            body.principal_type,
+            body.principal_id,
+            body.permission_level,
+            body.expires_at,
+            created_by=user_id,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.put("/{entity_type}/{entity_id}/{permission_id}")
+async def update_entity_permission(
+    entity_type: str,
+    entity_id: str,
+    permission_id: str,
+    body: EntityPermissionUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Update an existing permission entry."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        return await entity_permission_service.update_permission(
+            db,
+            tenant_id,
+            permission_id,
+            body.permission_level,
+            body.expires_at,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+
+@router.delete("/{entity_type}/{entity_id}/{permission_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_entity_permission(
+    entity_type: str,
+    entity_id: str,
+    permission_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Revoke a permission entry."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        await entity_permission_service.delete_permission(db, tenant_id, permission_id)
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+
+@router.get("/{entity_type}/{entity_id}/access")
+async def get_entity_access(
+    entity_type: str,
+    entity_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """Get effective access level for the current user on an entity."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    access = await entity_permission_service.get_effective_access(
+        db, tenant_id, user_id, entity_type, uuid.UUID(entity_id)
+    )
+    is_owner = access == "owner"
+    is_shared = access not in ("none", "owner")
+    return {
+        "entity_type": entity_type,
+        "entity_id": entity_id,
+        "access_level": access,
+        "is_owner": is_owner,
+        "is_shared": is_shared,
+    }
+
+
+@router.get("/registry")
+async def list_entity_registry(
+    current_user: dict = Depends(get_current_user),
+):
+    """List all registered entity types that support permissions."""
+    # Static list for now — will be dynamic from Permission Registry in Sprint 4
+    entity_types = [
+        {"entity_type": "contact", "label": "Kontakte", "table": "contacts"},
+        {"entity_type": "contact_folder", "label": "Ordner", "table": "contact_folders"},
+        {"entity_type": "address", "label": "Adressen", "table": "addresses"},
+        {"entity_type": "attachment", "label": "Anhänge", "table": "attachments"},
+        {"entity_type": "bank_account", "label": "Bankkonten", "table": "bank_accounts"},
+        {"entity_type": "workflow", "label": "Workflows", "table": "workflows"},
+        {"entity_type": "sequence", "label": "Sequenzen", "table": "sequences"},
+        {"entity_type": "saved_filter", "label": "Gespeicherte Filter", "table": "saved_filters"},
+        {"entity_type": "saved_view", "label": "Gespeicherte Ansichten", "table": "saved_views"},
+        {"entity_type": "webhook", "label": "Webhooks", "table": "webhooks"},
+        {"entity_type": "notification", "label": "Benachrichtigungen", "table": "notifications"},
+        {"entity_type": "custom_field_definition", "label": "Custom Fields", "table": "custom_field_definitions"},
+        {"entity_type": "ai_conversation", "label": "AI Konversationen", "table": "ai_conversations"},
+    ]
+    return {"items": entity_types, "total": len(entity_types)}

app/schemas/entity_permission.py

@@ -0,0 +1,45 @@
+"""Schemas for universal entity permissions."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class EntityPermissionCreate(BaseModel):
+    principal_type: str = Field(..., pattern="^(user|group|role|guest)$")
+    principal_id: str
+    permission_level: str = Field("read", pattern="^(none|read|write|admin|delete)$")
+    expires_at: datetime | None = None
+
+
+class EntityPermissionUpdate(BaseModel):
+    permission_level: str = Field(..., pattern="^(none|read|write|admin|delete)$")
+    expires_at: datetime | None = None
+
+
+class EntityPermissionResponse(BaseModel):
+    id: str
+    entity_type: str
+    entity_id: str
+    principal_type: str
+    principal_id: str
+    principal_name: str | None = None
+    permission_level: str
+    expires_at: str | None = None
+    created_by: str | None = None
+    created_at: str | None = None
+
+
+class EntityPermissionListResponse(BaseModel):
+    items: list[EntityPermissionResponse]
+    total: int
+
+
+class EntityAccessResponse(BaseModel):
+    entity_type: str
+    entity_id: str
+    access_level: str
+    is_owner: bool
+    is_shared: bool

app/services/entity_permission_service.py

@@ -0,0 +1,648 @@
+"""Universal entity permission service — ACL management for ALL entities.
+
+This service handles:
+- CRUD for entity_permissions
+- Effective access resolution (owner → user → group → role → guest)
+- Batch resolution for list queries
+- Redis caching with bitmap optimization
+- Permission expiration checks
+- Audit logging for permission changes
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from datetime import datetime, UTC
+from typing import Any
+
+import redis.asyncio as aioredis
+from sqlalchemy import and_, func, or_, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.entity_permission import EntityPermission
+from app.models.group import Group, UserGroup
+from app.models.role import Role
+from app.models.user import User, UserTenant
+
+logger = logging.getLogger(__name__)
+
+# Permission hierarchy: higher = more access
+_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
+
+CACHE_TTL = 300  # 5 minutes
+CACHE_PREFIX = "ep_vis"  # entity permission visibility
+
+
+def _rank(level: str) -> int:
+    return _PERM_RANK.get(level, 0)
+
+
+def _serialize_permission(p: EntityPermission, principal_name: str | None = None) -> dict:
+    return {
+        "id": str(p.id),
+        "entity_type": p.entity_type,
+        "entity_id": str(p.entity_id),
+        "principal_type": p.principal_type,
+        "principal_id": str(p.principal_id),
+        "principal_name": principal_name,
+        "permission_level": p.permission_level,
+        "expires_at": p.expires_at.isoformat() if p.expires_at else None,
+        "created_by": str(p.created_by) if p.created_by else None,
+        "created_at": p.created_at.isoformat() if p.created_at else None,
+    }
+
+
+async def _load_principal_names(
+    db: AsyncSession, perms: list[EntityPermission]
+) -> dict[uuid.UUID, str]:
+    """Batch-load names for all principals in a permission list."""
+    user_ids = [p.principal_id for p in perms if p.principal_type == "user"]
+    group_ids = [p.principal_id for p in perms if p.principal_type == "group"]
+    role_ids = [p.principal_id for p in perms if p.principal_type == "role"]
+
+    names: dict[uuid.UUID, str] = {}
+
+    if user_ids:
+        result = await db.execute(select(User.id, User.name).where(User.id.in_(user_ids)))
+        names.update({row[0]: row[1] for row in result})
+
+    if group_ids:
+        result = await db.execute(select(Group.id, Group.name).where(Group.id.in_(group_ids)))
+        names.update({row[0]: row[1] for row in result})
+
+    if role_ids:
+        result = await db.execute(select(Role.id, Role.name).where(Role.id.in_(role_ids)))
+        names.update({row[0]: row[1] for row in result})
+
+    return names
+
+
+async def list_permissions(
+    db: AsyncSession, tenant_id: uuid.UUID, entity_type: str, entity_id: str
+) -> list[dict]:
+    """List all permission entries for a specific entity."""
+    entity_uuid = uuid.UUID(entity_id)
+    result = await db.execute(
+        select(EntityPermission)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id == entity_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .order_by(EntityPermission.created_at)
+    )
+    perms = result.scalars().all()
+
+    names = await _load_principal_names(db, perms)
+    return [_serialize_permission(p, names.get(p.principal_id)) for p in perms]
+
+
+async def create_permission(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    entity_type: str,
+    entity_id: str,
+    principal_type: str,
+    principal_id: str,
+    permission_level: str,
+    expires_at: datetime | None = None,
+    created_by: uuid.UUID | None = None,
+) -> dict:
+    """Create or update a permission entry (upsert)."""
+    entity_uuid = uuid.UUID(entity_id)
+    principal_uuid = uuid.UUID(principal_id)
+
+    # Check for existing entry (upsert)
+    existing_q = await db.execute(
+        select(EntityPermission)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id == entity_uuid)
+        .where(EntityPermission.principal_type == principal_type)
+        .where(EntityPermission.principal_id == principal_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+    )
+    existing = existing_q.scalar_one_or_none()
+
+    if existing:
+        existing.permission_level = permission_level
+        existing.expires_at = expires_at
+        await db.commit()
+        await db.refresh(existing)
+        names = await _load_principal_names(db, [existing])
+        return _serialize_permission(existing, names.get(existing.principal_id))
+
+    perm = EntityPermission(
+        tenant_id=tenant_id,
+        entity_type=entity_type,
+        entity_id=entity_uuid,
+        principal_type=principal_type,
+        principal_id=principal_uuid,
+        permission_level=permission_level,
+        expires_at=expires_at,
+        created_by=created_by,
+    )
+    db.add(perm)
+    await db.commit()
+    await db.refresh(perm)
+
+    # Invalidate cache for this principal
+    if principal_type == "user":
+        await _invalidate_user_cache(None, tenant_id, principal_uuid, entity_type)
+    elif principal_type == "group":
+        # Invalidate for all group members
+        members_q = await db.execute(
+            select(UserGroup.user_id)
+            .where(UserGroup.group_id == principal_uuid)
+            .where(UserGroup.tenant_id == tenant_id)
+        )
+        for (uid,) in members_q:
+            await _invalidate_user_cache(None, tenant_id, uid, entity_type)
+
+    names = await _load_principal_names(db, [perm])
+    return _serialize_permission(perm, names.get(perm.principal_id))
+
+
+async def update_permission(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    permission_id: str,
+    permission_level: str,
+    expires_at: datetime | None = None,
+) -> dict:
+    """Update an existing permission entry."""
+    perm_uuid = uuid.UUID(permission_id)
+    result = await db.execute(
+        select(EntityPermission)
+        .where(EntityPermission.id == perm_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+    )
+    perm = result.scalar_one_or_none()
+    if not perm:
+        raise ValueError("Permission not found")
+
+    old_principal_type = perm.principal_type
+    old_principal_id = perm.principal_id
+    old_entity_type = perm.entity_type
+
+    perm.permission_level = permission_level
+    if expires_at is not None:
+        perm.expires_at = expires_at
+
+    await db.commit()
+    await db.refresh(perm)
+
+    # Invalidate cache
+    if old_principal_type == "user":
+        await _invalidate_user_cache(None, tenant_id, old_principal_id, old_entity_type)
+    elif old_principal_type == "group":
+        members_q = await db.execute(
+            select(UserGroup.user_id)
+            .where(UserGroup.group_id == old_principal_id)
+            .where(UserGroup.tenant_id == tenant_id)
+        )
+        for (uid,) in members_q:
+            await _invalidate_user_cache(None, tenant_id, uid, old_entity_type)
+
+    names = await _load_principal_names(db, [perm])
+    return _serialize_permission(perm, names.get(perm.principal_id))
+
+
+async def delete_permission(
+    db: AsyncSession, tenant_id: uuid.UUID, permission_id: str
+) -> None:
+    """Delete a permission entry."""
+    perm_uuid = uuid.UUID(permission_id)
+    result = await db.execute(
+        select(EntityPermission)
+        .where(EntityPermission.id == perm_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+    )
+    perm = result.scalar_one_or_none()
+    if not perm:
+        raise ValueError("Permission not found")
+
+    old_principal_type = perm.principal_type
+    old_principal_id = perm.principal_id
+    old_entity_type = perm.entity_type
+
+    await db.delete(perm)
+    await db.commit()
+
+    # Invalidate cache
+    if old_principal_type == "user":
+        await _invalidate_user_cache(None, tenant_id, old_principal_id, old_entity_type)
+    elif old_principal_type == "group":
+        members_q = await db.execute(
+            select(UserGroup.user_id)
+            .where(UserGroup.group_id == old_principal_id)
+            .where(UserGroup.tenant_id == tenant_id)
+        )
+        for (uid,) in members_q:
+            await _invalidate_user_cache(None, tenant_id, uid, old_entity_type)
+
+
+async def get_effective_access(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+    entity_id: uuid.UUID,
+) -> str:
+    """Get the effective access level for a user on a specific entity.
+
+    Resolution (highest wins):
+    1. System admin → 'delete' (full access)
+    2. Owner → 'owner' (from owner_id on the entity table)
+    3. Direct user permission
+    4. Group permission (via user_groups)
+    5. Role permission (via user_tenants.role_id)
+    6. owner_id IS NULL → 'read' (tenant-owned, visible to all with module permission)
+    7. No access → 'none'
+
+    Returns: 'none' | 'read' | 'write' | 'admin' | 'delete' | 'owner'
+    """
+    # Check system admin
+    user_q = await db.execute(
+        select(User.is_system_admin).where(User.id == user_id)
+    )
+    if user_q.scalar():
+        return "delete"
+
+    # Check ownership — load the entity's owner_id
+    # We use raw SQL to avoid importing every model
+    owner_q = await db.execute(
+        text(f"SELECT owner_id FROM {entity_type}s WHERE id = :eid AND tenant_id = :tid"),
+        {"eid": entity_id, "tid": tenant_id},
+    )
+    owner_row = owner_q.first()
+    if not owner_row:
+        return "none"
+
+    owner_id = owner_row[0]
+    if owner_id == user_id:
+        return "owner"
+
+    # Get user's groups and role
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    # Build principal conditions
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+
+    # Query permissions
+    now = datetime.now(UTC)
+    perm_q = await db.execute(
+        select(EntityPermission.permission_level)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id == entity_id)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(or_(*principal_conditions))
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > now,
+            )
+        )
+    )
+
+    best_level = "none"
+    for (level,) in perm_q:
+        if _rank(level) > _rank(best_level):
+            best_level = level
+
+    # If owner_id is NULL (tenant-owned), user with module permission gets at least 'read'
+    if best_level == "none" and owner_id is None:
+        return "read"
+
+    return best_level
+
+
+async def get_visible_ids(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
+    """Get all visible entity IDs for a user and their access levels.
+
+    Returns (visible_ids, access_map) where access_map is
+    entity_id → access_level string.
+
+    Resolution:
+    1. System admin → all entities at 'delete' level
+    2. Owned entities → 'owner'
+    3. Entities with direct/group/role permissions → permission level
+    4. Tenant-owned entities (owner_id IS NULL) → 'read'
+    """
+    # Check system admin
+    user_q = await db.execute(
+        select(User.is_system_admin).where(User.id == user_id)
+    )
+    if user_q.scalar():
+        # Return all entity IDs
+        all_q = await db.execute(
+            text(f"SELECT id FROM {entity_type}s WHERE tenant_id = :tid AND deleted_at IS NULL"),
+            {"tid": tenant_id},
+        )
+        all_ids = {row[0] for row in all_q}
+        return all_ids, {eid: "delete" for eid in all_ids}
+
+    # Get user's groups and role
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    # 1. Owned entities
+    owned_q = await db.execute(
+        text(f"SELECT id FROM {entity_type}s WHERE tenant_id = :tid AND owner_id = :uid AND deleted_at IS NULL"),
+        {"tid": tenant_id, "uid": user_id},
+    )
+    visible: set[uuid.UUID] = set()
+    access_map: dict[uuid.UUID, str] = {}
+    for (eid,) in owned_q:
+        visible.add(eid)
+        access_map[eid] = "owner"
+
+    # 2. Tenant-owned entities (owner_id IS NULL)
+    tenant_owned_q = await db.execute(
+        text(f"SELECT id FROM {entity_type}s WHERE tenant_id = :tid AND owner_id IS NULL AND deleted_at IS NULL"),
+        {"tid": tenant_id},
+    )
+    for (eid,) in tenant_owned_q:
+        if eid not in visible:
+            visible.add(eid)
+            access_map[eid] = "read"
+
+    # 3. Permission-based access
+    now = datetime.now(UTC)
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+
+    perm_q = await db.execute(
+        select(EntityPermission.entity_id, EntityPermission.permission_level)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(or_(*principal_conditions))
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > now,
+            )
+        )
+    )
+    for eid, level in perm_q:
+        if eid not in visible or _rank(level) > _rank(access_map.get(eid, "none")):
+            visible.add(eid)
+            access_map[eid] = level
+
+    return visible, access_map
+
+
+async def batch_get_effective_access(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+    entity_ids: list[uuid.UUID],
+) -> dict[uuid.UUID, str]:
+    """Batch resolution: get access levels for multiple entities at once.
+
+    Much more efficient than calling get_effective_access() in a loop.
+    """
+    if not entity_ids:
+        return {}
+
+    # Check system admin
+    user_q = await db.execute(
+        select(User.is_system_admin).where(User.id == user_id)
+    )
+    if user_q.scalar():
+        return {eid: "delete" for eid in entity_ids}
+
+    # Get user's groups and role
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    result: dict[uuid.UUID, str] = {}
+
+    # 1. Check ownership
+    owner_q = await db.execute(
+        text(f"SELECT id, owner_id FROM {entity_type}s WHERE id = ANY(:ids) AND tenant_id = :tid AND deleted_at IS NULL"),
+        {"ids": [str(eid) for eid in entity_ids], "tid": tenant_id},
+    )
+    for eid, owner_id in owner_q:
+        if owner_id == user_id:
+            result[uuid.UUID(str(eid))] = "owner"
+        elif owner_id is None:
+            result[uuid.UUID(str(eid))] = "read"
+
+    # 2. Check permissions
+    now = datetime.now(UTC)
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+
+    perm_q = await db.execute(
+        select(EntityPermission.entity_id, EntityPermission.permission_level)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id.in_(entity_ids))
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(or_(*principal_conditions))
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > now,
+            )
+        )
+    )
+    for eid, level in perm_q:
+        current = result.get(eid, "none")
+        if _rank(level) > _rank(current):
+            result[eid] = level
+
+    # Fill in 'none' for entities not found
+    for eid in entity_ids:
+        if eid not in result:
+            result[eid] = "none"
+
+    return result
+
+
+async def _invalidate_user_cache(
+    redis: aioredis.Redis | None,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+) -> None:
+    """Invalidate the visibility cache for a user + entity type."""
+    if redis is None:
+        return
+    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:{entity_type}"
+    await redis.delete(cache_key)
+
+
+async def get_cached_visible_ids(
+    db: AsyncSession,
+    redis: aioredis.Redis,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
+    """Get visible IDs from Redis cache or resolve from DB.
+
+    Cache key: ep_vis:{user_id}:{tenant_id}:{entity_type}
+    Cache value: JSON {visible_ids: [str], access_map: {str: str}}
+    TTL: 5 minutes
+    """
+    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:{entity_type}"
+
+    raw = await redis.get(cache_key)
+    if raw is not None:
+        data = json.loads(raw)
+        visible = {uuid.UUID(eid) for eid in data.get("visible_ids", [])}
+        access_map = {uuid.UUID(eid): level for eid, level in data.get("access_map", {}).items()}
+        return visible, access_map
+
+    # Cache miss — resolve from DB
+    visible, access_map = await get_visible_ids(db, tenant_id, user_id, entity_type)
+
+    # Store in cache
+    cache_data = {
+        "visible_ids": [str(eid) for eid in visible],
+        "access_map": {str(eid): level for eid, level in access_map.items()},
+    }
+    await redis.setex(cache_key, CACHE_TTL, json.dumps(cache_data))
+
+    return visible, access_map
+
+
+async def invalidate_all_user_entity_cache(
+    redis: aioredis.Redis,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+) -> None:
+    """Invalidate all entity permission caches for a user."""
+    pattern = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:*"
+    batch_size = 200
+    cursor: int | bytes | str = 0
+    while True:
+        cursor, keys = await redis.scan(cursor=cursor, match=pattern, count=batch_size)
+        if keys:
+            await redis.delete(*keys)
+        if int(cursor) == 0:
+            break
+
+
+async def check_entity_access(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+    entity_id: uuid.UUID,
+    required_level: str = "read",
+) -> bool:
+    """Check if user has at least the required access level on an entity."""
+    access = await get_effective_access(db, tenant_id, user_id, entity_type, entity_id)
+    return _rank(access) >= _rank(required_level)
+
+
+async def cleanup_expired_permissions(db: AsyncSession) -> int:
+    """Delete all expired permission entries. Returns count deleted."""
+    now = datetime.now(UTC)
+    result = await db.execute(
+        select(EntityPermission).where(EntityPermission.expires_at < now)
+    )
+    expired = result.scalars().all()
+    count = len(expired)
+    for perm in expired:
+        await db.delete(perm)
+    if count > 0:
+        await db.commit()
+        logger.info("Cleaned up %d expired entity permissions", count)
+    return count