完成 MR 通知与 Hook/Channel 模块重构

gitlab_bot.py

@@ -17,18 +17,25 @@
 
 from dotenv import load_dotenv
 
-from src.approval_notification import ApprovalNotificationHooks, LogChannel
+from src.channels.log import LogChannel
 from src.config import (
     bot_gitlab_token,
     bot_gitlab_url,
     bot_gitlab_username,
     bot_host,
     bot_port,
+    merge_notification_db_path,
+    merge_notification_sending_timeout_seconds,
 )
-from src.issue_hook import IssueHooks
+from src.delivery.coordinator import NotificationDelivery
+from src.delivery.idempotent_channel import DurableIdempotentChannel
+from src.delivery.sqlite import NotificationDeliveryStore
+from src.hooks.approval_notification import ApprovalNotificationHooks
+from src.hooks.issue import IssueHooks
+from src.hooks.merge_notification import MergeRequestNotificationHooks
+from src.hooks.merge_request import MergeRequestHooks
+from src.hooks.note import NoteHooks
 from src.logs import print_event
-from src.merge_request_hook import MergeRequestHooks
-from src.note_hook import NoteHooks
 
 load_dotenv()  # isort:skip
 
@@ -58,7 +65,25 @@ def _load_gitlab_bot():
 issue_hooks = IssueHooks()
 merge_request_hooks = MergeRequestHooks()
 note_hooks = NoteHooks()
-approval_notification_hooks = ApprovalNotificationHooks(LogChannel())
+notification_channel = LogChannel()
+approval_notification_hooks = ApprovalNotificationHooks(notification_channel)
+notification_delivery_store = NotificationDeliveryStore(
+    merge_notification_db_path,
+    sending_timeout_seconds=merge_notification_sending_timeout_seconds,
+)
+merge_notification_channel = DurableIdempotentChannel(notification_channel, notification_delivery_store)
+merge_notification_delivery = NotificationDelivery(merge_notification_channel, notification_delivery_store)
+merge_request_notification_hooks = MergeRequestNotificationHooks(
+    merge_notification_channel,
+    delivery=merge_notification_delivery,
+)
+
+
+async def recover_merge_notification_deliveries(_app):
+    await merge_request_notification_hooks.recover()
+
+
+bot.app.on_startup.append(recover_merge_notification_deliveries)
 
 
 @bot.router.register("Issue Hook", action="open")
@@ -115,6 +140,11 @@ async def merge_request_unapproval_event(event, gl, *args, **kwargs):
     await approval_notification_hooks.handle(event, gl, *args, **kwargs)
 
 
+@bot.router.register("Merge Request Hook", action="merge")
+async def merge_request_merged_event(event, gl, *args, **kwargs):
+    await merge_request_notification_hooks.handle(event, gl, *args, **kwargs)
+
+
 @bot.router.register("Note Hook", noteable_type="MergeRequest")
 async def note_merge_request_event(event, gl, *args, **kwargs):
     if not ignore_event(event):

src/channels/__init__.py

@@ -0,0 +1,18 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .base import Channel
+from .log import LogChannel
+
+__all__ = ["Channel", "LogChannel"]

src/channels/base.py

@@ -0,0 +1,25 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from abc import ABC, abstractmethod
+
+from ..notifications.model import MergeRequestNotification
+
+
+class Channel(ABC):
+    """Asynchronous destination for normalized notifications."""
+
+    @abstractmethod
+    async def send(self, notification: MergeRequestNotification) -> None:
+        raise NotImplementedError

src/channels/log.py

@@ -0,0 +1,31 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+from dataclasses import asdict
+from typing import Optional
+
+from ..notifications.model import MergeRequestNotification
+from .base import Channel
+
+
+class LogChannel(Channel):
+    """Write normalized notifications as searchable JSON log records."""
+
+    def __init__(self, logger: Optional[logging.Logger] = None):
+        self.logger = logger or logging.getLogger(__name__)
+
+    async def send(self, notification: MergeRequestNotification) -> None:
+        self.logger.info(json.dumps(asdict(notification), ensure_ascii=False))

src/config.py

@@ -159,3 +159,12 @@
 )
 
 bot_gitlab_merge_request_reviewer_username = os.getenv("BOT_GITLAB_MERGE_REQUEST_REVIEWER_USERNAME", None)
