diff --git a/app/agent/tools/impl/delete_subscribe.py b/app/agent/tools/impl/delete_subscribe.py index ca00ae5c3..9013819cc 100644 --- a/app/agent/tools/impl/delete_subscribe.py +++ b/app/agent/tools/impl/delete_subscribe.py @@ -6,11 +6,15 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.runtime.events import eventmanager -from app.application.agentdata import SubscribePort as SubscribeOper -from app.adapters.external.server import MoviePilotServerHelper +from app.application.subscription.delete import ( + SubscribeDeletionActor, + get_delete_subscribe_scope, +) +from app.application.subscription.mutation import ( + SubscriptionActor, + get_subscription_mutation_scope, +) from app.runtime.log import logger -from app.schemas.types import EventType class DeleteSubscribeInput(BaseModel): @@ -45,32 +49,21 @@ class DeleteSubscribeTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: subscribe_id={subscribe_id}") try: - subscribe_oper = SubscribeOper() - # 获取订阅信息 - subscribe = await subscribe_oper.async_get(subscribe_id) + async with get_subscription_mutation_scope() as mutation: + subscribe = await mutation.get_accessible( + subscribe_id, + SubscriptionActor(name="agent", is_superuser=True), + ) if not subscribe: return f"订阅 ID {subscribe_id} 不存在" - # 在删除之前获取订阅信息（用于事件） - subscribe_info = subscribe.to_dict() - - await subscribe_oper.async_delete(subscribe_id) - # 分享订阅统计刷新本身已异步化，这里只需要在删除后触发即可。 - MoviePilotServerHelper.sub_done_async( - { - "media_source": subscribe.media_source, - "media_id": subscribe.media_id, - "music_type": subscribe.music_type, - "total_tracks": subscribe.total_tracks, - "season": subscribe.season, - } - ) - - # 发送事件 - await eventmanager.async_send_event( - EventType.SubscribeDeleted, - {"subscribe_id": subscribe_id, "subscribe_info": subscribe_info}, - ) + async with get_delete_subscribe_scope() as command: + deleted = await command.execute( + subscribe_id, + SubscribeDeletionActor(username="agent", is_superuser=True), + ) + if not deleted: + return f"订阅 ID {subscribe_id} 不存在" return f"成功删除订阅：{subscribe.name} ({subscribe.year})" except Exception as e: diff --git a/app/agent/tools/impl/update_subscribe.py b/app/agent/tools/impl/update_subscribe.py index 9240b6adf..874943bee 100644 --- a/app/agent/tools/impl/update_subscribe.py +++ b/app/agent/tools/impl/update_subscribe.py @@ -7,11 +7,12 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.runtime.events import eventmanager -from app.application.agentdata import SubscribePort as SubscribeOper +from app.application.subscription.mutation import ( + SubscriptionActor, + get_subscription_mutation_scope, +) from app.runtime.log import logger -from app.schemas.event import SubscribeModifiedEventData -from app.schemas.types import EventType, media_type_to_agent +from app.schemas.types import media_type_to_agent class UpdateSubscribeInput(BaseModel): @@ -172,8 +173,9 @@ class UpdateSubscribeTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: subscribe_id={subscribe_id}") try: - subscribe_oper = SubscribeOper() - subscribe = await subscribe_oper.async_get(subscribe_id) + actor = SubscriptionActor(name="agent", is_superuser=True) + async with get_subscription_mutation_scope() as mutation: + subscribe = await mutation.get_accessible(subscribe_id, actor) if not subscribe: return json.dumps( {"success": False, "message": f"订阅不存在: {subscribe_id}"}, @@ -206,9 +208,6 @@ class UpdateSubscribeTool(MoviePilotTool): ensure_ascii=False, ) - # 保存旧数据用于事件 - old_subscribe_dict = subscribe.to_dict() - # 构建更新字典 subscribe_dict = {} @@ -306,24 +305,21 @@ class UpdateSubscribeTool(MoviePilotTool): ensure_ascii=False, ) - # 更新订阅 - await subscribe_oper.async_update(subscribe_id, subscribe_dict) - - # 重新获取更新后的订阅数据 - updated_subscribe = await subscribe_oper.async_get(subscribe_id) - - # 发送订阅调整事件 - await eventmanager.async_send_event( - EventType.SubscribeModified, - SubscribeModifiedEventData( - subscribe_id=subscribe_id, - old_subscribe_info=old_subscribe_dict, - subscribe_info=updated_subscribe.to_dict() - if updated_subscribe - else {}, + # Agent 工具没有 FastAPI 请求会话，由组合根提供一次独占事务作用域； + # 更新和 durable intent 必须共享同一 AsyncSession。 + async with get_subscription_mutation_scope() as mutation: + change = await mutation.update( + subscribe_id, + subscribe_dict, + actor, scene="agent_update", - ).to_dict(), - ) + ) + if not change: + return json.dumps( + {"success": False, "message": f"订阅不存在: {subscribe_id}"}, + ensure_ascii=False, + ) + updated_subscribe = change.new # 构建返回结果 result = { @@ -335,23 +331,23 @@ class UpdateSubscribeTool(MoviePilotTool): if updated_subscribe: result["subscribe"] = { - "id": updated_subscribe.id, - "name": updated_subscribe.name, - "year": updated_subscribe.year, - "type": media_type_to_agent(updated_subscribe.type), - "music_type": updated_subscribe.music_type, - "total_tracks": updated_subscribe.total_tracks, - "media_source": updated_subscribe.media_source, - "media_id": updated_subscribe.media_id, - "season": updated_subscribe.season, - "state": updated_subscribe.state, - "total_episode": updated_subscribe.total_episode, - "manual_total_episode": updated_subscribe.manual_total_episode, - "lack_episode": updated_subscribe.lack_episode, - "start_episode": updated_subscribe.start_episode, - "quality": updated_subscribe.quality, - "resolution": updated_subscribe.resolution, - "effect": updated_subscribe.effect, + "id": updated_subscribe.get("id"), + "name": updated_subscribe.get("name"), + "year": updated_subscribe.get("year"), + "type": media_type_to_agent(updated_subscribe.get("type")), + "music_type": updated_subscribe.get("music_type"), + "total_tracks": updated_subscribe.get("total_tracks"), + "media_source": updated_subscribe.get("media_source"), + "media_id": updated_subscribe.get("media_id"), + "season": updated_subscribe.get("season"), + "state": updated_subscribe.get("state"), + "total_episode": updated_subscribe.get("total_episode"), + "manual_total_episode": updated_subscribe.get("manual_total_episode"), + "lack_episode": updated_subscribe.get("lack_episode"), + "start_episode": updated_subscribe.get("start_episode"), + "quality": updated_subscribe.get("quality"), + "resolution": updated_subscribe.get("resolution"), + "effect": updated_subscribe.get("effect"), } return json.dumps(result, ensure_ascii=False, indent=2) diff --git a/app/api/context.py b/app/api/context.py index 168e17d7d..4c3ec7ba2 100644 --- a/app/api/context.py +++ b/app/api/context.py @@ -6,7 +6,18 @@ from typing import cast from fastapi import Depends, Request from app.application.messaging.chat import AsyncAgentChatRepository, AsyncUnitOfWork -from app.startup.context import AgentChatRuntime, HostRuntime +from app.application.outbox import AsyncOutboxTransaction +from app.application.subscription.delete import SubscribeDeletionRepository +from app.application.subscription.identity import SubscribeIdentityDeletionRepository +from app.application.subscription.mutation import ( + SubscriptionHistoryMutationRepository, + SubscriptionMutationRepository, +) +from app.startup.context import ( + AgentChatRuntime, + HostRuntime, + SubscriptionRuntime, +) def get_host_runtime(request: Request) -> HostRuntime: @@ -46,3 +57,54 @@ def get_agent_chat_transaction( ) -> AsyncUnitOfWork: """构造绑定当前请求会话的 Agent 会话事务端口。""" return cast(AsyncUnitOfWork, runtime.transaction(session)) + + +def get_subscription_runtime( + runtime: HostRuntime = Depends(get_host_runtime), +) -> SubscriptionRuntime: + """从完整宿主运行时收窄到订阅写事务能力。""" + return runtime.subscription + + +async def get_subscription_session( + runtime: SubscriptionRuntime = Depends(get_subscription_runtime), +) -> AsyncGenerator[object, None]: + """从订阅运行时生成请求独占的异步会话。""" + async for session in runtime.async_session(): + yield session + + +def get_subscription_repository( + session: object = Depends(get_subscription_session), + runtime: SubscriptionRuntime = Depends(get_subscription_runtime), +) -> ( + SubscriptionMutationRepository + | SubscribeDeletionRepository + | SubscribeIdentityDeletionRepository +): + """构造绑定当前请求会话的订阅仓储。""" + return runtime.repository(session) + + +def get_subscription_history_repository( + session: object = Depends(get_subscription_session), + runtime: SubscriptionRuntime = Depends(get_subscription_runtime), +) -> SubscriptionHistoryMutationRepository: + """构造绑定当前请求会话的订阅历史仓储。""" + return runtime.history_repository(session) + + +def get_subscription_transaction( + session: object = Depends(get_subscription_session), + runtime: SubscriptionRuntime = Depends(get_subscription_runtime), +) -> AsyncUnitOfWork: + """构造绑定当前订阅请求会话的异步事务端口。""" + return cast(AsyncUnitOfWork, runtime.transaction(session)) + + +def get_subscription_outbox( + session: object = Depends(get_subscription_session), + runtime: SubscriptionRuntime = Depends(get_subscription_runtime), +) -> AsyncOutboxTransaction: + """构造与订阅写入共享请求会话的 outbox 端口。""" + return runtime.outbox(session) diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index e7322fa14..e1fbc3662 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -1,17 +1,36 @@ """订阅领域的请求级 command/query 依赖。""" +from typing import Any, cast + from fastapi import BackgroundTasks, Depends from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session from app.adapters.external.server import MoviePilotServerHelper +from app.api.context import ( + get_subscription_history_repository, + get_subscription_outbox, + get_subscription_repository, + get_subscription_transaction, +) from app.api.data import get_async_db, get_db -from app.api.dependencies.data import repository, transaction +from app.api.dependencies.data import repository +from app.application.outbox import AsyncOutboxTransaction from app.application.scheduling import Scheduler from app.application.servarr import ServarrSubscriptionService -from app.application.subscription.delete import DeleteSubscribeCommand +from app.application.subscription.delete import ( + AsyncUnitOfWork as DeleteUnitOfWork, + DeleteSubscribeCommand, + SubscribeDeletionRepository, +) from app.application.subscription.identity import DeleteSubscriptionsByIdentityCommand -from app.application.subscription.mutation import SubscriptionMutationService +from app.application.subscription.identity import SubscribeIdentityDeletionRepository +from app.application.subscription.mutation import ( + AsyncUnitOfWork as MutationUnitOfWork, + SubscriptionHistoryMutationRepository, + SubscriptionMutationRepository, + SubscriptionMutationService, +) from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.search import SearchSubscriptionsCommand from app.runtime.events import eventmanager @@ -20,25 +39,29 @@ from app.schemas.types import EventType async def _publish_subscribe_deleted( - subscribe_id: int, - subscribe_info: dict, + payload: dict[str, Any], ) -> None: """通过宿主事件总线发布已提交的订阅删除事件。""" - await eventmanager.async_send_event( - EventType.SubscribeDeleted, - {"subscribe_id": subscribe_id, "subscribe_info": subscribe_info}, - ) + await eventmanager.async_send_event(EventType.SubscribeDeleted, payload) + + +async def _publish_subscribe_modified(payload: dict[str, Any]) -> None: + """通过宿主事件总线发布已提交的订阅修改事件。""" + await eventmanager.async_send_event(EventType.SubscribeModified, payload) def get_delete_subscribe_command( - db: AsyncSession = Depends(get_async_db), + repository_port: object = Depends(get_subscription_repository), + unit_of_work: object = Depends(get_subscription_transaction), + outbox: AsyncOutboxTransaction = Depends(get_subscription_outbox), ) -> DeleteSubscribeCommand: """组装请求级订阅删除用例及其具体适配器。""" return DeleteSubscribeCommand( - repository=repository("subscribe", db), - unit_of_work=transaction("async", db), + repository=cast(SubscribeDeletionRepository, repository_port), + unit_of_work=cast(DeleteUnitOfWork, unit_of_work), publish_deleted=_publish_subscribe_deleted, report_deleted=MoviePilotServerHelper.sub_done_async, + outbox=outbox, ) @@ -54,14 +77,17 @@ def _log_subscribe_deleted_event_error( def get_delete_subscriptions_by_identity_command( - db: AsyncSession = Depends(get_async_db), + repository_port: object = Depends(get_subscription_repository), + unit_of_work: object = Depends(get_subscription_transaction), + outbox: AsyncOutboxTransaction = Depends(get_subscription_outbox), ) -> DeleteSubscriptionsByIdentityCommand: """组装请求级按媒体身份删除订阅用例。""" return DeleteSubscriptionsByIdentityCommand( - repository=repository("subscribe", db), - unit_of_work=transaction("async", db), + repository=cast(SubscribeIdentityDeletionRepository, repository_port), + unit_of_work=cast(DeleteUnitOfWork, unit_of_work), publish_deleted=_publish_subscribe_deleted, handle_event_error=_log_subscribe_deleted_event_error, + outbox=outbox, ) @@ -98,12 +124,20 @@ def get_subscription_query_service( def get_subscription_mutation_service( - db: AsyncSession = Depends(get_async_db), + repository_port: object = Depends(get_subscription_repository), + history_repository: SubscriptionHistoryMutationRepository = Depends( + get_subscription_history_repository + ), + unit_of_work: object = Depends(get_subscription_transaction), + outbox: AsyncOutboxTransaction = Depends(get_subscription_outbox), ) -> SubscriptionMutationService: """组装异步订阅写服务。""" return SubscriptionMutationService( - repository=repository("subscribe", db), - history_repository=repository("subscribe_history", db), + repository=cast(SubscriptionMutationRepository, repository_port), + history_repository=history_repository, + unit_of_work=cast(MutationUnitOfWork, unit_of_work), + outbox=outbox, + publish_modified=_publish_subscribe_modified, ) diff --git a/app/api/endpoints/subscribe.py b/app/api/endpoints/subscribe.py index af28ca543..031e85949 100644 --- a/app/api/endpoints/subscribe.py +++ b/app/api/endpoints/subscribe.py @@ -14,8 +14,8 @@ from app.schemas.workflow import Subscribe as _SchemaSubscribe from app.api.response import ResponseAPIRouter from app.chain.subscribe import SubscribeChain from app.runtime.config import settings -from app.domain.context import MediaInfo from app.runtime.events import eventmanager +from app.domain.context import MediaInfo from app.domain.metainfo import MetaInfo from app.adapters.web.security.access import verify_token, verify_apitoken from app.application.subscription.delete import ( @@ -294,16 +294,16 @@ async def update_subscribe( ) if not change: return _SchemaResponse(success=False, message="订阅不存在") - # 发送订阅调整事件 - await eventmanager.async_send_event( - EventType.SubscribeModified, - SubscribeModifiedEventData( - subscribe_id=subscribe_in.id, - old_subscribe_info=change.old, - subscribe_info=change.new, - scene="update", - ).to_dict(), - ) + if not change.event_published: + await eventmanager.async_send_event( + EventType.SubscribeModified, + SubscribeModifiedEventData( + subscribe_id=subscribe_in.id, + old_subscribe_info=change.old, + subscribe_info=change.new, + scene="update", + ).to_dict(), + ) return _SchemaResponse(success=True) @@ -327,16 +327,16 @@ async def update_subscribe_status( change = await mutation.update_status(subid, state, actor) if not change: return _SchemaResponse(success=False, message="订阅不存在") - # 发送订阅调整事件 - await eventmanager.async_send_event( - EventType.SubscribeModified, - SubscribeModifiedEventData( - subscribe_id=subid, - old_subscribe_info=change.old, - subscribe_info=change.new, - scene="status", - ).to_dict(), - ) + if not change.event_published: + await eventmanager.async_send_event( + EventType.SubscribeModified, + SubscribeModifiedEventData( + subscribe_id=subid, + old_subscribe_info=change.old, + subscribe_info=change.new, + scene="status", + ).to_dict(), + ) return _SchemaResponse(success=True) @@ -388,15 +388,16 @@ async def reset_subscribes( ) change = await mutation.reset(subid, actor) if change: - await eventmanager.async_send_event( - EventType.SubscribeModified, - SubscribeModifiedEventData( - subscribe_id=subid, - old_subscribe_info=change.old, - subscribe_info=change.new, - scene="reset", - ).to_dict(), - ) + if not change.event_published: + await eventmanager.async_send_event( + EventType.SubscribeModified, + SubscribeModifiedEventData( + subscribe_id=subid, + old_subscribe_info=change.old, + subscribe_info=change.new, + scene="reset", + ).to_dict(), + ) return _SchemaResponse(success=True) return _SchemaResponse(success=False, message="订阅不存在") diff --git a/app/application/outbox.py b/app/application/outbox.py index c8903f2bb..aac4a3345 100644 --- a/app/application/outbox.py +++ b/app/application/outbox.py @@ -53,6 +53,20 @@ class OutboxRepository(Protocol): """记录有限退避或 dead-letter 终态。""" +class AsyncOutboxTransaction(Protocol): + """异步业务事务暂存并收口 durable intent 的最小端口。""" + + async def stage(self, intent: OutboxIntent, now: datetime) -> None: + """把 intent 加入调用方当前事务，但不自行提交。""" + + async def complete_by_event_key( + self, + event_key: str, + completed_at: datetime, + ) -> None: + """即时投递成功后按稳定幂等键标记 intent 完成。""" + + class OutboxDispatcher: """认领并派发 outbox，按 event key 依赖 handler 幂等。""" diff --git a/app/application/subscription/delete.py b/app/application/subscription/delete.py index 932ebb1e0..4ba3a59fc 100644 --- a/app/application/subscription/delete.py +++ b/app/application/subscription/delete.py @@ -1,7 +1,13 @@ """订阅删除应用用例及其依赖端口。""" +from contextlib import AbstractAsyncContextManager from dataclasses import dataclass -from typing import Awaitable, Callable, Mapping, Protocol +from datetime import datetime, timezone +from typing import Any, Awaitable, Callable, Mapping, Protocol, cast +from uuid import uuid4 + +from app.application.outbox import AsyncOutboxTransaction, OutboxIntent +from app.schemas.event import SubscribeDeletedEventData @dataclass(frozen=True) @@ -48,10 +54,7 @@ class AsyncUnitOfWork(Protocol): ... -SubscribeDeletedPublisher = Callable[ - [int, Mapping[str, object]], - Awaitable[None], -] +SubscribeDeletedPublisher = Callable[[dict[str, Any]], Awaitable[None]] SubscribeDeletedReporter = Callable[[Mapping[str, object]], object] @@ -64,12 +67,14 @@ class DeleteSubscribeCommand: unit_of_work: AsyncUnitOfWork, publish_deleted: SubscribeDeletedPublisher, report_deleted: SubscribeDeletedReporter, + outbox: AsyncOutboxTransaction | None = None, ) -> None: """注入数据访问、事务与提交后副作用端口。""" self._repository = repository self._unit_of_work = unit_of_work self._publish_deleted = publish_deleted self._report_deleted = report_deleted + self._outbox = outbox async def execute( self, @@ -85,23 +90,38 @@ class DeleteSubscribeCommand: candidate = await self._repository.get_candidate(subscribe_id) if not self._can_delete(candidate, actor): return False + assert candidate is not None await self._repository.stage_delete(subscribe_id) + event_payload = build_subscribe_deleted_payload( + subscribe_id, + candidate.event_payload, + ) + event_key = event_payload["idempotency_key"] try: + if self._outbox: + await self._outbox.stage( + OutboxIntent( + event_key=event_key, + topic="subscribe.deleted", + payload=event_payload, + ), + datetime.now(timezone.utc), + ) await self._unit_of_work.commit() except Exception: await self._unit_of_work.rollback() raise - event_payload = dict(candidate.event_payload) - await self._publish_deleted(subscribe_id, event_payload) - self._report_deleted( - { - "media_source": event_payload.get("media_source"), - "media_id": event_payload.get("media_id"), - "season": event_payload.get("season"), - } - ) + await self._publish_deleted(event_payload) + if self._outbox: + await self._outbox.complete_by_event_key( + event_key, + datetime.now(timezone.utc), + ) + # 上报适配器会自行白名单过滤公开字段；传完整删除前快照可保留音乐实体维度， + # 避免 Agent 与 API 入口收敛后丢失 music_type / total_tracks。 + self._report_deleted(dict(candidate.event_payload)) return True @staticmethod @@ -115,3 +135,36 @@ class DeleteSubscribeCommand: if actor.is_superuser: return True return bool(candidate.username) and candidate.username == actor.username + + +def build_subscribe_deleted_payload( + subscribe_id: int, + subscribe_info: Mapping[str, object], +) -> dict[str, Any]: + """构造兼容旧字段并携带幂等键的订阅删除事件快照。""" + event_key = f"subscribe.deleted:{subscribe_id}:{uuid4().hex}:v1" + return cast( + dict[str, Any], + SubscribeDeletedEventData( + subscribe_id=subscribe_id, + subscribe_info=dict(subscribe_info), + idempotency_key=event_key, + ).model_dump(mode="json"), + ) + + +DeleteSubscribeScope = Callable[[], AbstractAsyncContextManager[DeleteSubscribeCommand]] +_configured_delete_scope: DeleteSubscribeScope | None = None + + +def configure_delete_subscribe_scope(provider: DeleteSubscribeScope) -> None: + """由启动组合根登记非 HTTP 入口使用的订阅删除事务作用域。""" + global _configured_delete_scope + _configured_delete_scope = provider + + +def get_delete_subscribe_scope() -> AbstractAsyncContextManager[DeleteSubscribeCommand]: + """返回一次独占会话的订阅删除命令作用域。""" + if _configured_delete_scope is None: + raise RuntimeError("订阅删除事务作用域尚未配置") + return _configured_delete_scope() diff --git a/app/application/subscription/identity.py b/app/application/subscription/identity.py index 2f25f0c43..063d4ae0d 100644 --- a/app/application/subscription/identity.py +++ b/app/application/subscription/identity.py @@ -1,12 +1,15 @@ """按媒体身份批量删除订阅的应用用例。""" -from typing import Callable, Protocol +from datetime import datetime, timezone +from typing import Any, Callable, Protocol +from app.application.outbox import AsyncOutboxTransaction, OutboxIntent from app.application.subscription.delete import ( AsyncUnitOfWork, SubscribeDeletedPublisher, SubscribeDeletionActor, SubscribeDeletionCandidate, + build_subscribe_deleted_payload, ) from app.schemas.types import MediaSource @@ -24,8 +27,8 @@ class SubscribeIdentityDeletionRepository(Protocol): """读取匹配媒体身份的去重订阅快照。""" ... - async def delete(self, subscribe_id: int) -> None: - """把指定订阅登记为待删除。""" + async def stage_delete(self, subscribe_id: int) -> None: + """把指定订阅登记为待删除，但不自行提交事务。""" ... @@ -41,12 +44,14 @@ class DeleteSubscriptionsByIdentityCommand: unit_of_work: AsyncUnitOfWork, publish_deleted: SubscribeDeletedPublisher, handle_event_error: SubscribeDeletionEventErrorHandler, + outbox: AsyncOutboxTransaction | None = None, ) -> None: """注入数据访问、事务、事件和事件错误处理端口。""" self._repository = repository self._unit_of_work = unit_of_work self._publish_deleted = publish_deleted self._handle_event_error = handle_event_error + self._outbox = outbox async def execute( self, @@ -68,21 +73,40 @@ class DeleteSubscriptionsByIdentityCommand: for candidate in candidates if self._can_delete(candidate, actor) ] + events: list[tuple[SubscribeDeletionCandidate, dict[str, Any]]] = [] for candidate in deletions: await self._repository.stage_delete(candidate.subscribe_id) + event_payload = build_subscribe_deleted_payload( + candidate.subscribe_id, + candidate.event_payload, + ) + events.append((candidate, event_payload)) try: + if self._outbox: + now = datetime.now(timezone.utc) + for _, event_payload in events: + await self._outbox.stage( + OutboxIntent( + event_key=event_payload["idempotency_key"], + topic="subscribe.deleted", + payload=event_payload, + ), + now, + ) await self._unit_of_work.commit() except Exception: await self._unit_of_work.rollback() raise - for candidate in deletions: + for candidate, event_payload in events: try: - await self._publish_deleted( - candidate.subscribe_id, - dict(candidate.event_payload), - ) + await self._publish_deleted(event_payload) + if self._outbox: + await self._outbox.complete_by_event_key( + event_payload["idempotency_key"], + datetime.now(timezone.utc), + ) except Exception as error: self._handle_event_error(candidate.subscribe_id, error) return len(deletions) diff --git a/app/application/subscription/mutation.py b/app/application/subscription/mutation.py index 6f6f44357..3aa96ed17 100644 --- a/app/application/subscription/mutation.py +++ b/app/application/subscription/mutation.py @@ -1,7 +1,14 @@ """订阅写操作用例及其数据端口。""" +from collections.abc import Awaitable, Callable +from contextlib import AbstractAsyncContextManager from dataclasses import dataclass +from datetime import datetime, timezone from typing import Any, Protocol +from uuid import uuid4 + +from app.application.outbox import AsyncOutboxTransaction, OutboxIntent +from app.schemas.event import SubscribeModifiedEventData class SubscriptionMutationRepository(Protocol): @@ -13,6 +20,13 @@ class SubscriptionMutationRepository(Protocol): async def async_update(self, subscribe_id: int, payload: dict[str, Any]) -> Any | None: """更新订阅。""" + async def async_stage_update( + self, + subscribe_id: int, + payload: dict[str, Any], + ) -> Any | None: + """在调用方事务中暂存更新但不提交。""" + def get(self, subscribe_id: int) -> Any | None: """同步按 ID 获取订阅。""" @@ -27,6 +41,19 @@ class SubscriptionHistoryMutationRepository(Protocol): """删除订阅历史。""" +class AsyncUnitOfWork(Protocol): + """订阅修改用例使用的异步事务端口。""" + + async def commit(self) -> None: + """提交当前订阅修改事务。""" + + async def rollback(self) -> None: + """回滚当前订阅修改事务。""" + + +SubscribeModifiedPublisher = Callable[[dict[str, Any]], Awaitable[None]] + + @dataclass(frozen=True) class SubscriptionActor: """订阅写操作的权限主体。""" @@ -41,6 +68,7 @@ class SubscriptionMutation: old: dict[str, Any] new: dict[str, Any] + event_published: bool = False class SubscriptionMutationService: @@ -50,10 +78,16 @@ class SubscriptionMutationService: self, repository: SubscriptionMutationRepository, history_repository: SubscriptionHistoryMutationRepository | None = None, + unit_of_work: AsyncUnitOfWork | None = None, + outbox: AsyncOutboxTransaction | None = None, + publish_modified: SubscribeModifiedPublisher | None = None, ) -> None: - """注入订阅和订阅历史数据端口。""" + """注入订阅数据、事务与 durable 事件端口。""" self._repository = repository self._history_repository = history_repository + self._unit_of_work = unit_of_work + self._outbox = outbox + self._publish_modified = publish_modified async def get_accessible( self, @@ -79,16 +113,56 @@ class SubscriptionMutationService: payload: dict[str, Any], actor: SubscriptionActor, existing: Any | None = None, + scene: str = "update", ) -> SubscriptionMutation | None: - """更新当前主体可访问的订阅并返回前后快照。""" + """更新订阅，并在同一事务暂存可恢复的 SubscribeModified 事件。""" subscribe = existing or await self.get_accessible(subscribe_id, actor) if subscribe and not self.can_access(subscribe, actor): return None if not subscribe: return None old = subscribe.to_dict() - updated = await self._repository.async_update(subscribe_id, payload) - return SubscriptionMutation(old=old, new=updated.to_dict() if updated else {}) + if not self._unit_of_work: + updated = await self._repository.async_update(subscribe_id, payload) + return SubscriptionMutation(old=old, new=updated.to_dict() if updated else {}) + + if not self._outbox or not self._publish_modified: + raise RuntimeError("订阅修改事务缺少 outbox 或事件发布端口") + try: + updated = await self._repository.async_stage_update(subscribe_id, payload) + if not updated: + return None + event_payload = SubscribeModifiedEventData( + subscribe_id=subscribe_id, + old_subscribe_info=old, + subscribe_info=updated.to_dict(), + scene=scene, + ).to_dict() + event_key = _modified_event_key(subscribe_id, scene) + event_payload["idempotency_key"] = event_key + await self._outbox.stage( + OutboxIntent( + event_key=event_key, + topic="subscribe.modified", + payload=event_payload, + ), + datetime.now(timezone.utc), + ) + await self._unit_of_work.commit() + except Exception: + await self._unit_of_work.rollback() + raise + + await self._publish_modified(event_payload) + await self._outbox.complete_by_event_key( + event_key, + datetime.now(timezone.utc), + ) + return SubscriptionMutation( + old=old, + new=event_payload["subscribe_info"], + event_published=True, + ) async def update_status( self, @@ -97,7 +171,12 @@ class SubscriptionMutationService: actor: SubscriptionActor, ) -> SubscriptionMutation | None: """更新订阅状态并返回前后快照。""" - return await self.update(subscribe_id, {"state": state}, actor) + return await self.update( + subscribe_id, + {"state": state}, + actor, + scene="status", + ) async def reset( self, @@ -120,9 +199,13 @@ class SubscriptionMutationService: "manual_total_episode": 0, "state": "R", } - old = subscribe.to_dict() - updated = await self._repository.async_update(subscribe_id, payload) - return SubscriptionMutation(old=old, new=updated.to_dict() if updated else {}) + return await self.update( + subscribe_id, + payload, + actor, + existing=subscribe, + scene="reset", + ) async def delete_history( self, @@ -147,3 +230,30 @@ class SubscriptionMutationService: return True username = getattr(subscribe, "username", None) return bool(username) and username == actor.name + + +def _modified_event_key(subscribe_id: int, scene: str) -> str: + """为一次订阅修改生成重试期间稳定且跨多次相同变更不碰撞的幂等键。""" + return f"subscribe.modified:{subscribe_id}:{scene}:{uuid4().hex}:v1" + + +SubscriptionMutationScope = Callable[ + [], + AbstractAsyncContextManager[SubscriptionMutationService], +] +_configured_mutation_scope: SubscriptionMutationScope | None = None + + +def configure_subscription_mutation_scope( + provider: SubscriptionMutationScope, +) -> None: + """由启动组合根登记 Agent 等非 HTTP 入口使用的事务作用域。""" + global _configured_mutation_scope + _configured_mutation_scope = provider + + +def get_subscription_mutation_scope() -> AbstractAsyncContextManager[SubscriptionMutationService]: + """返回一次独占会话的订阅修改服务作用域。""" + if _configured_mutation_scope is None: + raise RuntimeError("订阅修改事务作用域尚未配置") + return _configured_mutation_scope() diff --git a/app/db/oper/subscribe.py b/app/db/oper/subscribe.py index 2419da654..02860a784 100644 --- a/app/db/oper/subscribe.py +++ b/app/db/oper/subscribe.py @@ -496,6 +496,22 @@ class SubscribeOper(DbOper): await subscribe.async_update(self._db, payload) return subscribe + async def async_stage_update( + self, + sid: int, + payload: dict, + ) -> Optional[Subscribe]: + """在调用方 AsyncSession 中暂存订阅更新并 flush，不提交事务。""" + if not isinstance(self._db, AsyncSession): + raise RuntimeError("异步订阅修改需要调用方提供 AsyncSession") + subscribe = await self.async_get(sid) + if not subscribe: + return None + for key, value in _normalize_integer_flags(payload).items(): + setattr(subscribe, key, value) + await self._db.flush() + return subscribe + async def async_update_filter_groups( self, sid: int, filter_groups: List[str] ) -> Optional[Subscribe]: diff --git a/app/runtime/event/contracts.py b/app/runtime/event/contracts.py index 521b3b512..22fd19062 100644 --- a/app/runtime/event/contracts.py +++ b/app/runtime/event/contracts.py @@ -54,6 +54,8 @@ class EventContract: _PAYLOAD_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = { EventType.ConfigChanged: event_schemas.ConfigChangeEventData, EventType.AgentTokensUsage: event_schemas.AgentTokensUsageEventData, + EventType.SubscribeAdded: event_schemas.SubscribeAddedEventData, + EventType.SubscribeDeleted: event_schemas.SubscribeDeletedEventData, EventType.SubscribeModified: event_schemas.SubscribeModifiedEventData, ChainEventType.PluginDataReset: event_schemas.PluginDataResetEventData, ChainEventType.AuthVerification: event_schemas.AuthCredentials, diff --git a/app/schemas/event.py b/app/schemas/event.py index 5d7da7311..08586a543 100644 --- a/app/schemas/event.py +++ b/app/schemas/event.py @@ -680,6 +680,7 @@ class SubscribeModifiedEventData(BaseEventData): subscribe_info: Dict[str, Any] = Field(default_factory=dict, description="更新后订阅快照") scene: str = Field(default="update", description="触发场景：update/status/reset/agent_update") fields: List[str] = Field(default_factory=list, description="真实变更字段") + idempotency_key: Optional[str] = Field(default=None, description="宿主生成的幂等键") @model_validator(mode="after") def compute_fields(self): @@ -700,13 +701,33 @@ class SubscribeModifiedEventData(BaseEventData): """ 输出公开事件 payload，避免内部属性被未来扩展意外暴露。 """ - return { + payload = { "subscribe_id": self.subscribe_id, "old_subscribe_info": self.old_subscribe_info, "subscribe_info": self.subscribe_info, "scene": self.scene, "fields": list(self.fields), } + if self.idempotency_key: + payload["idempotency_key"] = self.idempotency_key + return payload + + +class SubscribeAddedEventData(BaseEventData): + """SubscribeAdded 广播事件的可恢复公开 payload。""" + + subscribe_id: int = Field(description="订阅 ID") + username: Optional[str] = Field(default=None, description="发起订阅的用户") + mediainfo: Dict[str, Any] = Field(default_factory=dict, description="媒体信息快照") + idempotency_key: Optional[str] = Field(default=None, description="宿主生成的幂等键") + + +class SubscribeDeletedEventData(BaseEventData): + """SubscribeDeleted 广播事件的可恢复公开 payload。""" + + subscribe_id: int = Field(description="订阅 ID") + subscribe_info: Dict[str, Any] = Field(default_factory=dict, description="删除前订阅快照") + idempotency_key: Optional[str] = Field(default=None, description="宿主生成的幂等键") class SubscribeCompletionCheckEventData(ChainEventData): diff --git a/app/schemas/exports.py b/app/schemas/exports.py index 06eaad182..9556ba3cc 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -323,7 +323,9 @@ SCHEMA_EXPORTS = { 'StorageUsage': ('app.schemas.file', 'StorageUsage'), 'SubscrbieInfo': ('app.schemas.subscribe', 'SubscrbieInfo'), 'Subscribe': ('app.schemas.workflow', 'Subscribe'), + 'SubscribeAddedEventData': ('app.schemas.event', 'SubscribeAddedEventData'), 'SubscribeCompletionCheckEventData': ('app.schemas.event', 'SubscribeCompletionCheckEventData'), + 'SubscribeDeletedEventData': ('app.schemas.event', 'SubscribeDeletedEventData'), 'SubscribeDownloadFileInfo': ('app.schemas.subscribe', 'SubscribeDownloadFileInfo'), 'SubscribeEpisodeInfo': ('app.schemas.subscribe', 'SubscribeEpisodeInfo'), 'SubscribeEpisodesRefreshEventData': ('app.schemas.event', 'SubscribeEpisodesRefreshEventData'), diff --git a/app/startup/context.py b/app/startup/context.py index e516761ee..6eabbcde2 100644 --- a/app/startup/context.py +++ b/app/startup/context.py @@ -8,6 +8,13 @@ from app.application.messaging.chat import ( AsyncAgentChatRepository, AsyncUnitOfWork, ) +from app.application.outbox import AsyncOutboxTransaction +from app.application.subscription.delete import SubscribeDeletionRepository +from app.application.subscription.identity import SubscribeIdentityDeletionRepository +from app.application.subscription.mutation import ( + SubscriptionHistoryMutationRepository, + SubscriptionMutationRepository, +) class AgentChatRepositoryFactory(Protocol): @@ -26,6 +33,37 @@ class AsyncUnitOfWorkFactory(Protocol): ... +class AsyncOutboxFactory(Protocol): + """由请求会话构造异步 outbox 事务端口的工厂。""" + + def __call__(self, session: object) -> AsyncOutboxTransaction: + """绑定请求会话并返回 outbox 暂存与收口端口。""" + ... + + +class SubscriptionRepositoryFactory(Protocol): + """由请求会话构造订阅写仓储的工厂。""" + + def __call__( + self, + session: object, + ) -> ( + SubscriptionMutationRepository + | SubscribeDeletionRepository + | SubscribeIdentityDeletionRepository + ): + """绑定请求会话并返回订阅领域仓储。""" + ... + + +class SubscriptionHistoryRepositoryFactory(Protocol): + """由请求会话构造订阅历史写仓储的工厂。""" + + def __call__(self, session: object) -> SubscriptionHistoryMutationRepository: + """绑定请求会话并返回订阅历史仓储。""" + ... + + class AsyncSessionProvider(Protocol): """FastAPI 请求级异步会话提供器。""" @@ -70,9 +108,21 @@ class AgentChatRuntime: transaction: AsyncUnitOfWorkFactory +@dataclass(frozen=True, slots=True) +class SubscriptionRuntime: + """订阅 API 可见的请求级写事务运行时。""" + + async_session: AsyncSessionProvider + repository: SubscriptionRepositoryFactory + history_repository: SubscriptionHistoryRepositoryFactory + transaction: AsyncUnitOfWorkFactory + outbox: AsyncOutboxFactory + + @dataclass(frozen=True, slots=True) class HostRuntime: """宿主组合根构建且在一个 FastAPI lifespan 内共享的运行时对象。""" agent_chat: AgentChatRuntime + subscription: SubscriptionRuntime compatibility_api_data: CompatibilityApiData diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index 6aa3be881..fa0b00cae 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -55,7 +55,7 @@ from app.application.security.userconfig import ( ) 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.startup.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 from app.application.workflow import WorkflowQueryService, configure_workflow_query @@ -103,8 +103,11 @@ from app.startup.managed_resources_initializer import ( init_managed_resources, stop_managed_resources, ) -from app.startup.subscription import TransactionalSubscribeWriter -from app.startup.context import AgentChatRuntime, HostRuntime +from app.startup.subscription import ( + TransactionalSubscribeWriter, + configure_transactional_subscription_scopes, +) +from app.startup.context import AgentChatRuntime, HostRuntime, SubscriptionRuntime from app.adapters.web.security.access import set_superuser_token_payload_provider from app.application.security.auth import build_superuser_token_payload from app.application.image import configure_wallpaper_providers @@ -205,7 +208,15 @@ def _build_outbox_dispatcher() -> OutboxDispatcher: "subscribe.added": lambda message: EventManager().send_event( EventType.SubscribeAdded, message.payload, - ) + ), + "subscribe.modified": lambda message: EventManager().send_event( + EventType.SubscribeModified, + message.payload, + ), + "subscribe.deleted": lambda message: EventManager().send_event( + EventType.SubscribeDeleted, + message.payload, + ), }, close=session.close, ) @@ -455,6 +466,13 @@ async def init_modules() -> HostRuntime: repository=AgentChatOper, transaction=SqlAlchemyAsyncUnitOfWork, ), + subscription=SubscriptionRuntime( + async_session=get_async_db, + repository=SubscribeOper, + history_repository=SubscribeHistoryOper, + transaction=SqlAlchemyAsyncUnitOfWork, + outbox=SqlAlchemyAsyncOutboxStager, + ), compatibility_api_data=api_data, ) configure_api_data_runtime(host_runtime.compatibility_api_data) @@ -515,6 +533,7 @@ async def init_modules() -> HostRuntime: async_session=async_session_scope, ) ) + configure_transactional_subscription_scopes() # 托管资源只在这里装配声明与 adapter，具体资源仍由首个消费者显式激活。 init_managed_resources() # 应用服务不反向依赖 Chain，由启动组合层注入壁纸来源。 diff --git a/app/startup/subscription.py b/app/startup/subscription.py index 393497ac8..91611be7c 100644 --- a/app/startup/subscription.py +++ b/app/startup/subscription.py @@ -1,8 +1,9 @@ """订阅写入事务适配器的启动装配。""" from collections.abc import Callable -from contextlib import AbstractAsyncContextManager +from contextlib import AbstractAsyncContextManager, asynccontextmanager from datetime import datetime, timezone +from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session @@ -14,12 +15,25 @@ from app.application.subscription.write import ( CreateSubscriptionCommand, subscription_added_event_key, ) +from app.application.subscription.delete import ( + DeleteSubscribeCommand, + configure_delete_subscribe_scope, +) +from app.application.subscription.mutation import ( + SubscriptionMutationService, + configure_subscription_mutation_scope, +) +from app.adapters.external.server import MoviePilotServerHelper from app.db.oper.subscribe import SubscribeOper +from app.db.oper.subscribehistory import SubscribeHistoryOper +from app.db.session import async_session_scope from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork from app.startup.outbox import ( SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxRepository, ) +from app.runtime.events import EventManager +from app.schemas.types import EventType class TransactionalSubscribeWriter: @@ -98,3 +112,45 @@ class TransactionalSubscribeWriter: username, delivered, ) + + +async def _publish_modified(payload: dict[str, Any]) -> None: + """发布事务已提交的订阅修改事件。""" + await EventManager().async_send_event(EventType.SubscribeModified, payload) + + +async def _publish_deleted(payload: dict[str, Any]) -> None: + """发布事务已提交的订阅删除事件。""" + await EventManager().async_send_event(EventType.SubscribeDeleted, payload) + + +@asynccontextmanager +async def subscription_mutation_scope(): + """为非 HTTP 入口创建独占订阅修改会话、UoW 与 outbox。""" + async with async_session_scope() as session: + yield SubscriptionMutationService( + repository=SubscribeOper(session), + history_repository=SubscribeHistoryOper(session), + unit_of_work=SqlAlchemyAsyncUnitOfWork(session), + outbox=SqlAlchemyAsyncOutboxStager(session), + publish_modified=_publish_modified, + ) + + +@asynccontextmanager +async def delete_subscribe_scope(): + """为非 HTTP 入口创建独占订阅删除会话、UoW 与 outbox。""" + async with async_session_scope() as session: + yield DeleteSubscribeCommand( + repository=SubscribeOper(session), + unit_of_work=SqlAlchemyAsyncUnitOfWork(session), + publish_deleted=_publish_deleted, + report_deleted=MoviePilotServerHelper.sub_done_async, + outbox=SqlAlchemyAsyncOutboxStager(session), + ) + + +def configure_transactional_subscription_scopes() -> None: + """登记 Agent 等非 HTTP 入口复用的订阅事务作用域。""" + configure_subscription_mutation_scope(subscription_mutation_scope) + configure_delete_subscribe_scope(delete_subscribe_scope) diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md index 4f7c51c20..39114bbab 100644 --- a/docs/adr/0007-background-action-reliability.md +++ b/docs/adr/0007-background-action-reliability.md @@ -19,6 +19,10 @@ MoviePilot 保持模块化单体，不把所有后台动作迁到分布式队列 `durable-required` 是目标语义，不代表当前实现已经 durable。ARCH-251 前，Event Registry 中标记该值的 事件仍应在风险报告中说明崩溃窗口。 +截至 2026-08-22，宿主正式装配的 `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted` 广播已由 +业务事务内的 outbox intent 提供 at-least-once 恢复；payload 保持插件 dict ABI，并增加可选幂等键。 +这不覆盖第三方插件自行发送的裸事件，也不代表订阅通知和外部统计上报已经全部 durable。 + ## Event 映射 Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同语义分组列出每个事件，不省略事件名。 @@ -93,5 +97,6 @@ pending 状态交由下次启动，不以取消异常写成成功。 ## 验证与演进 - Event Registry 的 `delivery` 字段与本 ADR 同步进入 runtime baseline。 -- ARCH-251 只实现一个 E2 pilot，并通过 commit 后崩溃、重复 claim、并发 claim 和 dead-letter 测试。 +- ARCH-251 从一个 E2 pilot 扩展到三种订阅生命周期事件，并通过 commit 后崩溃、重复 claim、并发 claim + 和 dead-letter 测试；下载与整理结果事件仍需逐条迁移。 - ARCH-252 将 Scheduler 的定义、触发和执行状态拆分，但不提升不需要 durable 的 E0 信号。 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index da0936de0..748e3e311 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -729,6 +729,21 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas - 66 个订阅/调度专项测试和 40 个数据库、迁移、Session/outbox 测试通过（1 个环境条件 skip）； fresh schema 先 create_all 再升级与重复迁移均保持幂等。 +**扩展实施记录（2026-08-22）**： + +- 宿主自有的 `SubscribeModified`、`SubscribeDeleted` 生产路径已扩展到同一 outbox：订阅行更新/删除与 + version 1 intent 使用同一 `AsyncSession`、UoW 和 commit；即时广播失败时 intent 保持 pending，恢复 + dispatcher 分别按 `subscribe.modified`、`subscribe.deleted` topic 重放。 +- API、Agent 更新/删除工具以及按媒体身份批量删除均复用请求级或独占事务作用域。API 中保留的 + `event_published=False` 分支只服务测试替身和旧依赖注入，不是正式装配路径；正式 `HostRuntime` + 同时提供订阅 repository、history repository、transaction 与 outbox factory。 +- `SubscribeAddedEventData`、`SubscribeModifiedEventData`、`SubscribeDeletedEventData` 已进入 Event Contract； + 对插件仍投递原有 dict 字段，只新增可选 `idempotency_key`，不把 Pydantic 实例传给插件。 +- 保证边界只覆盖主仓可追踪的宿主生产者。运行时安装在 `app/plugins/**` 的第三方插件未被主仓改写； + 插件若自行直接发送同名事件，该发送仍由插件负责，无法与插件自己的数据库写入自动组成原子事务。 +- 订阅外部统计上报仍是 post-commit 副作用，不在事件 intent 的重放 handler 中；因此当前可以宣称三种 + 订阅事件具备宿主级 at-least-once 恢复，但不能宣称订阅通知和所有外部上报均已 durable。 + **禁止**：本阶段不引入 Celery、Kafka、RabbitMQ 等新基础设施。 #### ARCH-252：Scheduler 拆成声明、执行和状态 diff --git a/mypy.ini b/mypy.ini index dc64dabe2..a113c8241 100644 --- a/mypy.ini +++ b/mypy.ini @@ -15,5 +15,10 @@ files = app/runtime/observability/__init__.py, app/runtime/event/contracts.py, app/runtime/extensions/module/contracts.py, + app/application/outbox.py, + app/application/subscription/delete.py, + app/application/subscription/identity.py, + app/application/subscription/mutation.py, app/startup/context.py, - app/api/context.py + app/api/context.py, + app/api/dependencies/subscription.py diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 84b9a5d68..5c6555b00 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6244, - "edge_sha256": "f187bdbb5e88ce9a6b2ff559b10e5e6cd5a60d14693484ea17663cebf438cc92", + "edge_count": 6271, + "edge_sha256": "0119519add8e4499684044fc69f3003c54cfca20bf04f4614a419374692d5a94", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -715,20 +715,16 @@ "app.agent.tools.impl.delete_rule_group -> app.runtime.log", "app.agent.tools.impl.delete_rule_group -> app.schemas", "app.agent.tools.impl.delete_rule_group -> app.schemas.types", - "app.agent.tools.impl.delete_subscribe -> app.adapters", - "app.agent.tools.impl.delete_subscribe -> app.adapters.external", - "app.agent.tools.impl.delete_subscribe -> app.adapters.external.server", "app.agent.tools.impl.delete_subscribe -> app.agent", "app.agent.tools.impl.delete_subscribe -> app.agent.tools", "app.agent.tools.impl.delete_subscribe -> app.agent.tools.base", "app.agent.tools.impl.delete_subscribe -> app.agent.tools.tags", "app.agent.tools.impl.delete_subscribe -> app.application", - "app.agent.tools.impl.delete_subscribe -> app.application.agentdata", + "app.agent.tools.impl.delete_subscribe -> app.application.subscription", + "app.agent.tools.impl.delete_subscribe -> app.application.subscription.delete", + "app.agent.tools.impl.delete_subscribe -> app.application.subscription.mutation", "app.agent.tools.impl.delete_subscribe -> app.runtime", - "app.agent.tools.impl.delete_subscribe -> app.runtime.events", "app.agent.tools.impl.delete_subscribe -> app.runtime.log", - "app.agent.tools.impl.delete_subscribe -> app.schemas", - "app.agent.tools.impl.delete_subscribe -> app.schemas.types", "app.agent.tools.impl.delete_transfer_history -> app.agent", "app.agent.tools.impl.delete_transfer_history -> app.agent.tools", "app.agent.tools.impl.delete_transfer_history -> app.agent.tools.base", @@ -1440,12 +1436,11 @@ "app.agent.tools.impl.update_subscribe -> app.agent.tools.base", "app.agent.tools.impl.update_subscribe -> app.agent.tools.tags", "app.agent.tools.impl.update_subscribe -> app.application", - "app.agent.tools.impl.update_subscribe -> app.application.agentdata", + "app.agent.tools.impl.update_subscribe -> app.application.subscription", + "app.agent.tools.impl.update_subscribe -> app.application.subscription.mutation", "app.agent.tools.impl.update_subscribe -> app.runtime", - "app.agent.tools.impl.update_subscribe -> app.runtime.events", "app.agent.tools.impl.update_subscribe -> app.runtime.log", "app.agent.tools.impl.update_subscribe -> app.schemas", - "app.agent.tools.impl.update_subscribe -> app.schemas.event", "app.agent.tools.impl.update_subscribe -> app.schemas.types", "app.agent.tools.impl.update_system_settings -> app.agent", "app.agent.tools.impl.update_system_settings -> app.agent.tools", @@ -1486,6 +1481,11 @@ "app.api.context -> app.application", "app.api.context -> app.application.messaging", "app.api.context -> app.application.messaging.chat", + "app.api.context -> app.application.outbox", + "app.api.context -> app.application.subscription", + "app.api.context -> app.application.subscription.delete", + "app.api.context -> app.application.subscription.identity", + "app.api.context -> app.application.subscription.mutation", "app.api.context -> app.startup", "app.api.context -> app.startup.context", "app.api.dependencies.agent -> app.api", @@ -1562,10 +1562,12 @@ "app.api.dependencies.subscription -> app.adapters.external", "app.api.dependencies.subscription -> app.adapters.external.server", "app.api.dependencies.subscription -> app.api", + "app.api.dependencies.subscription -> app.api.context", "app.api.dependencies.subscription -> app.api.data", "app.api.dependencies.subscription -> app.api.dependencies", "app.api.dependencies.subscription -> app.api.dependencies.data", "app.api.dependencies.subscription -> app.application", + "app.api.dependencies.subscription -> app.application.outbox", "app.api.dependencies.subscription -> app.application.scheduling", "app.api.dependencies.subscription -> app.application.servarr", "app.api.dependencies.subscription -> app.application.subscription", @@ -2703,11 +2705,20 @@ "app.application.subscription.contract -> app.schemas", "app.application.subscription.contract -> app.schemas.media", "app.application.subscription.contract -> app.schemas.types", + "app.application.subscription.delete -> app.application", + "app.application.subscription.delete -> app.application.outbox", + "app.application.subscription.delete -> app.schemas", + "app.application.subscription.delete -> app.schemas.event", "app.application.subscription.identity -> app.application", + "app.application.subscription.identity -> app.application.outbox", "app.application.subscription.identity -> app.application.subscription", "app.application.subscription.identity -> app.application.subscription.delete", "app.application.subscription.identity -> app.schemas", "app.application.subscription.identity -> app.schemas.types", + "app.application.subscription.mutation -> app.application", + "app.application.subscription.mutation -> app.application.outbox", + "app.application.subscription.mutation -> app.schemas", + "app.application.subscription.mutation -> app.schemas.event", "app.application.subscription.query -> app.domain", "app.application.subscription.query -> app.domain.context", "app.application.subscription.query -> app.domain.meta", @@ -5836,6 +5847,11 @@ "app.startup.context -> app.application", "app.startup.context -> app.application.messaging", "app.startup.context -> app.application.messaging.chat", + "app.startup.context -> app.application.outbox", + "app.startup.context -> app.application.subscription", + "app.startup.context -> app.application.subscription.delete", + "app.startup.context -> app.application.subscription.identity", + "app.startup.context -> app.application.subscription.mutation", "app.startup.database -> app.adapters", "app.startup.database -> app.adapters.system", "app.startup.database -> app.adapters.system.backup", @@ -6073,13 +6089,24 @@ "app.startup.scheduler_initializer -> app.application", "app.startup.scheduler_initializer -> app.application.scheduling", "app.startup.scheduler_initializer -> app.scheduler", + "app.startup.subscription -> app.adapters", + "app.startup.subscription -> app.adapters.external", + "app.startup.subscription -> app.adapters.external.server", "app.startup.subscription -> app.application", "app.startup.subscription -> app.application.subscription", + "app.startup.subscription -> app.application.subscription.delete", + "app.startup.subscription -> app.application.subscription.mutation", "app.startup.subscription -> app.application.subscription.write", "app.startup.subscription -> app.db", "app.startup.subscription -> app.db.oper", "app.startup.subscription -> app.db.oper.subscribe", + "app.startup.subscription -> app.db.oper.subscribehistory", + "app.startup.subscription -> app.db.session", "app.startup.subscription -> app.db.uow", + "app.startup.subscription -> app.runtime", + "app.startup.subscription -> app.runtime.events", + "app.startup.subscription -> app.schemas", + "app.startup.subscription -> app.schemas.types", "app.startup.subscription -> app.startup", "app.startup.subscription -> app.startup.outbox", "app.startup.transfer_initializer -> app.chain", diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 5ad3cd184..776cce6c3 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1581,10 +1581,10 @@ "EventType.SubscribeAdded": { "delivery": "durable_required", "error_behavior": "notify", - "legacy_reason": "现有插件 payload 尚未收敛为稳定 model，保留原始 dict ABI", + "legacy_reason": null, "mode": "broadcast", "ordering": "priority_queue", - "payload_contract": "legacy_dict", + "payload_contract": "SubscribeAddedEventData", "sensitive_fields": [], "visibility": "plugin_public" }, @@ -1601,10 +1601,10 @@ "EventType.SubscribeDeleted": { "delivery": "durable_required", "error_behavior": "notify", - "legacy_reason": "现有插件 payload 尚未收敛为稳定 model，保留原始 dict ABI", + "legacy_reason": null, "mode": "broadcast", "ordering": "priority_queue", - "payload_contract": "legacy_dict", + "payload_contract": "SubscribeDeletedEventData", "sensitive_fields": [], "visibility": "plugin_public" }, @@ -1958,11 +1958,11 @@ }, "ChainEventType.WorkflowExecution": { "consumers": [], - "producers": [ - { - "caller": "app.workflow.actions.send_event", - "count": 1 - } + "producers": [ + { + "caller": "app.workflow.actions.send_event", + "count": 1 + } ] }, "EventType.AgentTokensUsage": { @@ -2200,11 +2200,15 @@ "consumers": [], "producers": [ { - "caller": "app.agent.tools.impl.delete_subscribe", + "caller": "app.api.dependencies.subscription", "count": 1 }, { - "caller": "app.api.dependencies.subscription", + "caller": "app.startup.modules_initializer", + "count": 1 + }, + { + "caller": "app.startup.subscription", "count": 1 } ] @@ -2213,12 +2217,20 @@ "consumers": [], "producers": [ { - "caller": "app.agent.tools.impl.update_subscribe", + "caller": "app.api.dependencies.subscription", "count": 1 }, { "caller": "app.api.endpoints.subscribe", "count": 3 + }, + { + "caller": "app.startup.modules_initializer", + "count": 1 + }, + { + "caller": "app.startup.subscription", + "count": 1 } ] }, @@ -2308,7 +2320,7 @@ ] } }, - "producer_count": 67 + "producer_count": 70 }, "module_method_specs": { "download_file": { diff --git a/tests/test_agent_delete_subscribe.py b/tests/test_agent_delete_subscribe.py new file mode 100644 index 000000000..805fea840 --- /dev/null +++ b/tests/test_agent_delete_subscribe.py @@ -0,0 +1,63 @@ +"""Agent 删除订阅工具的事务作用域委托测试。""" + +import asyncio +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from app.agent.tools.impl.delete_subscribe import DeleteSubscribeTool + + +@asynccontextmanager +async def _scope(value): + """把测试替身包装成工具使用的异步作用域。""" + yield value + + +def test_agent_delete_subscribe_uses_transactional_delete_command(): + """Agent 删除必须委托带 UoW/outbox 的应用命令，不能直接调用 Oper。""" + subscribe = SimpleNamespace(id=7, name="测试订阅", year="2026") + mutation = SimpleNamespace(get_accessible=AsyncMock(return_value=subscribe)) + command = SimpleNamespace(execute=AsyncMock(return_value=True)) + + with patch( + "app.agent.tools.impl.delete_subscribe.get_subscription_mutation_scope", + side_effect=lambda: _scope(mutation), + ), patch( + "app.agent.tools.impl.delete_subscribe.get_delete_subscribe_scope", + side_effect=lambda: _scope(command), + ): + result = asyncio.run( + DeleteSubscribeTool(session_id="session-1", user_id="10001").run( + subscribe_id=7 + ) + ) + + assert result == "成功删除订阅：测试订阅 (2026)" + mutation.get_accessible.assert_awaited_once() + command.execute.assert_awaited_once() + subscribe_id, actor = command.execute.await_args.args + assert subscribe_id == 7 + assert actor.is_superuser is True + + +def test_agent_delete_subscribe_skips_command_when_record_is_missing(): + """预读未命中时保持原有不存在提示，且不创建删除副作用。""" + mutation = SimpleNamespace(get_accessible=AsyncMock(return_value=None)) + delete_scope = AsyncMock() + + with patch( + "app.agent.tools.impl.delete_subscribe.get_subscription_mutation_scope", + side_effect=lambda: _scope(mutation), + ), patch( + "app.agent.tools.impl.delete_subscribe.get_delete_subscribe_scope", + delete_scope, + ): + result = asyncio.run( + DeleteSubscribeTool(session_id="session-1", user_id="10001").run( + subscribe_id=404 + ) + ) + + assert result == "订阅 ID 404 不存在" + delete_scope.assert_not_called() diff --git a/tests/test_agent_update_subscribe.py b/tests/test_agent_update_subscribe.py index 2d3ba8ea0..5c1a6f046 100644 --- a/tests/test_agent_update_subscribe.py +++ b/tests/test_agent_update_subscribe.py @@ -1,9 +1,11 @@ import asyncio import json -from unittest.mock import AsyncMock, patch +from contextlib import asynccontextmanager +from unittest.mock import patch from app.agent.tools.impl.update_subscribe import UpdateSubscribeTool -from app.schemas.types import EventType, MediaType +from app.application.subscription.mutation import SubscriptionMutation +from app.schemas.types import MediaType def test_agent_update_subscribe_sends_modified_event_payload_with_agent_scene(): @@ -13,13 +15,11 @@ def test_agent_update_subscribe_sends_modified_event_payload_with_agent_scene(): subscribe = _AgentSubscribe(id=9, name="旧标题", state="R", total_episode=8) oper = _SubscribeOperStub(subscribe) + mutation = _MutationServiceStub(oper) with patch( - "app.agent.tools.impl.update_subscribe.SubscribeOper", - return_value=oper, - ), patch( - "app.agent.tools.impl.update_subscribe.eventmanager.async_send_event", - new=AsyncMock(), - ) as send_event: + "app.agent.tools.impl.update_subscribe.get_subscription_mutation_scope", + side_effect=lambda: _mutation_scope(mutation), + ): result = asyncio.run( UpdateSubscribeTool(session_id="session-1", user_id="10001").run( subscribe_id=9, @@ -31,14 +31,7 @@ def test_agent_update_subscribe_sends_modified_event_payload_with_agent_scene(): payload = json.loads(result) assert payload["success"] is True assert oper.updates == [(9, {"name": "新标题", "state": "S"})] - send_event.assert_awaited_once() - event_type, event_payload = send_event.await_args.args - assert event_type == EventType.SubscribeModified - assert event_payload["subscribe_id"] == 9 - assert event_payload["scene"] == "agent_update" - assert event_payload["fields"] == ["name", "state"] - assert event_payload["old_subscribe_info"]["name"] == "旧标题" - assert event_payload["subscribe_info"]["name"] == "新标题" + assert mutation.calls == [(9, {"name": "新标题", "state": "S"}, "agent_update")] def test_agent_update_subscribe_ignores_unchanged_total_episode(): @@ -54,13 +47,11 @@ def test_agent_update_subscribe_ignores_unchanged_total_episode(): ) oper = _SubscribeOperStub(subscribe) + mutation = _MutationServiceStub(oper) with patch( - "app.agent.tools.impl.update_subscribe.SubscribeOper", - return_value=oper, - ), patch( - "app.agent.tools.impl.update_subscribe.eventmanager.async_send_event", - new=AsyncMock(), - ) as send_event: + "app.agent.tools.impl.update_subscribe.get_subscription_mutation_scope", + side_effect=lambda: _mutation_scope(mutation), + ): result = asyncio.run( UpdateSubscribeTool(session_id="session-1", user_id="10001").run( subscribe_id=160, @@ -71,7 +62,7 @@ def test_agent_update_subscribe_ignores_unchanged_total_episode(): payload = json.loads(result) assert payload == {"success": False, "message": "没有提供要更新的字段"} assert oper.updates == [] - send_event.assert_not_awaited() + assert mutation.calls == [] def test_agent_update_subscribe_only_updates_other_fields_with_unchanged_total_episode(): @@ -88,13 +79,11 @@ def test_agent_update_subscribe_only_updates_other_fields_with_unchanged_total_e ) oper = _SubscribeOperStub(subscribe) + mutation = _MutationServiceStub(oper) with patch( - "app.agent.tools.impl.update_subscribe.SubscribeOper", - return_value=oper, - ), patch( - "app.agent.tools.impl.update_subscribe.eventmanager.async_send_event", - new=AsyncMock(), - ) as send_event: + "app.agent.tools.impl.update_subscribe.get_subscription_mutation_scope", + side_effect=lambda: _mutation_scope(mutation), + ): result = asyncio.run( UpdateSubscribeTool(session_id="session-1", user_id="10001").run( subscribe_id=160, @@ -108,9 +97,7 @@ def test_agent_update_subscribe_only_updates_other_fields_with_unchanged_total_e assert payload["updated_fields"] == ["best_version"] assert payload["subscribe"]["manual_total_episode"] == 0 assert oper.updates == [(160, {"best_version": 1})] - send_event.assert_awaited_once() - _, event_payload = send_event.await_args.args - assert event_payload["fields"] == ["best_version"] + assert mutation.calls == [(160, {"best_version": 1}, "agent_update")] def test_agent_update_subscribe_marks_changed_total_episode_as_manual(): @@ -126,12 +113,10 @@ def test_agent_update_subscribe_marks_changed_total_episode_as_manual(): ) oper = _SubscribeOperStub(subscribe) + mutation = _MutationServiceStub(oper) with patch( - "app.agent.tools.impl.update_subscribe.SubscribeOper", - return_value=oper, - ), patch( - "app.agent.tools.impl.update_subscribe.eventmanager.async_send_event", - new=AsyncMock(), + "app.agent.tools.impl.update_subscribe.get_subscription_mutation_scope", + side_effect=lambda: _mutation_scope(mutation), ): result = asyncio.run( UpdateSubscribeTool(session_id="session-1", user_id="10001").run( @@ -186,3 +171,33 @@ class _SubscribeOperStub: self.updates.append((subscribe_id, dict(payload))) self.subscribe.__dict__.update(payload) return self.subscribe + + +class _MutationServiceStub: + """让 Agent 工具测试观察事务化修改服务收到的最终 payload。""" + + def __init__(self, oper): + """保存内存 Oper 与调用记录。""" + self.oper = oper + self.calls = [] + + async def get_accessible(self, subscribe_id, _actor): + """模拟事务作用域内的权限读取。""" + return await self.oper.async_get(subscribe_id) + + async def update(self, subscribe_id, payload, _actor, scene="update"): + """模拟事务化更新并返回稳定快照。""" + old = self.oper.subscribe.to_dict() + updated = await self.oper.async_update(subscribe_id, payload) + self.calls.append((subscribe_id, dict(payload), scene)) + return SubscriptionMutation( + old=old, + new=updated.to_dict(), + event_published=True, + ) + + +@asynccontextmanager +async def _mutation_scope(service): + """把测试修改服务包装成 Agent 使用的异步事务作用域。""" + yield service diff --git a/tests/test_host_runtime_context.py b/tests/test_host_runtime_context.py index 23b204fc4..8ec582e07 100644 --- a/tests/test_host_runtime_context.py +++ b/tests/test_host_runtime_context.py @@ -17,7 +17,7 @@ from app.api.data import ( get_api_data_ports, ) from app.startup import lifecycle -from app.startup.context import AgentChatRuntime, HostRuntime +from app.startup.context import AgentChatRuntime, HostRuntime, SubscriptionRuntime class _Repository: @@ -42,6 +42,20 @@ class _UnitOfWork: """模拟回滚。""" +class _Outbox: + """记录绑定会话的异步 outbox 替身。""" + + def __init__(self, session: object) -> None: + """保存与订阅仓储相同的请求会话。""" + self.session = session + + async def stage(self, intent, now) -> None: + """模拟暂存 durable intent。""" + + async def complete_by_event_key(self, event_key, completed_at) -> None: + """模拟收口 durable intent。""" + + def _runtime() -> HostRuntime: """构造不加载数据库引擎或 PluginManager 的假宿主运行时。""" async def async_session(): @@ -66,6 +80,13 @@ def _runtime() -> HostRuntime: repository=_Repository, transaction=_UnitOfWork, ), + subscription=SubscriptionRuntime( + async_session=async_session, + repository=_Repository, + history_repository=_Repository, + transaction=_UnitOfWork, + outbox=_Outbox, + ), compatibility_api_data=compatibility, ) diff --git a/tests/test_subscribe_delete_by_identity_command.py b/tests/test_subscribe_delete_by_identity_command.py index 7f71c5c0d..1db026b36 100644 --- a/tests/test_subscribe_delete_by_identity_command.py +++ b/tests/test_subscribe_delete_by_identity_command.py @@ -65,8 +65,9 @@ def _candidate(subscribe_id, username): def _command(candidates, calls, commit_error=None, failing_event_id=None): """构造带可观察事件错误处理的批量删除用例。""" - async def publish(subscribe_id, payload): + async def publish(payload): """记录事件并按订阅编号注入失败。""" + subscribe_id = payload["subscribe_id"] calls.append(("event", subscribe_id, payload)) if subscribe_id == failing_event_id: raise RuntimeError("event failed") diff --git a/tests/test_subscribe_delete_command.py b/tests/test_subscribe_delete_command.py index f91b40d56..faf1070fd 100644 --- a/tests/test_subscribe_delete_command.py +++ b/tests/test_subscribe_delete_command.py @@ -50,6 +50,25 @@ class _UnitOfWork: self.calls.append(("rollback",)) +class _Outbox: + """记录订阅删除 intent 暂存和收口顺序的 outbox 替身。""" + + def __init__(self, calls, stage_error=None): + """保存共享调用序列与可选暂存异常。""" + self.calls = calls + self.stage_error = stage_error + + async def stage(self, intent, _now): + """记录 intent，并按需模拟持久化失败。""" + self.calls.append(("outbox_stage", intent)) + if self.stage_error: + raise self.stage_error + + async def complete_by_event_key(self, event_key, _completed_at): + """记录即时事件成功后的 intent 收口。""" + self.calls.append(("outbox_complete", event_key)) + + def _candidate(username="alice"): """构造带完整事件身份字段的订阅删除候选。""" return SubscribeDeletionCandidate( @@ -66,11 +85,18 @@ def _candidate(username="alice"): ) -def _command(candidate, calls, commit_error=None, event_error=None, report_error=None): +def _command( + candidate, + calls, + commit_error=None, + event_error=None, + report_error=None, + outbox=None, +): """构造可观察事件与上报失败的订阅删除用例。""" - async def publish(subscribe_id, subscribe_info): + async def publish(payload): """记录删除事件并按需失败。""" - calls.append(("event", subscribe_id, subscribe_info)) + calls.append(("event", payload["subscribe_id"], payload)) if event_error: raise event_error @@ -85,6 +111,7 @@ def _command(candidate, calls, commit_error=None, event_error=None, report_error unit_of_work=_UnitOfWork(calls, commit_error), publish_deleted=publish, report_deleted=report, + outbox=outbox, ) @@ -101,12 +128,9 @@ async def test_owner_delete_commits_before_event_and_report(): assert deleted is True assert [call[0] for call in calls] == ["get", "delete", "commit", "event", "report"] - assert calls[3][2] == _candidate().event_payload - assert calls[4][1] == { - "media_source": "tmdb", - "media_id": "123", - "season": 2, - } + assert calls[3][2]["subscribe_info"] == _candidate().event_payload + assert calls[3][2]["idempotency_key"].startswith("subscribe.deleted:7:") + assert calls[4][1] == _candidate().event_payload @pytest.mark.asyncio @@ -185,6 +209,56 @@ async def test_report_failure_happens_after_commit_and_event(): assert [call[0] for call in calls] == ["get", "delete", "commit", "event", "report"] +@pytest.mark.asyncio +async def test_delete_stages_outbox_before_commit_and_completes_after_event(): + """订阅删除、intent 与即时事件必须按原子提交和成功收口顺序执行。""" + calls = [] + command = _command(_candidate(), calls, outbox=_Outbox(calls)) + + assert await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) is True + + assert [call[0] for call in calls] == [ + "get", + "delete", + "outbox_stage", + "commit", + "event", + "outbox_complete", + "report", + ] + intent = calls[2][1] + assert intent.topic == "subscribe.deleted" + assert intent.event_key == calls[4][2]["idempotency_key"] + assert calls[5][1] == intent.event_key + + +@pytest.mark.asyncio +async def test_delete_outbox_stage_failure_rolls_back_business_delete(): + """订阅删除 intent 无法暂存时不得提交业务删除。""" + calls = [] + command = _command( + _candidate(), + calls, + outbox=_Outbox(calls, stage_error=RuntimeError("outbox failed")), + ) + + with pytest.raises(RuntimeError, match="outbox failed"): + await command.execute( + 7, + SubscribeDeletionActor(username="alice", is_superuser=False), + ) + + assert [call[0] for call in calls] == [ + "get", + "delete", + "outbox_stage", + "rollback", + ] + + @pytest.mark.asyncio async def test_repository_candidate_uses_loaded_orm_snapshot(monkeypatch): """DB 适配器只向应用层暴露权限字段和完整列快照。""" diff --git a/tests/test_subscription_mutation_outbox.py b/tests/test_subscription_mutation_outbox.py new file mode 100644 index 000000000..80fe21c20 --- /dev/null +++ b/tests/test_subscription_mutation_outbox.py @@ -0,0 +1,185 @@ +"""订阅修改 UoW 与 durable outbox 边界测试。""" + +import pytest + +from app.application.subscription.mutation import ( + SubscriptionActor, + SubscriptionMutationService, +) + + +class _Subscribe: + """提供稳定前后快照的订阅替身。""" + + def __init__(self) -> None: + """初始化可修改字段与 owner。""" + self.id = 7 + self.username = "alice" + self.name = "旧标题" + + def to_dict(self) -> dict: + """返回当前订阅快照。""" + return {"id": self.id, "username": self.username, "name": self.name} + + +class _Repository: + """记录订阅读取、兼容更新和事务内暂存顺序。""" + + def __init__(self, subscribe: _Subscribe, calls: list) -> None: + """保存订阅对象与共享调用序列。""" + self.subscribe = subscribe + self.calls = calls + + async def async_get(self, subscribe_id: int): + """返回指定订阅。""" + self.calls.append(("get", subscribe_id)) + return self.subscribe + + async def async_update(self, subscribe_id: int, payload: dict): + """模拟旧兼容自动提交路径。""" + self.calls.append(("legacy_update", subscribe_id, payload)) + for key, value in payload.items(): + setattr(self.subscribe, key, value) + return self.subscribe + + async def async_stage_update(self, subscribe_id: int, payload: dict): + """模拟调用方事务内的更新暂存。""" + self.calls.append(("stage_update", subscribe_id, payload)) + for key, value in payload.items(): + setattr(self.subscribe, key, value) + return self.subscribe + + def get(self, subscribe_id: int): + """提供协议要求的同步读取。""" + return self.subscribe if subscribe_id == self.subscribe.id else None + + +class _UnitOfWork: + """记录订阅修改事务提交和回滚。""" + + def __init__(self, calls: list) -> None: + """保存共享调用序列。""" + self.calls = calls + + async def commit(self) -> None: + """记录提交。""" + self.calls.append(("commit",)) + + async def rollback(self) -> None: + """记录回滚。""" + self.calls.append(("rollback",)) + + +class _Outbox: + """记录修改事件 intent 暂存和完成。""" + + def __init__(self, calls: list, stage_error: Exception | None = None) -> None: + """保存共享调用序列与可选暂存异常。""" + self.calls = calls + self.stage_error = stage_error + + async def stage(self, intent, _now) -> None: + """记录 intent 并按需失败。""" + self.calls.append(("outbox_stage", intent)) + if self.stage_error: + raise self.stage_error + + async def complete_by_event_key(self, event_key: str, _completed_at) -> None: + """记录即时事件成功后的完成键。""" + self.calls.append(("outbox_complete", event_key)) + + +def _service(calls: list, *, event_error: Exception | None = None, outbox=None): + """构造拥有请求级 UoW 和 outbox 的订阅修改服务。""" + subscribe = _Subscribe() + + async def publish(payload: dict) -> None: + """记录公开事件并按需失败。""" + calls.append(("event", payload)) + if event_error: + raise event_error + + return SubscriptionMutationService( + repository=_Repository(subscribe, calls), + unit_of_work=_UnitOfWork(calls), + outbox=outbox or _Outbox(calls), + publish_modified=publish, + ) + + +@pytest.mark.asyncio +async def test_modified_event_is_staged_with_update_and_completed_after_publish(): + """订阅修改与 intent 同事务提交，事件成功后才标记完成。""" + calls = [] + service = _service(calls) + + change = await service.update( + 7, + {"name": "新标题"}, + SubscriptionActor(name="alice", is_superuser=False), + scene="update", + ) + + assert change is not None + assert change.event_published is True + assert change.old["name"] == "旧标题" + assert change.new["name"] == "新标题" + assert [call[0] for call in calls] == [ + "get", + "stage_update", + "outbox_stage", + "commit", + "event", + "outbox_complete", + ] + intent = calls[2][1] + assert intent.topic == "subscribe.modified" + assert intent.event_key.startswith("subscribe.modified:7:update:") + assert calls[4][1]["idempotency_key"] == intent.event_key + assert calls[5][1] == intent.event_key + + +@pytest.mark.asyncio +async def test_modified_outbox_stage_failure_rolls_back_update(): + """修改事件 intent 无法暂存时业务更新不得提交。""" + calls = [] + service = _service( + calls, + outbox=_Outbox(calls, stage_error=RuntimeError("outbox failed")), + ) + + with pytest.raises(RuntimeError, match="outbox failed"): + await service.update( + 7, + {"name": "新标题"}, + SubscriptionActor(name="alice", is_superuser=False), + ) + + assert [call[0] for call in calls] == [ + "get", + "stage_update", + "outbox_stage", + "rollback", + ] + + +@pytest.mark.asyncio +async def test_modified_event_failure_keeps_committed_intent_pending(): + """提交后的事件失败向调用方传播，且不得错误收口待恢复 intent。""" + calls = [] + service = _service(calls, event_error=RuntimeError("event failed")) + + with pytest.raises(RuntimeError, match="event failed"): + await service.update( + 7, + {"name": "新标题"}, + SubscriptionActor(name="alice", is_superuser=False), + ) + + assert [call[0] for call in calls] == [ + "get", + "stage_update", + "outbox_stage", + "commit", + "event", + ]