# -*- coding: utf-8 -*-
"""Location: ./mcpgateway/services/team_management_service.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

Team Management Service.
This module provides team creation, management, and membership operations
for the multi-team collaboration system.

Examples:
    >>> from unittest.mock import Mock
    >>> service = TeamManagementService(Mock())
    >>> isinstance(service, TeamManagementService)
    True
    >>> hasattr(service, 'db')
    True
"""

# Standard
import asyncio
import base64
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple, Union

# Third-Party
import orjson
from sqlalchemy import and_, desc, func, or_, select
from sqlalchemy.orm import selectinload, Session

# First-Party
from mcpgateway.cache.admin_stats_cache import admin_stats_cache
from mcpgateway.cache.auth_cache import auth_cache, get_auth_cache
from mcpgateway.common.validators import SecurityValidator
from mcpgateway.config import settings
from mcpgateway.db import EmailTeam, EmailTeamJoinRequest, EmailTeamMember, EmailTeamMemberHistory, EmailUser, utc_now
from mcpgateway.services.logging_service import LoggingService
from mcpgateway.utils.create_slug import slugify
from mcpgateway.utils.pagination import unified_paginate
from mcpgateway.utils.redis_client import get_redis_client

# Initialize logging
logging_service = LoggingService()
logger = logging_service.get_logger(__name__)


def get_user_team_count(db: Session, user_email: str) -> int:
    """Get the number of active teams a user belongs to.

    Args:
        db: SQLAlchemy database session
        user_email: Email address of the user

    Returns:
        int: Number of active team memberships
    """
    return db.query(EmailTeamMember).filter(EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)).count()


class TeamManagementError(Exception):
    """Base class for team management-related errors.

    Examples:
        >>> error = TeamManagementError("Test error")
        >>> str(error)
        'Test error'
        >>> isinstance(error, Exception)
        True
    """


class InvalidRoleError(TeamManagementError):
    """Raised when an invalid role is specified.

    Examples:
        >>> error = InvalidRoleError("Invalid role: guest")
        >>> str(error)
        'Invalid role: guest'
        >>> isinstance(error, TeamManagementError)
        True
    """


class TeamNotFoundError(TeamManagementError):
    """Raised when a team does not exist.

    Examples:
        >>> error = TeamNotFoundError("Team not found: team-123")
        >>> str(error)
        'Team not found: team-123'
        >>> isinstance(error, TeamManagementError)
        True
    """


class UserNotFoundError(TeamManagementError):
    """Raised when a user does not exist.

    Examples:
        >>> error = UserNotFoundError("User not found: user@example.com")
        >>> str(error)
        'User not found: user@example.com'
        >>> isinstance(error, TeamManagementError)
        True
    """


class MemberAlreadyExistsError(TeamManagementError):
    """Raised when a user is already a member of the team.

    Examples:
        >>> error = MemberAlreadyExistsError("User user@example.com is already a member of team team-123")
        >>> str(error)
        'User user@example.com is already a member of team team-123'
        >>> isinstance(error, TeamManagementError)
        True
    """


class TeamMemberLimitExceededError(TeamManagementError):
    """Raised when a team has reached its maximum member limit.

    Examples:
        >>> error = TeamMemberLimitExceededError("Team has reached maximum member limit of 10")
        >>> str(error)
        'Team has reached maximum member limit of 10'
        >>> isinstance(error, TeamManagementError)
        True
    """


class TeamMemberAddError(TeamManagementError):
    """Raised when adding a member to a team fails due to database or system errors.

    Examples:
        >>> error = TeamMemberAddError("Failed to add member due to database error")
        >>> str(error)
        'Failed to add member due to database error'
        >>> isinstance(error, TeamManagementError)
        True
    """


class _Unset:
    """Sentinel type: distinguishes 'caller omitted the argument' from 'caller passed None'."""

    __slots__ = ()

    def __repr__(self) -> str:
        """Return string representation of the unset sentinel.

        Returns:
            str: The literal string ``"UNSET"``.
        """
        return "UNSET"

    def __bool__(self) -> bool:
        """Make the sentinel falsy so truthiness checks treat it as not provided.

        Returns:
            bool: Always ``False``.
        """
        return False


UNSET = _Unset()


def get_effective_max_members(team: "EmailTeam") -> int:
    """Return the effective member limit for a team.

    If the team has an explicit ``max_members`` value stored in the DB, that
    value is used.  Otherwise the global ``MAX_MEMBERS_PER_TEAM`` setting is
    returned so that changing the environment variable takes effect for all
    teams that have not been individually overridden.

    Args:
        team: The team whose effective member limit should be resolved.

    Returns:
        The member limit (always an int; 0 means unlimited).
    """
    if team.max_members is not None:
        return team.max_members
    return settings.max_members_per_team


def check_team_member_capacity(db: "Session", team: "EmailTeam", *, extra_count: int = 0) -> None:
    """Raise if the team is at or over its member capacity.

    Args:
        db: Database session for querying active member count.
        team: The team to check.
        extra_count: Additional slots to reserve (e.g. pending invitations).

    Raises:
        TeamMemberLimitExceededError: If the team is at capacity.
    """
    effective_max = get_effective_max_members(team)
    if effective_max <= 0:
        return
    member_count = db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team.id, EmailTeamMember.is_active.is_(True)).count()
    if (member_count + extra_count) >= effective_max:
        raise TeamMemberLimitExceededError(f"Team has reached maximum member limit of {effective_max}")


