diff --git a/app/adapters/external/server.py b/app/adapters/external/server.py index 9e2293192..1eb2c4c65 100644 --- a/app/adapters/external/server.py +++ b/app/adapters/external/server.py @@ -10,6 +10,7 @@ from app.runtime.config import settings from app.domain.context import MediaInfo, MusicInfo from app.domain.meta.metabase import MetaBase from app.runtime.log import logger +from app.runtime.observability import observe_compat_facade from app.schemas.types import ( MUSIC_ENTITY_RECORDING, MediaType, @@ -38,6 +39,7 @@ def configure_server_application_services( _server_sharing_service = sharing_service +@observe_compat_facade("MoviePilotServerHelper") class MoviePilotServerHelper: """ MoviePilot 服务端请求辅助工具。 @@ -870,6 +872,20 @@ class MoviePilotServerHelper: res = await cls.async_subscribe_add(payload) return bool(res is not None and res.status_code == 200) + @classmethod + def sub_reg_durable(cls, sub: dict) -> bool: + """同步上报新增统计；明确禁用时视为无需投递。""" + if not settings.SUBSCRIBE_STATISTIC_SHARE: + return True + return cls.sub_reg(sub) + + @classmethod + async def async_sub_reg_durable(cls, sub: dict) -> bool: + """异步上报新增统计；明确禁用时视为无需投递。""" + if not settings.SUBSCRIBE_STATISTIC_SHARE: + return True + return await cls.async_sub_reg(sub) + @classmethod def sub_done(cls, sub: dict) -> bool: """ @@ -894,6 +910,20 @@ class MoviePilotServerHelper: res = await cls.async_subscribe_done(payload) return bool(res is not None and res.status_code == 200) + @classmethod + def sub_done_durable(cls, sub: dict) -> bool: + """同步上报完成统计；明确禁用时视为无需投递。""" + if not settings.SUBSCRIBE_STATISTIC_SHARE: + return True + return cls.sub_done(sub) + + @classmethod + async def async_sub_done_durable(cls, sub: dict) -> bool: + """异步上报完成统计；明确禁用时视为无需投递。""" + if not settings.SUBSCRIBE_STATISTIC_SHARE: + return True + return await cls.async_sub_done(sub) + @classmethod def sub_reg_async(cls, sub: dict) -> bool: """ diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index dc0357150..f3a05e5e0 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -62,7 +62,7 @@ def get_delete_subscribe_command( repository=cast(SubscribeDeletionRepository, repository_port), unit_of_work=cast(DeleteUnitOfWork, unit_of_work), publish_deleted=_publish_subscribe_deleted, - report_deleted=MoviePilotServerHelper.async_sub_done, + report_deleted=MoviePilotServerHelper.async_sub_done_durable, outbox=outbox, ) diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index 4a42ecd9d..04f49cf92 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -159,10 +159,13 @@ class CreateSubscriptionCommand: staged = self._repository.stage_add(identity, payload, username) if staged.created: if self._outbox: - self._outbox.stage( - _subscribe_added_intent(staged.subscribe_id, payload, username), - datetime.now(timezone.utc), - ) + now = datetime.now(timezone.utc) + for intent in _subscribe_added_intents( + staged.subscribe_id, + payload, + username, + ): + self._outbox.stage(intent, now) self._unit_of_work.commit() except Exception: self._unit_of_work.rollback() @@ -202,10 +205,13 @@ class AsyncCreateSubscriptionCommand: ) if staged.created: if self._outbox: - await self._outbox.stage( - _subscribe_added_intent(staged.subscribe_id, payload, username), - datetime.now(timezone.utc), - ) + now = datetime.now(timezone.utc) + for intent in _subscribe_added_intents( + staged.subscribe_id, + payload, + username, + ): + await self._outbox.stage(intent, now) await self._unit_of_work.commit() except Exception: await self._unit_of_work.rollback() @@ -215,20 +221,29 @@ class AsyncCreateSubscriptionCommand: return staged.subscribe_id, staged.message -def _subscribe_added_intent( +def _subscribe_added_intents( subscribe_id: int, payload: dict, username: str | None, -) -> OutboxIntent: - """构造版本化订阅新增事件，event key 同时作为 handler 幂等键。""" - return OutboxIntent( - event_key=subscription_added_event_key(subscribe_id, payload), - topic="subscribe.added", - payload={ - "subscribe_id": subscribe_id, - "username": username, - "mediainfo": dict(payload), - }, +) -> tuple[OutboxIntent, OutboxIntent]: + """构造订阅新增事件与外部统计的同事务 durable intents。""" + event_key = subscription_added_event_key(subscribe_id, payload) + event_payload = { + "subscribe_id": subscribe_id, + "username": username, + "mediainfo": dict(payload), + } + return ( + OutboxIntent( + event_key=event_key, + topic="subscribe.added", + payload=event_payload, + ), + OutboxIntent( + event_key=subscription_added_report_key(subscribe_id, payload), + topic="subscribe.added.report", + payload={"subscribe_info": dict(payload)}, + ), ) @@ -241,6 +256,11 @@ def subscription_added_event_key(subscribe_id: int, payload: dict) -> str: ) +def subscription_added_report_key(subscribe_id: int, payload: dict) -> str: + """返回与新增事件身份一致但可独立重试的统计幂等键。""" + return f"{subscription_added_event_key(subscribe_id, payload)}:report" + + _configured_subscribe_writer: Callable[[], SubscribeWriter] | None = None diff --git a/app/chain/subscribe.py b/app/chain/subscribe.py index 149ce93f4..711567dfd 100644 --- a/app/chain/subscribe.py +++ b/app/chain/subscribe.py @@ -958,9 +958,10 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): "username": context.username, "mediainfo": context.mediainfo.to_dict(), }) - MoviePilotServerHelper.sub_reg_async( + if not MoviePilotServerHelper.sub_reg_durable( self.__subscribe_report_payload(context) - ) + ): + raise RuntimeError("订阅新增统计上报未确认") async def __async_post_subscribe_added( self, @@ -993,9 +994,10 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase): "username": context.username, "mediainfo": context.mediainfo.to_dict(), }) - await MoviePilotServerHelper.async_sub_reg( + if not await MoviePilotServerHelper.async_sub_reg_durable( self.__subscribe_report_payload(context) - ) + ): + raise RuntimeError("订阅新增统计上报未确认") @staticmethod def __build_subscribe_create_context( diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index bb51fde56..f1947b192 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -237,11 +237,18 @@ def _build_outbox_dispatcher() -> OutboxDispatcher: """创建一次恢复批次独占的 Session、Repository 和事件 handler。""" def dispatch_subscribe_deleted_report(message) -> None: """重放订阅删除统计；未确认时抛错以进入有限重试。""" - if not MoviePilotServerHelper.sub_done( + if not MoviePilotServerHelper.sub_done_durable( message.payload.get("subscribe_info") or {} ): raise RuntimeError("订阅删除统计上报未确认") + def dispatch_subscribe_added_report(message) -> None: + """重放订阅新增统计；未确认时抛错以进入有限重试。""" + if not MoviePilotServerHelper.sub_reg_durable( + message.payload.get("subscribe_info") or {} + ): + raise RuntimeError("订阅新增统计上报未确认") + session = SessionFactory() return OutboxDispatcher( repository=SqlAlchemyOutboxRepository(session), @@ -250,6 +257,7 @@ def _build_outbox_dispatcher() -> OutboxDispatcher: EventType.SubscribeAdded, message.payload, ), + "subscribe.added.report": dispatch_subscribe_added_report, "subscribe.modified": lambda message: EventManager().send_event( EventType.SubscribeModified, message.payload, diff --git a/app/startup/subscription.py b/app/startup/subscription.py index bf7ac718f..5861892ba 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_report_key, ) from app.application.subscription.delete import ( DeleteSubscribeCommand, @@ -76,6 +77,10 @@ class TransactionalSubscribeWriter: subscription_added_event_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) finally: @@ -105,6 +110,10 @@ class TransactionalSubscribeWriter: subscription_added_event_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), + ) return await command.execute( identity, @@ -145,7 +154,7 @@ async def delete_subscribe_scope(): repository=SubscribeOper(session), unit_of_work=SqlAlchemyAsyncUnitOfWork(session), publish_deleted=_publish_deleted, - report_deleted=MoviePilotServerHelper.async_sub_done, + report_deleted=MoviePilotServerHelper.async_sub_done_durable, outbox=SqlAlchemyAsyncOutboxStager(session), ) diff --git a/tests/test_outbox.py b/tests/test_outbox.py index 2f7fab758..0e7a66812 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -31,10 +31,13 @@ def test_subscription_and_outbox_intent_commit_together() -> None: result = command.execute({}, {"name": "demo"}, "user") assert result == (42, "ok") - assert calls == ["subscription", "outbox", "commit"] - intent = outbox.stage.call_args.args[0] + assert calls == ["subscription", "outbox", "outbox", "commit"] + intent = outbox.stage.call_args_list[0].args[0] assert intent.event_key == "subscribe.added:42:unknown:unknown:v1" assert intent.payload["subscribe_id"] == 42 + report_intent = outbox.stage.call_args_list[1].args[0] + assert report_intent.topic == "subscribe.added.report" + assert report_intent.event_key.endswith(":report") def test_outbox_stage_failure_rolls_back_business_transaction() -> None: diff --git a/tests/test_server_helper.py b/tests/test_server_helper.py index 22bb10603..41ff73eb2 100644 --- a/tests/test_server_helper.py +++ b/tests/test_server_helper.py @@ -9,6 +9,7 @@ from app.adapters.external.server import ( ) from app.application.server.report import ServerReportService from app.application.server.share import ServerSharingService +from app.runtime.config import settings from app.schemas.types import MediaSource @@ -237,3 +238,19 @@ class MoviePilotServerHelperTests(unittest.TestCase): "tmdbid": 99, }) ) + + def test_durable_subscribe_report_treats_disabled_sharing_as_success(self): + """用户关闭统计分享时，durable intent 应视为无需远端投递。""" + with patch.object(settings, "SUBSCRIBE_STATISTIC_SHARE", False), patch.object( + MoviePilotServerHelper, + "sub_reg", + ) as reporter: + self.assertTrue(MoviePilotServerHelper.sub_reg_durable({"media_id": "1"})) + reporter.assert_not_called() + + with patch.object(settings, "SUBSCRIBE_STATISTIC_SHARE", False), patch.object( + MoviePilotServerHelper, + "sub_done", + ) as reporter: + self.assertTrue(MoviePilotServerHelper.sub_done_durable({"media_id": "1"})) + reporter.assert_not_called()