xiaoxia-saas/packages/adapters/in_memory/user_repository.py

"""
用户仓储 In-Memory 实现
"""

import copy
from typing import Dict, Optional

from packages.domain.entities import User
from packages.ports.user_repository import UserRepository


class InMemoryUserRepository(UserRepository):
    """用户仓储内存实现"""

    def __init__(self):
        self._users: Dict[str, User] = {}
        self._email_index: Dict[str, str] = {}  # email -> user_id
        self._username_index: Dict[str, str] = {}  # username -> user_id
        self._verification_token_index: Dict[str, str] = {}  # token -> user_id
        self._reset_token_index: Dict[str, str] = {}  # token -> user_id
        self._wechat_openid_index: Dict[str, str] = {}  # openid -> user_id
        self._wechat_unionid_index: Dict[str, str] = {}  # unionid -> user_id
        self._phone_index: Dict[str, str] = {}  # phone -> user_id

    def save(self, user: User) -> None:
        """保存用户（存储独立副本，避免外部修改影响内部状态）"""
        # 第一步：唯一性约束检查（O(1)，全部检查通过再动数据）
        # email 是必填字段，做空值防御
        if not user.email:
            raise ValueError("User email cannot be empty")
        new_email = user.email.lower()
        existing_id = self._email_index.get(new_email)
        if existing_id and existing_id != user.id:
            raise ValueError(f"Email already in use: {user.email}")

        new_username = user.username.lower() if user.username else None
        if new_username:
            existing_id = self._username_index.get(new_username)
            if existing_id and existing_id != user.id:
                raise ValueError(f"Username already in use: {user.username}")

        if user.phone:
            existing_id = self._phone_index.get(user.phone)
            if existing_id and existing_id != user.id:
                raise ValueError(f"Phone already in use: {user.phone}")

        if user.wechat_openid:
            existing_id = self._wechat_openid_index.get(user.wechat_openid)
            if existing_id and existing_id != user.id:
                raise ValueError(f"WeChat openid already in use: {user.wechat_openid}")

        if user.wechat_unionid:
            existing_id = self._wechat_unionid_index.get(user.wechat_unionid)
            if existing_id and existing_id != user.id:
                raise ValueError(f"WeChat unionid already in use: {user.wechat_unionid}")

        # 验证令牌和重置令牌也做唯一性防御（防止生成器异常导致重复）
        if user.email_verification_token:
            existing_id = self._verification_token_index.get(user.email_verification_token)
            if existing_id and existing_id != user.id:
                raise ValueError("Email verification token already in use")

        if user.password_reset_token:
            existing_id = self._reset_token_index.get(user.password_reset_token)
            if existing_id and existing_id != user.id:
                raise ValueError("Password reset token already in use")

        # 第二步：如果是更新，用旧对象属性清理旧索引（O(1)，因存储的是独立副本）
        old_user = self._users.get(user.id)
        if old_user is not None:
            self._remove_indexes_of_user(old_user)

        # 第三步：存储独立副本 + 写入新索引
        stored_user = copy.copy(user)
        self._users[user.id] = stored_user
        self._email_index[new_email] = user.id
        if new_username:
            self._username_index[new_username] = user.id
        if user.email_verification_token:
            self._verification_token_index[user.email_verification_token] = user.id
        if user.password_reset_token:
            self._reset_token_index[user.password_reset_token] = user.id
        if user.wechat_openid:
            self._wechat_openid_index[user.wechat_openid] = user.id
        if user.wechat_unionid:
            self._wechat_unionid_index[user.wechat_unionid] = user.id
        if user.phone:
            self._phone_index[user.phone] = user.id

    def _remove_indexes_of_user(self, user: User) -> None:
        """利用已知用户对象属性清理所有索引，时间复杂度 O(1)."""
        if user.email:
            self._email_index.pop(user.email.lower(), None)
        if user.username:
            self._username_index.pop(user.username.lower(), None)
        if user.email_verification_token:
            self._verification_token_index.pop(user.email_verification_token, None)
        if user.password_reset_token:
            self._reset_token_index.pop(user.password_reset_token, None)
        if user.wechat_openid:
            self._wechat_openid_index.pop(user.wechat_openid, None)
        if user.wechat_unionid:
            self._wechat_unionid_index.pop(user.wechat_unionid, None)
        if user.phone:
            self._phone_index.pop(user.phone, None)

    def find_by_id(self, user_id: str) -> Optional[User]:
        """根据 ID 查找用户"""
        return self._users.get(user_id)

    def find_by_email(self, email: str) -> Optional[User]:
        """根据邮箱查找用户"""
        if not email:
            return None
        user_id = self._email_index.get(email.lower())
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_username(self, username: str) -> Optional[User]:
        """根据用户名查找用户"""
        if not username:
            return None
        user_id = self._username_index.get(username.lower())
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_verification_token(self, token: str) -> Optional[User]:
        """根据邮箱验证令牌查找用户"""
        user_id = self._verification_token_index.get(token)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_password_reset_token(self, token: str) -> Optional[User]:
        """根据密码重置令牌查找用户"""
        user_id = self._reset_token_index.get(token)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_wechat_openid(self, openid: str) -> Optional[User]:
        """根据微信 openid 查找用户"""
        user_id = self._wechat_openid_index.get(openid)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_wechat_unionid(self, unionid: str) -> Optional[User]:
        """根据微信 unionid 查找用户"""
        if not unionid:
            return None
        user_id = self._wechat_unionid_index.get(unionid)
        if user_id:
            return self._users.get(user_id)
        return None

    def find_by_phone(self, phone: str) -> Optional[User]:
        """根据手机号查找用户"""
        if not phone:
            return None
        user_id = self._phone_index.get(phone)
        if user_id:
            return self._users.get(user_id)
        return None

    def delete(self, user_id: str) -> bool:
        """删除用户"""
        user = self._users.get(user_id)
        if user is None:
            return False

        # O(1) 清理所有索引
        self._remove_indexes_of_user(user)

        # 删除用户
        del self._users[user_id]
        return True