leocrm/app/services/entity_permission_service.py

"""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.core.audit import log_audit
from app.core.notifications import create_notification
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
from app.models.contact import Contact
from app.models.address import Address
from app.models.attachment import Attachment
from app.models.bank_account import BankAccount
from app.models.workflow import Workflow
from app.models.sequence import Sequence
from app.models.saved_filter import SavedFilter
from app.models.saved_view import SavedView
from app.models.webhook import Webhook
from app.models.custom_field_definition import CustomFieldDefinition
from app.models.contact_folder import ContactFolder

logger = logging.getLogger(__name__)

# ── Entity Model Registry ────────────────────────────────────────────────────
# Maps entity_type string to SQLAlchemy model class.
# This replaces insecure text(f"SELECT ... FROM {entity_type}s") queries
# with safe SQLAlchemy model-based queries (prevents SQL injection).
ENTITY_MODELS: dict[str, type] = {
    "contact": Contact,
    "address": Address,
    "attachment": Attachment,
    "bank_account": BankAccount,
    "workflow": Workflow,
    "sequence": Sequence,
    "saved_filter": SavedFilter,
    "saved_view": SavedView,
    "webhook": Webhook,
    "custom_field_definition": CustomFieldDefinition,
    "contact_folder": ContactFolder,
}

# Core models with OwnedMixin (Phase 2 additions)
try:
    from app.models.entity_attachment import EntityAttachment
    ENTITY_MODELS["entity_attachment"] = EntityAttachment
except ImportError:
    pass
try:
    from app.models.entity_history import EntityHistory
    ENTITY_MODELS["entity_history"] = EntityHistory
except ImportError:
    pass

# Plugin models if available
try:
    from app.plugins.builtins.dms.models import File as DmsFile
    ENTITY_MODELS["file"] = DmsFile
except ImportError:
    pass
try:
    from app.plugins.builtins.dms.models import Folder as DmsFolder
    ENTITY_MODELS["folder"] = DmsFolder
except ImportError:
    pass
try:
    from app.plugins.builtins.calendar.models import CalendarEntry
    ENTITY_MODELS["calendar_event"] = CalendarEntry
except ImportError:
    pass
try:
    from app.plugins.builtins.calendar.models import Calendar
    ENTITY_MODELS["calendar"] = Calendar
except ImportError:
    pass
try:
    from app.plugins.builtins.calendar.models import Subtask
    ENTITY_MODELS["subtask"] = Subtask
except ImportError:
    pass
try:
    from app.plugins.builtins.tasks.models import Task
    ENTITY_MODELS["task"] = Task
except ImportError:
    pass
try:
    from app.plugins.builtins.mail.models import MailAccount
    ENTITY_MODELS["mailbox"] = MailAccount
except ImportError:
    pass


def _get_entity_model(entity_type: str) -> type:
    """Get SQLAlchemy model class for entity_type, or raise ValueError."""
    model = ENTITY_MODELS.get(entity_type)
    if model is None:
        raise ValueError(f"Unknown entity type: {entity_type}")
    return model

# 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,
    redis: Any = 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])
        # Audit log for permission update
        await log_audit(
            db, tenant_id, created_by,
            action='permission_update',
            entity_type='entity_permission',
            entity_id=existing.id,
            changes={'permission_level': permission_level, 'principal_type': principal_type, 'principal_id': principal_id}
        )
        # Notify user if direct permission
        if principal_type == 'user':
            await create_notification(
                db, tenant_id, principal_uuid,
                type='permission_granted',
                title='Neue Berechtigung',
                body=f'{entity_type} wurde mit dir geteilt',
                entity_type=entity_type,
                entity_id=entity_uuid,
            )
        # Invalidate cache for upsert
        if principal_type == "user":
            await _invalidate_user_cache(redis, tenant_id, principal_uuid, entity_type)
        elif principal_type == "group":
            members_q2 = await db.execute(
                select(UserGroup.user_id)
                .where(UserGroup.group_id == principal_uuid)
                .where(UserGroup.tenant_id == tenant_id)
            )
            for (uid2,) in members_q2:
                await _invalidate_user_cache(redis, tenant_id, uid2, entity_type)
        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(redis, 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(redis, tenant_id, uid, entity_type)

    # Audit log for new permission
    await log_audit(
        db, tenant_id, created_by,
        action='permission_grant',
        entity_type='entity_permission',
        entity_id=perm.id,
        changes={'permission_level': permission_level, 'principal_type': principal_type, 'principal_id': principal_id}
    )
    # Notify user if direct permission
    if principal_type == 'user':
        await create_notification(
            db, tenant_id, principal_uuid,
            type='permission_granted',
            title='Neue Berechtigung',
            body=f'{entity_type} wurde mit dir geteilt',
            entity_type=entity_type,
            entity_id=entity_uuid,
        )

    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,
    redis: Any = 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(redis, 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(redis, tenant_id, uid, old_entity_type)

    # Audit log for permission update
    await log_audit(
        db, tenant_id, None,
        action='permission_update',
        entity_type='entity_permission',
        entity_id=perm.id,
        changes={'permission_level': permission_level, 'principal_type': old_principal_type, 'principal_id': str(old_principal_id)}
    )

    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, redis: Any = None
) -> 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
    old_entity_id = perm.entity_id

    # Audit log for permission revoke
    await log_audit(
        db, tenant_id, None,
        action='permission_revoke',
        entity_type='entity_permission',
        entity_id=perm.id,
        changes={'permission_level': perm.permission_level, 'principal_type': old_principal_type, 'principal_id': str(old_principal_id)}
    )
    # Notify user if direct permission
    if old_principal_type == 'user':
        await create_notification(
            db, tenant_id, old_principal_id,
            type='permission_revoked',
            title='Berechtigung entfernt',
            body=f'{old_entity_type} wurde nicht mehr mit dir geteilt',
            entity_type=old_entity_type,
            entity_id=old_entity_id,
        )

    await db.delete(perm)
    await db.commit()

    # Invalidate cache
    if old_principal_type == "user":
        await _invalidate_user_cache(redis, 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(redis, 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. Guest permission (principal_type='guest')
    7. owner_id IS NULL → 'read' (tenant-owned, visible to all with module permission)
    8. 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 via SQLAlchemy model (safe from SQL injection)
    model = _get_entity_model(entity_type)
    owner_q = await db.execute(
        select(model.owner_id).where(model.id == entity_id).where(model.tenant_id == 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,
            )
        )
    # Guest permission — only if user_id is actually a guest_user_id
    # (Guest users have no groups/roles, only direct entity_permissions)
    principal_conditions.append(
        and_(
            EntityPermission.principal_type == "guest",
            EntityPermission.principal_id == user_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
        model = _get_entity_model(entity_type)
        admin_q = select(model.id).where(model.tenant_id == tenant_id)
        if hasattr(model, 'deleted_at'):
            admin_q = admin_q.where(model.deleted_at.is_(None))
        all_q = await db.execute(admin_q)
        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
    model = _get_entity_model(entity_type)
    owned_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id == user_id)
    if hasattr(model, 'deleted_at'):
        owned_q_builder = owned_q_builder.where(model.deleted_at.is_(None))
    owned_q = await db.execute(owned_q_builder)
    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_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id.is_(None))
    if hasattr(model, 'deleted_at'):
        tenant_q_builder = tenant_q_builder.where(model.deleted_at.is_(None))
    tenant_owned_q = await db.execute(tenant_q_builder)
    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,
            )
        )
    # Guest permission — guest users have no groups/roles
    principal_conditions.append(
        and_(
            EntityPermission.principal_type == "guest",
            EntityPermission.principal_id == user_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 via SQLAlchemy model (safe from SQL injection)
    model = _get_entity_model(entity_type)
    batch_q = select(model.id, model.owner_id).where(model.id.in_(entity_ids)).where(model.tenant_id == tenant_id)
    if hasattr(model, 'deleted_at'):
        batch_q = batch_q.where(model.deleted_at.is_(None))
    owner_q = await db.execute(batch_q)
    for eid, owner_id in owner_q:
        if owner_id == user_id:
            result[eid] = "owner"
        elif owner_id is None:
            result[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,
            )
        )
    # Guest permission — guest users have no groups/roles
    principal_conditions.append(
        and_(
            EntityPermission.principal_type == "guest",
            EntityPermission.principal_id == user_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 list_all_permissions(
    db: AsyncSession, tenant_id: uuid.UUID
) -> list[dict]:
    """List ALL permission entries for a tenant (global view)."""
    result = await db.execute(
        select(EntityPermission)
        .where(EntityPermission.tenant_id == tenant_id)
        .order_by(EntityPermission.entity_type, 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 get_permission_analytics(
    db: AsyncSession,
    tenant_id: uuid.UUID,
) -> dict:
    """Get permission analytics for a tenant.

    Returns:
        total_permissions: Total number of permission entries
        total_shared_entities: Number of unique entities with permissions
        permissions_by_level: Breakdown by permission level
        permissions_by_entity_type: Breakdown by entity type
        recent_changes: Last 10 permission changes
    """
    from sqlalchemy import func as sa_func

    # Total permissions
    total_q = await db.execute(
        select(sa_func.count(EntityPermission.id))
        .where(EntityPermission.tenant_id == tenant_id)
    )
    total_permissions = total_q.scalar() or 0

    # Total unique shared entities
    unique_q = await db.execute(
        select(sa_func.count(sa_func.distinct(
            EntityPermission.entity_type + ":" + EntityPermission.entity_id.cast(String)
        )))
        .where(EntityPermission.tenant_id == tenant_id)
    )
    total_shared_entities = unique_q.scalar() or 0

    # Permissions by level
    level_q = await db.execute(
        select(EntityPermission.permission_level, sa_func.count(EntityPermission.id))
        .where(EntityPermission.tenant_id == tenant_id)
        .group_by(EntityPermission.permission_level)
    )
    permissions_by_level = {row[0]: row[1] for row in level_q}

    # Permissions by entity type
    type_q = await db.execute(
        select(EntityPermission.entity_type, sa_func.count(EntityPermission.id))
        .where(EntityPermission.tenant_id == tenant_id)
        .group_by(EntityPermission.entity_type)
    )
    permissions_by_entity_type = {row[0]: row[1] for row in type_q}

    # Recent changes (last 10)
    recent_q = await db.execute(
        select(EntityPermission)
        .where(EntityPermission.tenant_id == tenant_id)
        .order_by(EntityPermission.updated_at.desc())
        .limit(10)
    )
    recent = recent_q.scalars().all()
    recent_changes = [
        {
            "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),
            "permission_level": p.permission_level,
            "updated_at": p.updated_at.isoformat() if p.updated_at else None,
        }
        for p in recent
    ]

    return {
        "total_permissions": total_permissions,
        "total_shared_entities": total_shared_entities,
        "permissions_by_level": permissions_by_level,
        "permissions_by_entity_type": permissions_by_entity_type,
        "recent_changes": recent_changes,
    }


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