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

Token Catalog Service.
This module provides comprehensive API token management with scoping,
revocation, usage tracking, and analytics for email-based users.

Examples:
    >>> from mcpgateway.services.token_catalog_service import TokenCatalogService
    >>> service = TokenCatalogService(None)  # Mock database for doctest
    >>> # Service provides full token lifecycle management
"""

# Standard
import asyncio
from datetime import datetime, timedelta, timezone
import hashlib
import math
from typing import Dict, List, Optional
import uuid

# Third-Party
from sqlalchemy import and_, case, func, or_, select
from sqlalchemy.orm import Session

# First-Party
from mcpgateway.common.validators import SecurityValidator
from mcpgateway.config import settings
from mcpgateway.db import EmailApiToken, EmailUser, Permissions, TokenRevocation, TokenUsageLog, utc_now
from mcpgateway.services.logging_service import LoggingService
from mcpgateway.utils.create_jwt_token import create_jwt_token

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

# Strong references to background tasks to prevent GC before completion
_background_tasks: set[asyncio.Task] = set()


class TokenScope:
    """Token scoping configuration for fine-grained access control.

    This class encapsulates token scoping parameters including
    server restrictions, permissions, IP limitations, and usage quotas.

    Attributes:
        server_id (Optional[str]): Limit token to specific server
        permissions (List[str]): Specific permission scopes
        ip_restrictions (List[str]): IP address/CIDR restrictions
        time_restrictions (dict): Time-based access limitations
        usage_limits (dict): Rate limiting and quota settings

    Examples:
        >>> scope = TokenScope(
        ...     server_id="prod-server-123",
        ...     permissions=["tools.read", "resources.read"],
        ...     ip_restrictions=["192.168.1.0/24"],
        ...     time_restrictions={"business_hours_only": True}
        ... )
        >>> scope.is_server_scoped()
        True
        >>> scope.has_permission("tools.read")
        True
        >>> scope.has_permission("tools.write")
        False
        >>> scope.has_permission("resources.read")
        True
        >>>
        >>> # Test empty scope
        >>> empty_scope = TokenScope()
        >>> empty_scope.is_server_scoped()
        False
        >>> empty_scope.has_permission("anything")
        False
        >>>
        >>> # Test global scope
        >>> global_scope = TokenScope(permissions=["*"])
        >>> global_scope.has_permission("*")
        True
    """

    def __init__(
        self,
        server_id: Optional[str] = None,
        permissions: Optional[List[str]] = None,
        ip_restrictions: Optional[List[str]] = None,
        time_restrictions: Optional[dict] = None,
        usage_limits: Optional[dict] = None,
    ):
        """Initialize TokenScope with specified restrictions and limits.

        Args:
            server_id: Optional server ID to scope token to specific server
            permissions: List of permissions granted to this token
            ip_restrictions: List of IP addresses/ranges allowed to use token
            time_restrictions: Dictionary of time-based access restrictions
            usage_limits: Dictionary of usage limits for the token
        """
        self.server_id = server_id
        self.permissions = permissions or []
        self.ip_restrictions = ip_restrictions or []
        self.time_restrictions = time_restrictions or {}
        self.usage_limits = usage_limits or {}

    def is_server_scoped(self) -> bool:
        """Check if token is scoped to a specific server.

        Returns:
            bool: True if scoped to a server, False otherwise.
        """
        return self.server_id is not None

    def has_permission(self, permission: str) -> bool:
        """Check if scope includes specific permission.

        Args:
            permission: Permission string to check for.

        Returns:
            bool: True if permission is included, False otherwise.
        """
        return permission in self.permissions

    def to_dict(self) -> dict:
        """Convert scope to dictionary for JSON storage.

        Returns:
            dict: Dictionary representation of the token scope.

        Examples:
            >>> scope = TokenScope(server_id="server-123", permissions=["read", "write"])
            >>> result = scope.to_dict()
            >>> result["server_id"]
            'server-123'
            >>> result["permissions"]
            ['read', 'write']
            >>> isinstance(result, dict)
            True
        """
        return {"server_id": self.server_id, "permissions": self.permissions, "ip_restrictions": self.ip_restrictions, "time_restrictions": self.time_restrictions, "usage_limits": self.usage_limits}

    @classmethod
    def from_dict(cls, data: dict) -> "TokenScope":
        """Create TokenScope from dictionary.

        Args:
            data: Dictionary containing scope configuration.

        Returns:
            TokenScope: New TokenScope instance.

        Examples:
            >>> data = {
            ...     "server_id": "server-456",
            ...     "permissions": ["tools.read", "tools.execute"],
            ...     "ip_restrictions": ["10.0.0.0/8"]
            ... }
            >>> scope = TokenScope.from_dict(data)
            >>> scope.server_id
            'server-456'
            >>> scope.permissions
            ['tools.read', 'tools.execute']
            >>> scope.is_server_scoped()
            True
            >>> scope.has_permission("tools.read")
            True
            >>>
            >>> # Test empty dict
            >>> empty_scope = TokenScope.from_dict({})
            >>> empty_scope.server_id is None
            True
            >>> empty_scope.permissions
            []
        """
        return cls(
            server_id=data.get("server_id"),
            permissions=data.get("permissions", []),
            ip_restrictions=data.get("ip_restrictions", []),
            time_restrictions=data.get("time_restrictions", {}),
            usage_limits=data.get("usage_limits", {}),
        )


class TokenCatalogService:
    """Service for managing user API token catalogs.

    This service provides comprehensive token lifecycle management including
    creation, scoping, revocation, usage tracking, and analytics. It handles
    JWT-based API tokens with fine-grained access control, team support,
    and comprehensive audit logging.

    Key features:
    - Token creation with customizable scopes and permissions
    - Team-based token management with role-based access
    - Token revocation and blacklisting
    - Usage tracking and analytics
    - IP and time-based restrictions
    - Automatic cleanup of expired tokens

    Attributes:
        db (Session): SQLAlchemy database session for token operations

    Examples:
        >>> from mcpgateway.services.token_catalog_service import TokenCatalogService
        >>> service = TokenCatalogService(None)  # Mock database for doctest
        >>> service.db is None
        True
    """

    def __init__(self, db: Session):
        """Initialize TokenCatalogService with database session.

        Args:
            db: SQLAlchemy database session for token operations
        """
        self.db = db

    async def _generate_token(
        self, user_email: str, jti: str, team_id: Optional[str] = None, expires_at: Optional[datetime] = None, scope: Optional["TokenScope"] = None, user: Optional[object] = None
    ) -> str:
        """Generate a JWT token for API access.

        This internal method creates a properly formatted JWT token with all
        necessary claims including user identity, scopes, team membership,
        and expiration. The token follows ContextForge JWT structure.

        Args:
            user_email: User's email address for the token subject
            jti: JWT ID for token uniqueness
            team_id: Optional team ID for team-scoped tokens
            expires_at: Optional expiration datetime
            scope: Optional token scope information for access control
            user: Optional user object to extract admin privileges

        Returns:
            str: Signed JWT token string ready for API authentication

        Raises:
            ValueError: If expires_at is in the past (cannot create already-expired tokens)

        Note:
            This is an internal method. Use create_token() to generate
            tokens with proper database tracking and validation.
        """
        # Calculate expiration in minutes from expires_at
        expires_in_minutes = 0
        if expires_at:
            now = datetime.now(timezone.utc)
            delta = expires_at - now
            delta_seconds = delta.total_seconds()

            # Guard: reject already-expired expiration times
            if delta_seconds <= 0:
                raise ValueError("Token expiration time is in the past. Cannot create already-expired tokens.")

            # Use ceiling to ensure we always have at least 1 minute expiration
            # This prevents <60s from rounding to 0 and creating non-expiring tokens
            expires_in_minutes = max(1, math.ceil(delta_seconds / 60))

        # Build user data dict
        user_data = {
            "email": user_email,
            "full_name": "API Token User",
            "is_admin": user.is_admin if user else False,
            "auth_provider": "api_token",
        }

        # Build teams list — None means "all teams" (admin bypass when is_admin=true),
        # [] means "public-only" (see normalize_token_teams() in auth.py)
        teams = [team_id] if team_id else None

        # Build scopes dict
        # Empty permissions = defer to RBAC at runtime (not wildcard access)
        scopes_dict = None
        if scope:
            scopes_dict = {
                "server_id": scope.server_id,
                "permissions": scope.permissions if scope.permissions is not None else [],
                "ip_restrictions": scope.ip_restrictions or [],
                "time_restrictions": scope.time_restrictions or {},
            }
        else:
            scopes_dict = {
                "server_id": None,
                "permissions": [],  # Empty = inherit from RBAC at runtime
                "ip_restrictions": [],
                "time_restrictions": {},
            }

        # Auto-inject servers.use for tokens with explicit MCP-related permissions.
        # Without servers.use, the token scoping middleware blocks /rpc and /mcp
        # transport access, making MCP-method permissions useless.
        permissions = scopes_dict["permissions"]
        if permissions and "*" not in permissions and "servers.use" not in permissions:
            if any(p.startswith(Permissions.MCP_METHOD_PREFIXES) for p in permissions):
                scopes_dict["permissions"] = [*permissions, "servers.use"]

        # Generate JWT token using the centralized token creation utility
        # Pass structured data to the enhanced create_jwt_token function
        return await create_jwt_token(
            data={"sub": user_email, "jti": jti, "token_use": "api"},  # nosec B105 - token type marker, not a password
            expires_in_minutes=expires_in_minutes,
            user_data=user_data,
            teams=teams,
            scopes=scopes_dict,
        )

    def _hash_token(self, token: str) -> str:
        """Create secure hash of token for storage.

        Args:
            token: Raw token string

        Returns:
            str: SHA-256 hash of token

        Examples:
            >>> service = TokenCatalogService(None)
            >>> hash_val = service._hash_token("test_token")
            >>> len(hash_val) == 64
            True
        """
        return hashlib.sha256(token.encode()).hexdigest()

    def _validate_scope_containment(
        self,
        requested_permissions: Optional[List[str]],
        caller_permissions: Optional[List[str]],
    ) -> None:
        """Validate that requested permissions don't exceed caller's permissions.

        SECURITY: This is fail-secure. If caller_permissions is empty/None,
        custom scopes are DENIED. Users without explicit permissions can only
        create tokens with empty scope (inherit at runtime).

        Args:
            requested_permissions: Permissions requested for new/updated token
            caller_permissions: Caller's effective permissions (RBAC + current token scopes)

        Raises:
            ValueError: If requested permissions exceed caller's permissions
        """
        # No requested permissions = empty scope, always allowed
        if not requested_permissions:
            return

        # FAIL-SECURE: If caller has no permissions, deny any custom scope
        if not caller_permissions:
            raise ValueError("Cannot specify custom token permissions. " + "You have no explicit permissions to delegate. " + "Create a token without scope to inherit permissions at runtime.")

        # Wildcard caller can grant anything
        if "*" in caller_permissions:
            return

        # Wildcard request requires wildcard caller
        if "*" in requested_permissions:
            raise ValueError("Cannot create token with wildcard permissions. " + "Your effective permissions do not include wildcard access.")

        # Check each requested permission
        for req_perm in requested_permissions:
            if req_perm in caller_permissions:
                continue

            # Check for category wildcard (e.g., "tools.*" allows "tools.read")
            if "." in req_perm:
                category = req_perm.split(".")[0]
                if f"{category}.*" in caller_permissions:
                    continue

            raise ValueError(f"Cannot grant permission '{req_perm}' - not in your effective permissions.")

    async def create_token(
        self,
        user_email: str,
        name: str,
        description: Optional[str] = None,
        scope: Optional[TokenScope] = None,
        expires_in_days: Optional[int] = None,
        tags: Optional[List[str]] = None,
        team_id: Optional[str] = None,
        caller_permissions: Optional[List[str]] = None,
        is_admin: bool = False,
        caller_token_teams: Optional[List[str]] = None,
        caller_token_teams_provided: bool = False,
        is_active: bool = True,
        caller_email: Optional[str] = None,
    ) -> tuple[EmailApiToken, str]:
        """
        Create a new API token with team-level scoping and additional configurations.

        This method generates a JWT-based API token with team-level scoping and optional security configurations,
        such as expiration, permissions, IP restrictions, and usage limits. The token is associated with a user
        and a specific team, ensuring access control and multi-tenancy support.

        The function will:
        - Validate the existence of the user.
        - Ensure the user is an active member of the specified team (unless admin bypass applies).
        - Verify that the token name is unique for the user+team combination.
        - Generate a JWT with the specified scoping parameters (e.g., permissions, IP, etc.).
        - Store the token in the database with the relevant details and return the token and raw JWT string.

        Args:
            user_email (str): The email address of the user requesting the token.
            name (str): A unique, human-readable name for the token (must be unique per user+team).
            description (Optional[str]): A description for the token (default is None).
            scope (Optional[TokenScope]): The scoping configuration for the token, including permissions,
                server ID, IP restrictions, etc. (default is None).
            expires_in_days (Optional[int]): The expiration time in days for the token (None means no expiration).
            tags (Optional[List[str]]): A list of organizational tags for the token (default is an empty list).
            team_id (Optional[str]): The team ID to which the token should be scoped. This is required for team-level scoping.
            caller_permissions (Optional[List[str]]): The permissions of the caller creating the token. Used for
                scope containment validation to ensure the new token cannot have broader permissions than the caller.
                Also used with is_admin and caller_token_teams to determine admin bypass eligibility.
            is_admin (bool): Whether the caller is a platform admin. Used with caller_permissions and
                caller_token_teams for defense-in-depth validation of admin bypass (default is False).
            caller_token_teams (Optional[List[str]]): The caller's token narrowing scope (from JWT ``teams`` claim
                via ``current_user["token_teams"]``). Required for admin bypass evaluation: only un-narrowed
                sessions (``caller_token_teams_provided=True`` and ``caller_token_teams is None``) may bypass
                team-membership checks. Narrowed (``["team-a"]``) and public-only (``[]``) sessions never bypass,
                even if their effective permissions include ``"*"`` via a global ``platform_admin`` role.
            caller_token_teams_provided (bool): Whether the caller resolved ``caller_token_teams`` from the
                authenticated session. Defaults to ``False``: callers that did not opt in cannot satisfy the
                admin bypass, even if they pass ``is_admin=True`` and ``caller_permissions=["*"]``. Routers
                that intend to allow the bypass must set this to ``True`` (default is False).
            is_active (bool): Whether the token should be created as active (default is True).
            caller_email (Optional[str]): The email of the caller creating the token. When admin delegation
                is used (``user_email != caller_email``), team membership checks are enforced even for
                unrestricted admins to prevent privilege escalation.

        Returns:
            tuple[EmailApiToken, str]: A tuple where the first element is the `EmailApiToken` database record and
            the second element is the raw JWT token string. The `EmailApiToken` contains the database record with the
            token details.

        Raises:
            ValueError: If any of the following validation checks fail:
                - The `user_email` does not correspond to an existing user.
                - The `team_id` is missing or the user is not an active member of the specified team.
                - A token with the same name already exists for the given user and team.
                - Invalid token configuration (e.g., invalid expiration date).

        Examples:
            >>> # This method requires database operations, shown for reference
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # token, raw_token = await service.create_token(...)
            >>> # Returns (EmailApiToken, raw_token_string) tuple
        """
        # # Enforce team-level scoping requirement
        # if not team_id:
        #     raise ValueError("team_id is required for token creation. " "Please select a specific team before creating a token. " "You cannot create tokens while viewing 'All Teams'.")

        # Validate user exists
        user = self.db.execute(select(EmailUser).where(EmailUser.email == user_email)).scalar_one_or_none()

        if not user:
            raise ValueError(f"User not found: {user_email}")

        # Validate scope containment (fail-secure if no caller_permissions)
        if scope and scope.permissions:
            self._validate_scope_containment(scope.permissions, caller_permissions)

        # Validate team exists and user is active member
        if team_id:
            # First-Party
            from mcpgateway.db import EmailTeam, EmailTeamMember  # pylint: disable=import-outside-toplevel

            # Check if team exists
            team = self.db.execute(select(EmailTeam).where(EmailTeam.id == team_id)).scalar_one_or_none()

            if not team:
                raise ValueError(f"Team not found: {team_id}")

            # Admin bypass: Only un-narrowed platform admins may create team tokens
            # without being active team members. This supports service account
            # workflows and centralized token management.
            #
            # Triple-gated defense-in-depth — ALL must hold for bypass:
            #   1. is_admin=True               — caller flagged as platform admin by router
            #   2. caller_token_teams_provided — router opted into bypass evaluation and
            #      caller_token_teams is None  — session is un-narrowed (no JWT `teams` claim)
            #   3. caller_permissions == ["*"] — effective permissions are wildcard
            #
            # Why all three: a narrowed admin (is_admin=True, token_teams=["other"]) can still
            # have caller_permissions=["*"] from the global `platform_admin` role, because
            # `_get_caller_permissions` falls through to PermissionService for narrowed
            # sessions. Without the token_teams gate, narrowed admins would silently bypass
            # membership for teams they were never granted access to. Defaulting
            # caller_token_teams_provided=False also keeps callers that haven't been audited
            # for this contract from accidentally satisfying the bypass.
            is_unrestricted_admin = is_admin and caller_token_teams_provided and caller_token_teams is None and caller_permissions is not None and caller_permissions == ["*"]

            # When delegating (caller_email != user_email), enforce team membership
            # even for unrestricted admins to prevent privilege escalation.
            if not is_unrestricted_admin or (caller_email and caller_email != user_email):
                # Verify user is an active member of the team
                membership = self.db.execute(
                    select(EmailTeamMember).where(and_(EmailTeamMember.team_id == team_id, EmailTeamMember.user_email == user_email, EmailTeamMember.is_active.is_(True)))
                ).scalar_one_or_none()

                if not membership:
                    raise ValueError(f"User {user_email} is not an active member of team {team_id}. Only team members can create tokens for the team.")

        # Check for duplicate active token name for this user within the same team scope,
        # matching DB constraint uq_email_api_tokens_user_name_team (user_email, name, team_id).
        # team_id=None tokens are scoped to the global (no-team) bucket.
        if team_id:
            name_check = and_(EmailApiToken.user_email == user_email, EmailApiToken.name == name, EmailApiToken.team_id == team_id, EmailApiToken.is_active.is_(True))
        else:
            name_check = and_(EmailApiToken.user_email == user_email, EmailApiToken.name == name, EmailApiToken.team_id.is_(None), EmailApiToken.is_active.is_(True))
        existing_token = self.db.execute(select(EmailApiToken).where(name_check)).scalar_one_or_none()

        if existing_token:
            scope_label = f"team '{team_id}'" if team_id else "the global scope (no team)"
            raise ValueError(f"Token with name '{name}' already exists for user {user_email} in {scope_label}. Token names must be unique per user per team. Please choose a different name.")

        # CALCULATE EXPIRATION DATE
        expires_at = None
        if expires_in_days:
            expires_at = utc_now() + timedelta(days=expires_in_days)

        # Enforce expiration requirement if configured
        if settings.require_token_expiration and not expires_at:
            raise ValueError("Token expiration is required by server policy (REQUIRE_TOKEN_EXPIRATION=true). Please specify an expiration date for the token.")

        jti = str(uuid.uuid4())  # Unique JWT ID
        # Generate JWT token with all necessary claims
        raw_token = await self._generate_token(user_email=user_email, jti=jti, team_id=team_id, expires_at=expires_at, scope=scope, user=user)  # Pass user object to include admin status

        # Hash token for secure storage
        token_hash = self._hash_token(raw_token)

        # Create database record
        api_token = EmailApiToken(
            id=str(uuid.uuid4()),
            user_email=user_email,
            team_id=team_id,  # Store team association
            name=name,
            jti=jti,
            description=description,
            token_hash=token_hash,  # Store hash, not raw token
            expires_at=expires_at,
            tags=tags or [],
            # Store scoping information
            server_id=scope.server_id if scope else None,
            resource_scopes=scope.permissions if scope else [],
            ip_restrictions=scope.ip_restrictions if scope else [],
            time_restrictions=scope.time_restrictions if scope else {},
            usage_limits=scope.usage_limits if scope else {},
            # Token status
            is_active=is_active,
            created_at=utc_now(),
            last_used=None,
        )

        self.db.add(api_token)
        self.db.commit()
        self.db.refresh(api_token)

        token_type = f"team-scoped (team: {team_id})" if team_id else "public-only"
        logger.info(f"Created {token_type} API token '{name}' for user {SecurityValidator.sanitize_log_message(user_email)}. Token ID: {api_token.id}, Expires: {expires_at or 'Never'}")
        return api_token, raw_token

    async def count_user_tokens(self, user_email: str, include_inactive: bool = False) -> int:
        """Count API tokens for a user.

        Args:
            user_email: User's email address
            include_inactive: Include inactive/expired tokens

        Returns:
            int: Total number of matching tokens
        """
        # pylint: disable=not-callable
        query = select(func.count(EmailApiToken.id)).where(EmailApiToken.user_email == user_email)

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        result = self.db.execute(query)
        return result.scalar() or 0

    async def get_user_team_ids(self, user_email: str) -> List[str]:
        """Get all team IDs the user is a member of.

        Uses TeamManagementService.get_user_teams which is cached and consistent
        with how other services (servers, tools, resources) resolve team visibility.

        Args:
            user_email: User's email address

        Returns:
            List[str]: Team IDs the user belongs to
        """
        # First-Party
        from mcpgateway.services.team_management_service import TeamManagementService  # pylint: disable=import-outside-toplevel

        team_service = TeamManagementService(self.db)
        user_teams = await team_service.get_user_teams(user_email)
        return [team.id for team in user_teams]

    async def count_user_and_team_tokens(self, user_email: str, include_inactive: bool = False) -> int:
        """Count API tokens for a user plus team tokens from teams the user belongs to.

        This combines personal tokens (created by the user) with team-scoped tokens
        from all teams where the user is an active member.

        Args:
            user_email: User's email address
            include_inactive: Include inactive/expired tokens

        Returns:
            int: Total number of matching tokens
        """
        team_ids = await self.get_user_team_ids(user_email)

        # Build query: tokens created by user OR tokens in user's teams
        conditions = [EmailApiToken.user_email == user_email]
        if team_ids:
            conditions.append(EmailApiToken.team_id.in_(team_ids))

        # pylint: disable=not-callable
        query = select(func.count(EmailApiToken.id)).where(or_(*conditions))

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        result = self.db.execute(query)
        return result.scalar() or 0

    async def count_team_tokens(self, team_id: str, include_inactive: bool = False) -> int:
        """Count API tokens for a team.

        Args:
            team_id: Team ID to count tokens for
            include_inactive: Include inactive/expired tokens

        Returns:
            int: Total number of matching tokens
        """
        # pylint: disable=not-callable
        query = select(func.count(EmailApiToken.id)).where(EmailApiToken.team_id == team_id)

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        result = self.db.execute(query)
        return result.scalar() or 0

    async def list_user_tokens(self, user_email: str, include_inactive: bool = False, limit: int = 100, offset: int = 0) -> List[EmailApiToken]:
        """List API tokens for a user.

        Args:
            user_email: User's email address
            include_inactive: Include inactive/expired tokens
            limit: Maximum tokens to return
            offset: Number of tokens to skip

        Returns:
            List[EmailApiToken]: User's API tokens

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns List[EmailApiToken] for user
        """
        # Validate parameters
        if limit <= 0 or limit > 1000:
            limit = 50  # Use default
        offset = max(offset, 0)  # Use default
        query = select(EmailApiToken).where(EmailApiToken.user_email == user_email)

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        query = query.order_by(EmailApiToken.created_at.desc()).limit(limit).offset(offset)

        result = self.db.execute(query)
        return result.scalars().all()

    async def list_team_tokens(
        self,
        team_id: str,
        user_email: str,
        include_inactive: bool = False,
        limit: int = 100,
        offset: int = 0,
        caller_permissions: Optional[List[str]] = None,
        is_admin: bool = False,
        caller_token_teams: Optional[List[str]] = None,
        caller_token_teams_provided: bool = False,
    ) -> List[EmailApiToken]:
        """List API tokens for a team (accessible by any active team member or un-narrowed admin).

        Args:
            team_id: Team ID to list tokens for
            user_email: User's email (must be an active member of the team unless admin bypass applies)
            include_inactive: Include inactive/expired tokens
            limit: Maximum tokens to return
            offset: Number of tokens to skip
            caller_permissions: Caller's effective permissions (for admin bypass check)
            is_admin: Whether the caller is a platform admin (for admin bypass check)
            caller_token_teams: Caller's token narrowing scope from JWT ``teams`` claim. Must be ``None``
                (un-narrowed) for admin bypass to apply. Narrowed (``["team-a"]``) and public-only (``[]``)
                sessions never bypass membership, even when ``caller_permissions == ["*"]`` from a global
                ``platform_admin`` role.
            caller_token_teams_provided: Whether the caller resolved ``caller_token_teams`` from the
                authenticated session. Defaults to ``False``: callers that did not opt in cannot satisfy
                the admin bypass. Routers that intend to allow the bypass must set this to ``True``.

        Returns:
            List[EmailApiToken]: Team's API tokens

        Raises:
            ValueError: If user is not an active member of the team (unless admin bypass applies)
        """
        # Triple-gated admin bypass — see create_token() for the security rationale.
        is_unrestricted_admin = is_admin and caller_token_teams_provided and caller_token_teams is None and caller_permissions is not None and caller_permissions == ["*"]

        if not is_unrestricted_admin:
            team_ids = await self.get_user_team_ids(user_email)

            if team_id not in team_ids:
                raise ValueError(f"User {user_email} is not an active member of team {team_id}")

        # Validate parameters
        if limit <= 0 or limit > 1000:
            limit = 50
        offset = max(offset, 0)

        query = select(EmailApiToken).where(EmailApiToken.team_id == team_id)

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        query = query.order_by(EmailApiToken.created_at.desc()).limit(limit).offset(offset)
        result = self.db.execute(query)
        return result.scalars().all()

    async def list_user_and_team_tokens(self, user_email: str, include_inactive: bool = False, limit: int = 100, offset: int = 0) -> List[EmailApiToken]:
        """List API tokens for a user plus team tokens from teams the user belongs to.

        This combines personal tokens (created by the user) with team-scoped tokens
        from all teams where the user is an active member.

        Args:
            user_email: User's email address
            include_inactive: Include inactive/expired tokens
            limit: Maximum tokens to return
            offset: Number of tokens to skip

        Returns:
            List[EmailApiToken]: Combined list of user's personal tokens and team tokens

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns List[EmailApiToken] including personal and team tokens
        """
        # Validate parameters
        if limit <= 0 or limit > 1000:
            limit = 50
        offset = max(offset, 0)

        team_ids = await self.get_user_team_ids(user_email)

        # Build query: tokens created by user OR tokens in user's teams
        conditions = [EmailApiToken.user_email == user_email]
        if team_ids:
            conditions.append(EmailApiToken.team_id.in_(team_ids))

        query = select(EmailApiToken).where(or_(*conditions))

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        query = query.order_by(EmailApiToken.created_at.desc()).limit(limit).offset(offset)

        result = self.db.execute(query)
        return result.scalars().all()

    async def get_token(self, token_id: str, user_email: Optional[str] = None) -> Optional[EmailApiToken]:
        """Get a specific token by ID.

        Args:
            token_id: Token ID
            user_email: Optional user email filter for security

        Returns:
            Optional[EmailApiToken]: Token if found and authorized

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns Optional[EmailApiToken] if found and authorized
        """
        query = select(EmailApiToken).where(EmailApiToken.id == token_id)

        if user_email:
            query = query.where(EmailApiToken.user_email == user_email)

        result = self.db.execute(query)
        return result.scalar_one_or_none()

    async def update_token(
        self,
        token_id: str,
        user_email: str,
        name: Optional[str] = None,
        description: Optional[str] = None,
        scope: Optional[TokenScope] = None,
        tags: Optional[List[str]] = None,
        caller_permissions: Optional[List[str]] = None,
        is_active: Optional[bool] = None,
    ) -> Optional[EmailApiToken]:
        """Update an existing token with scope containment validation.

        Args:
            token_id: Token ID to update
            user_email: Owner's email for security
            name: New token name
            description: New description
            scope: New scoping configuration
            tags: New tags
            caller_permissions: Caller's effective permissions for scope containment
            is_active: New token active status

        Returns:
            Optional[EmailApiToken]: Updated token if found

        Raises:
            ValueError: If token not found or name conflicts

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns Optional[EmailApiToken] if updated successfully
        """
        token = await self.get_token(token_id, user_email)
        if not token:
            raise ValueError("Token not found or not authorized")

        # Validate scope containment for scope changes
        if scope and scope.permissions:
            self._validate_scope_containment(scope.permissions, caller_permissions)

        # Check for duplicate name if changing
        if name and name != token.name:
            existing = self.db.execute(
                select(EmailApiToken).where(and_(EmailApiToken.user_email == user_email, EmailApiToken.name == name, EmailApiToken.id != token_id, EmailApiToken.is_active.is_(True)))
            ).scalar_one_or_none()

            if existing:
                raise ValueError(f"Token name '{name}' already exists")

            token.name = name

        if description is not None:
            token.description = description

        if tags is not None:
            token.tags = tags

        if is_active is not None:
            token.is_active = is_active

        if scope:
            token.server_id = scope.server_id
            token.resource_scopes = scope.permissions
            token.ip_restrictions = scope.ip_restrictions
            token.time_restrictions = scope.time_restrictions
            token.usage_limits = scope.usage_limits

        self.db.commit()
        self.db.refresh(token)

        logger.info(f"Updated token '{token.name}' for user {SecurityValidator.sanitize_log_message(user_email)}")

        return token

    async def revoke_token(self, token_id: str, user_email: str, revoked_by: str, reason: Optional[str] = None) -> bool:
        """Revoke a token owned by the specified user or in a team the user belongs to.

        Args:
            token_id: Token ID to revoke
            user_email: Caller's email - must own the token or be a member of the token's team
            revoked_by: Email of user performing revocation (for audit)
            reason: Optional reason for revocation

        Returns:
            bool: True if token was revoked, False if not found or not authorized

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns bool: True if token was revoked successfully
        """
        # First try ownership match
        token = await self.get_token(token_id, user_email)

        # If not owned by caller, check if token is in a team the caller is an owner of
        if not token:
            token = await self.get_token(token_id)
            if not token or not token.team_id:
                return False
            # Only team owners (admins) can revoke other members' team tokens
            # First-Party
            from mcpgateway.services.team_management_service import TeamManagementService  # pylint: disable=import-outside-toplevel

            team_service = TeamManagementService(self.db)
            role = await team_service.get_user_role_in_team(user_email, token.team_id)
            if role != "owner":
                return False

        # Mark token as inactive
        token.is_active = False

        # Add to blacklist
        revocation = TokenRevocation(jti=token.jti, revoked_by=revoked_by, reason=reason)

        self.db.add(revocation)
        self.db.commit()

        # Invalidate auth cache synchronously so revoked tokens are rejected immediately
        # (fire-and-forget via create_task risks a race where the next request arrives
        # before the invalidation task runs, allowing the revoked token through).
        try:
            # First-Party
            from mcpgateway.cache.auth_cache import auth_cache  # pylint: disable=import-outside-toplevel

            await auth_cache.invalidate_revocation(token.jti)
        except Exception as cache_error:
            logger.debug(f"Failed to invalidate auth cache for revoked token: {cache_error}")

        logger.info(f"Revoked token '{token.name}' (JTI: {token.jti}) by {revoked_by}")

        return True

    async def admin_revoke_token(self, token_id: str, revoked_by: str, reason: Optional[str] = None) -> bool:
        """Admin-only: Revoke any token without ownership check.

        WARNING: This method bypasses ownership verification.
        Only call from admin-authenticated endpoints.

        Args:
            token_id: Token ID to revoke
            revoked_by: Admin email for audit
            reason: Revocation reason

        Returns:
            bool: True if token was revoked, False if not found

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns bool: True if token was revoked successfully
        """
        # No user filter - admin can revoke any token
        token = await self.get_token(token_id)
        if not token:
            return False

        token.is_active = False
        revocation = TokenRevocation(jti=token.jti, revoked_by=revoked_by, reason=reason)
        self.db.add(revocation)
        self.db.commit()

        try:
            # First-Party
            from mcpgateway.cache.auth_cache import auth_cache  # pylint: disable=import-outside-toplevel

            await auth_cache.invalidate_revocation(token.jti)
        except Exception as cache_error:
            logger.debug(f"Failed to invalidate auth cache: {cache_error}")

        logger.info(f"Admin revoked token '{token.name}' (JTI: {token.jti}) by {revoked_by}")
        return True

    async def is_token_revoked(self, jti: str) -> bool:
        """Check if a token JTI is revoked.

        Args:
            jti: JWT ID to check

        Returns:
            bool: True if token is revoked

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns bool: True if token is revoked
        """
        revocation = self.db.execute(select(TokenRevocation).where(TokenRevocation.jti == jti)).scalar_one_or_none()

        return revocation is not None

    async def log_token_usage(
        self,
        jti: str,
        user_email: str,
        endpoint: Optional[str] = None,
        method: Optional[str] = None,
        ip_address: Optional[str] = None,
        user_agent: Optional[str] = None,
        status_code: Optional[int] = None,
        response_time_ms: Optional[int] = None,
        blocked: bool = False,
        block_reason: Optional[str] = None,
    ) -> None:
        """Log token usage for analytics and security.

        Args:
            jti: JWT ID of token used
            user_email: Token owner's email
            endpoint: API endpoint accessed
            method: HTTP method
            ip_address: Client IP address
            user_agent: Client user agent
            status_code: HTTP response status
            response_time_ms: Response time in milliseconds
            blocked: Whether request was blocked
            block_reason: Reason for blocking

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Logs token usage for analytics - no return value
        """
        usage_log = TokenUsageLog(
            token_jti=jti,
            user_email=user_email,
            endpoint=endpoint,
            method=method,
            ip_address=ip_address,
            user_agent=user_agent,
            status_code=status_code,
            response_time_ms=response_time_ms,
            blocked=blocked,
            block_reason=block_reason,
        )

        self.db.add(usage_log)
        self.db.commit()

    async def get_token_usage_stats(self, user_email: str, token_id: Optional[str] = None, days: int = 30) -> dict:
        """Get token usage statistics.

        Args:
            user_email: User's email address
            token_id: Optional specific token ID
            days: Number of days to analyze

        Returns:
            dict: Usage statistics

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns dict with usage statistics
        """
        start_date = utc_now() - timedelta(days=days)

        # Get token JTI if specific token requested
        token_jti = None
        if token_id:
            token = await self.get_token(token_id, user_email)
            if token:
                token_jti = token.jti

        # Use SQL aggregation for PostgreSQL, Python fallback for SQLite
        dialect_name = self.db.get_bind().dialect.name
        if dialect_name == "postgresql":
            return await self._get_usage_stats_postgresql(user_email, start_date, token_jti, days)
        return await self._get_usage_stats_python(user_email, start_date, token_jti, days)

    async def _get_usage_stats_postgresql(self, user_email: str, start_date: datetime, token_jti: Optional[str], days: int) -> dict:
        """Compute usage stats using PostgreSQL SQL aggregation.

        Args:
            user_email: User's email address
            start_date: Start date for analysis
            token_jti: Optional token JTI filter
            days: Number of days being analyzed

        Returns:
            dict: Usage statistics computed via SQL
        """
        # Build filter conditions
        conditions = [TokenUsageLog.user_email == user_email, TokenUsageLog.timestamp >= start_date]
        if token_jti:
            conditions.append(TokenUsageLog.token_jti == token_jti)

        base_filter = and_(*conditions)

        # Main stats query using SQL aggregation
        # Match Python behavior:
        # - status_code must be non-null AND non-zero AND < 400 for success count
        # - response_time_ms must be non-null AND non-zero for average (Python: if log.response_time_ms)
        stats_query = (
            select(
                func.count().label("total"),  # pylint: disable=not-callable
                func.sum(
                    case(
                        (and_(TokenUsageLog.status_code.isnot(None), TokenUsageLog.status_code > 0, TokenUsageLog.status_code < 400), 1),
                        else_=0,
                    )
                ).label("successful"),
                func.sum(case((TokenUsageLog.blocked.is_(True), 1), else_=0)).label("blocked"),
                # Only average non-null and non-zero response times (NULL values are ignored by AVG)
                func.avg(
                    case(
                        (and_(TokenUsageLog.response_time_ms.isnot(None), TokenUsageLog.response_time_ms > 0), TokenUsageLog.response_time_ms),
                        else_=None,
                    )
                ).label("avg_response"),
            )
            .select_from(TokenUsageLog)
            .where(base_filter)
        )

        result = self.db.execute(stats_query).fetchone()

        total_requests = result.total or 0
        successful_requests = result.successful or 0
        blocked_requests = result.blocked or 0
        avg_response_time = float(result.avg_response) if result.avg_response else 0.0

        # Top endpoints query using SQL GROUP BY
        # Match Python behavior: exclude None AND empty string endpoints (Python: if log.endpoint)
        endpoints_query = (
            select(TokenUsageLog.endpoint, func.count().label("count"))  # pylint: disable=not-callable
            .where(and_(base_filter, TokenUsageLog.endpoint.isnot(None), TokenUsageLog.endpoint != ""))
            .group_by(TokenUsageLog.endpoint)
            .order_by(func.count().desc())  # pylint: disable=not-callable
            .limit(5)
        )

        endpoints_result = self.db.execute(endpoints_query).fetchall()
        top_endpoints = [(row.endpoint, row.count) for row in endpoints_result]

        return {
            "period_days": days,
            "total_requests": total_requests,
            "successful_requests": successful_requests,
            "blocked_requests": blocked_requests,
            "success_rate": successful_requests / total_requests if total_requests > 0 else 0,
            "average_response_time_ms": round(avg_response_time, 2),
            "top_endpoints": top_endpoints,
        }

    async def _get_usage_stats_python(self, user_email: str, start_date: datetime, token_jti: Optional[str], days: int) -> dict:
        """Compute usage stats using Python (fallback for SQLite).

        Args:
            user_email: User's email address
            start_date: Start date for analysis
            token_jti: Optional token JTI filter
            days: Number of days being analyzed

        Returns:
            dict: Usage statistics computed in Python
        """
        query = select(TokenUsageLog).where(and_(TokenUsageLog.user_email == user_email, TokenUsageLog.timestamp >= start_date))

        if token_jti:
            query = query.where(TokenUsageLog.token_jti == token_jti)

        usage_logs = self.db.execute(query).scalars().all()

        # Calculate statistics
        total_requests = len(usage_logs)
        successful_requests = sum(1 for log in usage_logs if log.status_code and log.status_code < 400)
        blocked_requests = sum(1 for log in usage_logs if log.blocked)

        # Average response time
        response_times = [log.response_time_ms for log in usage_logs if log.response_time_ms]
        avg_response_time = sum(response_times) / len(response_times) if response_times else 0

        # Most accessed endpoints
        endpoint_counts: dict = {}
        for log in usage_logs:
            if log.endpoint:
                endpoint_counts[log.endpoint] = endpoint_counts.get(log.endpoint, 0) + 1

        top_endpoints = sorted(endpoint_counts.items(), key=lambda x: x[1], reverse=True)[:5]

        return {
            "period_days": days,
            "total_requests": total_requests,
            "successful_requests": successful_requests,
            "blocked_requests": blocked_requests,
            "success_rate": successful_requests / total_requests if total_requests > 0 else 0,
            "average_response_time_ms": round(avg_response_time, 2),
            "top_endpoints": top_endpoints,
        }

    async def get_token_revocation(self, jti: str) -> Optional[TokenRevocation]:
        """Get token revocation information by JTI.

        Args:
            jti: JWT token ID

        Returns:
            Optional[TokenRevocation]: Revocation info if token is revoked

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns Optional[TokenRevocation] if token is revoked
        """
        result = self.db.execute(select(TokenRevocation).where(TokenRevocation.jti == jti))
        return result.scalar_one_or_none()

    async def get_token_revocations_batch(self, jtis: List[str]) -> Dict[str, TokenRevocation]:
        """Get token revocation information for multiple JTIs in a single query.

        Args:
            jtis: List of JWT token IDs

        Returns:
            Dict mapping JTI to TokenRevocation for revoked tokens only.
        """
        if not jtis:
            return {}
        result = self.db.execute(select(TokenRevocation).where(TokenRevocation.jti.in_(jtis)))
        return {rev.jti: rev for rev in result.scalars().all()}

    async def list_all_tokens(self, include_inactive: bool = False, limit: int = 100, offset: int = 0) -> List[EmailApiToken]:
        """List all API tokens (admin only).

        Args:
            include_inactive: Include inactive/expired tokens
            limit: Maximum tokens to return
            offset: Number of tokens to skip

        Returns:
            List[EmailApiToken]: All API tokens
        """
        if limit <= 0 or limit > 1000:
            limit = 50
        offset = max(offset, 0)

        query = select(EmailApiToken)

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        query = query.order_by(EmailApiToken.created_at.desc()).limit(limit).offset(offset)

        result = self.db.execute(query)
        return result.scalars().all()

    async def count_all_tokens(self, include_inactive: bool = False) -> int:
        """Count all API tokens (admin only).

        Args:
            include_inactive: Include inactive/expired tokens in count

        Returns:
            int: Total count of all tokens
        """
        query = select(func.count(EmailApiToken.id))  # pylint: disable=not-callable

        if not include_inactive:
            query = query.where(and_(EmailApiToken.is_active.is_(True), or_(EmailApiToken.expires_at.is_(None), EmailApiToken.expires_at > utc_now())))

        result = self.db.execute(query)
        return result.scalar() or 0

    async def cleanup_expired_tokens(self) -> int:
        """Clean up expired tokens using bulk UPDATE.

        Uses a single SQL UPDATE statement instead of loading tokens into memory
        and updating them one by one. This is more efficient and avoids memory
        issues when many tokens expire at once.

        Returns:
            int: Number of tokens cleaned up

        Examples:
            >>> service = TokenCatalogService(None)  # Would use real DB session
            >>> # Returns int: Number of tokens cleaned up
        """
        try:
            now = utc_now()
            count = self.db.query(EmailApiToken).filter(EmailApiToken.expires_at < now, EmailApiToken.is_active.is_(True)).update({"is_active": False}, synchronize_session=False)

            self.db.commit()

            if count > 0:
                logger.info(f"Cleaned up {count} expired tokens")

            return count

        except Exception as e:
            self.db.rollback()
            logger.error(f"Failed to cleanup expired tokens: {e}")
            return 0
