feat: make subscribe reports durable

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:
         """

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,
     )
 

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
 
 

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(

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,

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),
         )
 

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:

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()