/**
 * 用户服务控制器
 * 提供用户管理接口
 */
@RestController
@RequestMapping("/api/user")
@Slf4j
public class UserController {

    @Autowired
    private UserService userService;
    
    @Autowired
    private UserQueryService userQueryService;
    
    @Autowired
    private AuthService authService;

    /**
     * 用户注册
     */
    @PostMapping("/register")
    public Result<UserInfo> register(@RequestBody @Valid UserRegisterRequest request) {
        try {
            UserInfo userInfo = userService.register(request);
            return Result.success(userInfo);
        } catch (Exception e) {
            log.error("用户注册失败: error={}", e.getMessage(), e);
            return Result.error("用户注册失败: " + e.getMessage());
        }
    }

    /**
     * 用户登录
     */
    @PostMapping("/login")
    public Result<LoginResponse> login(@RequestBody @Valid UserLoginRequest request) {
        try {
            LoginResponse loginResponse = authService.login(request);
            return Result.success(loginResponse);
        } catch (Exception e) {
            log.error("用户登录失败: error={}", e.getMessage(), e);
            return Result.error("用户登录失败: " + e.getMessage());
        }
    }

    /**
     * 获取用户信息
     */
    @GetMapping("/{userId}")
    public Result<UserInfo> getUserInfo(@PathVariable String userId) {
        try {
            UserInfo userInfo = userQueryService.getUserInfo(userId);
            return Result.success(userInfo);
        } catch (Exception e) {
            log.error("获取用户信息失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            return Result.error("获取用户信息失败: " + e.getMessage());
        }
    }

    /**
     * 更新用户信息
     */
    @PutMapping("/{userId}")
    public Result<Void> updateUserInfo(
            @PathVariable String userId,
            @RequestBody @Valid UserUpdateRequest request) {
        try {
            userService.updateUserInfo(userId, request);
            return Result.success();
        } catch (Exception e) {
            log.error("更新用户信息失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            return Result.error("更新用户信息失败: " + e.getMessage());
        }
    }

    /**
     * 修改密码
     */
    @PostMapping("/{userId}/password")
    public Result<Void> changePassword(
            @PathVariable String userId,
            @RequestBody @Valid PasswordChangeRequest request) {
        try {
            userService.changePassword(userId, request);
            return Result.success();
        } catch (Exception e) {
            log.error("修改密码失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            return Result.error("修改密码失败: " + e.getMessage());
        }
    }

    /**
     * 获取用户列表
     */
    @GetMapping("/list")
    public Result<PageResult<UserInfo>> getUserList(
            @RequestParam(required = false) String keyword,
            @RequestParam(required = false) String status,
            @RequestParam(required = false) String startTime,
            @RequestParam(required = false) String endTime,
            @RequestParam(defaultValue = "1") Integer pageNum,
            @RequestParam(defaultValue = "10") Integer pageSize) {
        try {
            PageResult<UserInfo> result = userQueryService.getUserList(
                    keyword, status, startTime, endTime, pageNum, pageSize);
            return Result.success(result);
        } catch (Exception e) {
            log.error("获取用户列表失败: error={}", e.getMessage(), e);
            return Result.error("获取用户列表失败: " + e.getMessage());
        }
    }

    /**
     * 获取用户统计
     */
    @GetMapping("/statistics")
    public Result<UserStatistics> getUserStatistics(
            @RequestParam(required = false) String startTime,
            @RequestParam(required = false) String endTime) {
        try {
            UserStatistics statistics = userQueryService.getUserStatistics(startTime, endTime);
            return Result.success(statistics);
        } catch (Exception e) {
            log.error("获取用户统计失败: error={}", e.getMessage(), e);
            return Result.error("获取用户统计失败: " + e.getMessage());
        }
    }
}

/**
 * 用户服务实现
 * 负责用户注册、信息管理、密码修改
 */
@Service
@Slf4j
public class UserService {

    @Autowired
    private UserMapper userMapper;
    
    @Autowired
    private UserProfileMapper userProfileMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Autowired
    private PointsServiceClient pointsServiceClient;
    
    @Autowired
    private NotificationServiceClient notificationServiceClient;
    
    @Autowired
    private DistributedLock distributedLock;
    
    @Autowired
    private PasswordEncoder passwordEncoder;

    /**
     * 用户注册
     */
    @Transactional(rollbackFor = Exception.class)
    public UserInfo register(UserRegisterRequest request) {
        String lockKey = "user:register:" + request.getUsername();
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // 1. 获取分布式锁
            boolean lockAcquired = distributedLock.tryLock(lockKey, lockValue, 10, TimeUnit.SECONDS);
            if (!lockAcquired) {
                throw new BusinessException("注册中，请稍候");
            }
            
            // 2. 验证用户名是否已存在
            User existingUser = userMapper.selectOne(
                    new LambdaQueryWrapper<User>()
                            .eq(User::getUsername, request.getUsername()));
            if (existingUser != null) {
                throw new BusinessException("用户名已存在");
            }
            
            // 3. 验证手机号是否已存在
            if (request.getPhone() != null && !request.getPhone().isEmpty()) {
                User existingPhoneUser = userMapper.selectOne(
                        new LambdaQueryWrapper<User>()
                                .eq(User::getPhone, request.getPhone()));
                if (existingPhoneUser != null) {
                    throw new BusinessException("手机号已存在");
                }
            }
            
            // 4. 验证邮箱是否已存在
            if (request.getEmail() != null && !request.getEmail().isEmpty()) {
                User existingEmailUser = userMapper.selectOne(
                        new LambdaQueryWrapper<User>()
                                .eq(User::getEmail, request.getEmail()));
                if (existingEmailUser != null) {
                    throw new BusinessException("邮箱已存在");
                }
            }
            
            // 5. 生成用户ID
            String userId = generateUserId();
            
            // 6. 加密密码
            String encodedPassword = passwordEncoder.encode(request.getPassword());
            
            // 7. 创建用户
            User user = new User();
            user.setUserId(userId);
            user.setUsername(request.getUsername());
            user.setPassword(encodedPassword);
            user.setPhone(request.getPhone());
            user.setEmail(request.getEmail());
            user.setStatus("ACTIVE");
            user.setCreateTime(LocalDateTime.now());
            user.setUpdateTime(LocalDateTime.now());
            
            userMapper.insert(user);
            
            // 8. 创建用户资料
            UserProfile userProfile = new UserProfile();
            userProfile.setUserId(userId);
            userProfile.setNickname(request.getNickname());
            userProfile.setAvatar(request.getAvatar());
            userProfile.setGender(request.getGender());
            userProfile.setBirthday(request.getBirthday());
            userProfile.setCreateTime(LocalDateTime.now());
            userProfile.setUpdateTime(LocalDateTime.now());
            
            userProfileMapper.insert(userProfile);
            
            // 9. 初始化用户积分
            pointsServiceClient.initUserPoints(userId);
            
            // 10. 构建用户信息
            UserInfo userInfo = convertToUserInfo(user, userProfile);
            
            // 11. 缓存用户信息
            String cacheKey = "user:info:" + userId;
            redisTemplate.opsForValue().set(cacheKey, userInfo, 1, TimeUnit.HOURS);
            
            // 12. 发送用户注册事件
            sendUserRegisterEvent(userInfo);
            
            // 13. 发送注册通知
            notificationServiceClient.sendRegisterNotification(userId, request.getPhone(), request.getEmail());
            
            log.info("用户注册成功: userId={}, username={}", userId, request.getUsername());
            
            return userInfo;
            
        } catch (Exception e) {
            log.error("用户注册失败: username={}, error={}", 
                    request.getUsername(), e.getMessage(), e);
            throw new BusinessException("用户注册失败: " + e.getMessage());
        } finally {
            // 释放分布式锁
            distributedLock.releaseLock(lockKey, lockValue);
        }
    }

    /**
     * 更新用户信息
     */
    @Transactional(rollbackFor = Exception.class)
    public void updateUserInfo(String userId, UserUpdateRequest request) {
        try {
            // 1. 获取用户
            User user = userMapper.selectOne(
                    new LambdaQueryWrapper<User>()
                            .eq(User::getUserId, userId));
            if (user == null) {
                throw new BusinessException("用户不存在");
            }
            
            // 2. 更新用户基本信息
            if (request.getPhone() != null) {
                // 验证手机号是否已被其他用户使用
                User existingPhoneUser = userMapper.selectOne(
                        new LambdaQueryWrapper<User>()
                                .eq(User::getPhone, request.getPhone())
                                .ne(User::getUserId, userId));
                if (existingPhoneUser != null) {
                    throw new BusinessException("手机号已被使用");
                }
                user.setPhone(request.getPhone());
            }
            
            if (request.getEmail() != null) {
                // 验证邮箱是否已被其他用户使用
                User existingEmailUser = userMapper.selectOne(
                        new LambdaQueryWrapper<User>()
                                .eq(User::getEmail, request.getEmail())
                                .ne(User::getUserId, userId));
                if (existingEmailUser != null) {
                    throw new BusinessException("邮箱已被使用");
                }
                user.setEmail(request.getEmail());
            }
            
            user.setUpdateTime(LocalDateTime.now());
            userMapper.updateById(user);
            
            // 3. 更新用户资料
            UserProfile userProfile = userProfileMapper.selectOne(
                    new LambdaQueryWrapper<UserProfile>()
                            .eq(UserProfile::getUserId, userId));
            if (userProfile == null) {
                userProfile = new UserProfile();
                userProfile.setUserId(userId);
                userProfile.setCreateTime(LocalDateTime.now());
            }
            
            if (request.getNickname() != null) {
                userProfile.setNickname(request.getNickname());
            }
            if (request.getAvatar() != null) {
                userProfile.setAvatar(request.getAvatar());
            }
            if (request.getGender() != null) {
                userProfile.setGender(request.getGender());
            }
            if (request.getBirthday() != null) {
                userProfile.setBirthday(request.getBirthday());
            }
            userProfile.setUpdateTime(LocalDateTime.now());
            
            if (userProfile.getId() == null) {
                userProfileMapper.insert(userProfile);
            } else {
                userProfileMapper.updateById(userProfile);
            }
            
            // 4. 清除缓存
            String cacheKey = "user:info:" + userId;
            redisTemplate.delete(cacheKey);
            
            // 5. 发送用户信息变更事件
            sendUserInfoChangeEvent(userId);
            
            log.info("更新用户信息成功: userId={}", userId);
            
        } catch (Exception e) {
            log.error("更新用户信息失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            throw new BusinessException("更新用户信息失败: " + e.getMessage());
        }
    }

    /**
     * 修改密码
     */
    @Transactional(rollbackFor = Exception.class)
    public void changePassword(String userId, PasswordChangeRequest request) {
        try {
            // 1. 获取用户
            User user = userMapper.selectOne(
                    new LambdaQueryWrapper<User>()
                            .eq(User::getUserId, userId));
            if (user == null) {
                throw new BusinessException("用户不存在");
            }
            
            // 2. 验证旧密码
            if (!passwordEncoder.matches(request.getOldPassword(), user.getPassword())) {
                throw new BusinessException("旧密码错误");
            }
            
            // 3. 加密新密码
            String encodedPassword = passwordEncoder.encode(request.getNewPassword());
            
            // 4. 更新密码
            user.setPassword(encodedPassword);
            user.setUpdateTime(LocalDateTime.now());
            userMapper.updateById(user);
            
            // 5. 清除用户会话
            String sessionKey = "user:session:" + userId;
            redisTemplate.delete(sessionKey);
            
            // 6. 发送密码修改事件
            sendPasswordChangeEvent(userId);
            
            // 7. 发送密码修改通知
            notificationServiceClient.sendPasswordChangeNotification(userId, user.getPhone(), user.getEmail());
            
            log.info("修改密码成功: userId={}", userId);
            
        } catch (Exception e) {
            log.error("修改密码失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            throw new BusinessException("修改密码失败: " + e.getMessage());
        }
    }

    /**
     * 生成用户ID
     */
    private String generateUserId() {
        return "U" + System.currentTimeMillis() + 
                String.format("%06d", new Random().nextInt(1000000));
    }

    /**
     * 转换为用户信息
     */
    private UserInfo convertToUserInfo(User user, UserProfile userProfile) {
        UserInfo userInfo = new UserInfo();
        userInfo.setUserId(user.getUserId());
        userInfo.setUsername(user.getUsername());
        userInfo.setPhone(user.getPhone());
        userInfo.setEmail(user.getEmail());
        userInfo.setStatus(user.getStatus());
        userInfo.setCreateTime(user.getCreateTime());
        
        if (userProfile != null) {
            userInfo.setNickname(userProfile.getNickname());
            userInfo.setAvatar(userProfile.getAvatar());
            userInfo.setGender(userProfile.getGender());
            userInfo.setBirthday(userProfile.getBirthday());
        }
        
        return userInfo;
    }

    /**
     * 发送用户注册事件
     */
    private void sendUserRegisterEvent(UserInfo userInfo) {
        try {
            UserEvent event = new UserEvent();
            event.setEventType("USER_REGISTER");
            event.setUserId(userInfo.getUserId());
            event.setUserInfo(userInfo);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("user.register", userInfo.getUserId(), message);
            
        } catch (Exception e) {
            log.error("发送用户注册事件失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 发送用户信息变更事件
     */
    private void sendUserInfoChangeEvent(String userId) {
        try {
            UserInfoChangeEvent event = new UserInfoChangeEvent();
            event.setEventType("USER_INFO_CHANGE");
            event.setUserId(userId);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("user.info.change", userId, message);
            
        } catch (Exception e) {
            log.error("发送用户信息变更事件失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 发送密码修改事件
     */
    private void sendPasswordChangeEvent(String userId) {
        try {
            PasswordChangeEvent event = new PasswordChangeEvent();
            event.setEventType("PASSWORD_CHANGE");
            event.setUserId(userId);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("user.password.change", userId, message);
            
        } catch (Exception e) {
            log.error("发送密码修改事件失败: error={}", e.getMessage(), e);
        }
    }
}

/**
 * 用户查询服务实现
 * 负责用户查询、统计
 */
@Service
@Slf4j
public class UserQueryService {

    @Autowired
    private UserMapper userMapper;
    
    @Autowired
    private UserProfileMapper userProfileMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 获取用户信息
     */
    public UserInfo getUserInfo(String userId) {
        try {
            // 1. 从缓存获取
            String cacheKey = "user:info:" + userId;
            UserInfo cachedInfo = (UserInfo) redisTemplate.opsForValue().get(cacheKey);
            if (cachedInfo != null) {
                return cachedInfo;
            }
            
            // 2. 从数据库查询
            User user = userMapper.selectOne(
                    new LambdaQueryWrapper<User>()
                            .eq(User::getUserId, userId));
            if (user == null) {
                throw new BusinessException("用户不存在");
            }
            
            // 3. 查询用户资料
            UserProfile userProfile = userProfileMapper.selectOne(
                    new LambdaQueryWrapper<UserProfile>()
                            .eq(UserProfile::getUserId, userId));
            
            // 4. 构建用户信息
            UserInfo userInfo = convertToUserInfo(user, userProfile);
            
            // 5. 缓存用户信息
            redisTemplate.opsForValue().set(cacheKey, userInfo, 1, TimeUnit.HOURS);
            
            return userInfo;
            
        } catch (Exception e) {
            log.error("获取用户信息失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            throw new BusinessException("获取用户信息失败: " + e.getMessage());
        }
    }

    /**
     * 获取用户列表
     */
    public PageResult<UserInfo> getUserList(String keyword, String status, 
                                            String startTime, String endTime,
                                            Integer pageNum, Integer pageSize) {
        try {
            // 1. 构建查询条件
            LambdaQueryWrapper<User> wrapper = new LambdaQueryWrapper<>();
            if (keyword != null && !keyword.isEmpty()) {
                wrapper.and(w -> w.like(User::getUsername, keyword)
                        .or().like(User::getPhone, keyword)
                        .or().like(User::getEmail, keyword));
            }
            if (status != null && !status.isEmpty()) {
                wrapper.eq(User::getStatus, status);
            }
            if (startTime != null && !startTime.isEmpty()) {
                wrapper.ge(User::getCreateTime, LocalDateTime.parse(startTime));
            }
            if (endTime != null && !endTime.isEmpty()) {
                wrapper.le(User::getCreateTime, LocalDateTime.parse(endTime));
            }
            wrapper.orderByDesc(User::getCreateTime);
            
            // 2. 分页查询
            Page<User> page = new Page<>(pageNum, pageSize);
            Page<User> result = userMapper.selectPage(page, wrapper);
            
            // 3. 转换为用户信息列表
            List<UserInfo> userInfoList = result.getRecords().stream()
                    .map(user -> {
                        UserProfile userProfile = userProfileMapper.selectOne(
                                new LambdaQueryWrapper<UserProfile>()
                                        .eq(UserProfile::getUserId, user.getUserId()));
                        return convertToUserInfo(user, userProfile);
                    })
                    .collect(Collectors.toList());
            
            // 4. 构建分页结果
            PageResult<UserInfo> pageResult = new PageResult<>();
            pageResult.setList(userInfoList);
            pageResult.setTotal(result.getTotal());
            pageResult.setPageNum(pageNum);
            pageResult.setPageSize(pageSize);
            
            return pageResult;
            
        } catch (Exception e) {
            log.error("获取用户列表失败: error={}", e.getMessage(), e);
            throw new BusinessException("获取用户列表失败: " + e.getMessage());
        }
    }

    /**
     * 获取用户统计
     */
    public UserStatistics getUserStatistics(String startTime, String endTime) {
        try {
            // 1. 构建查询条件
            LambdaQueryWrapper<User> wrapper = new LambdaQueryWrapper<>();
            if (startTime != null && !startTime.isEmpty()) {
                wrapper.ge(User::getCreateTime, LocalDateTime.parse(startTime));
            }
            if (endTime != null && !endTime.isEmpty()) {
                wrapper.le(User::getCreateTime, LocalDateTime.parse(endTime));
            }
            
            // 2. 查询用户列表
            List<User> users = userMapper.selectList(wrapper);
            
            // 3. 统计用户数据
            UserStatistics statistics = new UserStatistics();
            statistics.setTotalCount(users.size());
            
            Map<String, Long> statusCount = users.stream()
                    .collect(Collectors.groupingBy(User::getStatus, Collectors.counting()));
            statistics.setStatusCount(statusCount);
            
            // 4. 按日期统计注册用户数
            Map<String, Long> dailyRegisterCount = users.stream()
                    .collect(Collectors.groupingBy(
                            user -> user.getCreateTime().toLocalDate().toString(),
                            Collectors.counting()));
            statistics.setDailyRegisterCount(dailyRegisterCount);
            
            return statistics;
            
        } catch (Exception e) {
            log.error("获取用户统计失败: error={}", e.getMessage(), e);
            throw new BusinessException("获取用户统计失败: " + e.getMessage());
        }
    }

    /**
     * 转换为用户信息
     */
    private UserInfo convertToUserInfo(User user, UserProfile userProfile) {
        UserInfo userInfo = new UserInfo();
        userInfo.setUserId(user.getUserId());
        userInfo.setUsername(user.getUsername());
        userInfo.setPhone(user.getPhone());
        userInfo.setEmail(user.getEmail());
        userInfo.setStatus(user.getStatus());
        userInfo.setCreateTime(user.getCreateTime());
        userInfo.setUpdateTime(user.getUpdateTime());
        
        if (userProfile != null) {
            userInfo.setNickname(userProfile.getNickname());
            userInfo.setAvatar(userProfile.getAvatar());
            userInfo.setGender(userProfile.getGender());
            userInfo.setBirthday(userProfile.getBirthday());
        }
        
        return userInfo;
    }
}

/**
 * 认证服务控制器
 * 提供认证管理接口
 */
@RestController
@RequestMapping("/api/auth")
@Slf4j
public class AuthController {

    @Autowired
    private AuthService authService;

    /**
     * 用户登录
     */
    @PostMapping("/login")
    public Result<LoginResponse> login(@RequestBody @Valid UserLoginRequest request) {
        try {
            LoginResponse loginResponse = authService.login(request);
            return Result.success(loginResponse);
        } catch (Exception e) {
            log.error("用户登录失败: error={}", e.getMessage(), e);
            return Result.error("用户登录失败: " + e.getMessage());
        }
    }

    /**
     * 用户登出
     */
    @PostMapping("/logout")
    public Result<Void> logout(@RequestHeader("Authorization") String token) {
        try {
            authService.logout(token);
            return Result.success();
        } catch (Exception e) {
            log.error("用户登出失败: error={}", e.getMessage(), e);
            return Result.error("用户登出失败: " + e.getMessage());
        }
    }

    /**
     * 刷新Token
     */
    @PostMapping("/refresh")
    public Result<LoginResponse> refreshToken(@RequestBody @Valid RefreshTokenRequest request) {
        try {
            LoginResponse loginResponse = authService.refreshToken(request.getRefreshToken());
            return Result.success(loginResponse);
        } catch (Exception e) {
            log.error("刷新Token失败: error={}", e.getMessage(), e);
            return Result.error("刷新Token失败: " + e.getMessage());
        }
    }

    /**
     * 验证Token
     */
    @GetMapping("/verify")
    public Result<UserInfo> verifyToken(@RequestHeader("Authorization") String token) {
        try {
            UserInfo userInfo = authService.verifyToken(token);
            return Result.success(userInfo);
        } catch (Exception e) {
            log.error("验证Token失败: error={}", e.getMessage(), e);
            return Result.error("验证Token失败: " + e.getMessage());
        }
    }
}

/**
 * 认证服务实现
 * 负责用户认证、Token生成、验证
 */
@Service
@Slf4j
public class AuthService {

    @Autowired
    private UserMapper userMapper;
    
    @Autowired
    private UserQueryService userQueryService;
    
    @Autowired
    private LoginRecordMapper loginRecordMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Autowired
    private PasswordEncoder passwordEncoder;
    
    @Autowired
    private JwtTokenUtil jwtTokenUtil;
    
    @Value("${jwt.token.expiration:3600}")
    private Long tokenExpiration;
    
    @Value("${jwt.refresh.token.expiration:86400}")
    private Long refreshTokenExpiration;

    /**
     * 用户登录
     */
    @Transactional(rollbackFor = Exception.class)
    public LoginResponse login(UserLoginRequest request) {
        try {
            // 1. 查询用户
            User user = userMapper.selectOne(
                    new LambdaQueryWrapper<User>()
                            .eq(User::getUsername, request.getUsername())
                            .or()
                            .eq(User::getPhone, request.getUsername())
                            .or()
                            .eq(User::getEmail, request.getUsername()));
            if (user == null) {
                throw new BusinessException("用户名或密码错误");
            }
            
            // 2. 验证用户状态
            if (!"ACTIVE".equals(user.getStatus())) {
                throw new BusinessException("用户已被禁用");
            }
            
            // 3. 验证密码
            if (!passwordEncoder.matches(request.getPassword(), user.getPassword())) {
                // 记录登录失败
                recordLoginFailure(user.getUserId(), request.getIp(), "密码错误");
                throw new BusinessException("用户名或密码错误");
            }
            
            // 4. 生成Token
            String accessToken = jwtTokenUtil.generateToken(user.getUserId(), user.getUsername(), tokenExpiration);
            String refreshToken = jwtTokenUtil.generateToken(user.getUserId(), user.getUsername(), refreshTokenExpiration);
            
            // 5. 缓存Token
            String tokenKey = "user:token:" + user.getUserId();
            redisTemplate.opsForValue().set(tokenKey, accessToken, tokenExpiration, TimeUnit.SECONDS);
            
            String refreshTokenKey = "user:refresh:token:" + user.getUserId();
            redisTemplate.opsForValue().set(refreshTokenKey, refreshToken, refreshTokenExpiration, TimeUnit.SECONDS);
            
            // 6. 缓存用户会话
            String sessionKey = "user:session:" + user.getUserId();
            UserSession session = new UserSession();
            session.setUserId(user.getUserId());
            session.setUsername(user.getUsername());
            session.setAccessToken(accessToken);
            session.setRefreshToken(refreshToken);
            session.setLoginTime(LocalDateTime.now());
            session.setLoginIp(request.getIp());
            redisTemplate.opsForValue().set(sessionKey, session, tokenExpiration, TimeUnit.SECONDS);
            
            // 7. 记录登录成功
            recordLoginSuccess(user.getUserId(), request.getIp());
            
            // 8. 获取用户信息
            UserInfo userInfo = userQueryService.getUserInfo(user.getUserId());
            
            // 9. 构建登录响应
            LoginResponse loginResponse = new LoginResponse();
            loginResponse.setAccessToken(accessToken);
            loginResponse.setRefreshToken(refreshToken);
            loginResponse.setTokenType("Bearer");
            loginResponse.setExpiresIn(tokenExpiration);
            loginResponse.setUserInfo(userInfo);
            
            // 10. 发送登录事件
            sendLoginEvent(user.getUserId(), request.getIp());
            
            log.info("用户登录成功: userId={}, username={}", user.getUserId(), user.getUsername());
            
            return loginResponse;
            
        } catch (Exception e) {
            log.error("用户登录失败: username={}, error={}", 
                    request.getUsername(), e.getMessage(), e);
            throw new BusinessException("用户登录失败: " + e.getMessage());
        }
    }

    /**
     * 用户登出
     */
    public void logout(String token) {
        try {
            // 1. 解析Token
            String userId = jwtTokenUtil.getUserIdFromToken(token);
            if (userId == null) {
                throw new BusinessException("Token无效");
            }
            
            // 2. 清除Token缓存
            String tokenKey = "user:token:" + userId;
            redisTemplate.delete(tokenKey);
            
            String refreshTokenKey = "user:refresh:token:" + userId;
            redisTemplate.delete(refreshTokenKey);
            
            // 3. 清除用户会话
            String sessionKey = "user:session:" + userId;
            redisTemplate.delete(sessionKey);
            
            // 4. 发送登出事件
            sendLogoutEvent(userId);
            
            log.info("用户登出成功: userId={}", userId);
            
        } catch (Exception e) {
            log.error("用户登出失败: error={}", e.getMessage(), e);
            throw new BusinessException("用户登出失败: " + e.getMessage());
        }
    }

    /**
     * 刷新Token
     */
    public LoginResponse refreshToken(String refreshToken) {
        try {
            // 1. 验证Refresh Token
            if (!jwtTokenUtil.validateToken(refreshToken)) {
                throw new BusinessException("Refresh Token无效");
            }
            
            // 2. 获取用户ID
            String userId = jwtTokenUtil.getUserIdFromToken(refreshToken);
            if (userId == null) {
                throw new BusinessException("Refresh Token无效");
            }
            
            // 3. 验证Refresh Token是否在缓存中
            String refreshTokenKey = "user:refresh:token:" + userId;
            String cachedRefreshToken = (String) redisTemplate.opsForValue().get(refreshTokenKey);
            if (!refreshToken.equals(cachedRefreshToken)) {
                throw new BusinessException("Refresh Token无效");
            }
            
            // 4. 获取用户信息
            User user = userMapper.selectOne(
                    new LambdaQueryWrapper<User>()
                            .eq(User::getUserId, userId));
            if (user == null) {
                throw new BusinessException("用户不存在");
            }
            
            // 5. 生成新的Token
            String newAccessToken = jwtTokenUtil.generateToken(user.getUserId(), user.getUsername(), tokenExpiration);
            String newRefreshToken = jwtTokenUtil.generateToken(user.getUserId(), user.getUsername(), refreshTokenExpiration);
            
            // 6. 更新Token缓存
            String tokenKey = "user:token:" + userId;
            redisTemplate.opsForValue().set(tokenKey, newAccessToken, tokenExpiration, TimeUnit.SECONDS);
            redisTemplate.opsForValue().set(refreshTokenKey, newRefreshToken, refreshTokenExpiration, TimeUnit.SECONDS);
            
            // 7. 获取用户信息
            UserInfo userInfo = userQueryService.getUserInfo(userId);
            
            // 8. 构建登录响应
            LoginResponse loginResponse = new LoginResponse();
            loginResponse.setAccessToken(newAccessToken);
            loginResponse.setRefreshToken(newRefreshToken);
            loginResponse.setTokenType("Bearer");
            loginResponse.setExpiresIn(tokenExpiration);
            loginResponse.setUserInfo(userInfo);
            
            log.info("刷新Token成功: userId={}", userId);
            
            return loginResponse;
            
        } catch (Exception e) {
            log.error("刷新Token失败: error={}", e.getMessage(), e);
            throw new BusinessException("刷新Token失败: " + e.getMessage());
        }
    }

    /**
     * 验证Token
     */
    public UserInfo verifyToken(String token) {
        try {
            // 1. 验证Token
            if (!jwtTokenUtil.validateToken(token)) {
                throw new BusinessException("Token无效");
            }
            
            // 2. 获取用户ID
            String userId = jwtTokenUtil.getUserIdFromToken(token);
            if (userId == null) {
                throw new BusinessException("Token无效");
            }
            
            // 3. 验证Token是否在缓存中
            String tokenKey = "user:token:" + userId;
            String cachedToken = (String) redisTemplate.opsForValue().get(tokenKey);
            if (!token.equals(cachedToken)) {
                throw new BusinessException("Token已失效");
            }
            
            // 4. 获取用户信息
            UserInfo userInfo = userQueryService.getUserInfo(userId);
            
            return userInfo;
            
        } catch (Exception e) {
            log.error("验证Token失败: error={}", e.getMessage(), e);
            throw new BusinessException("验证Token失败: " + e.getMessage());
        }
    }

    /**
     * 记录登录成功
     */
    private void recordLoginSuccess(String userId, String ip) {
        try {
            LoginRecord record = new LoginRecord();
            record.setUserId(userId);
            record.setLoginIp(ip);
            record.setLoginTime(LocalDateTime.now());
            record.setStatus("SUCCESS");
            record.setCreateTime(LocalDateTime.now());
            
            loginRecordMapper.insert(record);
            
        } catch (Exception e) {
            log.error("记录登录成功失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 记录登录失败
     */
    private void recordLoginFailure(String userId, String ip, String reason) {
        try {
            LoginRecord record = new LoginRecord();
            record.setUserId(userId);
            record.setLoginIp(ip);
            record.setLoginTime(LocalDateTime.now());
            record.setStatus("FAILED");
            record.setFailureReason(reason);
            record.setCreateTime(LocalDateTime.now());
            
            loginRecordMapper.insert(record);
            
        } catch (Exception e) {
            log.error("记录登录失败失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 发送登录事件
     */
    private void sendLoginEvent(String userId, String ip) {
        try {
            LoginEvent event = new LoginEvent();
            event.setEventType("USER_LOGIN");
            event.setUserId(userId);
            event.setLoginIp(ip);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("user.login", userId, message);
            
        } catch (Exception e) {
            log.error("发送登录事件失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 发送登出事件
     */
    private void sendLogoutEvent(String userId) {
        try {
            LogoutEvent event = new LogoutEvent();
            event.setEventType("USER_LOGOUT");
            event.setUserId(userId);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("user.logout", userId, message);
            
        } catch (Exception e) {
            log.error("发送登出事件失败: error={}", e.getMessage(), e);
        }
    }
}

/**
 * 积分服务控制器
 * 提供积分管理接口
 */
@RestController
@RequestMapping("/api/points")
@Slf4j
public class PointsController {

    @Autowired
    private PointsService pointsService;
    
    @Autowired
    private PointsQueryService pointsQueryService;

    /**
     * 获取用户积分
     */
    @GetMapping("/{userId}")
    public Result<PointsInfo> getUserPoints(@PathVariable String userId) {
        try {
            PointsInfo pointsInfo = pointsQueryService.getUserPoints(userId);
            return Result.success(pointsInfo);
        } catch (Exception e) {
            log.error("获取用户积分失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            return Result.error("获取用户积分失败: " + e.getMessage());
        }
    }

    /**
     * 增加积分
     */
    @PostMapping("/add")
    public Result<Void> addPoints(@RequestBody @Valid PointsAddRequest request) {
        try {
            pointsService.addPoints(request);
            return Result.success();
        } catch (Exception e) {
            log.error("增加积分失败: error={}", e.getMessage(), e);
            return Result.error("增加积分失败: " + e.getMessage());
        }
    }

    /**
     * 扣减积分
     */
    @PostMapping("/deduct")
    public Result<Void> deductPoints(@RequestBody @Valid PointsDeductRequest request) {
        try {
            pointsService.deductPoints(request);
            return Result.success();
        } catch (Exception e) {
            log.error("扣减积分失败: error={}", e.getMessage(), e);
            return Result.error("扣减积分失败: " + e.getMessage());
        }
    }

    /**
     * 获取积分记录
     */
    @GetMapping("/{userId}/records")
    public Result<PageResult<PointsRecord>> getPointsRecords(
            @PathVariable String userId,
            @RequestParam(required = false) String changeType,
            @RequestParam(defaultValue = "1") Integer pageNum,
            @RequestParam(defaultValue = "10") Integer pageSize) {
        try {
            PageResult<PointsRecord> result = pointsQueryService.getPointsRecords(
                    userId, changeType, pageNum, pageSize);
            return Result.success(result);
        } catch (Exception e) {
            log.error("获取积分记录失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            return Result.error("获取积分记录失败: " + e.getMessage());
        }
    }
}

/**
 * 积分服务实现
 * 负责积分增加、扣减、查询
 */
@Service
@Slf4j
public class PointsService {

    @Autowired
    private UserPointsMapper userPointsMapper;
    
    @Autowired
    private PointsRecordMapper pointsRecordMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Autowired
    private DistributedLock distributedLock;

    /**
     * 初始化用户积分
     */
    @Transactional(rollbackFor = Exception.class)
    public void initUserPoints(String userId) {
        try {
            // 检查是否已初始化
            UserPoints existingPoints = userPointsMapper.selectOne(
                    new LambdaQueryWrapper<UserPoints>()
                            .eq(UserPoints::getUserId, userId));
            if (existingPoints != null) {
                return;
            }
            
            // 创建用户积分记录
            UserPoints userPoints = new UserPoints();
            userPoints.setUserId(userId);
            userPoints.setTotalPoints(0);
            userPoints.setAvailablePoints(0);
            userPoints.setUsedPoints(0);
            userPoints.setCreateTime(LocalDateTime.now());
            userPoints.setUpdateTime(LocalDateTime.now());
            
            userPointsMapper.insert(userPoints);
            
            log.info("初始化用户积分成功: userId={}", userId);
            
        } catch (Exception e) {
            log.error("初始化用户积分失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            throw new BusinessException("初始化用户积分失败: " + e.getMessage());
        }
    }

    /**
     * 增加积分
     */
    @Transactional(rollbackFor = Exception.class)
    public void addPoints(PointsAddRequest request) {
        String lockKey = "points:add:" + request.getUserId();
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // 1. 获取分布式锁
            boolean lockAcquired = distributedLock.tryLock(lockKey, lockValue, 10, TimeUnit.SECONDS);
            if (!lockAcquired) {
                throw new BusinessException("积分增加中，请稍候");
            }
            
            // 2. 查询用户积分
            UserPoints userPoints = userPointsMapper.selectOne(
                    new LambdaQueryWrapper<UserPoints>()
                            .eq(UserPoints::getUserId, request.getUserId()));
            if (userPoints == null) {
                initUserPoints(request.getUserId());
                userPoints = userPointsMapper.selectOne(
                        new LambdaQueryWrapper<UserPoints>()
                                .eq(UserPoints::getUserId, request.getUserId()));
            }
            
            // 3. 增加积分
            userPoints.setTotalPoints(userPoints.getTotalPoints() + request.getPoints());
            userPoints.setAvailablePoints(userPoints.getAvailablePoints() + request.getPoints());
            userPoints.setUpdateTime(LocalDateTime.now());
            
            userPointsMapper.updateById(userPoints);
            
            // 4. 记录积分变更
            PointsRecord record = new PointsRecord();
            record.setUserId(request.getUserId());
            record.setChangeType("ADD");
            record.setChangePoints(request.getPoints());
            record.setBeforePoints(userPoints.getAvailablePoints() - request.getPoints());
            record.setAfterPoints(userPoints.getAvailablePoints());
            record.setSourceType(request.getSourceType());
            record.setSourceId(request.getSourceId());
            record.setRemark(request.getRemark());
            record.setCreateTime(LocalDateTime.now());
            
            pointsRecordMapper.insert(record);
            
            // 5. 清除缓存
            String cacheKey = "points:info:" + request.getUserId();
            redisTemplate.delete(cacheKey);
            
            // 6. 发送积分增加事件
            sendPointsAddEvent(request.getUserId(), request.getPoints(), request.getSourceType());
            
            log.info("增加积分成功: userId={}, points={}, sourceType={}", 
                    request.getUserId(), request.getPoints(), request.getSourceType());
            
        } catch (Exception e) {
            log.error("增加积分失败: userId={}, error={}", 
                    request.getUserId(), e.getMessage(), e);
            throw new BusinessException("增加积分失败: " + e.getMessage());
        } finally {
            // 释放分布式锁
            distributedLock.releaseLock(lockKey, lockValue);
        }
    }

    /**
     * 扣减积分
     */
    @Transactional(rollbackFor = Exception.class)
    public void deductPoints(PointsDeductRequest request) {
        String lockKey = "points:deduct:" + request.getUserId();
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // 1. 获取分布式锁
            boolean lockAcquired = distributedLock.tryLock(lockKey, lockValue, 10, TimeUnit.SECONDS);
            if (!lockAcquired) {
                throw new BusinessException("积分扣减中，请稍候");
            }
            
            // 2. 查询用户积分
            UserPoints userPoints = userPointsMapper.selectOne(
                    new LambdaQueryWrapper<UserPoints>()
                            .eq(UserPoints::getUserId, request.getUserId()));
            if (userPoints == null) {
                throw new BusinessException("用户积分不存在");
            }
            
            // 3. 验证积分是否足够
            if (userPoints.getAvailablePoints() < request.getPoints()) {
                throw new BusinessException("积分不足: availablePoints=" + userPoints.getAvailablePoints() + ", points=" + request.getPoints());
            }
            
            // 4. 扣减积分
            userPoints.setAvailablePoints(userPoints.getAvailablePoints() - request.getPoints());
            userPoints.setUsedPoints(userPoints.getUsedPoints() + request.getPoints());
            userPoints.setUpdateTime(LocalDateTime.now());
            
            userPointsMapper.updateById(userPoints);
            
            // 5. 记录积分变更
            PointsRecord record = new PointsRecord();
            record.setUserId(request.getUserId());
            record.setChangeType("DEDUCT");
            record.setChangePoints(request.getPoints());
            record.setBeforePoints(userPoints.getAvailablePoints() + request.getPoints());
            record.setAfterPoints(userPoints.getAvailablePoints());
            record.setSourceType(request.getSourceType());
            record.setSourceId(request.getSourceId());
            record.setRemark(request.getRemark());
            record.setCreateTime(LocalDateTime.now());
            
            pointsRecordMapper.insert(record);
            
            // 6. 清除缓存
            String cacheKey = "points:info:" + request.getUserId();
            redisTemplate.delete(cacheKey);
            
            // 7. 发送积分扣减事件
            sendPointsDeductEvent(request.getUserId(), request.getPoints(), request.getSourceType());
            
            log.info("扣减积分成功: userId={}, points={}, sourceType={}", 
                    request.getUserId(), request.getPoints(), request.getSourceType());
            
        } catch (Exception e) {
            log.error("扣减积分失败: userId={}, error={}", 
                    request.getUserId(), e.getMessage(), e);
            throw new BusinessException("扣减积分失败: " + e.getMessage());
        } finally {
            // 释放分布式锁
            distributedLock.releaseLock(lockKey, lockValue);
        }
    }

    /**
     * 发送积分增加事件
     */
    private void sendPointsAddEvent(String userId, Integer points, String sourceType) {
        try {
            PointsEvent event = new PointsEvent();
            event.setEventType("POINTS_ADD");
            event.setUserId(userId);
            event.setPoints(points);
            event.setSourceType(sourceType);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("points.add", userId, message);
            
        } catch (Exception e) {
            log.error("发送积分增加事件失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 发送积分扣减事件
     */
    private void sendPointsDeductEvent(String userId, Integer points, String sourceType) {
        try {
            PointsEvent event = new PointsEvent();
            event.setEventType("POINTS_DEDUCT");
            event.setUserId(userId);
            event.setPoints(points);
            event.setSourceType(sourceType);
            event.setEventTime(LocalDateTime.now());
            
            String message = JSON.toJSONString(event);
            kafkaTemplate.send("points.deduct", userId, message);
            
        } catch (Exception e) {
            log.error("发送积分扣减事件失败: error={}", e.getMessage(), e);
        }
    }
}

/**
 * 积分查询服务实现
 * 负责积分查询
 */
@Service
@Slf4j
public class PointsQueryService {

    @Autowired
    private UserPointsMapper userPointsMapper;
    
    @Autowired
    private PointsRecordMapper pointsRecordMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 获取用户积分
     */
    public PointsInfo getUserPoints(String userId) {
        try {
            // 1. 从缓存获取
            String cacheKey = "points:info:" + userId;
            PointsInfo cachedInfo = (PointsInfo) redisTemplate.opsForValue().get(cacheKey);
            if (cachedInfo != null) {
                return cachedInfo;
            }
            
            // 2. 从数据库查询
            UserPoints userPoints = userPointsMapper.selectOne(
                    new LambdaQueryWrapper<UserPoints>()
                            .eq(UserPoints::getUserId, userId));
            if (userPoints == null) {
                throw new BusinessException("用户积分不存在");
            }
            
            // 3. 构建积分信息
            PointsInfo pointsInfo = new PointsInfo();
            pointsInfo.setUserId(userPoints.getUserId());
            pointsInfo.setTotalPoints(userPoints.getTotalPoints());
            pointsInfo.setAvailablePoints(userPoints.getAvailablePoints());
            pointsInfo.setUsedPoints(userPoints.getUsedPoints());
            
            // 4. 缓存积分信息
            redisTemplate.opsForValue().set(cacheKey, pointsInfo, 1, TimeUnit.HOURS);
            
            return pointsInfo;
            
        } catch (Exception e) {
            log.error("获取用户积分失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            throw new BusinessException("获取用户积分失败: " + e.getMessage());
        }
    }

    /**
     * 获取积分记录
     */
    public PageResult<PointsRecord> getPointsRecords(String userId, String changeType,
                                                     Integer pageNum, Integer pageSize) {
        try {
            // 1. 构建查询条件
            LambdaQueryWrapper<PointsRecord> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(PointsRecord::getUserId, userId);
            if (changeType != null && !changeType.isEmpty()) {
                wrapper.eq(PointsRecord::getChangeType, changeType);
            }
            wrapper.orderByDesc(PointsRecord::getCreateTime);
            
            // 2. 分页查询
            Page<PointsRecord> page = new Page<>(pageNum, pageSize);
            Page<PointsRecord> result = pointsRecordMapper.selectPage(page, wrapper);
            
            // 3. 构建分页结果
            PageResult<PointsRecord> pageResult = new PageResult<>();
            pageResult.setList(result.getRecords());
            pageResult.setTotal(result.getTotal());
            pageResult.setPageNum(pageNum);
            pageResult.setPageSize(pageSize);
            
            return pageResult;
            
        } catch (Exception e) {
            log.error("获取积分记录失败: userId={}, error={}", 
                    userId, e.getMessage(), e);
            throw new BusinessException("获取积分记录失败: " + e.getMessage());
        }
    }
}

/**
 * 用户事件消费者
 * 消费Kafka用户事件消息
 */
@Component
@Slf4j
public class UserEventConsumer {

    @Autowired
    private UserService userService;
    
    @Autowired
    private PointsServiceClient pointsServiceClient;

    /**
     * 消费用户注册事件
     */
    @KafkaListener(topics = "user.register", groupId = "user-event-group")
    public void consumeUserRegister(String message) {
        try {
            UserEvent event = JSON.parseObject(message, UserEvent.class);
            
            // 处理用户注册事件，如发送欢迎消息、赠送积分等
            log.info("消费用户注册事件: userId={}", event.getUserId());
            
            // 注册赠送积分
            PointsAddRequest pointsRequest = new PointsAddRequest();
            pointsRequest.setUserId(event.getUserId());
            pointsRequest.setPoints(100);
            pointsRequest.setSourceType("REGISTER");
            pointsRequest.setSourceId(event.getUserId());
            pointsRequest.setRemark("注册赠送积分");
            pointsServiceClient.addPoints(pointsRequest);
            
        } catch (Exception e) {
            log.error("消费用户注册事件失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 消费用户登录事件
     */
    @KafkaListener(topics = "user.login", groupId = "user-event-group")
    public void consumeUserLogin(String message) {
        try {
            LoginEvent event = JSON.parseObject(message, LoginEvent.class);
            
            // 处理用户登录事件，如更新最后登录时间、发送通知等
            log.info("消费用户登录事件: userId={}, loginIp={}", 
                    event.getUserId(), event.getLoginIp());
            
        } catch (Exception e) {
            log.error("消费用户登录事件失败: error={}", e.getMessage(), e);
        }
    }

    /**
     * 消费用户信息变更事件
     */
    @KafkaListener(topics = "user.info.change", groupId = "user-event-group")
    public void consumeUserInfoChange(String message) {
        try {
            UserInfoChangeEvent event = JSON.parseObject(message, UserInfoChangeEvent.class);
            
            // 处理用户信息变更事件，如更新缓存、发送通知等
            log.info("消费用户信息变更事件: userId={}", event.getUserId());
            
        } catch (Exception e) {
            log.error("消费用户信息变更事件失败: error={}", e.getMessage(), e);
        }
    }
}