feat: persist subscription side-effect outbox

app/application/outbox.py

@@ -0,0 +1,127 @@
+"""持久副作用 outbox 的应用契约与有限重试 dispatcher。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, Protocol
+
+
+@dataclass(frozen=True, slots=True)
+class OutboxIntent:
+    """与业务事务一起暂存的版本化副作用意图。"""
+
+    event_key: str
+    topic: str
+    payload: dict[str, Any]
+    payload_version: int = 1
+
+
+@dataclass(frozen=True, slots=True)
+class ClaimedOutboxMessage:
+    """dispatcher 已获得 lease 的稳定消息投影。"""
+
+    message_id: int
+    event_key: str
+    topic: str
+    payload: dict[str, Any]
+    payload_version: int
+    attempt: int
+
+
+class OutboxRepository(Protocol):
+    """outbox 写入、claim 和终态更新所需的最小端口。"""
+
+    def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """在调用方当前事务中暂存意图，不自行提交。"""
+
+    def claim(self, now: datetime, lease_until: datetime) -> ClaimedOutboxMessage | None:
+        """原子认领一条到期消息。"""
+
+    def complete(self, message_id: int, completed_at: datetime) -> None:
+        """按消息 ID 标记完成。"""
+
+    def retry(
+        self,
+        message_id: int,
+        *,
+        next_retry_at: datetime,
+        last_error: str,
+        dead: bool,
+    ) -> None:
+        """记录有限退避或 dead-letter 终态。"""
+
+
+class OutboxDispatcher:
+    """认领并派发 outbox，按 event key 依赖 handler 幂等。"""
+
+    def __init__(
+        self,
+        repository: OutboxRepository,
+        handlers: dict[str, Callable[[ClaimedOutboxMessage], None]],
+        *,
+        max_attempts: int = 5,
+        lease_seconds: int = 60,
+        clock: Callable[[], datetime] | None = None,
+        close: Callable[[], None] | None = None,
+    ) -> None:
+        """注入持久端口、topic handler 和有界重试策略。"""
+        self._repository = repository
+        self._handlers = handlers
+        self._max_attempts = max_attempts
+        self._lease_seconds = lease_seconds
+        self._clock = clock or (lambda: datetime.now(timezone.utc))
+        self._close = close or (lambda: None)
+
+    def dispatch_one(self) -> bool:
+        """处理一条到期消息；无消息返回 False，handler 失败留待重试。"""
+        now = self._clock()
+        message = self._repository.claim(
+            now,
+            now + timedelta(seconds=self._lease_seconds),
+        )
+        if message is None:
+            return False
+        try:
+            handler = self._handlers[message.topic]
+            handler(message)
+        except Exception as error:
+            dead = message.attempt >= self._max_attempts
+            delay = min(3600, 2 ** max(0, message.attempt - 1))
+            self._repository.retry(
+                message.message_id,
+                next_retry_at=now + timedelta(seconds=delay),
+                last_error=str(error)[:4000],
+                dead=dead,
+            )
+            return True
+        self._repository.complete(message.message_id, now)
+        return True
+
+    def close(self) -> None:
+        """释放 dispatcher 工厂创建的短生命周期持久化资源。"""
+        self._close()
+
+
+_configured_dispatcher: Callable[[], OutboxDispatcher] | None = None
+
+
+def configure_outbox_dispatcher(provider: Callable[[], OutboxDispatcher]) -> None:
+    """由组合根登记短生命周期 dispatcher 工厂。"""
+    global _configured_dispatcher
+    _configured_dispatcher = provider
+
+
+def dispatch_pending_outbox(limit: int = 20) -> int:
+    """恢复有限数量到期 intent，供 Scheduler 与启动补偿复用。"""
+    if _configured_dispatcher is None:
+        raise RuntimeError("Outbox dispatcher 尚未配置")
+    dispatcher = _configured_dispatcher()
+    try:
+        processed = 0
+        while processed < limit and dispatcher.dispatch_one():
+            processed += 1
+        return processed
+    finally:
+        dispatcher.close()

app/application/subscription/write.py

@@ -16,8 +16,10 @@ app/application/history.py 里整理历史的写入路径同构。
 """
 
 from collections.abc import Awaitable, Callable
