leocrm/app/services/delegation_service.py

"""Permission delegation service — CRUD + active check for permission handovers.

Allows users to temporarily delegate their permissions to other users
for a specified time period and scope.
"""

from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.permission_delegation import PermissionDelegation

logger = logging.getLogger(__name__)


def _serialize_delegation(d: PermissionDelegation) -> dict:
    return {
        "id": str(d.id),
        "from_user_id": str(d.from_user_id),
        "to_user_id": str(d.to_user_id),
        "start_at": d.start_at.isoformat() if d.start_at else None,
        "end_at": d.end_at.isoformat() if d.end_at else None,
        "scope": d.scope,
        "active": d.active,
        "tenant_id": str(d.tenant_id),
        "created_at": d.created_at.isoformat() if d.created_at else None,
        "updated_at": d.updated_at.isoformat() if d.updated_at else None,
    }


async def list_delegations(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    direction: str = "all",
) -> list[dict]:
    """List delegations for a tenant.

    Args:
        direction: 'from' (delegations I created), 'to' (delegations to me), 'all' (both)
    """
    query = select(PermissionDelegation).where(PermissionDelegation.tenant_id == tenant_id)

    if user_id:
        if direction == "from":
            query = query.where(PermissionDelegation.from_user_id == user_id)
        elif direction == "to":
            query = query.where(PermissionDelegation.to_user_id == user_id)
        else:
            query = query.where(
                or_(
                    PermissionDelegation.from_user_id == user_id,
                    PermissionDelegation.to_user_id == user_id,
                )
            )

    query = query.order_by(PermissionDelegation.created_at.desc())
    result = await db.execute(query)
    delegations = result.scalars().all()
    return [_serialize_delegation(d) for d in delegations]


async def create_delegation(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    from_user_id: uuid.UUID,
    to_user_id: uuid.UUID,
    start_at: datetime,
    end_at: datetime,
    scope: dict | None = None,
) -> dict:
    """Create a new permission delegation."""
    if end_at <= start_at:
        raise ValueError("end_at must be after start_at")

    delegation = PermissionDelegation(
        tenant_id=tenant_id,
        from_user_id=from_user_id,
        to_user_id=to_user_id,
        start_at=start_at,
        end_at=end_at,
        scope=scope,
    )
    db.add(delegation)
    await db.commit()
    await db.refresh(delegation)
    return _serialize_delegation(delegation)


async def update_delegation(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    delegation_id: str,
    **kwargs: Any,
) -> dict:
    """Update an existing delegation."""
    result = await db.execute(
        select(PermissionDelegation)
        .where(PermissionDelegation.id == uuid.UUID(delegation_id))
        .where(PermissionDelegation.tenant_id == tenant_id)
    )
    delegation = result.scalar_one_or_none()
    if delegation is None:
        raise ValueError(f"Delegation {delegation_id} not found")

    updatable_fields = {"start_at", "end_at", "scope", "active"}
    for key, value in kwargs.items():
        if key in updatable_fields and value is not None:
            setattr(delegation, key, value)

    await db.commit()
    await db.refresh(delegation)
    return _serialize_delegation(delegation)


async def delete_delegation(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    delegation_id: str,
) -> None:
    """Delete a delegation."""
    result = await db.execute(
        select(PermissionDelegation)
        .where(PermissionDelegation.id == uuid.UUID(delegation_id))
        .where(PermissionDelegation.tenant_id == tenant_id)
    )
    delegation = result.scalar_one_or_none()
    if delegation is None:
        raise ValueError(f"Delegation {delegation_id} not found")
    await db.delete(delegation)
    await db.commit()


async def is_delegation_active(
    db: AsyncSession,
    user_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> bool:
    """Check if a user has any active delegations (as delegatee).

    Returns True if there is at least one active delegation where
    this user is the to_user_id and the current time is within [start_at, end_at].
    """
    now = datetime.now(UTC)
    result = await db.execute(
        select(PermissionDelegation)
        .where(PermissionDelegation.to_user_id == user_id)
        .where(PermissionDelegation.tenant_id == tenant_id)
        .where(PermissionDelegation.active == True)  # noqa: E712
        .where(PermissionDelegation.start_at <= now)
        .where(PermissionDelegation.end_at > now)
    )
    delegation = result.scalar_one_or_none()
    return delegation is not None


async def get_active_delegations(
    db: AsyncSession,
    user_id: uuid.UUID,
    tenant_id: uuid.UUID,
) -> list[dict]:
    """Get all active delegations for a user (as delegatee)."""
    now = datetime.now(UTC)
    result = await db.execute(
        select(PermissionDelegation)
        .where(PermissionDelegation.to_user_id == user_id)
        .where(PermissionDelegation.tenant_id == tenant_id)
        .where(PermissionDelegation.active == True)  # noqa: E712
        .where(PermissionDelegation.start_at <= now)
        .where(PermissionDelegation.end_at > now)
    )
    delegations = result.scalars().all()
    return [_serialize_delegation(d) for d in delegations]


async def deactivate_expired_delegations(db: AsyncSession) -> int:
    """Deactivate all delegations that have passed their end_at."""
    now = datetime.now(UTC)
    result = await db.execute(
        select(PermissionDelegation)
        .where(PermissionDelegation.active == True)  # noqa: E712
        .where(PermissionDelegation.end_at <= now)
    )
    expired = result.scalars().all()
    count = len(expired)
    for delegation in expired:
        delegation.active = False
    if count > 0:
        await db.commit()
        logger.info("Deactivated %d expired delegations", count)
    return count