feat:新增napcat断线后重连重新拉取历史消息的机制

plugin-templates/MaiBot-Napcat-Adapter/runtime/builder.py

@@ -14,6 +14,7 @@ from ..services import (
     NapCatActionService,
     NapCatBanStateStore,
     NapCatBanTracker,
+    NapCatHistoryRecoveryStore,
     NapCatOfficialBotGuard,
     NapCatQueryService,
 )
@@ -66,6 +67,7 @@ class NapCatRuntimeBuilder:
         action_service = NapCatActionService(self._logger, transport)
         query_service = NapCatQueryService(action_service, self._logger)
         ban_state_store = NapCatBanStateStore(self._logger)
+        history_recovery_store = NapCatHistoryRecoveryStore(self._logger)
         inbound_codec = NapCatInboundCodec(self._logger, query_service)
         notice_codec = NapCatNoticeCodec(self._logger, query_service)
         runtime_state = NapCatRuntimeStateManager(
@@ -92,6 +94,7 @@ class NapCatRuntimeBuilder:
             ban_tracker=ban_tracker,
             chat_filter=chat_filter,
             heartbeat_monitor=heartbeat_monitor,
+            history_recovery_store=history_recovery_store,
             inbound_codec=inbound_codec,
             notice_codec=notice_codec,
             official_bot_guard=official_bot_guard,

plugin-templates/MaiBot-Napcat-Adapter/runtime/bundle.py

@@ -14,6 +14,7 @@ from ..services import (
     NapCatActionService,
     NapCatBanStateStore,
     NapCatBanTracker,
+    NapCatHistoryRecoveryStore,
     NapCatOfficialBotGuard,
     NapCatQueryService,
 )
@@ -29,6 +30,7 @@ class NapCatRuntimeBundle:
     ban_tracker: NapCatBanTracker
     chat_filter: NapCatChatFilter
     heartbeat_monitor: NapCatHeartbeatMonitor
+    history_recovery_store: NapCatHistoryRecoveryStore
     inbound_codec: NapCatInboundCodec
     notice_codec: NapCatNoticeCodec
     official_bot_guard: NapCatOfficialBotGuard

plugin-templates/MaiBot-Napcat-Adapter/runtime/router.py

@@ -2,11 +2,14 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import Any, Callable, Dict, Mapping, Optional, Protocol
 
 import asyncio
 
 from ..config import NapCatPluginSettings
+from ..constants import DEFAULT_HISTORY_RECOVERY_BATCH_SIZE, DEFAULT_HISTORY_RECOVERY_CHECKPOINT_LIMIT
+from ..services import NapCatChatCheckpoint
 from ..types import NapCatPayloadDict
 from .bundle import NapCatRuntimeBundle
 
@@ -27,6 +30,14 @@ class _GatewayCapabilityProtocol(Protocol):
         ...
 
 
+@dataclass(frozen=True)
+class _NapCatChatIdentity:
+    """描述一条 NapCat 消息所属的会话身份。"""
+
+    chat_type: str
+    chat_id: str
+
+
 class NapCatEventRouter:
     """协调 NapCat 运行时组件处理各类平台事件。"""
 
@@ -50,6 +61,7 @@ class NapCatEventRouter:
         self._gateway_name = gateway_name
         self._load_settings = load_settings
         self._runtime: Optional[NapCatRuntimeBundle] = None
+        self._recovery_task: Optional[asyncio.Task[None]] = None
 
     def bind_runtime(self, runtime: NapCatRuntimeBundle) -> None:
         """绑定当前路由器使用的运行时依赖。
@@ -64,6 +76,7 @@ class NapCatEventRouter:
         runtime = self._runtime
         if runtime is None:
             return
+        self._cancel_recovery_task()
         runtime.official_bot_guard.clear_cache()
 
     async def handle_transport_payload(self, payload: NapCatPayloadDict) -> None:
@@ -82,7 +95,7 @@ class NapCatEventRouter:
         if post_type == "meta_event":
             await self.handle_meta_event(payload)
 
-    async def handle_inbound_message(self, payload: NapCatPayloadDict) -> None:
+    async def handle_inbound_message(self, payload: NapCatPayloadDict) -> bool:
         """处理单条 NapCat 入站消息并注入 Host。
 
         Args:
@@ -101,25 +114,25 @@ class NapCatEventRouter:
 
         sender_user_id = str(payload.get("user_id") or sender.get("user_id") or "").strip()
         if not sender_user_id:
-            return
+            return False
 
         group_id = str(payload.get("group_id") or "").strip()
         if self_id and sender_user_id == self_id and settings.filters.ignore_self_message:
-            return
+            return False
         if not runtime.chat_filter.is_inbound_chat_allowed(sender_user_id, group_id, settings.chat):
-            return
+            return False
         if await runtime.official_bot_guard.should_reject(
             sender_user_id=sender_user_id,
             group_id=group_id,
             ban_qq_bot=settings.chat.ban_qq_bot,
         ):
-            return
+            return False
 
         try:
             message_dict = await runtime.inbound_codec.build_message_dict(payload, self_id, sender_user_id, sender)
         except ValueError as exc:
             self._logger.warning(f"NapCat 入站消息格式不受支持，已丢弃: {exc}")
-            return
+            return False
 
         route_metadata = self._build_route_metadata(self_id, settings.napcat_server.connection_id)
         external_message_id = str(payload.get("message_id") or "").strip()
@@ -132,6 +145,15 @@ class NapCatEventRouter:
         )
         if not accepted:
             self._logger.debug(f"Host 丢弃了 NapCat 入站消息: {external_message_id or '无消息 ID'}")
