Merge commit 'ca66b39b7ee36803864142c954b7f0ca0afbb7ac' into codex/arch/config-database-worker

app/startup/modules_initializer.py

@@ -299,6 +299,20 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
         ):
             raise RuntimeError("订阅完成统计上报未确认")
 
+    def dispatch_subscribe_notification(message) -> None:
+        """恢复订阅完成通知；消息快照无需重建领域对象。"""
+        snapshot = message.payload.get("message") or {}
+        if not isinstance(snapshot, dict):
+            raise RuntimeError("订阅完成通知快照格式无效")
+        CommandChain().post_message(Message.model_validate(snapshot))
+
+    def dispatch_subscribe_added_notification(message) -> None:
+        """恢复订阅新增通知；恢复使用提交前冻结的渲染消息快照。"""
+        snapshot = message.payload.get("message") or {}
+        if not isinstance(snapshot, dict):
+            raise RuntimeError("订阅新增通知快照格式无效")
+        CommandChain().post_message(Message.model_validate(snapshot))
+
     session = SessionFactory()
     return OutboxDispatcher(
         repository=SqlAlchemyOutboxRepository(session),
@@ -308,6 +322,7 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
                 message.payload,
             ),
             "subscribe.added.report": dispatch_subscribe_added_report,
+            "subscribe.added.notification": dispatch_subscribe_added_notification,
             "subscribe.modified": lambda message: EventManager().send_event(
                 EventType.SubscribeModified,
                 message.payload,
@@ -322,6 +337,7 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
                 message.payload,
             ),
             "subscribe.complete.report": dispatch_subscribe_complete_report,
+            "subscribe.complete.notification": dispatch_subscribe_notification,
             "download.added": lambda message: EventManager().send_event(
                 EventType.DownloadAdded,
                 restore_download_added(message.payload),

app/startup/subscription.py

@@ -14,6 +14,7 @@ from app.application.subscription.write import (
     AsyncCreateSubscriptionCommand,
     CreateSubscriptionCommand,
     subscription_added_event_key,
+    subscription_added_notification_key,
     subscription_added_report_key,
 )
 from app.application.subscription.delete import (
@@ -63,6 +64,7 @@ class TransactionalSubscribeWriter:
         payload: dict,
         username: str | None = None,
         after_commit: AfterCommitEffect | None = None,
+        notification: dict[str, object] | None = None,
     ) -> tuple[int, str]:
         """在独占同步会话内执行一次完整订阅新增事务。"""
         session = self._sync_session()
@@ -82,12 +84,23 @@ class TransactionalSubscribeWriter:
                         subscription_added_event_key(subscribe_id, payload),
                         datetime.now(timezone.utc),
                     )
+                    if notification:
+                        outbox.complete_by_event_key(
+                            subscription_added_notification_key(subscribe_id, payload),
+                            datetime.now(timezone.utc),
+                        )
                     outbox.complete_by_event_key(
                         subscription_added_report_key(subscribe_id, payload),
                         datetime.now(timezone.utc),
                     )
 
-            return command.execute(identity, payload, username, delivered)
+            return command.execute(
+                identity,
+                payload,
+                username,
+                delivered,
+                notification,
+            )
         finally:
             session.close()
 
@@ -97,6 +110,7 @@ class TransactionalSubscribeWriter:
         payload: dict,
         username: str | None = None,
         after_commit: AsyncAfterCommitEffect | None = None,
+        notification: dict[str, object] | None = None,
     ) -> tuple[int, str]:
         """在独占异步会话作用域内执行一次完整订阅新增事务。"""
         async with self._async_session() as session:
@@ -115,6 +129,11 @@ class TransactionalSubscribeWriter:
                         subscription_added_event_key(subscribe_id, payload),
                         datetime.now(timezone.utc),
                     )
+                    if notification:
+                        await outbox.complete_by_event_key(
+                            subscription_added_notification_key(subscribe_id, payload),
+                            datetime.now(timezone.utc),
+                        )
                     await outbox.complete_by_event_key(
                         subscription_added_report_key(subscribe_id, payload),
                         datetime.now(timezone.utc),
@@ -125,6 +144,7 @@ class TransactionalSubscribeWriter:
                 payload,
                 username,
                 delivered,
+                notification,
             )