+
+# merge notification delivery
+merge_notification_db_path = os.getenv(
+    "MERGE_NOTIFICATION_DB_PATH",
+    "data/merge-notifications.sqlite3",
+)
+merge_notification_sending_timeout_seconds = float(
+    os.getenv("MERGE_NOTIFICATION_SENDING_TIMEOUT_SECONDS", "300")
+)

src/delivery/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

src/delivery/coordinator.py

@@ -0,0 +1,99 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+from collections import Counter
+from typing import Any, List, Optional
+
+from ..channels.base import Channel
+from ..notifications.model import MergeRequestNotification
+from .sqlite import DeliveryRecord, NotificationDeliveryStore
+
+
+class NotificationDelivery:
+    """Coordinate durable delivery state, recovery, and observability."""
+
+    def __init__(
+        self,
+        channel: Channel,
+        store: NotificationDeliveryStore,
+        logger: Optional[logging.Logger] = None,
+    ):
+        self.channel = channel
+        self.store = store
+        self.logger = logger or logging.getLogger(__name__)
+        self.counters = Counter()
+
+    def _log(self, level: int, event: str, notification: MergeRequestNotification, **extra: Any) -> None:
+        payload = {
+            "event": "merge_notification_delivery",
+            "action": event,
+            "idempotency_key": notification.idempotency_key,
+            "project": notification.project.get("path") or notification.project.get("id"),
+            "mr_iid": notification.merge_request.get("iid"),
+        }
+        payload.update(extra)
+        self.logger.log(level, json.dumps(payload, ensure_ascii=False, sort_keys=True))
+        self.counters[event] += 1
+
+    async def deliver(self, notification: MergeRequestNotification) -> bool:
+        try:
+            decision = self.store.begin_delivery(notification)
+        except Exception as exc:
+            self.counters["failed"] += 1
+            self.logger.error("notification delivery state failed: %s", exc, exc_info=True)
+            return False
+        if not decision.should_send:
+            self._log(logging.INFO, "duplicate", notification, reason=decision.reason)
+            return False
+
+        try:
+            await self.channel.send(notification)
+        except Exception as exc:
+            self.store.mark_failed(notification.idempotency_key, str(exc))
+            self._log(logging.ERROR, "failed", notification, error=str(exc))
+            return False
+
+        reconcile = getattr(self.channel, "reconcile", None)
+        if callable(reconcile):
+            reconcile(notification)
+        self.store.mark_sent(notification.idempotency_key)
+        self._log(logging.INFO, "sent", notification, reason=decision.reason)
+        return True
+
+    async def recover(self) -> int:
+        recovered = 0
+        for record in self.store.recoverable_deliveries():
+            if await self.deliver(record.notification):
+                recovered += 1
+        if recovered:
+            self.logger.info(
+                json.dumps({"event": "merge_notification_recovery", "count": recovered}, sort_keys=True)
+            )
+        return recovered
+
+    async def replay_failed(self) -> int:
+        replayed = 0
+        for record in self.store.failed_deliveries():
+            if await self.deliver(record.notification):
+                replayed += 1
+        if replayed:
+            self.logger.info(
+                json.dumps({"event": "merge_notification_replay", "count": replayed}, sort_keys=True)
+            )
+        return replayed
+
+    def failure_backlog(self) -> List[DeliveryRecord]:
+        return self.store.failed_deliveries()

src/delivery/idempotent_channel.py

@@ -0,0 +1,38 @@
+# Copyright 2026 Lei Zhang
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ..channels.base import Channel
+from ..notifications.model import MergeRequestNotification
+from .sqlite import NotificationDeliveryStore
+
+
+class DurableIdempotentChannel(Channel):
+    """Add a durable idempotency ledger around an existing Channel."""
+
+    def __init__(self, channel: Channel, store: NotificationDeliveryStore):
+        self.channel = channel
+        self.store = store
+
+    async def send(self, notification: MergeRequestNotification) -> None:
+        if not self.store.claim_channel_effect(notification):
+            return
+        try:
+            await self.channel.send(notification)
+        except Exception as exc:
+            self.store.release_channel_effect(notification.idempotency_key, str(exc))
+            raise
+        self.store.mark_channel_accepted(notification.idempotency_key)
+
+    def reconcile(self, notification: MergeRequestNotification) -> None:
+        self.store.mark_channel_accepted(notification.idempotency_key)