refactor: unify durable event topics

app/application/outbox.py

@@ -2,15 +2,43 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
+from types import MappingProxyType
 from typing import Any, Protocol, TypeVar
 
+from app.schemas.types import EventType
+
 
 T = TypeVar("T")
 
 
+SUBSCRIBE_ADDED_TOPIC = "subscribe.added"
+SUBSCRIBE_MODIFIED_TOPIC = "subscribe.modified"
+SUBSCRIBE_DELETED_TOPIC = "subscribe.deleted"
+DOWNLOAD_ADDED_TOPIC = "download.added"
+TRANSFER_COMPLETED_TOPIC = "transfer.completed"
+TRANSFER_FAILED_TOPIC = "transfer.failed"
+
+DURABLE_EVENT_TOPICS: Mapping[EventType, str] = MappingProxyType({
+    EventType.SubscribeAdded: SUBSCRIBE_ADDED_TOPIC,
+    EventType.SubscribeModified: SUBSCRIBE_MODIFIED_TOPIC,
+    EventType.SubscribeDeleted: SUBSCRIBE_DELETED_TOPIC,
+    EventType.DownloadAdded: DOWNLOAD_ADDED_TOPIC,
+    EventType.TransferComplete: TRANSFER_COMPLETED_TOPIC,
+    EventType.TransferFailed: TRANSFER_FAILED_TOPIC,
+})
+
+
+def durable_event_topic(event_type: EventType) -> str:
+    """返回 durable-required 事件唯一登记的 outbox topic。"""
+    try:
+        return DURABLE_EVENT_TOPICS[event_type]
+    except KeyError as error:
+        raise ValueError(f"事件 {event_type.name} 未登记 durable topic") from error
+
+
 @dataclass(frozen=True, slots=True)
 class OutboxIntent:
     """与业务事务一起暂存的版本化副作用意图。"""
@@ -33,6 +61,18 @@ class ClaimedOutboxMessage:
     attempt: int
 
 
+def validate_durable_event_handlers(
+    handlers: Mapping[str, Callable[[ClaimedOutboxMessage], None]],
+) -> None:
+    """拒绝缺少任一 durable-required 事件恢复 handler 的 dispatcher。"""
+    missing = set(DURABLE_EVENT_TOPICS.values()) - set(handlers)
+    if missing:
+        raise RuntimeError(
+            "Outbox dispatcher 缺少 durable 事件 handler: "
+            + ", ".join(sorted(missing))
+        )
+
+
 class OutboxRepository(Protocol):
     """outbox 写入、claim 和终态更新所需的最小端口。"""
 

app/application/subscription/delete.py

@@ -12,6 +12,7 @@ from app.application.outbox import (
     OutboxIntent,
     SyncOutboxTransaction,
     SyncUnitOfWork,
+    SUBSCRIBE_DELETED_TOPIC,
 )
 from app.schemas.event import SubscribeDeletedEventData
 
