#微信海量通讯架构

// WebSocket 长连接：消息推送
public class WeChatPushService {

    // 维护用户长连接
    private ConcurrentHashMap<Long, WebSocketSession> userConnections = new ConcurrentHashMap<>();

    // 用户上线：建立长连接
    public void onUserOnline(Long userId, WebSocketSession session) {
        userConnections.put(userId, session);

        // 更新用户在线状态
        onlineStatusService.setOnline(userId);

        // 同步未读消息
        syncUnreadMessages(userId);
    }

    // 用户离线：关闭连接
    public void onUserOffline(Long userId) {
        userConnections.remove(userId);

        // 更新用户在线状态
        onlineStatusService.setOffline(userId);

        // 存储离线消息
        offlineMessageService.storeOffline(userId);
    }

    // 推送消息到用户
    public void pushMessage(Long userId, Message message) {
        WebSocketSession session = userConnections.get(userId);

        if (session != null && session.isOpen()) {
            // 在线：直接推送
            session.sendMessage(new TextMessage(toJson(message)));
        } else {
            // 离线：存储离线消息
            offlineMessageService.storeOffline(userId, message);
        }
    }
}

// 心跳服务
public class HeartbeatService {

    private static final int HEARTBEAT_INTERVAL_SECONDS = 30;
    private static final int HEARTBEAT_TIMEOUT_SECONDS = 60;

    // 客户端定时发送心跳
    @Scheduled(fixedRate = HEARTBEAT_INTERVAL_SECONDS * 1000)
    public void sendHeartbeat() {
        WebSocketSession session = getCurrentSession();
        if (session.isOpen()) {
            HeartbeatMessage heartbeat = new HeartbeatMessage();
            heartbeat.setUserId(getCurrentUserId());
            heartbeat.setTimestamp(System.currentTimeMillis());
            heartbeat.setClientVersion(BuildConfig.VERSION_NAME);

            session.sendMessage(new TextMessage(toJson(heartbeat)));
        }
    }

    // 服务器检测心跳超时
    @Scheduled(fixedRate = HEARTBEAT_TIMEOUT_SECONDS * 1000)
    public void checkHeartbeatTimeout() {
        long timeoutThreshold = System.currentTimeMillis()
            - HEARTBEAT_TIMEOUT_SECONDS * 1000L;

        for (Long userId : connectionManager.getOnlineUsers()) {
            long lastHeartbeat = connectionManager.getLastHeartbeat(userId);

            if (lastHeartbeat < timeoutThreshold) {
                // 心跳超时：标记为离线
                connectionManager.markOffline(userId);
                log.info("User {} heartbeat timeout, marking offline", userId);

                // 触发离线事件
                eventBus.publish(new UserOfflineEvent(userId));
            }
        }
    }
}

// 单聊消息路由
public class SingleChatRouter {

    @Autowired private MessageStore messageStore;
    @Autowired private MultiDeviceSyncService multiDeviceSync;

    // 发送单聊消息
    public SendResult sendMessage(Long senderId, Long receiverId, MessageContent content) {
        // 1. 生成全局唯一消息 ID
        String msgId = snowflakeIdGenerator.nextId();

        // 2. 构造消息
        Message message = new Message();
        message.setMsgId(msgId);
        message.setSenderId(senderId);
        message.setReceiverId(receiverId);
        message.setContent(content);
        message.setTimestamp(System.currentTimeMillis());
        message.setStatus(MessageStatus.SENDING);

        // 3. 持久化消息
        messageStore.save(message);

        // 4. 检查接收方是否在线
        if (onlineStatusService.isOnline(receiverId)) {
            // 在线：直接推送
            pushService.pushToUser(receiverId, message);
        } else {
            // 离线：存入离线消息队列
            offlineMessageService.store(receiverId, message);
        }

        // 5. 同步到发送方的所有设备
        multiDeviceSync.syncToSenderDevices(senderId, message);

        return SendResult.success(msgId);
    }
}

// 群聊消息处理
public class GroupChatService {

    @Autowired private GroupMemberService groupMemberService;
    @Autowired private MessageStore messageStore;
    @Autowired private PushService pushService;

    // 发送群聊消息
    public SendResult sendGroupMessage(Long senderId, Long groupId, MessageContent content) {
        // 1. 获取群成员列表
        List<Long> memberIds = groupMemberService.getActiveMembers(groupId);

        // 2. 构造消息
        String msgId = snowflakeIdGenerator.nextId();
        Message message = new Message();
        message.setMsgId(msgId);
        message.setSenderId(senderId);
        message.setGroupId(groupId);
        message.setContent(content);
        message.setTimestamp(System.currentTimeMillis());

        // 3. 持久化消息（只存一份）
        messageStore.saveGroupMessage(groupId, message);

        // 4. 统计未读：检查每个成员上次拉取的位置
        Map<Long, Long> unreadCounts = new HashMap<>();
        for (Long memberId : memberIds) {
            long lastReadMsgId = readPositionService.getLastRead(memberId, groupId);
            long unreadCount = messageStore.countAfter(groupId, lastReadMsgId);
            unreadCounts.put(memberId, unreadCount);
        }

        // 5. 推送消息给在线成员
        for (Long memberId : memberIds) {
            if (onlineStatusService.isOnline(memberId)) {
                // 在线成员：推送消息
                pushService.pushToUser(memberId, message);
            }
            // 离线成员：通过未读数提醒，下次上线拉取
        }

        return SendResult.success(msgId);
    }
}