class TeamManagementService:
    """Service for team management operations.

    This service handles team creation, membership management,
    role assignments, and team access control.

    Attributes:
        db (Session): SQLAlchemy database session

    Examples:
        >>> from unittest.mock import Mock
        >>> service = TeamManagementService(Mock())
        >>> service.__class__.__name__
        'TeamManagementService'
        >>> hasattr(service, 'db')
        True
    """

    def __init__(self, db: Session):
        """Initialize the team management service.

        Args:
            db: SQLAlchemy database session

        Examples:
            Basic initialization:
            >>> from mcpgateway.services.team_management_service import TeamManagementService
            >>> from unittest.mock import Mock
            >>> db_session = Mock()
            >>> service = TeamManagementService(db_session)
            >>> service.db is db_session
            True

            Service attributes:
            >>> hasattr(service, 'db')
            True
            >>> service.__class__.__name__
            'TeamManagementService'
        """
        self.db = db
        self._role_service = None  # Lazy initialization to avoid circular imports

    @property
    def role_service(self):
        """Lazy-initialized RoleService to avoid circular imports.

        Returns:
            RoleService: Instance of RoleService
        """
        if self._role_service is None:
            # First-Party
            from mcpgateway.services.role_service import RoleService  # pylint: disable=import-outside-toplevel

            self._role_service = RoleService(self.db)
        return self._role_service

    def _get_user_team_count(self, user_email: str) -> int:
        """Get the number of active teams a user belongs to.

        Args:
            user_email: Email address of the user

        Returns:
            int: Number of active team memberships
        """
        return get_user_team_count(self.db, user_email)

    @staticmethod
    def _get_rbac_role_name(membership_role: str) -> str:
        """Map a team membership role to the corresponding configurable RBAC role name.

        Args:
            membership_role: Team membership role ('owner' or 'member').

        Returns:
            str: The configured RBAC role name from settings.
        """
        return settings.default_team_owner_role if membership_role == "owner" else settings.default_team_member_role

    @staticmethod
    def _fire_and_forget(coro: Any) -> None:
        """Schedule a background coroutine and close it if scheduling fails.

        Args:
            coro: The coroutine to schedule as a background task.

        Raises:
            Exception: If asyncio.create_task fails (e.g. no running loop).
        """
        try:
            task = asyncio.create_task(coro)
            # Some tests patch create_task with a plain Mock return value. In that
            # case the coroutine is never actually scheduled and must be closed.
            if asyncio.iscoroutine(coro) and not isinstance(task, asyncio.Task):
                close = getattr(coro, "close", None)
                if callable(close):
                    close()
        except Exception:
            # If create_task() fails (e.g. no running loop), the coroutine has
            # already been created and must be closed to avoid runtime warnings.
            close = getattr(coro, "close", None)
            if callable(close):
                close()
            raise

    def _log_team_member_action(self, team_member_id: str, team_id: str, user_email: str, role: str, action: str, action_by: Optional[str]):
        """
        Log a team member action to EmailTeamMemberHistory.

        Args:
            team_member_id: ID of the EmailTeamMember
            team_id: Team ID
            user_email: Email of the affected user
            role: Role at the time of action
            action: Action type ("added", "removed", "reactivated", "role_changed")
            action_by: Email of the user who performed the action

        Examples:
            >>> from mcpgateway.services.team_management_service import TeamManagementService
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> service._log_team_member_action("tm-123", "team-123", "user@example.com", "member", "added", "admin@example.com")
        """
        history = EmailTeamMemberHistory(team_member_id=team_member_id, team_id=team_id, user_email=user_email, role=role, action=action, action_by=action_by, action_timestamp=utc_now())
        self.db.add(history)
        self.db.commit()

    async def _assign_team_rbac_role(self, user_email: str, team_id: str, membership_role: str, granted_by: Optional[str] = None) -> None:
        """Assign a team-scoped RBAC role matching the membership role.

        Looks up the configured RBAC role name for *membership_role* (owner
        or member) and assigns it if the user does not already have an
        active assignment.  Errors are logged but do not propagate.

        Args:
            user_email: Email of the user to assign the role to.
            team_id: Team ID for the scoped assignment.
            membership_role: Team membership role (``"owner"`` or ``"member"``).
            granted_by: Email of the user who triggered the assignment.
        """
        try:
            rbac_role_name = self._get_rbac_role_name(membership_role)
            team_rbac_role = await self.role_service.get_role_by_name(rbac_role_name, scope="team")
            if team_rbac_role:
                existing = await self.role_service.get_user_role_assignment(user_email=user_email, role_id=team_rbac_role.id, scope="team", scope_id=team_id)
                if not existing or not existing.is_active:
                    await self.role_service.assign_role_to_user(user_email=user_email, role_id=team_rbac_role.id, scope="team", scope_id=team_id, granted_by=granted_by or user_email)
                    logger.info(f"Assigned {rbac_role_name} role to {SecurityValidator.sanitize_log_message(user_email)} for team {SecurityValidator.sanitize_log_message(team_id)}")
                else:
                    logger.debug(f"User {SecurityValidator.sanitize_log_message(user_email)} already has active {rbac_role_name} role for team {SecurityValidator.sanitize_log_message(team_id)}")
            else:
                logger.warning(f"Role '{rbac_role_name}' not found. User {SecurityValidator.sanitize_log_message(user_email)} added without RBAC role.")
        except Exception as role_error:
            logger.warning(f"Failed to assign role to {SecurityValidator.sanitize_log_message(user_email)}: {role_error}")

    def _invalidate_membership_caches(self, user_email: str, team_id: str, *, include_admin_stats: bool = True) -> None:
        """Invalidate auth and admin caches after a membership change.

        Errors are logged at debug level but do not propagate.

        Args:
            user_email: Email of the affected user.
            team_id: Team ID whose caches should be invalidated.
            include_admin_stats: Whether to also invalidate admin stats (skip on removal).
        """
        try:
            self._fire_and_forget(auth_cache.invalidate_team(user_email))
            self._fire_and_forget(auth_cache.invalidate_user_role(user_email, team_id))
            self._fire_and_forget(auth_cache.invalidate_user_teams(user_email))
            self._fire_and_forget(auth_cache.invalidate_team_membership(user_email))
            if include_admin_stats:
                self._fire_and_forget(admin_stats_cache.invalidate_teams())
        except Exception as cache_error:
            logger.debug(f"Failed to invalidate membership caches for {SecurityValidator.sanitize_log_message(user_email)}: {cache_error}")

    def _check_user_team_limit(self, user_email: str) -> None:
        """Raise if the user has reached the maximum team membership limit.

        Args:
            user_email: Email of the user to check.

        Raises:
            TeamManagementError: If the user is at the configured limit.
        """
        max_teams = getattr(settings, "max_teams_per_user", 50)
        if self._get_user_team_count(user_email) >= max_teams:
            raise TeamManagementError(f"User has reached the maximum team limit of {max_teams}")

    async def create_team(
        self, name: str, description: Optional[str], created_by: str, visibility: Optional[str] = "public", max_members: Optional[int] = None, skip_limits: bool = False
    ) -> EmailTeam:
        """Create a new team.

        Args:
            name: Team name
            description: Team description
            created_by: Email of the user creating the team
            visibility: Team visibility (private, team, public)
            max_members: Maximum number of team members allowed
            skip_limits: Skip max_teams_per_user check (for admin bypass)

        Returns:
            EmailTeam: The created team

        Raises:
            ValueError: If team name is taken or invalid
            Exception: If team creation fails

        Examples:
            Team creation parameter validation:
            >>> from mcpgateway.services.team_management_service import TeamManagementService

            Test team name validation:
            >>> team_name = "My Development Team"
            >>> len(team_name) > 0
            True
            >>> len(team_name) <= 255
            True
            >>> bool(team_name.strip())
            True

            Test visibility validation:
            >>> visibility = "private"
            >>> valid_visibilities = ["private", "public"]
            >>> visibility in valid_visibilities
            True
            >>> "invalid" in valid_visibilities
            False

            Test max_members validation:
            >>> max_members = 50
            >>> isinstance(max_members, int)
            True
            >>> max_members > 0
            True

            Test creator validation:
            >>> created_by = "admin@example.com"
            >>> "@" in created_by
            True
            >>> len(created_by) > 0
            True

            Test description handling:
            >>> description = "A team for software development"
            >>> description is not None
            True
            >>> isinstance(description, str)
            True

            >>> # Test None description
            >>> description_none = None
            >>> description_none is None
            True
        """
        try:
            # Validate visibility
            valid_visibilities = ["private", "public"]
            if visibility not in valid_visibilities:
                raise ValueError(f"Invalid visibility. Must be one of: {', '.join(valid_visibilities)}")

            # Check max teams per user
            if not skip_limits:
                max_teams = getattr(settings, "max_teams_per_user", 50)
                if self._get_user_team_count(created_by) >= max_teams:
                    raise ValueError(f"User has reached the maximum team limit of {max_teams}")

            # Enforce max_members cap for non-admins (only when explicitly provided)
            if not skip_limits and max_members is not None:
                max_limit = settings.max_members_per_team
                if max_members > max_limit:
                    raise ValueError(f"max_members cannot exceed the configured limit of {max_limit}")

            # If max_members is not explicitly provided, leave it as None in the DB.
            # The effective limit will be resolved at check time from settings.max_members_per_team,
            # so changing the env var affects all teams that don't have an explicit override.

            # Check for existing inactive team with same name

            potential_slug = slugify(name)
            existing_inactive_team = self.db.query(EmailTeam).filter(EmailTeam.slug == potential_slug, EmailTeam.is_active.is_(False)).first()

            if existing_inactive_team:
                # Reactivate the existing team with new details
                existing_inactive_team.name = name
                existing_inactive_team.description = description
                existing_inactive_team.created_by = created_by
                existing_inactive_team.visibility = visibility
                existing_inactive_team.max_members = max_members
                existing_inactive_team.is_active = True
                existing_inactive_team.updated_at = utc_now()
                team = existing_inactive_team

                # Check if the creator already has an inactive membership
                existing_membership = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team.id, EmailTeamMember.user_email == created_by).first()

                if existing_membership:
                    # Reactivate existing membership as owner
                    existing_membership.role = "owner"
                    existing_membership.joined_at = utc_now()
                    existing_membership.is_active = True
                    membership = existing_membership
                else:
                    # Create new membership
                    membership = EmailTeamMember(team_id=team.id, user_email=created_by, role="owner", joined_at=utc_now(), is_active=True)
                    self.db.add(membership)

                logger.info(f"Reactivated existing team with slug {potential_slug}")
            else:
                # Create the team (slug will be auto-generated by event listener)
                team = EmailTeam(name=name, description=description, created_by=created_by, is_personal=False, visibility=visibility, max_members=max_members, is_active=True)
                self.db.add(team)

                self.db.flush()  # Get the team ID

                # Add the creator as owner
                membership = EmailTeamMember(team_id=team.id, user_email=created_by, role="owner", joined_at=utc_now(), is_active=True)
                self.db.add(membership)

            self.db.commit()

            # Invalidate member count cache for the new team
            await self.invalidate_team_member_count_cache(str(team.id))

            # Invalidate auth cache for creator's team membership
            # Without this, the cache won't know the user belongs to this new team
            try:
                await auth_cache.invalidate_user_teams(created_by)
                await auth_cache.invalidate_team_membership(created_by)
                await auth_cache.invalidate_user_role(created_by, str(team.id))
                await admin_stats_cache.invalidate_teams()
            except Exception as cache_error:
                logger.debug(f"Failed to invalidate cache on team create: {cache_error}")

            logger.info(f"Created team '{SecurityValidator.sanitize_log_message(team.name)}' by {created_by}")
            return team

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to create team '{name}': {e}")
            raise

    async def get_team_by_id(self, team_id: str) -> Optional[EmailTeam]:
        """Get a team by ID.

        Args:
            team_id: Team ID to lookup

        Returns:
            EmailTeam: The team or None if not found

        Examples:
            >>> import asyncio
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> asyncio.iscoroutinefunction(service.get_team_by_id)
            True
        """
        try:
            team = self.db.query(EmailTeam).filter(EmailTeam.id == team_id, EmailTeam.is_active.is_(True)).first()
            self.db.commit()  # Release transaction to avoid idle-in-transaction
            return team

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get team by ID {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return None

    async def get_team_by_slug(self, slug: str) -> Optional[EmailTeam]:
        """Get a team by slug.

        Args:
            slug: Team slug to lookup

        Returns:
            EmailTeam: The team or None if not found

        Examples:
            >>> import asyncio
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> asyncio.iscoroutinefunction(service.get_team_by_slug)
            True
        """
        try:
            team = self.db.query(EmailTeam).filter(EmailTeam.slug == slug, EmailTeam.is_active.is_(True)).first()
            self.db.commit()  # Release transaction to avoid idle-in-transaction
            return team

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get team by slug {slug}: {e}")
            return None

    async def update_team(
        self,
        team_id: str,
        name: Optional[str] = None,
        description: Optional[str] = None,
        visibility: Optional[str] = None,
        max_members: Union[int, None, _Unset] = UNSET,
        updated_by: Optional[str] = None,
        skip_limits: bool = False,
    ) -> bool:
        """Update team information.

        Args:
            team_id: ID of the team to update
            name: New team name
            description: New team description
            visibility: New visibility setting
            max_members: New maximum member limit. Pass ``None`` to clear an
                explicit per-team override (reverts to global default).  Omit
                (or pass ``UNSET``) to leave the current value unchanged.
            updated_by: Email of user making the update
            skip_limits: Skip the max_members_per_team cap check (platform admins only)

        Returns:
            bool: True if update succeeded, False otherwise

        Raises:
            ValueError: If visibility setting is invalid

        Examples:
            >>> import asyncio
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> asyncio.iscoroutinefunction(service.update_team)
            True
        """
        try:
            team = await self.get_team_by_id(team_id)
            if not team:
                logger.warning(f"Team {SecurityValidator.sanitize_log_message(team_id)} not found for update")
                return False

            # Prevent updating personal teams
            if team.is_personal:
                logger.warning(f"Cannot update personal team {SecurityValidator.sanitize_log_message(team_id)}")
                return False

            # Update fields if provided
            if name is not None:
                team.name = name
                # Slug will be updated by event listener if name changes

            if description is not None:
                team.description = description

            if visibility is not None:
                valid_visibilities = ["private", "public"]
                if visibility not in valid_visibilities:
                    raise ValueError(f"Invalid visibility. Must be one of: {', '.join(valid_visibilities)}")
                team.visibility = visibility

            # UNSET means "not provided" — leave unchanged.
            # None means "explicitly clear the per-team override" — store NULL.
            # An int means "set an explicit per-team limit".
            if max_members is not UNSET:
                if max_members is not None and not skip_limits:
                    max_limit = settings.max_members_per_team
                    if max_members > max_limit:
                        raise ValueError(f"max_members cannot exceed the configured limit of {max_limit}")
                team.max_members = max_members

            team.updated_at = utc_now()
            self.db.commit()

            logger.info(f"Updated team {SecurityValidator.sanitize_log_message(team_id)} by {updated_by}")
            return True

        except ValueError:
            raise  # Let ValueError propagate to caller for proper error handling
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to update team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return False

    async def delete_team(self, team_id: str, deleted_by: str) -> bool:
        """Delete a team (soft delete).

        Args:
            team_id: ID of the team to delete
            deleted_by: Email of user performing deletion

        Returns:
            bool: True if deletion succeeded, False otherwise

        Raises:
            ValueError: If attempting to delete a personal team

        Examples:
            >>> import asyncio
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> asyncio.iscoroutinefunction(service.delete_team)
            True
        """
        try:
            team = await self.get_team_by_id(team_id)
            if not team:
                logger.warning(f"Team {SecurityValidator.sanitize_log_message(team_id)} not found for deletion")
                return False

            # Prevent deleting personal teams
            if team.is_personal:
                logger.warning(f"Cannot delete personal team {SecurityValidator.sanitize_log_message(team_id)}")
                raise ValueError("Personal teams cannot be deleted")

            # Soft delete the team
            team.is_active = False
            team.updated_at = utc_now()

            # Get all active memberships before deactivating (for history logging)
            memberships = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.is_active.is_(True)).all()

            # Log history for each membership (before bulk update)
            for membership in memberships:
                self._log_team_member_action(membership.id, team_id, membership.user_email, membership.role, "team-deleted", deleted_by)

            # Bulk update: deactivate all memberships in single query instead of looping
            self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.is_active.is_(True)).update({EmailTeamMember.is_active: False}, synchronize_session=False)

            self.db.commit()

            # Invalidate all role caches for this team
            try:
                self._fire_and_forget(auth_cache.invalidate_team_roles(team_id))
                self._fire_and_forget(admin_stats_cache.invalidate_teams())
                # Also invalidate team cache, teams list cache, and team membership cache for each member
                for membership in memberships:
                    self._fire_and_forget(auth_cache.invalidate_team(membership.user_email))
                    self._fire_and_forget(auth_cache.invalidate_user_teams(membership.user_email))
                    self._fire_and_forget(auth_cache.invalidate_team_membership(membership.user_email))
            except Exception as cache_error:
                logger.debug(f"Failed to invalidate caches on team delete: {cache_error}")

            logger.info(f"Deleted team {SecurityValidator.sanitize_log_message(team_id)} by {deleted_by}")
            return True

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to delete team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return False

    async def add_member_to_team(self, team_id: str, user_email: str, role: str = "member", invited_by: Optional[str] = None, grant_source: Optional[str] = None) -> EmailTeamMember:
        """Add a member to a team.

        Args:
            team_id: ID of the team
            user_email: Email of the user to add
            role: Role to assign (owner, member)
            invited_by: Email of user who added this member
            grant_source: Origin of grant (e.g., 'sso', 'manual', 'bootstrap', 'auto')

        Returns:
            EmailTeamMember: The created or reactivated team member object

        Raises:
            InvalidRoleError: If role is invalid
            TeamNotFoundError: If team does not exist
            TeamManagementError: If team is a personal team
            UserNotFoundError: If user does not exist
            MemberAlreadyExistsError: If user is already a member
            TeamMemberLimitExceededError: If team has reached maximum member limit
            TeamMemberAddError: If adding member fails due to database or system errors

        Examples:
            >>> import asyncio
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> asyncio.iscoroutinefunction(service.add_member_to_team)
            True
            >>> # After adding, EmailTeamMemberHistory is updated
            >>> # service._log_team_member_action("tm-123", "team-123", "user@example.com", "member", "added", "admin@example.com")
        """
        # Validate role
        valid_roles = ["owner", "member"]
        if role not in valid_roles:
            raise InvalidRoleError(f"Invalid role '{role}'. Must be one of: {', '.join(valid_roles)}")

        # Check if team exists
        team = await self.get_team_by_id(team_id)
        if not team:
            logger.warning(f"Team {SecurityValidator.sanitize_log_message(team_id)} not found")
            raise TeamNotFoundError("Team not found")

        # Prevent adding members to personal teams
        if team.is_personal:
            logger.warning(f"Cannot add members to personal team {SecurityValidator.sanitize_log_message(team_id)}")
            raise TeamManagementError("Cannot add members to personal teams")

        # Check if user exists
        user = self.db.query(EmailUser).filter(EmailUser.email == user_email).first()
        if not user:
            logger.warning(f"User {SecurityValidator.sanitize_log_message(user_email)} not found")
            raise UserNotFoundError("User not found")

        self._check_user_team_limit(user_email)

        # Check if user is already a member
        existing_membership = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email).first()

        if existing_membership and existing_membership.is_active:
            logger.warning(f"User {SecurityValidator.sanitize_log_message(user_email)} is already a member of team {SecurityValidator.sanitize_log_message(team_id)}")
            raise MemberAlreadyExistsError("User is already a member of this team")

        # Check team member limit (explicit per-team value or global default)
        check_team_member_capacity(self.db, team)

        # Add or reactivate membership
        try:
            if existing_membership:
                existing_membership.is_active = True
                existing_membership.role = role
                existing_membership.joined_at = utc_now()
                existing_membership.invited_by = invited_by
                if grant_source is not None:
                    existing_membership.grant_source = grant_source
                self.db.commit()
                self._log_team_member_action(existing_membership.id, team_id, user_email, role, "reactivated", invited_by)
                member = existing_membership
            else:
                membership = EmailTeamMember(team_id=team_id, user_email=user_email, role=role, joined_at=utc_now(), invited_by=invited_by, grant_source=grant_source, is_active=True)
                self.db.add(membership)
                self.db.commit()
                self._log_team_member_action(membership.id, team_id, user_email, role, "added", invited_by)
                member = membership

            await self._assign_team_rbac_role(user_email, team_id, role, granted_by=invited_by)
            self._invalidate_membership_caches(user_email, team_id)
            await self.invalidate_team_member_count_cache(str(team_id))

            logger.info(f"Added {SecurityValidator.sanitize_log_message(user_email)} to team {SecurityValidator.sanitize_log_message(team_id)} with role {role}")
            return member

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to add {SecurityValidator.sanitize_log_message(user_email)} to team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            raise TeamMemberAddError("Failed to add member to team") from e

    async def remove_member_from_team(self, team_id: str, user_email: str, removed_by: Optional[str] = None) -> bool:
        """Remove a member from a team.

        Args:
            team_id: ID of the team
            user_email: Email of the user to remove
            removed_by: Email of user performing the removal

        Returns:
            bool: True if member was removed successfully, False otherwise

        Raises:
            ValueError: If attempting to remove the last owner

        Examples:
            Team membership management with role-based access control.
            After removal, EmailTeamMemberHistory is updated via _log_team_member_action.
        """
        try:
            team = await self.get_team_by_id(team_id)
            if not team:
                logger.warning(f"Team {SecurityValidator.sanitize_log_message(team_id)} not found")
                return False

            # Prevent removing members from personal teams
            if team.is_personal:
                logger.warning(f"Cannot remove members from personal team {SecurityValidator.sanitize_log_message(team_id)}")
                return False

            # Find the membership
            membership = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)).first()

            if not membership:
                logger.warning(f"User {SecurityValidator.sanitize_log_message(user_email)} is not a member of team {SecurityValidator.sanitize_log_message(team_id)}")
                return False

            # Prevent removing the last owner
            if membership.role == "owner":
                owner_count = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.role == "owner", EmailTeamMember.is_active.is_(True)).count()

                if owner_count <= 1:
                    logger.warning(f"Cannot remove the last owner from team {SecurityValidator.sanitize_log_message(team_id)}")
                    raise ValueError("Cannot remove the last owner from a team")

            # Remove membership (soft delete)
            membership.is_active = False
            self.db.commit()
            self._log_team_member_action(membership.id, team_id, user_email, membership.role, "removed", removed_by)

            # Revoke all team-scoped RBAC roles from removed member defensively
            # (revoke both owner and member roles to handle edge cases)
            try:
                for role_name in (settings.default_team_owner_role, settings.default_team_member_role):
                    rbac_role = await self.role_service.get_role_by_name(role_name, scope="team")
                    if rbac_role:
                        revoked = await self.role_service.revoke_role_from_user(user_email=user_email, role_id=rbac_role.id, scope="team", scope_id=team_id)
                        if revoked:
                            logger.info(f"Revoked {role_name} role from {SecurityValidator.sanitize_log_message(user_email)} for team {SecurityValidator.sanitize_log_message(team_id)}")
            except Exception as role_error:
                logger.warning(f"Failed to revoke roles from {SecurityValidator.sanitize_log_message(user_email)}: {role_error}")

            self._invalidate_membership_caches(user_email, team_id, include_admin_stats=False)
            await self.invalidate_team_member_count_cache(str(team_id))

            logger.info(f"Removed {SecurityValidator.sanitize_log_message(user_email)} from team {SecurityValidator.sanitize_log_message(team_id)} by {removed_by}")
            return True

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to remove {SecurityValidator.sanitize_log_message(user_email)} from team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return False

    async def update_member_role(self, team_id: str, user_email: str, new_role: str, updated_by: Optional[str] = None) -> bool:
        """Update a team member's role.

        Args:
            team_id: ID of the team
            user_email: Email of the user whose role to update
            new_role: New role to assign
            updated_by: Email of user making the change

        Returns:
            bool: True if role was updated successfully, False otherwise

        Raises:
            ValueError: If role is invalid or removing last owner role

        Examples:
            Role management within teams for access control.
            After role update, EmailTeamMemberHistory is updated via _log_team_member_action.
        """
        try:
            # Validate role
            valid_roles = ["owner", "member"]
            if new_role not in valid_roles:
                raise ValueError(f"Invalid role. Must be one of: {', '.join(valid_roles)}")

            team = await self.get_team_by_id(team_id)
            if not team:
                logger.warning(f"Team {SecurityValidator.sanitize_log_message(team_id)} not found")
                return False

            # Prevent updating roles in personal teams
            if team.is_personal:
                logger.warning(f"Cannot update roles in personal team {SecurityValidator.sanitize_log_message(team_id)}")
                return False

            # Find the membership
            membership = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)).first()

            if not membership:
                logger.warning(f"User {SecurityValidator.sanitize_log_message(user_email)} is not a member of team {SecurityValidator.sanitize_log_message(team_id)}")
                return False

            # Prevent changing the role of the last owner to non-owner
            if membership.role == "owner" and new_role != "owner":
                owner_count = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.role == "owner", EmailTeamMember.is_active.is_(True)).count()

                if owner_count <= 1:
                    logger.warning(f"Cannot remove owner role from the last owner of team {SecurityValidator.sanitize_log_message(team_id)}")
                    raise ValueError("Cannot remove owner role from the last owner of a team")

            # Update the role
            old_role = membership.role
            membership.role = new_role
            self.db.commit()
            self._log_team_member_action(membership.id, team_id, user_email, new_role, "role_changed", updated_by)

            # Handle RBAC role changes when team membership role changes
            if old_role != new_role:
                try:
                    # Get both role types
                    team_member_role = await self.role_service.get_role_by_name(settings.default_team_member_role, scope="team")
                    team_owner_role = await self.role_service.get_role_by_name(settings.default_team_owner_role, scope="team")

                    # Handle role transitions
                    if old_role == "member" and new_role == "owner":
                        # member -> owner: revoke member role, assign owner role
                        if team_member_role:
                            await self.role_service.revoke_role_from_user(user_email=user_email, role_id=team_member_role.id, scope="team", scope_id=team_id)
                        if team_owner_role:
                            await self.role_service.assign_role_to_user(user_email=user_email, role_id=team_owner_role.id, scope="team", scope_id=team_id, granted_by=updated_by or user_email)
                        logger.info(
                            f"Transitioned RBAC role from {settings.default_team_member_role} to {settings.default_team_owner_role} for {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}"
                        )

                    elif old_role == "owner" and new_role == "member":
                        # owner -> member: revoke owner role, assign member role
                        if team_owner_role:
                            await self.role_service.revoke_role_from_user(user_email=user_email, role_id=team_owner_role.id, scope="team", scope_id=team_id)
                        if team_member_role:
                            await self.role_service.assign_role_to_user(user_email=user_email, role_id=team_member_role.id, scope="team", scope_id=team_id, granted_by=updated_by or user_email)
                        logger.info(
                            f"Transitioned RBAC role from {settings.default_team_owner_role} to {settings.default_team_member_role} for {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}"
                        )

                except Exception as role_error:
                    logger.warning(f"Failed to update RBAC roles for {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}: {role_error}")
                    # Don't fail the membership role update if RBAC role update fails

            # Invalidate role cache
            try:
                self._fire_and_forget(auth_cache.invalidate_user_role(user_email, team_id))
            except Exception as cache_error:
                logger.debug(f"Failed to invalidate cache on role update: {cache_error}")

            logger.info(f"Updated role of {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)} to {new_role} by {updated_by}")
            return True

        except ValueError:
            raise  # Let ValueError propagate to caller for proper error handling
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to update role of {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return False

    async def get_member(self, team_id: str, user_email: str) -> Optional[EmailTeamMember]:
        """Get a single team member by team ID and user email.

        Args:
            team_id: ID of the team
            user_email: Email of the user

        Returns:
            EmailTeamMember if found and active, None otherwise
        """
        try:
            return self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)).first()
        except Exception as e:
            logger.error(f"Failed to get member {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return None

    async def get_user_teams(self, user_email: str, include_personal: bool = True) -> List[EmailTeam]:
        """Get all teams a user belongs to.

        Uses caching to reduce database queries (called 20+ times per request).
        Cache can be disabled via AUTH_CACHE_TEAMS_ENABLED=false.

        Args:
            user_email: Email of the user
            include_personal: Whether to include personal teams

        Returns:
            List[EmailTeam]: List of teams the user belongs to

        Examples:
            User dashboard showing team memberships.
        """
        # Check cache first
        cache = self._get_auth_cache()
        cache_key = f"{user_email}:{include_personal}"

        if cache:
            cached_team_ids = await cache.get_user_teams(cache_key)
            if cached_team_ids is not None:
                if not cached_team_ids:  # Empty list = user has no teams
                    return []
                # Fetch full team objects by IDs (fast indexed lookup)
                try:
                    teams = self.db.query(EmailTeam).filter(EmailTeam.id.in_(cached_team_ids), EmailTeam.is_active.is_(True)).all()
                    self.db.commit()  # Release transaction to avoid idle-in-transaction
                    return teams
                except Exception as e:
                    self.db.rollback()
                    logger.warning(f"Failed to fetch teams by IDs from cache: {e}")
                    # Fall through to full query

        # Cache miss or caching disabled - do full query
        try:
            query = self.db.query(EmailTeam).join(EmailTeamMember).filter(EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True), EmailTeam.is_active.is_(True))

            if not include_personal:
                query = query.filter(EmailTeam.is_personal.is_(False))

            teams = query.all()
            self.db.commit()  # Release transaction to avoid idle-in-transaction

            # Update cache with team IDs
            if cache:
                team_ids = [t.id for t in teams]
                await cache.set_user_teams(cache_key, team_ids)

            return teams

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get teams for user {SecurityValidator.sanitize_log_message(user_email)}: {e}")
            return []

    async def verify_team_for_user(self, user_email, team_id=None):
        """
        Retrieve a team ID for a user based on their membership and optionally a specific team ID.

        This function attempts to fetch all teams associated with the given user email.
        If no `team_id` is provided, it returns the ID of the user's personal team (if any).
        If a `team_id` is provided, it checks whether the user is a member of that team.
        If the user is not a member of the specified team, it returns a JSONResponse with an error message.

        Args:
            user_email (str): The email of the user whose teams are being queried.
            team_id (str or None, optional): Specific team ID to check for membership. Defaults to None.

        Returns:
            str or JSONResponse or None:
                - If `team_id` is None, returns the ID of the user's personal team or None if not found.
                - If `team_id` is provided and the user is a member of that team, returns `team_id`.
                - If `team_id` is provided but the user is not a member of that team, returns a JSONResponse with error.
                - Returns None if an error occurs and no `team_id` was initially provided.

        Raises:
            None explicitly, but any exceptions during the process are caught and logged.

        Examples:
            Verifies user team if team_id provided otherwise finds its personal id.
        """
        try:
            # Get all teams the user belongs to in a single query
            try:
                query = self.db.query(EmailTeam).join(EmailTeamMember).filter(EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True), EmailTeam.is_active.is_(True))
                user_teams = query.all()
                self.db.commit()  # Release transaction to avoid idle-in-transaction
            except Exception as e:
                self.db.rollback()
                logger.error(f"Failed to get teams for user {SecurityValidator.sanitize_log_message(user_email)}: {e}")
                return []

            if not team_id:
                # If no team_id is provided, try to get the personal team
                personal_team = next((t for t in user_teams if getattr(t, "is_personal", False)), None)
                team_id = personal_team.id if personal_team else None
            else:
                # Check if the provided team_id exists among the user's teams
                is_team_present = any(team.id == team_id for team in user_teams)
                if not is_team_present:
                    return []
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to verify team for user {SecurityValidator.sanitize_log_message(user_email)}: {e}")
            if not team_id:
                team_id = None

        return team_id

    @staticmethod
    def _escape_like(value: str) -> str:
        """Escape LIKE wildcards for prefix search.

        Args:
            value: Raw value to escape for LIKE matching.

        Returns:
            Escaped string safe for LIKE patterns.
        """
        return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

    async def get_team_members(
        self,
        team_id: str,
        cursor: Optional[str] = None,
        limit: Optional[int] = None,
        page: Optional[int] = None,
        per_page: Optional[int] = None,
        search: Optional[str] = None,
    ) -> Union[List[Tuple[EmailUser, EmailTeamMember]], Tuple[List[Tuple[EmailUser, EmailTeamMember]], Optional[str]], Dict[str, Any]]:
        """Get all members of a team with optional cursor or page-based pagination.

        Note: This method returns ORM objects and cannot be cached since callers
        depend on ORM attributes and methods.

        Args:
            team_id: ID of the team
            cursor: Opaque cursor token for cursor-based pagination
            limit: Maximum number of members to return (for cursor-based, default: 50)
            page: Page number for page-based pagination (1-indexed). Mutually exclusive with cursor.
            per_page: Items per page for page-based pagination (default: 30)
            search: Optional search term to filter by email or full name

        Returns:
            - If cursor is provided: Tuple (members, next_cursor)
            - If page is provided: Dict with keys 'data', 'pagination', 'links'
            - If neither: List of all members (backward compatibility)

        Examples:
            Team member management and role display.
        """
        try:
            # Build base query - for pagination, select EmailTeamMember and eager-load user
            # For backward compat (no pagination), select both entities as tuple
            # Build optional search filter
            search_filter = None
            if search and search.strip():
                search_term = f"{self._escape_like(search.strip())}%"
                search_filter = or_(
                    EmailUser.email.ilike(search_term, escape="\\"),
                    EmailUser.full_name.ilike(search_term, escape="\\"),
                )

            if cursor is None and page is None and limit is None:
                # Backward compatibility: return tuples (no pagination requested)
                query = (
                    select(EmailUser, EmailTeamMember)
                    .join(EmailTeamMember, EmailUser.email == EmailTeamMember.user_email)
                    .where(EmailTeamMember.team_id == team_id, EmailTeamMember.is_active.is_(True))
                    .order_by(EmailUser.full_name, EmailUser.email)
                )
                if search_filter is not None:
                    query = query.where(search_filter)
                result = self.db.execute(query)
                members = list(result.all())
                self.db.commit()
                return members

            # For pagination: select EmailTeamMember and eager-load user to avoid N+1
            query = (
                select(EmailTeamMember)
                .options(selectinload(EmailTeamMember.user))
                .where(EmailTeamMember.team_id == team_id, EmailTeamMember.is_active.is_(True))
                .join(EmailUser, EmailUser.email == EmailTeamMember.user_email)
            )
            if search_filter is not None:
                query = query.where(search_filter)

            # PAGE-BASED PAGINATION (Admin UI) - use unified_paginate
            if page is not None:
                # Alphabetical ordering for user-friendly display
                query = query.order_by(EmailUser.full_name, EmailUser.email)
                pag_result = await unified_paginate(
                    db=self.db,
                    query=query,
                    page=page,
                    per_page=per_page or 30,
                    cursor=None,
                    limit=None,
                    base_url=f"/admin/teams/{team_id}/members",
                    query_params={},
                )
                self.db.commit()
                memberships = pag_result["data"]
                tuples = [(m.user, m) for m in memberships]
                return {
                    "data": tuples,
                    "pagination": pag_result["pagination"],
                    "links": pag_result["links"],
                }

            # CURSOR-BASED PAGINATION (API) - custom implementation using (joined_at, id)
            # unified_paginate uses created_at which doesn't exist on EmailTeamMember

            # Order by joined_at DESC, id DESC for keyset pagination
            query = query.order_by(desc(EmailTeamMember.joined_at), desc(EmailTeamMember.id))

            # Decode cursor and apply keyset filter
            if cursor:
                try:
                    cursor_json = base64.urlsafe_b64decode(cursor.encode()).decode()
                    cursor_data = orjson.loads(cursor_json)
                    last_id = cursor_data.get("id")
                    joined_str = cursor_data.get("joined_at")
                    if last_id and joined_str:
                        last_joined = datetime.fromisoformat(joined_str)
                        # Keyset filter: (joined_at < last) OR (joined_at = last AND id < last_id)
                        query = query.where(
                            or_(
                                EmailTeamMember.joined_at < last_joined,
                                and_(EmailTeamMember.joined_at == last_joined, EmailTeamMember.id < last_id),
                            )
                        )
                except (ValueError, TypeError) as e:
                    logger.warning(f"Invalid cursor for team members, ignoring: {e}")

            # Fetch limit + 1 to check for more results (cap at max_page_size)
            page_size = min(limit or 50, settings.pagination_max_page_size)
            query = query.limit(page_size + 1)
            memberships = list(self.db.execute(query).scalars().all())

            # Check if there are more results
            has_more = len(memberships) > page_size
            if has_more:
                memberships = memberships[:page_size]

            # Generate next cursor using (joined_at, id)
            next_cursor = None
            if has_more and memberships:
                last_member = memberships[-1]
                cursor_data = {
                    "joined_at": last_member.joined_at.isoformat() if last_member.joined_at else None,
                    "id": last_member.id,
                }
                next_cursor = base64.urlsafe_b64encode(orjson.dumps(cursor_data)).decode()

            self.db.commit()
            tuples = [(m.user, m) for m in memberships]
            return (tuples, next_cursor)

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get members for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

            # Return appropriate empty response based on mode
            if page is not None:
                return {"data": [], "pagination": {"page": page, "per_page": per_page or 30, "total": 0, "has_next": False, "has_prev": False}, "links": None}

            if cursor is not None:
                return ([], None)

            return []

    def count_team_owners(self, team_id: str) -> int:
        """Count the number of owners in a team using SQL COUNT.

        This is more efficient than loading all members and counting in Python.

        Args:
            team_id: ID of the team

        Returns:
            int: Number of active owners in the team
        """
        count = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.role == "owner", EmailTeamMember.is_active.is_(True)).count()
        self.db.commit()  # Release transaction to avoid idle-in-transaction
        return count

    def _get_auth_cache(self):
        """Get auth cache instance lazily.

        Returns:
            AuthCache instance or None if unavailable.
        """
        try:
            return get_auth_cache()
        except ImportError:
            return None

    def _get_admin_stats_cache(self):
        """Get admin stats cache instance lazily.

        Returns:
            AdminStatsCache instance or None if unavailable.
        """
        try:
            # First-Party
            from mcpgateway.cache.admin_stats_cache import get_admin_stats_cache  # pylint: disable=import-outside-toplevel

            return get_admin_stats_cache()
        except ImportError:
            return None

    async def get_user_role_in_team(self, user_email: str, team_id: str) -> Optional[str]:
        """Get a user's role in a specific team.

        Uses caching to reduce database queries (called 11+ times per team operation).

        Args:
            user_email: Email of the user
            team_id: ID of the team

        Returns:
            str: User's role or None if not a member

        Examples:
            Access control and permission checking.
        """
        # Check cache first
        cache = self._get_auth_cache()
        if cache:
            cached_role = await cache.get_user_role(user_email, team_id)
            if cached_role is not None:
                # Empty string means "not a member" (cached None)
                return cached_role if cached_role else None

        try:
            membership = self.db.query(EmailTeamMember).filter(EmailTeamMember.user_email == user_email, EmailTeamMember.team_id == team_id, EmailTeamMember.is_active.is_(True)).first()
            self.db.commit()  # Release transaction to avoid idle-in-transaction

            role = membership.role if membership else None

            # Store in cache
            if cache:
                await cache.set_user_role(user_email, team_id, role)

            return role

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get role for {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return None

    @staticmethod
    def _apply_team_list_filters(
        query: Any,
        *,
        include_personal: bool = False,
        include_inactive: bool = False,
        visibility_filter: Optional[str] = None,
        search_query: Optional[str] = None,
        personal_owner_email: Optional[str] = None,
        search_description: bool = False,
    ) -> Any:
        """Apply common filter predicates to a team list query.

        Args:
            query: SQLAlchemy select statement to filter.
            include_personal: Whether to include personal teams.
            include_inactive: Whether to include inactive teams.
            visibility_filter: Filter by visibility (private, public).
            search_query: Optional search term for name/slug (and description if *search_description*).
            personal_owner_email: When set (and *include_personal* is False), include this user's personal team.
            search_description: Whether to include ``EmailTeam.description`` in the search predicate.

        Returns:
            The filtered query.
        """
        if not include_personal:
            if personal_owner_email:
                query = query.where(
                    or_(
                        EmailTeam.is_personal.is_(False),
                        and_(EmailTeam.is_personal.is_(True), EmailTeam.created_by == personal_owner_email),
                    )
                )
            else:
                query = query.where(EmailTeam.is_personal.is_(False))

        if not include_inactive:
            query = query.where(EmailTeam.is_active.is_(True))

        if visibility_filter:
            query = query.where(EmailTeam.visibility == visibility_filter)

        if search_query:
            search_term = f"%{search_query}%"
            predicates = [EmailTeam.name.ilike(search_term), EmailTeam.slug.ilike(search_term)]
            if search_description:
                predicates.append(EmailTeam.description.ilike(search_term))
            query = query.where(or_(*predicates))

        return query

    async def list_teams(
        self,
        # Unified pagination params
        limit: int = 100,
        offset: int = 0,
        cursor: Optional[str] = None,
        page: Optional[int] = None,
        per_page: int = 50,
        include_inactive: bool = False,
        visibility_filter: Optional[str] = None,
        base_url: Optional[str] = None,
        include_personal: bool = False,
        search_query: Optional[str] = None,
        personal_owner_email: Optional[str] = None,
    ) -> Union[Tuple[List[EmailTeam], Optional[str]], Dict[str, Any]]:
        """List teams with pagination support (cursor or page based).

        Args:
            limit: Max items for cursor pagination
            offset: Offset for legacy/cursor pagination
            cursor: Cursor token
            page: Page number (1-indexed)
            per_page: Items per page
            include_inactive: Whether to include inactive teams
            visibility_filter: Filter by visibility (private, team, public)
            base_url: Base URL for pagination links
            include_personal: Whether to include personal teams
            search_query: Search term for name/slug/description
            personal_owner_email: When set (and include_personal=False), includes this user's personal team alongside non-personal teams

        Returns:
            Union[Tuple[List[EmailTeam], Optional[str]], Dict[str, Any]]:
                - Tuple (teams, next_cursor) if cursor/offset based
                - Dict {data, pagination, links} if page based
        """
        query = self._apply_team_list_filters(
            select(EmailTeam),
            include_personal=include_personal,
            include_inactive=include_inactive,
            visibility_filter=visibility_filter,
            search_query=search_query,
            personal_owner_email=personal_owner_email,
            search_description=True,
        )

        # Choose ordering based on pagination mode:
        # - Page-based (UI): alphabetical by name for user-friendly display
        # - Cursor-based (API): created_at DESC, id DESC to match unified_paginate expectations
        if page is not None:
            query = query.order_by(EmailTeam.name, EmailTeam.id)
        else:
            query = query.order_by(desc(EmailTeam.created_at), desc(EmailTeam.id))

        # Base URL for pagination links (default to admin partial if not provided)
        if not base_url:
            base_url = f"{settings.app_root_path}/admin/teams/partial"

        # Apply offset manually for legacy offset-based pagination if not using page or cursor
        if not page and not cursor and offset > 0:
            query = query.offset(offset)

        result = await unified_paginate(
            db=self.db,
            query=query,
            cursor=cursor,
            limit=limit,
            page=page,
            per_page=per_page,
            base_url=base_url,
        )
        self.db.commit()  # Release transaction to avoid idle-in-transaction
        return result

    async def get_all_team_ids(
        self,
        include_inactive: bool = False,
        visibility_filter: Optional[str] = None,
        include_personal: bool = False,
        search_query: Optional[str] = None,
        personal_owner_email: Optional[str] = None,
    ) -> List[int]:
        """Get all team IDs matching criteria (unpaginated).

        Args:
            include_inactive: Whether to include inactive teams
            visibility_filter: Filter by visibility (private, team, public)
            include_personal: Whether to include personal teams
            search_query: Search term for name/slug
            personal_owner_email: When set (and include_personal=False), includes this user's personal team alongside non-personal teams

        Returns:
            List[int]: List of team IDs
        """
        query = self._apply_team_list_filters(
            select(EmailTeam.id),
            include_personal=include_personal,
            include_inactive=include_inactive,
            visibility_filter=visibility_filter,
            search_query=search_query,
            personal_owner_email=personal_owner_email,
        )

        result = self.db.execute(query)
        team_ids = [row[0] for row in result.all()]
        self.db.commit()  # Release transaction to avoid idle-in-transaction
        return team_ids

    async def get_teams_count(
        self,
        include_inactive: bool = False,
        visibility_filter: Optional[str] = None,
        include_personal: bool = False,
        search_query: Optional[str] = None,
        personal_owner_email: Optional[str] = None,
    ) -> int:
        """Get total count of teams matching criteria.

        Args:
            include_inactive: Whether to include inactive teams
            visibility_filter: Filter by visibility (private, team, public)
            include_personal: Whether to include personal teams
            search_query: Search term for name/slug
            personal_owner_email: When set (and include_personal=False), includes this user's personal team in the count

        Returns:
            int: Total count of matching teams
        """
        query = self._apply_team_list_filters(
            select(func.count(EmailTeam.id)),  # pylint: disable=not-callable
            include_personal=include_personal,
            include_inactive=include_inactive,
            visibility_filter=visibility_filter,
            search_query=search_query,
            personal_owner_email=personal_owner_email,
        )

        result = self.db.execute(query)
        count = result.scalar() or 0
        self.db.commit()  # Release transaction to avoid idle-in-transaction
        return count

    async def discover_public_teams(self, user_email: str, skip: int = 0, limit: Optional[int] = None) -> List[EmailTeam]:
        """Discover public teams that user can join.

        Args:
            user_email: Email of the user discovering teams
            skip: Number of teams to skip for pagination
            limit: Maximum number of teams to return (None for unlimited)

        Returns:
            List[EmailTeam]: List of public teams user can join

        Raises:
            Exception: If discovery fails
        """
        try:
            # Optimized: Use subquery instead of loading all IDs into memory (2 queries → 1)
            user_team_subquery = select(EmailTeamMember.team_id).where(EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)).scalar_subquery()

            query = self.db.query(EmailTeam).filter(EmailTeam.visibility == "public", EmailTeam.is_active.is_(True), EmailTeam.is_personal.is_(False), ~EmailTeam.id.in_(user_team_subquery))

            query = query.offset(skip)
            if limit is not None:
                query = query.limit(limit)
            teams = query.all()
            self.db.commit()  # Release transaction to avoid idle-in-transaction
            return teams

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to discover public teams for {SecurityValidator.sanitize_log_message(user_email)}: {e}")
            return []

    async def create_join_request(self, team_id: str, user_email: str, message: Optional[str] = None) -> "EmailTeamJoinRequest":
        """Create a request to join a public team.

        Args:
            team_id: ID of the team to join
            user_email: Email of the user requesting to join
            message: Optional message to team owners

        Returns:
            EmailTeamJoinRequest: Created join request

        Raises:
            ValueError: If team not found, not public, or user already member/has pending request
        """
        try:
            # Validate team
            team = await self.get_team_by_id(team_id)
            if not team:
                raise ValueError("Team not found")

            if team.visibility != "public":
                raise ValueError("Can only request to join public teams")

            # Check if user is already a member
            existing_member = self.db.query(EmailTeamMember).filter(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)).first()

            if existing_member:
                raise ValueError("User is already a member of this team")

            # Check max teams per user
            max_teams = getattr(settings, "max_teams_per_user", 50)
            if self._get_user_team_count(user_email) >= max_teams:
                raise ValueError(f"User has reached the maximum team limit of {max_teams}")

            # Check for existing requests (any status)
            existing_request = self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.team_id == team_id, EmailTeamJoinRequest.user_email == user_email).first()

            if existing_request:
                if existing_request.status == "pending" and not existing_request.is_expired():
                    raise ValueError("User already has a pending join request for this team")

                # Update existing request (cancelled, rejected, expired) to pending
                existing_request.message = message or ""
                existing_request.status = "pending"
                existing_request.requested_at = utc_now()
                existing_request.expires_at = utc_now() + timedelta(days=7)
                existing_request.reviewed_at = None
                existing_request.reviewed_by = None
                existing_request.notes = None
                join_request = existing_request
            else:
                # Create new join request
                join_request = EmailTeamJoinRequest(team_id=team_id, user_email=user_email, message=message, expires_at=utc_now() + timedelta(days=7))
                self.db.add(join_request)

            self.db.commit()
            self.db.refresh(join_request)

            logger.info(f"Created join request for user {SecurityValidator.sanitize_log_message(user_email)} to team {SecurityValidator.sanitize_log_message(team_id)}")
            return join_request

        except ValueError:
            self.db.rollback()
            raise
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to create join request: {e}")
            raise

    async def list_join_requests(self, team_id: str) -> List["EmailTeamJoinRequest"]:
        """List pending join requests for a team.

        Args:
            team_id: ID of the team

        Returns:
            List[EmailTeamJoinRequest]: List of pending join requests
        """
        try:
            requests = (
                self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.team_id == team_id, EmailTeamJoinRequest.status == "pending").order_by(EmailTeamJoinRequest.requested_at.desc()).all()
            )
            return requests

        except Exception as e:
            logger.error(f"Failed to list join requests for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
            return []

    async def approve_join_request(self, request_id: str, approved_by: str) -> Optional[EmailTeamMember]:
        """Approve a team join request.

        Args:
            request_id: ID of the join request
            approved_by: Email of the user approving the request

        Returns:
            EmailTeamMember: New team member or None if request not found

        Raises:
            ValueError: If request not found, expired, or already processed
        """
        try:
            # Get join request
            join_request = self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.id == request_id, EmailTeamJoinRequest.status == "pending").first()

            if not join_request:
                raise ValueError("Join request not found or already processed")

            if join_request.is_expired():
                join_request.status = "expired"
                self.db.commit()
                raise ValueError("Join request has expired")

            # Check max teams per user
            max_teams = getattr(settings, "max_teams_per_user", 50)
            if self._get_user_team_count(join_request.user_email) >= max_teams:
                raise ValueError(f"User has reached the maximum team limit of {max_teams}")

            # Check team member limit
            team = await self.get_team_by_id(join_request.team_id)
            if not team:
                raise ValueError(f"Team {join_request.team_id} not found or inactive")
            check_team_member_capacity(self.db, team)

            # Add user to team
            member = EmailTeamMember(team_id=join_request.team_id, user_email=join_request.user_email, role="member", invited_by=approved_by, joined_at=utc_now())  # New joiners are always members

            self.db.add(member)
            # Update join request status
            join_request.status = "approved"
            join_request.reviewed_at = utc_now()
            join_request.reviewed_by = approved_by

            self.db.flush()
            self._log_team_member_action(member.id, join_request.team_id, join_request.user_email, member.role, "added", approved_by)

            self.db.refresh(member)

            await self._assign_team_rbac_role(join_request.user_email, join_request.team_id, member.role, granted_by=approved_by)
            self._invalidate_membership_caches(join_request.user_email, join_request.team_id)
            await self.invalidate_team_member_count_cache(str(join_request.team_id))

            logger.info(f"Approved join request {request_id}: user {join_request.user_email} joined team {join_request.team_id}")
            return member

        except ValueError:
            self.db.rollback()
            raise
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to approve join request {request_id}: {e}")
            raise

    async def reject_join_request(self, request_id: str, rejected_by: str) -> bool:
        """Reject a team join request.

        Args:
            request_id: ID of the join request
            rejected_by: Email of the user rejecting the request

        Returns:
            bool: True if request was rejected successfully

        Raises:
            ValueError: If request not found or already processed
        """
        try:
            # Get join request
            join_request = self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.id == request_id, EmailTeamJoinRequest.status == "pending").first()

            if not join_request:
                raise ValueError("Join request not found or already processed")

            # Update join request status
            join_request.status = "rejected"
            join_request.reviewed_at = utc_now()
            join_request.reviewed_by = rejected_by

            self.db.commit()

            logger.info(f"Rejected join request {request_id}: user {join_request.user_email} for team {join_request.team_id}")
            return True

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to reject join request {request_id}: {e}")
            raise

    async def get_user_join_requests(self, user_email: str, team_id: Optional[str] = None) -> List["EmailTeamJoinRequest"]:
        """Get join requests made by a user.

        Args:
            user_email: Email of the user
            team_id: Optional team ID to filter requests

        Returns:
            List[EmailTeamJoinRequest]: List of join requests made by the user

        Examples:
            Get all requests made by a user or for a specific team.
        """
        try:
            query = self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.user_email == user_email)

            if team_id:
                query = query.filter(EmailTeamJoinRequest.team_id == team_id)

            requests = query.all()
            return requests

        except Exception as e:
            logger.error(f"Failed to get join requests for user {SecurityValidator.sanitize_log_message(user_email)}: {e}")
            return []

    async def cancel_join_request(self, request_id: str, user_email: str) -> bool:
        """Cancel a join request.

        Args:
            request_id: ID of the join request to cancel
            user_email: Email of the user canceling the request

        Returns:
            bool: True if canceled successfully, False otherwise

        Examples:
            Allow users to cancel their pending join requests.
        """
        try:
            # Get the join request
            join_request = (
                self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.id == request_id, EmailTeamJoinRequest.user_email == user_email, EmailTeamJoinRequest.status == "pending").first()
            )

            if not join_request:
                logger.warning(f"Join request {request_id} not found for user {SecurityValidator.sanitize_log_message(user_email)} or not pending")
                return False

            # Update join request status
            join_request.status = "cancelled"
            join_request.reviewed_at = utc_now()
            join_request.reviewed_by = user_email

            self.db.commit()

            logger.info(f"Cancelled join request {request_id} by user {SecurityValidator.sanitize_log_message(user_email)}")
            return True

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to cancel join request {request_id}: {e}")
            return False

    # ==================================================================================
    # Batch Query Methods (N+1 Query Elimination - Issue #1892)
    # ==================================================================================

    def get_member_counts_batch(self, team_ids: List[str]) -> Dict[str, int]:
        """Get member counts for multiple teams in a single query.

        This is a synchronous method following the existing service pattern.
        Note: Like other sync SQLAlchemy calls, this will block the event
        loop in async contexts. For typical team counts this is acceptable.

        Args:
            team_ids: List of team UUIDs

        Returns:
            Dict mapping team_id to member count

        Raises:
            Exception: Re-raises any database errors after rollback

        Examples:
            >>> from unittest.mock import Mock
            >>> service = TeamManagementService(Mock())
            >>> service.get_member_counts_batch([])
            {}
        """
        if not team_ids:
            return {}

        try:
            # Single query for all teams
            results = (
                self.db.query(EmailTeamMember.team_id, func.count(EmailTeamMember.id).label("count"))  # pylint: disable=not-callable
                .filter(EmailTeamMember.team_id.in_(team_ids), EmailTeamMember.is_active.is_(True))
                .group_by(EmailTeamMember.team_id)
                .all()
            )

            self.db.commit()  # Release transaction to avoid idle-in-transaction

            # Build result dict, defaulting to 0 for teams with no members
            counts = {str(row.team_id): row.count for row in results}
            return {tid: counts.get(tid, 0) for tid in team_ids}
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get member counts for teams: {e}")
            raise

    def get_user_roles_batch(self, user_email: str, team_ids: List[str]) -> Dict[str, Optional[str]]:
        """Get a user's role in multiple teams in a single query.

        Args:
            user_email: Email of the user
            team_ids: List of team UUIDs

        Returns:
            Dict mapping team_id to role (or None if not a member)

        Raises:
            Exception: Re-raises any database errors after rollback
        """
        if not team_ids:
            return {}

        try:
            # Single query for all teams
            results = (
                self.db.query(EmailTeamMember.team_id, EmailTeamMember.role)
                .filter(EmailTeamMember.user_email == user_email, EmailTeamMember.team_id.in_(team_ids), EmailTeamMember.is_active.is_(True))
                .all()
            )

            self.db.commit()  # Release transaction to avoid idle-in-transaction

            # Build result dict - teams with no membership return None
            roles = {str(row.team_id): row.role for row in results}
            return {tid: roles.get(tid) for tid in team_ids}
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get user roles for {SecurityValidator.sanitize_log_message(user_email)}: {e}")
            raise

    def get_pending_join_requests_batch(self, user_email: str, team_ids: List[str]) -> Dict[str, Optional[Any]]:
        """Get pending join requests for a user across multiple teams in a single query.

        Args:
            user_email: Email of the user
            team_ids: List of team UUIDs to check

        Returns:
            Dict mapping team_id to pending EmailTeamJoinRequest (or None if no pending request)

        Raises:
            Exception: Re-raises any database errors after rollback
        """
        if not team_ids:
            return {}

        try:
            # Single query for all pending requests across teams
            results = (
                self.db.query(EmailTeamJoinRequest).filter(EmailTeamJoinRequest.user_email == user_email, EmailTeamJoinRequest.team_id.in_(team_ids), EmailTeamJoinRequest.status == "pending").all()
            )

            self.db.commit()  # Release transaction to avoid idle-in-transaction

            # Build result dict - only one pending request per team expected
            pending_reqs = {str(req.team_id): req for req in results}
            return {tid: pending_reqs.get(tid) for tid in team_ids}
        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to get pending join requests for {SecurityValidator.sanitize_log_message(user_email)}: {e}")
            raise

    # ==================================================================================
    # Cached Batch Methods (Redis caching for member counts)
    # ==================================================================================

    def _get_member_count_cache_key(self, team_id: str) -> str:
        """Build cache key using settings.cache_prefix for consistency.

        Args:
            team_id: Team UUID to build cache key for

        Returns:
            Cache key string in format "{prefix}team:member_count:{team_id}"
        """
        cache_prefix = getattr(settings, "cache_prefix", "mcpgw:")
        return f"{cache_prefix}team:member_count:{team_id}"

    async def get_member_counts_batch_cached(self, team_ids: List[str]) -> Dict[str, int]:
        """Get member counts for multiple teams, using Redis cache with DB fallback.

        Caching behavior is controlled by settings:
        - team_member_count_cache_enabled: Enable/disable caching (default: True)
        - team_member_count_cache_ttl: Cache TTL in seconds (default: 300)

        Args:
            team_ids: List of team UUIDs

        Returns:
            Dict mapping team_id to member count

        Raises:
            Exception: Re-raises any database errors after rollback
        """
        if not team_ids:
            return {}

        cache_enabled = getattr(settings, "team_member_count_cache_enabled", True)
        cache_ttl = getattr(settings, "team_member_count_cache_ttl", 300)

        # If caching disabled, go straight to batch DB query
        if not cache_enabled:
            return self.get_member_counts_batch(team_ids)

        try:
            redis_client = await get_redis_client()
        except Exception:
            redis_client = None

        result: Dict[str, int] = {}
        cache_misses: List[str] = []

        # Step 1: Check Redis cache for all team IDs
        if redis_client:
            try:
                cache_keys = [self._get_member_count_cache_key(tid) for tid in team_ids]
                cached_values = await redis_client.mget(cache_keys)

                for tid, cached in zip(team_ids, cached_values):
                    if cached is not None:
                        result[tid] = int(cached)
                    else:
                        cache_misses.append(tid)
            except Exception as e:
                logger.warning(f"Redis cache read failed, falling back to DB: {e}")
                cache_misses = list(team_ids)
        else:
            # No Redis available, fall back to DB
            cache_misses = list(team_ids)

        # Step 2: Query database for cache misses
        if cache_misses:
            try:
                db_results = (
                    self.db.query(EmailTeamMember.team_id, func.count(EmailTeamMember.id).label("count"))  # pylint: disable=not-callable
                    .filter(EmailTeamMember.team_id.in_(cache_misses), EmailTeamMember.is_active.is_(True))
                    .group_by(EmailTeamMember.team_id)
                    .all()
                )

                self.db.commit()

                db_counts = {str(row.team_id): row.count for row in db_results}

                # Fill in results and cache them
                for tid in cache_misses:
                    count = db_counts.get(tid, 0)
                    result[tid] = count

                    # Step 3: Cache the result with configured TTL
                    if redis_client:
                        try:
                            await redis_client.setex(self._get_member_count_cache_key(tid), cache_ttl, str(count))
                        except Exception as e:
                            logger.warning(f"Redis cache write failed for team {tid}: {e}")

            except Exception as e:
                self.db.rollback()
                logger.error(f"Failed to get member counts for teams: {e}")
                raise

        return result

    async def invalidate_team_member_count_cache(self, team_id: str) -> None:
        """Invalidate the cached member count for a team.

        Call this after any membership changes (add/remove/update).
        No-op if caching is disabled or Redis unavailable.

        Args:
            team_id: Team UUID to invalidate
        """
        cache_enabled = getattr(settings, "team_member_count_cache_enabled", True)
        if not cache_enabled:
            return

        try:
            redis_client = await get_redis_client()
            if redis_client:
                await redis_client.delete(self._get_member_count_cache_key(team_id))
        except Exception as e:
            logger.warning(f"Failed to invalidate member count cache for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")