+from datetime import datetime, timezone
 from typing import Optional, Protocol, Tuple
 
+from app.application.outbox import OutboxIntent
 from app.domain.context import MediaInfo, MusicInfo
 from app.schemas.media import resolve_media_identity
 from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType
@@ -30,6 +32,20 @@ AfterCommitEffect = Callable[[int], None]
 AsyncAfterCommitEffect = Callable[[int], Awaitable[None]]
 
 
+class SubscriptionOutboxStager(Protocol):
+    """同步订阅事务暂存 durable intent 的最小端口。"""
+
+    def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """把意图加入当前业务事务。"""
+
+
+class AsyncSubscriptionOutboxStager(Protocol):
+    """异步订阅事务暂存 durable intent 的最小端口。"""
+
+    async def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """把意图加入当前异步业务事务。"""
+
+
 class SubscribeWriter(Protocol):
     """订阅写入应用服务使用的数据端口。"""
 
@@ -124,10 +140,12 @@ class CreateSubscriptionCommand:
         self,
         repository: SubscriptionStagingRepository,
         unit_of_work: UnitOfWork,
+        outbox: SubscriptionOutboxStager | None = None,
     ) -> None:
         """注入无提交仓储和事务所有者。"""
         self._repository = repository
         self._unit_of_work = unit_of_work
+        self._outbox = outbox
 
     def execute(
         self,
@@ -140,6 +158,11 @@ class CreateSubscriptionCommand:
         try:
             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),
+                    )
                 self._unit_of_work.commit()
         except Exception:
             self._unit_of_work.rollback()
@@ -156,10 +179,12 @@ class AsyncCreateSubscriptionCommand:
         self,
         repository: SubscriptionStagingRepository,
         unit_of_work: AsyncUnitOfWork,
+        outbox: AsyncSubscriptionOutboxStager | None = None,
     ) -> None:
         """注入无提交异步仓储和事务所有者。"""
         self._repository = repository
         self._unit_of_work = unit_of_work
+        self._outbox = outbox
 
     async def execute(
         self,
@@ -176,6 +201,11 @@ class AsyncCreateSubscriptionCommand:
                 username,
             )
             if staged.created:
+                if self._outbox:
+                    await self._outbox.stage(
+                        _subscribe_added_intent(staged.subscribe_id, payload, username),
+                        datetime.now(timezone.utc),
+                    )
                 await self._unit_of_work.commit()
         except Exception:
             await self._unit_of_work.rollback()
@@ -185,6 +215,32 @@ class AsyncCreateSubscriptionCommand:
         return staged.subscribe_id, staged.message
 
 
+def _subscribe_added_intent(
+    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),
+        },
+    )
+
+
+def subscription_added_event_key(subscribe_id: int, payload: dict) -> str:
+    """由订阅 ID 与媒体身份构造重试期间稳定、重建后不碰撞的幂等键。"""
+    return (
+        f"subscribe.added:{subscribe_id}:"
+        f"{payload.get('media_source') or 'unknown'}:"
+        f"{payload.get('media_id') or 'unknown'}:v1"
+    )
+
+
 _configured_subscribe_writer: Callable[[], SubscribeWriter] | None = None
 
 

app/chain/subscribe.py

@@ -924,6 +924,10 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             )
         eventmanager.send_event(EventType.SubscribeAdded, {
             "subscribe_id": subscribe_id,
+            "idempotency_key": (
+                f"subscribe.added:{subscribe_id}:"
+                f"{context.media_source}:{context.media_id}:v1"
+            ),
             "username": context.username,
             "mediainfo": context.mediainfo.to_dict(),
         })
@@ -955,6 +959,10 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
             )
         await eventmanager.async_send_event(EventType.SubscribeAdded, {
             "subscribe_id": subscribe_id,
+            "idempotency_key": (
+                f"subscribe.added:{subscribe_id}:"
+                f"{context.media_source}:{context.media_id}:v1"
+            ),
             "username": context.username,
             "mediainfo": context.mediainfo.to_dict(),
         })

app/db/models/__init__.py