// 消息存储分层
public class MessageStorageService {

    // 热度判断规则
    // 热数据：最近 7 天，访问频繁 -> SSD 存储
    // 温数据：7-90 天，偶尔访问 -> HDD 存储
    // 冷数据：90 天以上，极少访问 -> 低成本存储（如对象存储）

    // 写入消息
    public void saveMessage(Message message) {
        // 写入热数据层（SSD）
        hotStorage.save(message);

        // 异步归档到温数据层
        scheduleService.schedule(
            () -> archiveToWarmStorage(message),
            message.getTimestamp() + TimeUnit.DAYS.toMillis(7)
        );
    }

    // 查询消息
    public List<Message> queryMessages(Long userId, long startTime, long endTime) {
        // 1. 查询热数据层
        List<Message> hotMessages = hotStorage.query(userId, startTime, endTime);

        // 2. 查询温数据层
        List<Message> warmMessages = warmStorage.query(userId, startTime, endTime);

        // 3. 查询冷数据层（如需要）
        if (endTime - startTime > TimeUnit.DAYS.toMillis(90)) {
            List<Message> coldMessages = coldStorage.query(userId, startTime, endTime);
            return mergeAndSort(hotMessages, warmMessages, coldMessages);
        }

        return mergeAndSort(hotMessages, warmMessages);
    }

    // 归档到温数据层
    private void archiveToWarmStorage(Message message) {
        warmStorage.save(message);
        hotStorage.delete(message.getMsgId());
    }
}

// 消息同步协议
public class MessageSyncService {

    // 同步请求：客户端告诉服务器「我已经同步到哪里了」
    public SyncResponse syncMessages(SyncRequest request) {
        Long userId = request.getUserId();
        long syncKey = request.getSyncKey(); // 客户端已同步的最大消息 ID

        // 1. 查询新消息
        List<Message> newMessages = messageStore.queryAfter(userId, syncKey);

        // 2. 生成本次同步的 syncKey
        long newSyncKey = newMessages.isEmpty()
            ? syncKey
            : newMessages.get(newMessages.size() - 1).getMsgId();

        // 3. 返回新消息和新的 syncKey
        return SyncResponse.builder()
            .messages(newMessages)
            .syncKey(newSyncKey)
            .hasMore(newMessages.size() >= MAX_BATCH_SIZE)
            .build();
    }

    // 客户端收到同步响应后：
    // 1. 显示新消息
    // 2. 保存新的 syncKey 到本地
    // 3. 如果 hasMore=true，继续请求下一批
}

// 朋友圈 Feed 拉取
public class MomentsFeedService {

    @Autowired private PrivacyService privacyService;
    @Autowired private ContentService contentService;

    // 拉取朋友圈 Feed
    public List<FeedItem> pullFeed(Long userId, long sinceId, int limit) {
        // 1. 获取好友列表
        List<Long> friendIds = friendService.getFriends(userId);

        // 2. 查询好友最近发布的内容
        List<FeedItem> candidateFeeds = contentService.queryFeeds(
            friendIds, sinceId, limit * 2 // 多查一些，过滤后可能不够
        );

        // 3. 隐私过滤：移除用户无权看到的内容
        List<FeedItem> filteredFeeds = candidateFeeds.stream()
            .filter(feed -> privacyService.canView(userId, feed))
            .limit(limit)
            .collect(Collectors.toList());

        // 4. 排序（按发布时间倒序）
        filteredFeeds.sort((a, b) -> b.getTimestamp().compareTo(a.getTimestamp()));

        return filteredFeeds;
    }

    // 发布朋友圈
    public void publishMoment(Long userId, MomentContent content, PrivacySetting privacy) {
        // 1. 保存内容
        Moment moment = new Moment();
        moment.setId(snowflakeIdGenerator.nextId());
        moment.setUserId(userId);
        moment.setContent(content);
        moment.setPrivacy(privacy);
        moment.setCreatedAt(System.currentTimeMillis());
        momentRepository.save(moment);

        // 2. 通知好友有新内容（异步）
        List<Long> friendIds = friendService.getFriends(userId);
        for (Long friendId : friendIds) {
            // 推送到好友的未读计数
            unreadCountService.increment(friendId, UnreadType.MOMENTS);
        }
    }
}

