from typing import Optional, List, Dict, Any from datetime import datetime import sys # Add shared modules to path sys.path.append('/app') from app.models.user import User, UserStatus from app.schemas.user import UserResponse, UserUpdate from bson import ObjectId class UserService: @staticmethod async def get_user_by_id(user_id: str) -> Optional[User]: """Get user by ID""" try: return await User.get(ObjectId(user_id)) except Exception: return None @staticmethod async def get_user_by_email(email: str) -> Optional[User]: """Get user by email""" return await User.find_one(User.email == email) @staticmethod async def get_user_by_username(username: str) -> Optional[User]: """Get user by username""" return await User.find_one(User.username == username.lower()) @staticmethod async def update_user(user_id: str, update_data: UserUpdate) -> Optional[User]: """Update user information""" try: user = await User.get(ObjectId(user_id)) if not user: return None # Update fields if provided if update_data.username is not None: # Check if username is already taken existing_user = await User.find_one( User.username == update_data.username.lower(), User.id != user.id ) if existing_user: raise ValueError("Username already taken") user.username = update_data.username.lower() if update_data.status is not None: user.status = update_data.status if update_data.email_verified is not None: user.email_verified = update_data.email_verified if update_data.email_verified and user.status == UserStatus.PENDING_VERIFICATION: user.status = UserStatus.ACTIVE user.updated_at = datetime.utcnow() await user.save() return user except Exception as e: raise ValueError(f"Failed to update user: {str(e)}") @staticmethod async def deactivate_user(user_id: str, reason: str = "manual") -> bool: """Deactivate a user account""" try: user = await User.get(ObjectId(user_id)) if not user: return False user.status = UserStatus.INACTIVE user.updated_at = datetime.utcnow() await user.save() # Revoke all active sessions and tokens from app.services.auth_service import AuthService await AuthService.logout_user(user_id, logout_all=True) # Send deactivation event from shared.kafka_client import kafka_client, EventTypes, Topics await kafka_client.send_message( Topics.AUTH_EVENTS, { "event_type": EventTypes.ACCOUNT_DEACTIVATED, "user_id": user_id, "email": user.email, "username": user.username, "reason": reason, "timestamp": datetime.utcnow().isoformat() }, key=user_id ) return True except Exception: return False @staticmethod async def activate_user(user_id: str) -> bool: """Activate a user account""" try: user = await User.get(ObjectId(user_id)) if not user: return False user.status = UserStatus.ACTIVE user.updated_at = datetime.utcnow() await user.save() # Send activation event from shared.kafka_client import kafka_client, EventTypes, Topics await kafka_client.send_message( Topics.AUTH_EVENTS, { "event_type": EventTypes.ACCOUNT_ACTIVATED, "user_id": user_id, "email": user.email, "username": user.username, "timestamp": datetime.utcnow().isoformat() }, key=user_id ) return True except Exception: return False @staticmethod async def suspend_user(user_id: str, reason: str = "manual") -> bool: """Suspend a user account""" try: user = await User.get(ObjectId(user_id)) if not user: return False user.status = UserStatus.SUSPENDED user.updated_at = datetime.utcnow() await user.save() # Revoke all active sessions and tokens from app.services.auth_service import AuthService await AuthService.logout_user(user_id, logout_all=True) return True except Exception: return False @staticmethod async def get_user_stats(user_id: str) -> Dict[str, Any]: """Get user statistics""" try: user = await User.get(ObjectId(user_id)) if not user: return {} # Get session count from app.models.auth_session import AuthSession active_sessions = await AuthSession.find( AuthSession.user_id == user.id, AuthSession.is_active == True ).count() # Get refresh token count from app.models.refresh_token import RefreshToken active_tokens = await RefreshToken.find( RefreshToken.user_id == user.id, RefreshToken.is_active == True ).count() return { "user_id": str(user.id), "email": user.email, "username": user.username, "status": user.status.value, "email_verified": user.email_verified, "created_at": user.created_at, "last_successful_login": user.last_successful_login, "failed_login_attempts": user.failed_login_attempts, "is_locked": user.is_locked(), "active_sessions": active_sessions, "active_tokens": active_tokens } except Exception: return {} @staticmethod async def search_users( query: str = "", status: Optional[UserStatus] = None, limit: int = 50, offset: int = 0 ) -> Dict[str, Any]: """Search users with filters""" try: # Build query filters filters = [] if query: filters.append({ "$or": [ {"email": {"$regex": query, "$options": "i"}}, {"username": {"$regex": query, "$options": "i"}} ] }) if status: filters.append({"status": status}) # Combine filters query_filter = {"$and": filters} if filters else {} # Execute query users = await User.find(query_filter).skip(offset).limit(limit).to_list() total_count = await User.find(query_filter).count() # Convert to response format user_responses = [] for user in users: user_responses.append(UserResponse( id=str(user.id), email=user.email, username=user.username, status=user.status, email_verified=user.email_verified, created_at=user.created_at, last_successful_login=user.last_successful_login )) return { "users": [user.dict() for user in user_responses], "total_count": total_count, "offset": offset, "limit": limit } except Exception as e: return { "users": [], "total_count": 0, "offset": offset, "limit": limit, "error": str(e) } @staticmethod async def get_user_login_history(user_id: str, limit: int = 10) -> List[Dict[str, Any]]: """Get user login history from sessions""" try: from app.models.auth_session import AuthSession sessions = await AuthSession.find( AuthSession.user_id == ObjectId(user_id) ).sort(-AuthSession.created_at).limit(limit).to_list() history = [] for session in sessions: history.append({ "session_id": str(session.id), "login_time": session.created_at, "ip_address": session.ip_address, "user_agent": session.user_agent, "device_info": session.device_info, "last_accessed": session.last_accessed, "is_active": session.is_active and not session.is_expired(), "revoked_at": session.revoked_at, "revocation_reason": session.revocation_reason }) return history except Exception: return []