@@ -15,6 +15,7 @@ _MODEL_EXPORTS = {
     "DownloadHistory": ("app.db.models.downloadhistory", "DownloadHistory"),
     "MediaServerItem": ("app.db.models.mediaserver", "MediaServerItem"),
     "Message": ("app.db.models.message", "Message"),
+    "OutboxMessage": ("app.db.models.outbox", "OutboxMessage"),
     "PassKey": ("app.db.models.passkey", "PassKey"),
     "PluginData": ("app.db.models.plugindata", "PluginData"),
     "Site": ("app.db.models.site", "Site"),

app/db/models/outbox.py

@@ -0,0 +1,30 @@
+"""持久副作用 outbox 模型。"""
+
+from typing import Any, Optional
+
+from sqlalchemy import Index, Integer, JSON, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base, get_id_column
+
+
+class OutboxMessage(Base):
+    """记录与业务事务原子提交、可认领重试的副作用意图。"""
+
+    id = get_id_column()
+    event_key: Mapped[str] = mapped_column(String(255), nullable=False)
+    topic: Mapped[str] = mapped_column(String(100), nullable=False)
+    payload_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    next_retry_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    lease_until: Mapped[Optional[str]] = mapped_column(String(40))
+    last_error: Mapped[Optional[str]] = mapped_column(Text)
+    created_at: Mapped[str] = mapped_column(String(40), nullable=False)
+    completed_at: Mapped[Optional[str]] = mapped_column(String(40))
+
+    __table_args__ = (
+        UniqueConstraint("event_key", name="uq_outboxmessage_event_key"),
+        Index("ix_outboxmessage_claim", "status", "next_retry_at", "lease_until"),
+    )

app/scheduler.py

@@ -29,6 +29,7 @@ from app.runtime.extensions.plugin_manager import PluginManager
 from app.db.oper.agenttask import AgentTaskOper
 from app.db.oper.systemconfig import SystemConfigOper
 from app.application.database import get_database_governance
+from app.application.outbox import dispatch_pending_outbox
 from app.application.image import WallpaperHelper
 from app.application.messaging.message import MessageHelper
 from app.runtime.progress import AsyncProgressHelper, ProgressHelper
@@ -354,6 +355,21 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass):
             )
 
             self._register_database_backup_job()