+            return False
+
+        await self._record_inbound_checkpoint(
+            payload=payload,
+            self_id=self_id,
+            external_message_id=external_message_id or str(message_dict.get("message_id") or "").strip(),
+            scope=settings.napcat_server.connection_id,
+        )
+        return True
 
     async def handle_notice_event(self, payload: NapCatPayloadDict) -> None:
         """处理 NapCat ``notice`` 事件并注入 Host。
@@ -232,6 +254,8 @@ class NapCatEventRouter:
                 await runtime.runtime_state.report_connected(self_id, settings.napcat_server)
                 await runtime.heartbeat_monitor.start(self_id, settings.napcat_server.heartbeat_interval)
                 await runtime.ban_tracker.start()
+                await runtime.history_recovery_store.load()
+                self._schedule_history_recovery(self_id=self_id, scope=settings.napcat_server.connection_id)
                 return
             except asyncio.CancelledError:
                 raise
@@ -279,6 +303,274 @@ class NapCatEventRouter:
             raise RuntimeError("NapCat 运行时尚未初始化")
         return runtime
 
+    def _schedule_history_recovery(self, self_id: str, scope: str) -> None:
+        """在连接恢复后调度一次历史补拉任务。"""
+
+        self._cancel_recovery_task()
+        runtime = self._runtime
+        if runtime is None:
+            return
+
+        self._recovery_task = asyncio.create_task(
+            self._recover_recent_history(self_id=self_id, scope=scope),
+            name="napcat_adapter.history_recovery",
+        )
+
+    def _cancel_recovery_task(self) -> None:
+        """取消当前仍在运行的历史补拉任务。"""
+
+        recovery_task = self._recovery_task
+        self._recovery_task = None
+        if recovery_task is not None and not recovery_task.done():
+            recovery_task.cancel()
+
+    async def _recover_recent_history(self, *, self_id: str, scope: str) -> None:
+        """按 checkpoint 列表逐个尝试补拉断线期间遗漏的消息。"""
+
+        runtime = self._require_runtime()
+        checkpoints = await runtime.history_recovery_store.list_checkpoints(
+            self_id,
+            scope=scope,
+            limit=DEFAULT_HISTORY_RECOVERY_CHECKPOINT_LIMIT,
+        )
+        if not checkpoints:
+            return
+
+        recovered_count = 0
+        for checkpoint in checkpoints:
+            recovered_count += await self._recover_chat_history_from_checkpoint(
+                self_id=self_id,
+                scope=scope,
+                checkpoint=checkpoint,
+            )
+
+        if recovered_count > 0:
+            self._logger.info(f"NapCat 历史补拉完成，共补回 {recovered_count} 条消息")
+
+    async def _recover_chat_history_from_checkpoint(
+        self,
+        *,
+        self_id: str,
+        scope: str,
+        checkpoint: NapCatChatCheckpoint,
+    ) -> int:
+        """针对单个会话执行一次小批量历史补拉。"""
+
+        runtime = self._require_runtime()
+        history_messages = await self._query_history_messages(checkpoint, limit=DEFAULT_HISTORY_RECOVERY_BATCH_SIZE)
+        if not history_messages:
+            return 0
+
+        ordered_messages = sorted(
+            history_messages,
+            key=lambda item: (
+                self._extract_message_timestamp(item),
+                self._extract_message_seq(item),
+                str(item.get("message_id") or "").strip(),
+            ),
+        )
+
+        recovered_count = 0
+        for history_payload in ordered_messages:
+            external_message_id = str(history_payload.get("message_id") or "").strip()
+            if not external_message_id:
+                continue
+            if external_message_id == checkpoint.last_message_id:
+                continue
+            if await runtime.history_recovery_store.has_recovered_message_seen(
+                account_id=self_id,
+                scope=scope,
+                chat_type=checkpoint.chat_type,
+                chat_id=checkpoint.chat_id,
+                external_message_id=external_message_id,
+            ):
+                continue
+            if not self._is_message_after_checkpoint(history_payload, checkpoint):
+                continue
+            accepted = await self._reinject_history_payload(history_payload, self_id=self_id)
+            if not accepted:
+                continue
+            await runtime.history_recovery_store.mark_recovered_message_seen(
+                account_id=self_id,
+                scope=scope,
+                chat_type=checkpoint.chat_type,
+                chat_id=checkpoint.chat_id,
+                external_message_id=external_message_id,
+            )
+            recovered_count += 1
+
+        return recovered_count
+
+    async def _query_history_messages(
+        self,
+        checkpoint: NapCatChatCheckpoint,
+        *,
+        limit: int,
+    ) -> list[NapCatPayloadDict]:
+        """查询某个会话在 checkpoint 之后的一小批历史消息。"""
+
+        runtime = self._require_runtime()
+        payload_collections: list[list[NapCatPayloadDict]] = []
+        if checkpoint.last_message_seq is not None:
+            payload_collections.append(
+                await self._fetch_history_messages(
+                    chat_type=checkpoint.chat_type,
+                    chat_id=checkpoint.chat_id,
+                    message_seq=checkpoint.last_message_seq,
+                    limit=limit,
+                )
+            )
+        payload_collections.append(
+            await self._fetch_history_messages(
+                chat_type=checkpoint.chat_type,
+                chat_id=checkpoint.chat_id,
+                message_seq=None,
+                limit=limit,
+            )
+        )
+
+        merged_payloads: list[NapCatPayloadDict] = []
+        seen_message_ids: set[str] = set()
+        for payloads in payload_collections:
+            for payload in payloads:
+                external_message_id = str(payload.get("message_id") or "").strip()
+                dedupe_key = external_message_id or repr(sorted(payload.items()))
+                if dedupe_key in seen_message_ids:
+                    continue
+                seen_message_ids.add(dedupe_key)
+                merged_payloads.append(payload)
+        return merged_payloads
+
+    async def _fetch_history_messages(
+        self,
+        *,
+        chat_type: str,
+        chat_id: str,
+        message_seq: int | None,
+        limit: int,
+    ) -> list[NapCatPayloadDict]:
+        """调用查询服务获取一批历史消息。"""
+
+        runtime = self._require_runtime()
+        if chat_type == "group":
+            history_payloads = await runtime.query_service.get_group_message_history(
+                chat_id,
+                message_seq=message_seq,
+                count=limit,
+                reverse_order=False,
+            )
+        elif chat_type == "private":
+            history_payloads = await runtime.query_service.get_friend_message_history(
+                chat_id,
+                message_seq=message_seq,
+                count=limit,
+                reverse_order=False,
+            )
+        else:
+            return []
+
+        if history_payloads is None:
+            return []
+        return [dict(payload) for payload in history_payloads if isinstance(payload, Mapping)]
+
+    async def _reinject_history_payload(self, payload: NapCatPayloadDict, *, self_id: str) -> bool:
+        """将补拉到的历史消息重新送回实时入站路径。"""
+
+        try:
+            normalized_payload = dict(payload)
+            if self_id and not str(normalized_payload.get("self_id") or "").strip():
+                normalized_payload["self_id"] = self_id
+            return await self.handle_inbound_message(normalized_payload)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            external_message_id = str(payload.get("message_id") or "").strip() or "unknown"
+            self._logger.warning(f"NapCat 历史消息补拉注入失败: message_id={external_message_id} error={exc}")
+            return False
+
+    async def _record_inbound_checkpoint(
+        self,
+        *,
+        payload: NapCatPayloadDict,
+        self_id: str,
+        external_message_id: str,
+        scope: str,
+    ) -> None:
+        """在消息被 Host 接受后更新该会话的最新 checkpoint。"""
+
+        runtime = self._require_runtime()
+        chat_identity = self._extract_chat_identity(payload)
+        if chat_identity is None:
+            return
+
+        await runtime.history_recovery_store.record_checkpoint(
+            account_id=self_id,
+            scope=scope,
+            chat_type=chat_identity.chat_type,
+            chat_id=chat_identity.chat_id,
+            message_id=external_message_id,
+            message_time=self._extract_message_timestamp(payload),
+            message_seq=self._extract_message_seq(payload),
+        )
+
+    @staticmethod
+    def _extract_chat_identity(payload: Mapping[str, Any]) -> _NapCatChatIdentity | None:
+        """从 NapCat 载荷中提取会话身份。"""
+
+        group_id = str(payload.get("group_id") or "").strip()
+        user_id = str(payload.get("user_id") or "").strip()
+
+        if group_id:
+            return _NapCatChatIdentity(chat_type="group", chat_id=group_id)
+        if user_id:
+            return _NapCatChatIdentity(chat_type="private", chat_id=user_id)
+        return None
+
+    @staticmethod
+    def _extract_message_seq(payload: Mapping[str, Any]) -> int | None:
+        """从 NapCat 载荷中提取历史接口可复用的消息序号。"""
+
+        for field_name in ("message_seq", "messageSeq", "msg_seq"):
+            raw_value = payload.get(field_name)
+            if raw_value is None or str(raw_value).strip() == "":
+                continue
+            try:
+                return int(raw_value)
+            except (TypeError, ValueError):
+                continue
+        return None
+
+    @staticmethod
+    def _extract_message_timestamp(payload: Mapping[str, Any]) -> float:
+        """从 NapCat 载荷中提取消息时间戳。"""
+
+        raw_timestamp = payload.get("time")
+        if isinstance(raw_timestamp, (int, float)):
+            return float(raw_timestamp)
+        return 0.0
+
+    @classmethod
+    def _is_message_after_checkpoint(
+        cls,
+        payload: Mapping[str, Any],
+        checkpoint: NapCatChatCheckpoint,
+    ) -> bool:
+        """判断历史消息是否位于 checkpoint 之后。"""
+
+        payload_message_id = str(payload.get("message_id") or "").strip()
+        if payload_message_id == checkpoint.last_message_id:
+            return False
+
+        payload_message_seq = cls._extract_message_seq(payload)
+        if payload_message_seq is not None and checkpoint.last_message_seq is not None:
+            return payload_message_seq > checkpoint.last_message_seq
+
+        payload_timestamp = cls._extract_message_timestamp(payload)
+        if payload_timestamp != checkpoint.last_message_time:
+            return payload_timestamp > checkpoint.last_message_time
+
+        return True
+
     @staticmethod
     def _build_route_metadata(self_id: str, connection_id: str) -> Dict[str, Any]:
         """构造注入 Host 时使用的路由元数据。