leocrm/app/services/user_service.py

"""User management service."""

from __future__ import annotations

import uuid
from typing import Any

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

from app.core.auth import hash_password
from app.models.user import User, UserTenant

# Sentinel used to distinguish "not provided" from "explicitly set to None".
# When ``role_id`` is _UNSET the service leaves the existing value untouched.
# When ``role_id`` is None the service clears the FK (falls back to role string).
_UNSET: Any = object()


class UserService:
    """Handles user CRUD operations.

    All queries are tenant-scoped through the UserTenant association table.
    User.email is globally unique; tenant membership and role live in UserTenant.
    """

    async def list_users(
        self,
        db: AsyncSession,
        tenant_id: uuid.UUID,
        page: int = 1,
        page_size: int = 25,
        search: str | None = None,
    ) -> dict[str, Any]:
        """List users in a tenant with pagination and search."""
        offset = (page - 1) * page_size

        base = (
            select(User, UserTenant)
            .join(UserTenant, UserTenant.user_id == User.id)
            .where(UserTenant.tenant_id == tenant_id)
        )
        count_q = (
            select(func.count())
            .select_from(UserTenant)
            .where(UserTenant.tenant_id == tenant_id)
        )

        if search:
            search_filter = or_(
                User.name.ilike(f"%{search}%"),
                User.email.ilike(f"%{search}%"),
            )
            base = base.where(search_filter)
            count_q = count_q.join(User, User.id == UserTenant.user_id).where(search_filter)

        total = (await db.execute(count_q)).scalar() or 0

        q = base.offset(offset).limit(page_size).order_by(User.created_at.desc())
        result = await db.execute(q)
        rows = result.all()

        return {
            "items": [self._user_to_dict(u, ut) for u, ut in rows],
            "total": total,
            "page": page,
            "page_size": page_size,
        }

    async def get_user(
        self,
        db: AsyncSession,
        tenant_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> tuple[User, UserTenant] | None:
        """Get a single user by ID within tenant scope.

        Returns (User, UserTenant) tuple or None.
        """
        q = (
            select(User, UserTenant)
            .join(UserTenant, UserTenant.user_id == User.id)
            .where(User.id == user_id, UserTenant.tenant_id == tenant_id)
        )
        result = await db.execute(q)
        row = result.first()
        if row is None:
            return None
        return row[0], row[1]

    async def create_user(
        self,
        db: AsyncSession,
        tenant_id: uuid.UUID,
        email: str,
        name: str,
        password: str,
        role: str = "viewer",
        role_id: uuid.UUID | None = None,
        is_active: bool = True,
    ) -> User:
        """Create a new user and add them to the specified tenant.

        If role_id is provided it links the UserTenant to a custom Role record.
        The ``role`` string is the built-in role (admin/editor/viewer).
        """

        # ── Hook: user.before_create (Action) ──
        from app.core.hooks import do_action
        await do_action("user.before_create", email=email, name=name, role=role, tenant_id=tenant_id)

        user = User(
            email=email,
            name=name,
            password_hash=hash_password(password),
            is_active=is_active,
            preferences={},
        )
        db.add(user)
        await db.flush()

        # Add user-tenant membership with role
        ut = UserTenant(
            user_id=user.id,
            tenant_id=tenant_id,
            is_default=True,
            role=role,
            role_id=role_id,
        )
        db.add(ut)
        await db.flush()

        # ── Hook: user.after_create (Action) ──
        from app.core.hooks import do_action
        await do_action("user.after_create", user_id=str(user.id), email=email, name=name, role=role, tenant_id=tenant_id)

        return user

    async def update_user(
        self,
        db: AsyncSession,
        tenant_id: uuid.UUID,
        user_id: uuid.UUID,
        name: str | None = None,
        role: str | None = None,
        role_id: uuid.UUID | None | Any = _UNSET,
        is_active: bool | None = None,
        first_name: str | None = None,
        last_name: str | None = None,
        avatar_url: str | None = None,
        email: str | None = None,
        current_password: str | None = None,
        new_password: str | None = None,
    ) -> tuple[User, UserTenant] | None:
        """Update a user and their tenant membership.

        ``role_id`` uses a sentinel to distinguish three states:
        - ``_UNSET`` (default): leave the existing role_id unchanged
        - ``None``: clear the FK (fall back to the built-in ``role`` string)
        - ``uuid.UUID``: link to a custom Role record

        Returns (User, UserTenant) tuple or None if not found.
        """
        q = (
            select(User, UserTenant)
            .join(UserTenant, UserTenant.user_id == User.id)
            .where(User.id == user_id, UserTenant.tenant_id == tenant_id)
        )
        result = await db.execute(q)
        row = result.first()
        if row is None:
            return None

        user, user_tenant = row[0], row[1]

        if name is not None:
            user.name = name
        if role is not None:
            user_tenant.role = role
        if role_id is not _UNSET:
            user_tenant.role_id = role_id
        if is_active is not None:
            user.is_active = is_active
        if first_name is not None:
            user.first_name = first_name
        if last_name is not None:
            user.last_name = last_name
        if avatar_url is not None:
            user.avatar_url = avatar_url
        if email is not None:
            user.email = email
        if new_password is not None and current_password is not None:
            # Verify current password
            from app.core.auth import verify_password
            if not verify_password(current_password, user.password_hash):
                raise ValueError("Current password is incorrect")
            from app.core.auth import hash_password
            user.password_hash = hash_password(new_password)

        await db.flush()
        return user, user_tenant

    async def delete_user(
        self,
        db: AsyncSession,
        tenant_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> bool:
        """Remove a user from a tenant (delete UserTenant membership).

        If this is the user's only tenant membership, the User record is
        also deleted. Otherwise only the UserTenant row is removed.
        """
        ut_q = select(UserTenant).where(
            UserTenant.user_id == user_id,
            UserTenant.tenant_id == tenant_id,
        )
        ut_result = await db.execute(ut_q)
        user_tenant = ut_result.scalar_one_or_none()
        if user_tenant is None:
            return False

        # Count total tenant memberships for this user
        count_q = select(func.count()).select_from(UserTenant).where(
            UserTenant.user_id == user_id
        )
        count_result = await db.execute(count_q)
        membership_count = count_result.scalar() or 0

        await db.delete(user_tenant)

        if membership_count <= 1:
            # User's only tenant — delete the User record too
            user_q = select(User).where(User.id == user_id)
            user_result = await db.execute(user_q)
            user = user_result.scalar_one_or_none()
            if user is not None:
                await db.delete(user)

        await db.flush()
        return True

    def _user_to_dict(
        self, user: User, user_tenant: UserTenant | None = None
    ) -> dict[str, Any]:
        """Convert user + user_tenant to response dict."""
        result: dict[str, Any] = {
            "id": str(user.id),
            "email": user.email,
            "name": user.name,
            "is_active": user.is_active,
        }
        if user_tenant is not None:
            result["role"] = user_tenant.role
            result["role_id"] = str(user_tenant.role_id) if user_tenant.role_id else None
            result["tenant_id"] = str(user_tenant.tenant_id)
        else:
            result["role"] = "viewer"
            result["role_id"] = None
            result["tenant_id"] = None
        return result


user_service = UserService()