+            self._jobs["outbox_dispatch"] = {
+                "name": "恢复待投递副作用",
+                "func": dispatch_pending_outbox,
+                "running": False,
+            }
+            self._scheduler.add_job(
+                self.start,
+                "interval",
+                id="outbox_dispatch",
+                name="恢复待投递副作用",
+                seconds=30,
+                next_run_time=datetime.now(pytz.timezone(settings.TZ)),
+                kwargs={"job_id": "outbox_dispatch"},
+                replace_existing=True,
+            )
 
             # CookieCloud定时同步
             if (

app/startup/modules_initializer.py

@@ -54,6 +54,8 @@ 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.startup.outbox import SqlAlchemyOutboxRepository
 from app.application.site.query import SiteQueryService, configure_site_query_service
 from app.application.site.health import SiteHealthService, configure_site_health_service
 from app.application.workflow import WorkflowQueryService, configure_workflow_query
@@ -94,7 +96,7 @@ from app.db.oper.workflow import WorkflowOper
 from app.command import CommandChain
 from app.schemas.message import Message
 from app.schemas.message import MessageType
-from app.schemas.types import SystemConfigKey
+from app.schemas.types import EventType, SystemConfigKey
 from app.startup.agent_initializer import init_agent, stop_agent
 from app.startup.database import build_database_governance
 from app.startup.managed_resources_initializer import (
@@ -194,6 +196,21 @@ def configure_runtime_data_providers() -> None:
     )
 
 
+def _build_outbox_dispatcher() -> OutboxDispatcher:
+    """创建一次恢复批次独占的 Session、Repository 和事件 handler。"""
+    session = SessionFactory()
+    return OutboxDispatcher(
+        repository=SqlAlchemyOutboxRepository(session),
+        handlers={
+            "subscribe.added": lambda message: EventManager().send_event(
+                EventType.SubscribeAdded,
+                message.payload,
+            )
+        },
+        close=session.close,
+    )
+
+
 def configure_wallpaper_services() -> None:
     """把需要 Chain 编排的壁纸来源注入图片服务。"""
     configure_wallpaper_providers(
@@ -454,6 +471,7 @@ async def init_modules() -> HostRuntime:
         user=lambda: UserOper(),
     )
     configure_system_config(SystemConfigService(repository=SystemConfigOper()))
+    configure_outbox_dispatcher(_build_outbox_dispatcher)
     configure_transfer_retry_config(
         lambda: TransferRetryConfig(
             max_failed_retries=settings.TRANSFER_MAX_FAILED_RETRIES,

app/startup/outbox.py

@@ -0,0 +1,168 @@
+"""启动组合层使用的 SQLAlchemy outbox 持久化适配器。"""
+
+from datetime import datetime
+
+from sqlalchemy import or_, select, update
+from sqlalchemy.orm import Session
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.application.outbox import ClaimedOutboxMessage, OutboxIntent
+from app.db.base import execute_dml
+from app.db.models.outbox import OutboxMessage
+
+
+def _iso(value: datetime) -> str:
+    """将带时区时间统一序列化为可排序 ISO 字符串。"""
+    return value.isoformat()
+
+
+class SqlAlchemyOutboxRepository:
+    """使用调用方 Session 原子暂存并条件认领 outbox。"""
+
+    def __init__(self, session: Session) -> None:
+        """保存由调用方拥有的 SQLAlchemy Session。"""
+        self._session = session
+
+    def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """加入当前事务并 flush，使唯一键冲突在业务 commit 前暴露。"""
+        self._session.add(
+            OutboxMessage(
+                event_key=intent.event_key,
+                topic=intent.topic,
+                payload_version=intent.payload_version,
+                payload=intent.payload,
+                status="pending",
+                attempt=0,
+                next_retry_at=_iso(now),
+                created_at=_iso(now),
+            )
+        )
+        self._session.flush()
+
+    def claim(
+        self,
+        now: datetime,
+        lease_until: datetime,
+    ) -> ClaimedOutboxMessage | None:
+        """条件更新候选行；并发丢失竞争时返回 None。"""
+        now_text = _iso(now)
+        candidate = self._session.execute(
+            select(OutboxMessage)
+            .where(
+                OutboxMessage.status.in_(("pending", "processing")),
+                OutboxMessage.next_retry_at <= now_text,
+                or_(
+                    OutboxMessage.lease_until.is_(None),
+                    OutboxMessage.lease_until <= now_text,
+                ),
+            )
+            .order_by(OutboxMessage.id)
+            .limit(1)
+        ).scalars().first()
+        if candidate is None:
+            return None
+        next_attempt = candidate.attempt + 1
+        claimed = execute_dml(
+            self._session,
+            update(OutboxMessage)
+            .where(
+                OutboxMessage.id == candidate.id,
+                OutboxMessage.attempt == candidate.attempt,
+                or_(
+                    OutboxMessage.lease_until.is_(None),
+                    OutboxMessage.lease_until <= now_text,
+                ),
+            )
+            .values(
+                status="processing",
+                attempt=next_attempt,
+                lease_until=_iso(lease_until),
+            ),
+        )
+        self._session.commit()
+        if not claimed:
+            return None
+        return ClaimedOutboxMessage(
+            message_id=candidate.id,
+            event_key=candidate.event_key,
+            topic=candidate.topic,
+            payload=dict(candidate.payload),
+            payload_version=candidate.payload_version,
+            attempt=next_attempt,
+        )
+
+    def complete(self, message_id: int, completed_at: datetime) -> None:
+        """持久化完成终态并释放 lease。"""
+        self._session.execute(
+            update(OutboxMessage)
+            .where(OutboxMessage.id == message_id)
+            .values(status="completed", completed_at=_iso(completed_at), lease_until=None)
+        )
+        self._session.commit()
+
+    def complete_by_event_key(self, event_key: str, completed_at: datetime) -> None:
+        """即时 post-commit 全部成功时按幂等键收口对应 intent。"""
+        self._session.execute(
+            update(OutboxMessage)
+            .where(OutboxMessage.event_key == event_key)
+            .values(status="completed", completed_at=_iso(completed_at), lease_until=None)
+        )
+        self._session.commit()
+
+    def retry(
+        self,
+        message_id: int,
+        *,
+        next_retry_at: datetime,
+        last_error: str,
+        dead: bool,
+    ) -> None:
+        """持久化下一次退避或不可自动重试的 dead 终态。"""
+        self._session.execute(
+            update(OutboxMessage)
+            .where(OutboxMessage.id == message_id)
+            .values(
+                status="dead" if dead else "pending",
+                next_retry_at=_iso(next_retry_at),
+                lease_until=None,
+                last_error=last_error,
+            )
+        )
+        self._session.commit()
+
+
+class SqlAlchemyAsyncOutboxStager:
+    """只负责把 outbox 意图加入调用方异步事务。"""
+
+    def __init__(self, session: AsyncSession) -> None:
+        """保存由异步订阅命令拥有的 Session。"""
+        self._session = session
+
+    async def stage(self, intent: OutboxIntent, now: datetime) -> None:
+        """暂存并 flush，确保业务行与意图由同一次 commit 决定。"""
+        self._session.add(
+            OutboxMessage(
+                event_key=intent.event_key,
+                topic=intent.topic,
+                payload_version=intent.payload_version,
+                payload=intent.payload,
+                status="pending",
+                attempt=0,
+                next_retry_at=_iso(now),
+                created_at=_iso(now),
+            )
+        )
+        await self._session.flush()
+
+    async def complete_by_event_key(
+        self,
+        event_key: str,
+        completed_at: datetime,
+    ) -> None:
+        """异步 post-commit 全部成功时按幂等键收口 intent。"""
+        await self._session.execute(
+            update(OutboxMessage)
+            .where(OutboxMessage.event_key == event_key)
+            .values(status="completed", completed_at=_iso(completed_at), lease_until=None)
+        )
+        await self._session.commit()

app/startup/subscription.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Callable
 from contextlib import AbstractAsyncContextManager
+from datetime import datetime, timezone
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
@@ -11,9 +12,14 @@ from app.application.subscription.write import (
     AsyncAfterCommitEffect,
     AsyncCreateSubscriptionCommand,
     CreateSubscriptionCommand,
+    subscription_added_event_key,
 )
 from app.db.oper.subscribe import SubscribeOper
 from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
+from app.startup.outbox import (
+    SqlAlchemyAsyncOutboxStager,
+    SqlAlchemyOutboxRepository,
+)
 
 
 class TransactionalSubscribeWriter:
@@ -41,11 +47,23 @@ class TransactionalSubscribeWriter:
         """在独占同步会话内执行一次完整订阅新增事务。"""
         session = self._sync_session()
         try:
+            outbox = SqlAlchemyOutboxRepository(session)
             command = CreateSubscriptionCommand(
                 repository=SubscribeOper(session),
                 unit_of_work=SqlAlchemyUnitOfWork(session),
+                outbox=outbox,
             )
-            return command.execute(identity, payload, username, after_commit)
+
+            def delivered(subscribe_id: int) -> None:
+                """执行旧 post-commit 编排，全部成功后收口 durable intent。"""
+                if after_commit:
+                    after_commit(subscribe_id)
+                    outbox.complete_by_event_key(
+                        subscription_added_event_key(subscribe_id, payload),
+                        datetime.now(timezone.utc),
+                    )
+
+            return command.execute(identity, payload, username, delivered)
         finally:
             session.close()
 
@@ -58,13 +76,25 @@ class TransactionalSubscribeWriter:
     ) -> tuple[int, str]:
         """在独占异步会话作用域内执行一次完整订阅新增事务。"""
         async with self._async_session() as session:
+            outbox = SqlAlchemyAsyncOutboxStager(session)
             command = AsyncCreateSubscriptionCommand(
                 repository=SubscribeOper(session),
                 unit_of_work=SqlAlchemyAsyncUnitOfWork(session),
+                outbox=outbox,
             )
+
+            async def delivered(subscribe_id: int) -> None:
+                """异步执行旧编排，全部成功后收口 durable intent。"""
+                if after_commit:
+                    await after_commit(subscribe_id)
+                    await outbox.complete_by_event_key(
+                        subscription_added_event_key(subscribe_id, payload),
+                        datetime.now(timezone.utc),
+                    )
+
             return await command.execute(
                 identity,
                 payload,
                 username,
-                after_commit,
+                delivered,
             )