#数据双写

@Service
public class UserService {

    @Autowired private UserMapper newUserMapper;    // 新库
    @Autowired private UserMapper legacyUserMapper; // 旧库

    public void createUser(User user) {
        // 先写新库
        newUserMapper.insert(user);

        // 再写旧库
        legacyUserMapper.insert(user);
    }
}

@Service
public class UserService {

    @Autowired private UserMapper newUserMapper;
    @Autowired private KafkaTemplate kafkaTemplate;

    public void createUser(User user) {
        // 写新库
        newUserMapper.insert(user);

        // 发消息，异步同步到旧库
        kafkaTemplate.send("user-sync", user.getId(), user);
    }
}

// 消费者服务（独立部署）
@Service
public class UserSyncConsumer {

    @Autowired private UserMapper legacyUserMapper;

    @KafkaListener(topics = "user-sync")
    public void onUserCreated(User user) {
        // 写入旧库
        legacyUserMapper.insert(user);
    }
}

@Service
public class UserService {

    @Autowired private UserMapper newUserMapper;  // 新库

    @Transactional
    public void createUser(User user) {
        // 1. 写入新库
        newUserMapper.insert(user);

        // 2. 写入本地消息表（和业务数据在同一个事务）
        LocalMessage message = new LocalMessage();
        message.setId(UUID.randomUUID().toString());
        message.setTopic("user-sync");
        message.setPayload(JSON.toJSONString(user));
        message.setStatus("PENDING");
        message.setRetryCount(0);
        message.setCreateTime(LocalDateTime.now());
        newUserMapper.insertMessage(message);
    }
}

@Service
public class MessageDispatcher {

    @Autowired private UserMapper newUserMapper;
    @Autowired private UserMapper legacyUserMapper;
    @Autowired private KafkaTemplate kafkaTemplate;

    @Scheduled(fixedDelay = 1000)  // 每秒执行一次
    public void dispatch() {
        // 1. 查询待发送的消息
        List<LocalMessage> messages = newUserMapper.selectPendingMessages(100);

        for (LocalMessage msg : messages) {
            try {
                // 2. 发送到消息队列
                kafkaTemplate.send(msg.getTopic(), msg.getPayload());

                // 3. 更新消息状态
                msg.setStatus("SENT");
                msg.setSendTime(LocalDateTime.now());
                newUserMapper.updateMessage(msg);

            } catch (Exception e) {
                // 4. 发送失败，��加重试次数
                msg.setRetryCount(msg.getRetryCount() + 1);
                msg.setLastError(e.getMessage());

                if (msg.getRetryCount() >= 3) {
                    msg.setStatus("FAILED");
                }

                newUserMapper.updateMessage(msg);
            }
        }
    }
}

@Service
public class DoubleWriteMonitor {

    @Autowired private UserMapper newUserMapper;
    @Autowired private UserMapper legacyUserMapper;
    @Autowired private AlertService alertService;

    // 每分钟检查一次数据一致性
    @Scheduled(cron = "0 * * * * *")
    public void checkConsistency() {
        // 1. 抽样检查：随机抽取 100 条新库数据，对比旧库
        List<User> samples = newUserMapper.selectRandomSamples(100);

        int inconsistencyCount = 0;
        for (User newUser : samples) {
            User oldUser = legacyUserMapper.findById(newUser.getId());

            if (oldUser == null) {
                // 旧库缺失这条记录
                inconsistencyCount++;
                log.warn("数据不一致：旧库缺失 userId={}", newUser.getId());

            } else if (!equals(newUser, oldUser)) {
                // 数据值不一致
                inconsistencyCount++;
                log.warn("数据不一致：userId={}, 新库={}, 旧库={}",
                    newUser.getId(), newUser, oldUser);
            }
        }

        // 2. 告警阈值：超过 1% 的不一致率
        if (inconsistencyCount > samples.size() * 0.01) {
            alertService.alert("用户数据双写不一致率超过 1%，当前不一致数量：{}",
                inconsistencyCount);
        }
    }

    private boolean equals(User a, User b) {
        if (a == null || b == null) return false;
        return Objects.equals(a.getName(), b.getName())
            && Objects.equals(a.getEmail(), b.getEmail())
            && Objects.equals(a.getPhone(), b.getPhone());
    }
}