diff --git a/app/application/subscription/complete.py b/app/application/subscription/complete.py index 90ecabe4d..b5f7f750e 100644 --- a/app/application/subscription/complete.py +++ b/app/application/subscription/complete.py @@ -49,8 +49,9 @@ class CompleteSubscriptionCommand: mediainfo: Mapping[str, Any], notify: CompletionEffect, report: CompletionReporter, + notification: Mapping[str, Any] | None = None, ) -> None: - """在同一事务中写历史、删订阅并暂存完成事件与统计意图。""" + """在同一事务中写历史、删订阅并暂存完成事件、通知与统计意图。""" info = dict(subscribe_info) event_payload = { "subscribe_id": subscribe_id, @@ -60,6 +61,7 @@ class CompleteSubscriptionCommand: } event_key = event_payload["idempotency_key"] report_key = completion_report_key(subscribe_id, info) + notification_key = completion_notification_key(subscribe_id, info) report_payload = {"subscribe_info": _completion_report_payload(info, report_key)} try: self._repository.add_history(**info) @@ -74,6 +76,18 @@ class CompleteSubscriptionCommand: ), now, ) + if notification: + self._outbox.stage( + OutboxIntent( + event_key=notification_key, + topic="subscribe.complete.notification", + payload={ + "idempotency_key": notification_key, + "message": dict(notification), + }, + ), + now, + ) self._outbox.stage( OutboxIntent( event_key=report_key, @@ -88,6 +102,11 @@ class CompleteSubscriptionCommand: raise notify() + if self._outbox and notification: + self._outbox.complete_by_event_key( + notification_key, + datetime.now(timezone.utc), + ) self._publish(event_payload) if self._outbox: self._outbox.complete_by_event_key(event_key, datetime.now(timezone.utc)) @@ -111,6 +130,14 @@ def completion_report_key(subscribe_id: int, subscribe_info: Mapping[str, Any]) return f"{completion_event_key(subscribe_id, subscribe_info)}:report" +def completion_notification_key( + subscribe_id: int, + subscribe_info: Mapping[str, Any], +) -> str: + """构造订阅完成通知的稳定幂等键，避免恢复时重复生成不同消息。""" + return f"{completion_event_key(subscribe_id, subscribe_info)}:notification" + + def _completion_report_payload( subscribe_info: Mapping[str, Any], report_key: str, diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index 04f49cf92..ca34eec74 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -17,7 +17,7 @@ app/application/history.py 里整理历史的写入路径同构。 from collections.abc import Awaitable, Callable from datetime import datetime, timezone -from typing import Optional, Protocol, Tuple +from typing import Mapping, Optional, Protocol, Tuple from app.application.outbox import OutboxIntent from app.domain.context import MediaInfo, MusicInfo @@ -55,6 +55,7 @@ class SubscribeWriter(Protocol): payload: dict, username: Optional[str] = None, after_commit: Optional[AfterCommitEffect] = None, + notification: Mapping[str, object] | None = None, ) -> Tuple[int, str]: """同步新增订阅，并在事务成功后执行外部副作用。""" @@ -64,6 +65,7 @@ class SubscribeWriter(Protocol): payload: dict, username: Optional[str] = None, after_commit: Optional[AsyncAfterCommitEffect] = None, + notification: Mapping[str, object] | None = None, ) -> Tuple[int, str]: """异步新增订阅，并在事务成功后执行外部副作用。""" @@ -153,6 +155,7 @@ class CreateSubscriptionCommand: payload: dict, username: Optional[str] = None, after_commit: Optional[AfterCommitEffect] = None, + notification: Mapping[str, object] | None = None, ) -> Tuple[int, str]: """执行同步新增；事务失败回滚，提交后副作用失败不反向回滚。""" try: @@ -164,6 +167,7 @@ class CreateSubscriptionCommand: staged.subscribe_id, payload, username, + notification, ): self._outbox.stage(intent, now) self._unit_of_work.commit() @@ -195,6 +199,7 @@ class AsyncCreateSubscriptionCommand: payload: dict, username: Optional[str] = None, after_commit: Optional[AsyncAfterCommitEffect] = None, + notification: Mapping[str, object] | None = None, ) -> Tuple[int, str]: """执行异步新增；事务失败回滚，提交后副作用失败不反向回滚。""" try: @@ -210,6 +215,7 @@ class AsyncCreateSubscriptionCommand: staged.subscribe_id, payload, username, + notification, ): await self._outbox.stage(intent, now) await self._unit_of_work.commit() @@ -225,26 +231,44 @@ def _subscribe_added_intents( subscribe_id: int, payload: dict, username: str | None, -) -> tuple[OutboxIntent, OutboxIntent]: - """构造订阅新增事件与外部统计的同事务 durable intents。""" + notification: Mapping[str, object] | None = None, +) -> tuple[OutboxIntent, ...]: + """构造订阅新增事件、通知与外部统计的同事务 durable intents。""" event_key = subscription_added_event_key(subscribe_id, payload) event_payload = { "subscribe_id": subscribe_id, "username": username, "mediainfo": dict(payload), } - return ( + intents: list[OutboxIntent] = [ OutboxIntent( event_key=event_key, topic="subscribe.added", payload=event_payload, ), + ] + if notification: + intents.append( + OutboxIntent( + event_key=subscription_added_notification_key(subscribe_id, payload), + topic="subscribe.added.notification", + payload={ + "idempotency_key": subscription_added_notification_key( + subscribe_id, + payload, + ), + "message": dict(notification), + }, + ) + ) + intents.append( OutboxIntent( event_key=subscription_added_report_key(subscribe_id, payload), topic="subscribe.added.report", payload={"subscribe_info": dict(payload)}, - ), + ) ) + return tuple(intents) def subscription_added_event_key(subscribe_id: int, payload: dict) -> str: @@ -261,6 +285,11 @@ def subscription_added_report_key(subscribe_id: int, payload: dict) -> str: return f"{subscription_added_event_key(subscribe_id, payload)}:report" +def subscription_added_notification_key(subscribe_id: int, payload: dict) -> str: + """构造订阅新增通知的稳定幂等键。""" + return f"{subscription_added_event_key(subscribe_id, payload)}:notification" + + _configured_subscribe_writer: Callable[[], SubscribeWriter] | None = None @@ -345,6 +374,7 @@ def add_subscribe( mediainfo: MediaInfo | MusicInfo, subscribe_oper: Optional[SubscribeWriter] = None, after_commit: Optional[AfterCommitEffect] = None, + notification: Mapping[str, object] | None = None, **kwargs, ) -> Tuple[int, str]: """ @@ -361,13 +391,20 @@ def add_subscribe( return INCOMPLETE_IDENTITY identity, payload, username = translated oper = _get_subscribe_writer(subscribe_oper) + extra = {"notification": notification} if notification is not None else {} if after_commit is None: - return oper.add(identity=identity, payload=payload, username=username) + return oper.add( + identity=identity, + payload=payload, + username=username, + **extra, + ) return oper.add( identity=identity, payload=payload, username=username, after_commit=after_commit, + **extra, ) @@ -375,6 +412,7 @@ async def async_add_subscribe( mediainfo: MediaInfo | MusicInfo, subscribe_oper: Optional[SubscribeWriter] = None, after_commit: Optional[AsyncAfterCommitEffect] = None, + notification: Mapping[str, object] | None = None, **kwargs, ) -> Tuple[int, str]: """ @@ -391,13 +429,20 @@ async def async_add_subscribe( return INCOMPLETE_IDENTITY identity, payload, username = translated oper = _get_subscribe_writer(subscribe_oper) + extra = {"notification": notification} if notification is not None else {} if after_commit is None: - return await oper.async_add(identity=identity, payload=payload, username=username) + return await oper.async_add( + identity=identity, + payload=payload, + username=username, + **extra, + ) return await oper.async_add( identity=identity, payload=payload, username=username, after_commit=after_commit, + **extra, ) diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 25e12a51b..b5bc6e82e 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -44,6 +44,7 @@ from app.application.configuration import ( get_configured_system_config, ) from app.application.messaging.subscribe import SubscribeInteractionHandler +from app.application.messaging.message import MessageTemplateHelper from app.application.mediaserver import MediaServerHelper from app.application.subscription.write import add_subscribe, async_add_subscribe from app.application.subscription.complete import get_subscription_completion_scope @@ -109,6 +110,7 @@ class _SubscribePostCommitContext: userid: Optional[str] username: Optional[str] message: bool + notification: Optional[dict] = None @dataclass(slots=True) @@ -934,22 +936,8 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): context: _SubscribePostCommitContext, ) -> None: """同步执行提交后消息、事件和统计，异常不再触碰数据库事务。""" - if context.message: - self.post_message( - _SchemaMessage( - channel=context.channel, - source=context.source, - mtype=MessageType.Subscribe, - ctype=ContentType.SubscribeAdded, - image=context.mediainfo.get_message_image(), - link=self.__subscribe_added_link(context.mediainfo.type), - userid=context.userid, - username=context.username, - ), - meta=context.metainfo, - mediainfo=context.mediainfo, - username=context.username, - ) + if context.notification: + self.post_message(_SchemaMessage.model_validate(context.notification)) eventmanager.send_event(EventType.SubscribeAdded, { "subscribe_id": subscribe_id, "idempotency_key": ( @@ -970,21 +958,9 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): context: _SubscribePostCommitContext, ) -> None: """异步执行提交后消息、事件和统计，保持与同步入口相同顺序。""" - if context.message: + if context.notification: await self.async_post_message( - _SchemaMessage( - channel=context.channel, - source=context.source, - mtype=MessageType.Subscribe, - ctype=ContentType.SubscribeAdded, - image=context.mediainfo.get_message_image(), - link=self.__subscribe_added_link(context.mediainfo.type), - userid=context.userid, - username=context.username, - ), - meta=context.metainfo, - mediainfo=context.mediainfo, - username=context.username, + _SchemaMessage.model_validate(context.notification) ) await eventmanager.async_send_event(EventType.SubscribeAdded, { "subscribe_id": subscribe_id, @@ -1296,6 +1272,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): @staticmethod def __subscribe_post_commit_context( context: _SubscribeCreateContext, + notification: Optional[dict] = None, ) -> _SubscribePostCommitContext: """从创建阶段状态冻结提交后副作用需要的最小快照。""" return _SubscribePostCommitContext( @@ -1311,11 +1288,40 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): userid=context.userid, username=context.username, message=context.message, + notification=notification, ) + def __build_subscribe_notification( + self, + context: _SubscribeCreateContext, + ) -> Optional[dict]: + """在事务提交前冻结已渲染消息，供即时发送与 outbox 恢复共用。""" + if not context.message: + return None + message = _SchemaMessage( + channel=context.channel, + source=context.source, + mtype=MessageType.Subscribe, + ctype=ContentType.SubscribeAdded, + image=context.mediainfo.get_message_image(), + link=self.__subscribe_added_link(context.mediainfo.type), + userid=context.userid, + username=context.username, + ) + rendered = MessageTemplateHelper.render( + message, + meta=context.metainfo, + mediainfo=context.mediainfo, + username=context.username, + ) or message + return rendered.model_dump(mode="json") + def __persist_subscribe_create(self, context: _SubscribeCreateContext) -> Tuple[Optional[int], str]: """同步提交订阅，并在提交成功后按原顺序执行消息、事件和统计。""" - post_commit_context = self.__subscribe_post_commit_context(context) + post_commit_context = self.__subscribe_post_commit_context( + context, + self.__build_subscribe_notification(context), + ) def _after_commit(subscribe_id: int) -> None: """把同步提交后的副作用委托给单一顺序实现。""" @@ -1326,6 +1332,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): season=context.season, username=context.username, after_commit=_after_commit, + notification=post_commit_context.notification, **context.options, ) if not sid: @@ -1338,7 +1345,10 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): context: _SubscribeCreateContext, ) -> Tuple[Optional[int], str]: """异步提交订阅，并在提交成功后按原顺序执行消息、事件和统计。""" - post_commit_context = self.__subscribe_post_commit_context(context) + post_commit_context = self.__subscribe_post_commit_context( + context, + self.__build_subscribe_notification(context), + ) async def _after_commit(subscribe_id: int) -> None: """把异步提交后的副作用委托给单一顺序实现。""" @@ -1349,6 +1359,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): season=context.season, username=context.username, after_commit=_after_commit, + notification=post_commit_context.notification, **context.options, ) if not sid: @@ -3108,19 +3119,22 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): def notify() -> None: """提交成功后发送完成通知，保持历史消息 ABI。""" - self.post_message( - _SchemaMessage( - mtype=MessageType.Subscribe, - ctype=ContentType.SubscribeComplete, - image=mediainfo.get_message_image(), - link=link, - username=subscribe.username, - ), - meta=meta, - mediainfo=mediainfo, - msgstr=msgstr, - username=subscribe.username, - ) + self.post_message(_completion_message) + + _completion_message = _SchemaMessage( + mtype=MessageType.Subscribe, + ctype=ContentType.SubscribeComplete, + image=mediainfo.get_message_image(), + link=link, + username=subscribe.username, + ) + _completion_message = MessageTemplateHelper.render( + _completion_message, + meta=meta, + mediainfo=mediainfo, + msgstr=msgstr, + username=subscribe.username, + ) or _completion_message with get_subscription_completion_scope() as command: command.execute( @@ -3129,6 +3143,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): mediainfo=mediainfo.to_dict(), notify=notify, report=MoviePilotServerHelper.sub_done_durable, + notification=_completion_message.model_dump(mode="json"), ) def _interaction_handler(self) -> "SubscribeInteractionHandler": diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 52260d4d7..adbd05f39 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -259,6 +259,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), @@ -268,6 +282,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, @@ -282,6 +297,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), diff --git a/app/startup/subscription.py b/app/startup/subscription.py index 06ffdab79..7d643f79f 100644 --- a/app/startup/subscription.py +++ b/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, ) diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index b17c57563..5f39b7e80 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -805,6 +805,10 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas 投递原有 `subscribe_id`、`subscribe_info`、`mediainfo` 字段，仅增加可选 `idempotency_key`。 - 普通订阅新增/修改/删除路径的用户通知与第三方插件自行发送的事件仍不自动纳入宿主事务；本切片只覆盖 主仓可追踪的 `SubscribeChain` 完成生产者。 +- 2026-08-23 将主仓可追踪的订阅新增与完成用户通知冻结为已渲染 `Message` JSON 快照，并分别写入 + `subscribe.added.notification`、`subscribe.complete.notification` outbox intent。即时发送成功后按稳定 + 幂等键收口，崩溃或发送失败由现有 dispatcher 恢复；旧插件收到的事件字段和同步/异步入口保持不变。 + 第三方插件自行调用通知或自行写库的副作用仍不在宿主原子事务边界内。 - `DownloadAdded`、`TransferComplete`、`TransferFailed` 也已逐项接入，而不是复用一个不分业务语义的 “万能消息总线”。下载历史、下载文件清单或整理历史与各自 intent 在独占同步 Session/UoW 中原子提交； 即时广播失败时 intent 保持 pending，三种恢复 handler 均继续使用有限重试与 dead-letter 策略。 diff --git a/tests/test_outbox.py b/tests/test_outbox.py index 0e7a66812..a42e2e9a1 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -40,6 +40,34 @@ def test_subscription_and_outbox_intent_commit_together() -> None: assert report_intent.event_key.endswith(":report") +def test_subscription_notification_snapshot_is_part_of_same_transaction() -> None: + """订阅新增通知快照与事件、统计意图一起暂存，便于崩溃恢复。""" + calls = [] + repository = MagicMock() + repository.stage_add.side_effect = lambda *_args: calls.append("subscription") or _Staged() + outbox = MagicMock() + outbox.stage.side_effect = lambda *_args: calls.append("outbox") + unit_of_work = MagicMock() + unit_of_work.commit.side_effect = lambda: calls.append("commit") + command = CreateSubscriptionCommand(repository, unit_of_work, outbox=outbox) + + command.execute( + {}, + {"name": "demo"}, + "user", + notification={"title": "订阅成功", "text": "demo"}, + ) + + intents = [call.args[0] for call in outbox.stage.call_args_list] + assert [intent.topic for intent in intents] == [ + "subscribe.added", + "subscribe.added.notification", + "subscribe.added.report", + ] + assert intents[1].payload["message"]["text"] == "demo" + assert calls[-1] == "commit" + + def test_outbox_stage_failure_rolls_back_business_transaction() -> None: """intent 无法持久化时订阅行不得单独提交。""" repository = MagicMock() diff --git a/tests/test_subscription_completion_command.py b/tests/test_subscription_completion_command.py index 4d5df2cfb..3282b6c22 100644 --- a/tests/test_subscription_completion_command.py +++ b/tests/test_subscription_completion_command.py @@ -139,3 +139,28 @@ def test_completion_success_closes_event_then_report_intent(): ] assert calls[6][1]["idempotency_key"] == calls[2][1].event_key assert calls[8][1]["idempotency_key"] == calls[3][1].event_key + + +def test_completion_stages_and_closes_notification_snapshot() -> None: + """完成通知快照与业务事务同提交，成功即时投递后独立收口。""" + calls = [] + command, notify, report = _command(calls) + + command.execute( + 7, + {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2}, + {"title": "Test"}, + notify=notify, + report=report, + notification={"title": "完成", "text": "Test"}, + ) + + staged = [call[1] for call in calls if call[0] == "stage"] + assert [intent.topic for intent in staged] == [ + "subscribe.complete", + "subscribe.complete.notification", + "subscribe.complete.report", + ] + assert staged[1].payload["message"]["title"] == "完成" + completed = [call[1] for call in calls if call[0] == "complete"] + assert completed[0].endswith(":notification")