// 过载保护：削峰策略
public class OverloadProtectionService {

    private static final int MAX_QPS = 500000; // 系统最大处理能力

    // 消息发送限流
    public SendResult trySendMessage(Long userId, MessageContent content) {
        // 1. 检查用户发送频率
        long userQps = rateLimiter.getQps("user:" + userId);
        if (userQps > MAX_USER_QPS) {
            return SendResult.fail("发送太频繁，请稍后再试");
        }

        // 2. 检查全局 QPS
        long globalQps = rateLimiter.getQps("global");
        if (globalQps > MAX_QPS) {
            // 进入排队模式
            return enqueueForLater(userId, content);
        }

        // 3. 正常发送
        return doSendMessage(userId, content);
    }

    // 异步排队
    private SendResult enqueueForLater(Long userId, MessageContent content) {
        // 放入延迟队列，1-3 秒后再试
        delayQueue.offer(new DelayedMessage(userId, content, 2000));
        return SendResult.successWithDelay("消息已排队，稍后发送");
    }
}

// 降级策略：当系统压力大时，关闭非核心功能
public class DegradationService {

    // 检查是否需要降级
    public boolean shouldDegrade(String feature) {
        SystemLoad load = monitorService.getSystemLoad();

        // 负载超过 80%：关闭非核心功能
        if (load.getCpuUsage() > 80) {
            return degradationRules.get(feature).isDegradable();
        }

        return false;
    }

    // 朋友圈降级：图片质量下降
    public ImageQuality degradeImageQuality() {
        if (shouldDegrade("moments_image")) {
            // 高清 -> 普清
            return ImageQuality.STANDARD;
        }
        return ImageQuality.HIGH;
    }

    // 消息降级：语音转文字暂时关闭
    public boolean isSpeech2TextEnabled() {
        return !shouldDegrade("speech2text");
    }
}

// 就近接入：DNS 解析 + IP 库
public class AccessRouter {

    // 用户发起请求时，DNS 返回最近的接入机房 IP
    public InetSocketAddress resolveAccessPoint(Long userId) {
        // 1. 获取用户地理位置（通过 IP 库）
        String clientIp = getClientIp();
        GeoLocation location = ipGeoLocator.locate(clientIp);

        // 2. 根据地理位置选择最近的接入机房
        // 优先级：同机房 > 同城 > 同区域 > 跨区域
        AccessPoint accessPoint = accessPointSelector.select(
            location,
            Metric.RESPONSE_TIME // 按响应时间选
        );

        return new InetSocketAddress(
            accessPoint.getPublicIp(),
            accessPoint.getPort()
        );
    }

    // 跨机房消息路由
    public void routeMessage(Message message) {
        Long receiverId = message.getReceiverId();

        if (isLocalUser(receiverId)) {
            // 本机房用户：直接投递
            deliverLocally(message);
        } else {
            // 跨机房用户：通过专线投递
            String targetDataCenter = getUserDataCenter(receiverId);
            dcRouter.sendToDataCenter(targetDataCenter, message);
        }
    }
}

// 多机房数据同步
public class CrossDataCenterSync {

    // 用户数据变更时，同步到其他机房
    public void syncUserDataChange(UserDataChange change) {
        // 1. 在本机房更新
        localDataStore.update(change);

        // 2. 异步同步到其他机房
        List<String> otherDataCenters = getOtherDataCenters();
        for (String dc : otherDataCenters) {
            // 通过消息队列异步同步
            mqTemplate.send("dcsync:" + dc, change);
        }
    }

    // 处理同步消息
    @MQListener(topic = "dcsync:*")
    public void onSyncMessage(UserDataChange change) {
        // 应用变更到本地存储
        // 可能存在短暂不一致，但最终一致
        localDataStore.applyChange(change);
    }
}

// 小程序引擎
public class MiniProgramEngine {

    // 小程序运行在独立的 WebView 中
    // 与朋友圈、公众号等原生界面隔离

    // 小程序生命周期
    public void onLaunch(String appId) {
        // 1. 加载小程序包
        WxaPackage pkg = packageManager.loadPackage(appId);

        // 2. 初始化 WebView
        WebView webView = createIsolatedWebView();

        // 3. 加载小程序代码
        webView.loadUrl(pkg.getEntryUrl());

        // 4. 注册微信 JS-SDK
        registerWechatJSApi(webView);
    }

    // 微信 JS-SDK 提供的能力
    // - 微信支付
    // - 获取用户信息
    // - 分享
    // - 地理位置
    // - 扫码
    // - 蓝牙/NFC
    private void registerWechatJSApi(WebView webView) {
        JsInterface interface = new JsInterface() {
            @JavascriptInterface
            public String invoke(String api, String params) {
                return wechatApiBridge.handle(api, params);
            }
        };
        webView.addJavascriptInterface(interface, "WechatJSBridge");
    }
}