@@ -251,7 +252,7 @@ def _build_deletion_effects(
         report_payload=report_payload,
         event_intent=OutboxIntent(
             event_key=event_key,
-            topic="subscribe.deleted",
+            topic=SUBSCRIBE_DELETED_TOPIC,
             payload=event_payload,
         ),
         report_intent=OutboxIntent(

app/application/subscription/identity.py

@@ -3,7 +3,11 @@
 from datetime import datetime, timezone
 from typing import Any, Callable, Protocol
 
-from app.application.outbox import AsyncOutboxTransaction, OutboxIntent
+from app.application.outbox import (
+    AsyncOutboxTransaction,
+    OutboxIntent,
+    SUBSCRIBE_DELETED_TOPIC,
+)
 from app.application.subscription.delete import (
     AsyncUnitOfWork,
     SubscribeDeletedPublisher,
@@ -89,7 +93,7 @@ class DeleteSubscriptionsByIdentityCommand:
                     await self._outbox.stage(
                         OutboxIntent(
                             event_key=event_payload["idempotency_key"],
-                            topic="subscribe.deleted",
+                            topic=SUBSCRIBE_DELETED_TOPIC,
                             payload=event_payload,
                         ),
                         now,

app/application/subscription/mutation.py

@@ -7,7 +7,11 @@ from datetime import datetime, timezone
 from typing import Any, Protocol
 from uuid import uuid4
 
-from app.application.outbox import AsyncOutboxTransaction, OutboxIntent
+from app.application.outbox import (
+    AsyncOutboxTransaction,
+    OutboxIntent,
+    SUBSCRIBE_MODIFIED_TOPIC,
+)
 from app.schemas.event import SubscribeModifiedEventData
 
 
@@ -143,7 +147,7 @@ class SubscriptionMutationService:
             await self._outbox.stage(
                 OutboxIntent(
                     event_key=event_key,
-                    topic="subscribe.modified",
+                    topic=SUBSCRIBE_MODIFIED_TOPIC,
                     payload=event_payload,
                 ),
                 datetime.now(timezone.utc),

app/application/subscription/write.py

@@ -19,7 +19,7 @@ from collections.abc import Awaitable, Callable
 from datetime import datetime, timezone
 from typing import Mapping, Optional, Protocol, Tuple
 
-from app.application.outbox import OutboxIntent
+from app.application.outbox import OutboxIntent, SUBSCRIBE_ADDED_TOPIC
 from app.domain.context import MediaInfo, MusicInfo
 from app.schemas.media import resolve_media_identity
 from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType
@@ -243,7 +243,7 @@ def _subscribe_added_intents(
     intents: list[OutboxIntent] = [
         OutboxIntent(
             event_key=event_key,
-            topic="subscribe.added",
+            topic=SUBSCRIBE_ADDED_TOPIC,
             payload=event_payload,
         ),
     ]

app/chain/transfer.py

@@ -36,6 +36,7 @@ from app.application.history import (add_transfer_fail, add_transfer_success,
                                      clear_transfer_failures, describe_history_gate,
                                      evaluate_history_gate, is_skip_action,
                                      record_transfer_failure)
+from app.application.outbox import TRANSFER_COMPLETED_TOPIC, TRANSFER_FAILED_TOPIC
 from app.runtime.log import logger
 from app.schemas.event import StorageOperSelectionEventData
 from app.schemas.transfer import TransferInfo
@@ -456,7 +457,7 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo
                 if durable_transfer_failed:
                     event_payload = self._transfer_result_payload(task, transferinfo)
                     history = self.durable_event_writer.transfer_result(
-                        topic="transfer.failed",
+                        topic=TRANSFER_FAILED_TOPIC,
                         stage_history=lambda writer: add_transfer_fail(
                             fileitem=task.fileitem,
                             mode=transferinfo.transfer_type if transferinfo else "",
@@ -569,7 +570,7 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo
             if durable_transfer_complete:
                 event_payload = self._transfer_result_payload(task, transferinfo)
                 history = self.durable_event_writer.transfer_result(
-                    topic="transfer.completed",
+                    topic=TRANSFER_COMPLETED_TOPIC,
                     stage_history=lambda writer: add_transfer_success(
                         fileitem=task.fileitem,
                         mode=transferinfo.transfer_type if transferinfo else "",

app/db/adapters/chain.py

@@ -16,7 +16,11 @@ from app.application.chain.durable_events import (
     transfer_result_event_key,
 )
 from app.application.history import TransferHistoryRecord, TransferHistoryWriter
-from app.application.outbox import DurableEventCommand, OutboxIntent
+from app.application.outbox import (
+    DurableEventCommand,
+    DOWNLOAD_ADDED_TOPIC,
+    OutboxIntent,
+)
 from app.db.adapters.outbox import SqlAlchemyOutboxRepository
 from app.db.oper.downloadhistory import DownloadHistoryOper
 from app.db.oper.transferhistory import TransferHistoryOper
@@ -88,7 +92,7 @@ class TransactionalChainDurableEventWriter(ChainDurableEventWriter):
             command.execute(
                 intent=OutboxIntent(
                     event_key=event_key,
-                    topic="download.added",
+                    topic=DOWNLOAD_ADDED_TOPIC,
                     payload=snapshot_download_added(event_payload),
                 ),
                 stage_business=stage_business,

app/startup/initializers/modules.py

@@ -83,7 +83,12 @@ from app.application.security.userconfig import (
     configure_user_configuration,
 )
 from app.application.history import configure_transfer_history_provider
-from app.application.outbox import OutboxDispatcher, configure_outbox_dispatcher
+from app.application.outbox import (
+    OutboxDispatcher,
+    configure_outbox_dispatcher,
+    durable_event_topic,
+    validate_durable_event_handlers,
+)
 from app.db.adapters.outbox import SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxRepository
 from app.application.site.query import SiteQueryService, configure_site_query_service
 from app.application.site.health import SiteHealthService, configure_site_health_service
@@ -335,44 +340,58 @@ def _build_outbox_dispatcher() -> OutboxDispatcher:
             raise RuntimeError("订阅新增通知快照格式无效")
         CommandChain().post_message(Message.model_validate(snapshot))
 
+    handlers = {
+        durable_event_topic(
+            EventType.SubscribeAdded
+        ): lambda message: EventManager().send_event(
+            EventType.SubscribeAdded,
+            message.payload,
+        ),
+        "subscribe.added.report": dispatch_subscribe_added_report,
+        "subscribe.added.notification": dispatch_subscribe_added_notification,
+        durable_event_topic(
+            EventType.SubscribeModified
+        ): lambda message: EventManager().send_event(
+            EventType.SubscribeModified,
+            message.payload,
+        ),
+        durable_event_topic(
+            EventType.SubscribeDeleted
+        ): lambda message: EventManager().send_event(
+            EventType.SubscribeDeleted,
+            message.payload,
+        ),
+        "subscribe.deleted.report": dispatch_subscribe_deleted_report,
+        "subscribe.complete": lambda message: EventManager().send_event(
+            EventType.SubscribeComplete,
+            message.payload,
+        ),
+        "subscribe.complete.report": dispatch_subscribe_complete_report,
+        "subscribe.complete.notification": dispatch_subscribe_notification,
+        durable_event_topic(
+            EventType.DownloadAdded
+        ): lambda message: EventManager().send_event(
+            EventType.DownloadAdded,
+            restore_download_added(message.payload),
+        ),
+        durable_event_topic(
+            EventType.TransferComplete
+        ): lambda message: EventManager().send_event(
+            EventType.TransferComplete,
+            restore_transfer_result(message.payload),
+        ),
+        durable_event_topic(
+            EventType.TransferFailed
+        ): lambda message: EventManager().send_event(
+            EventType.TransferFailed,
+            restore_transfer_result(message.payload),
+        ),
+    }
+    validate_durable_event_handlers(handlers)
     session = SessionFactory()
     return OutboxDispatcher(
         repository=SqlAlchemyOutboxRepository(session),
-        handlers={
-            "subscribe.added": lambda message: EventManager().send_event(
-                EventType.SubscribeAdded,
-                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,
-            ),
-            "subscribe.deleted": lambda message: EventManager().send_event(
-                EventType.SubscribeDeleted,
-                message.payload,
-            ),
-            "subscribe.deleted.report": dispatch_subscribe_deleted_report,
-            "subscribe.complete": lambda message: EventManager().send_event(
-                EventType.SubscribeComplete,
-                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),
-            ),
-            "transfer.completed": lambda message: EventManager().send_event(
-                EventType.TransferComplete,
-                restore_transfer_result(message.payload),
-            ),
-            "transfer.failed": lambda message: EventManager().send_event(
-                EventType.TransferFailed,
-                restore_transfer_result(message.payload),
-            ),
-        },
+        handlers=handlers,
         close=session.close,
         failure_observer=lambda dead: record_metric(
             "scheduler.job.dead_letter" if dead else "scheduler.job.retry",