diff --git a/app/application/subscription/complete.py b/app/application/subscription/complete.py index 053ae31d2..c7e92e0f7 100644 --- a/app/application/subscription/complete.py +++ b/app/application/subscription/complete.py @@ -6,6 +6,7 @@ from collections.abc import Callable, Mapping from contextlib import AbstractContextManager from datetime import datetime, timezone from typing import Optional +from uuid import uuid4 from app.application.outbox import ( SUBSCRIBE_COMPLETED_TOPIC, @@ -55,15 +56,28 @@ class CompleteSubscriptionCommand: ) -> None: """在同一事务中写历史、删订阅并暂存完成事件、通知与统计意图。""" info = dict(subscribe_info) - event_key = completion_event_key(subscribe_id, info) + occurrence_id = uuid4().hex + event_key = completion_event_key( + subscribe_id, + info, + occurrence_id=occurrence_id, + ) event_payload: dict[str, JsonData] = { "subscribe_id": subscribe_id, "subscribe_info": info, "mediainfo": dict(mediainfo), "idempotency_key": event_key, } - report_key = completion_report_key(subscribe_id, info) - notification_key = completion_notification_key(subscribe_id, info) + report_key = completion_report_key( + subscribe_id, + info, + occurrence_id=occurrence_id, + ) + notification_key = completion_notification_key( + subscribe_id, + info, + occurrence_id=occurrence_id, + ) report_info = _completion_report_payload(info, report_key) report_payload: dict[str, JsonData] = {"subscribe_info": report_info} try: @@ -145,29 +159,46 @@ class CompleteSubscriptionCommand: def completion_event_key( subscribe_id: int, subscribe_info: Mapping[str, JsonData], + *, + occurrence_id: Optional[str] = None, ) -> str: - """构造跨重试稳定的订阅完成事件幂等键。""" + """由订阅身份和本次完成事实构造跨重试稳定且不复用的幂等键。""" + resolved_occurrence_id = occurrence_id or uuid4().hex return ( f"subscribe.complete:{subscribe_id}:" f"{subscribe_info.get('media_source') or 'unknown'}:" - f"{subscribe_info.get('media_id') or 'unknown'}:v1" + f"{subscribe_info.get('media_id') or 'unknown'}:{resolved_occurrence_id}:v1" ) def completion_report_key( subscribe_id: int, subscribe_info: Mapping[str, JsonData], + *, + occurrence_id: Optional[str] = None, ) -> str: """构造可独立重试的订阅完成统计幂等键。""" - return f"{completion_event_key(subscribe_id, subscribe_info)}:report" + event_key = completion_event_key( + subscribe_id, + subscribe_info, + occurrence_id=occurrence_id, + ) + return f"{event_key}:report" def completion_notification_key( subscribe_id: int, subscribe_info: Mapping[str, JsonData], + *, + occurrence_id: Optional[str] = None, ) -> str: """构造订阅完成通知的稳定幂等键，避免恢复时重复生成不同消息。""" - return f"{completion_event_key(subscribe_id, subscribe_info)}:notification" + event_key = completion_event_key( + subscribe_id, + subscribe_info, + occurrence_id=occurrence_id, + ) + return f"{event_key}:notification" def _completion_report_payload( diff --git a/app/application/subscription/contract.py b/app/application/subscription/contract.py index fc4eb75da..4ffb61ea9 100644 --- a/app/application/subscription/contract.py +++ b/app/application/subscription/contract.py @@ -6,6 +6,7 @@ import builtins from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, fields from typing import NoReturn, Optional, Protocol, cast +from uuid import uuid4 from app.domain.meta.metabase import MetaBase from app.domain.meta.metamusic import MetaMusic @@ -360,6 +361,51 @@ AfterCommitEffect = Callable[[int], bool | None] AsyncAfterCommitEffect = Callable[[int], Awaitable[bool | None]] +def subscription_added_event_key( + subscribe_id: int, + payload: Mapping[str, JsonData], + *, + occurrence_id: Optional[str] = None, +) -> str: + """由订阅 ID、媒体身份与本次创建事实构造可重试且不复用的幂等键。""" + resolved_occurrence_id = occurrence_id or uuid4().hex + return ( + f"subscribe.added:{subscribe_id}:" + f"{payload.get('media_source') or 'unknown'}:" + f"{payload.get('media_id') or 'unknown'}:{resolved_occurrence_id}:v1" + ) + + +def subscription_added_report_key( + subscribe_id: int, + payload: Mapping[str, JsonData], + *, + occurrence_id: Optional[str] = None, +) -> str: + """返回与新增事件身份一致但可独立重试的统计幂等键。""" + event_key = subscription_added_event_key( + subscribe_id, + payload, + occurrence_id=occurrence_id, + ) + return f"{event_key}:report" + + +def subscription_added_notification_key( + subscribe_id: int, + payload: Mapping[str, JsonData], + *, + occurrence_id: Optional[str] = None, +) -> str: + """构造订阅新增通知的稳定幂等键。""" + event_key = subscription_added_event_key( + subscribe_id, + payload, + occurrence_id=occurrence_id, + ) + return f"{event_key}:notification" + + class SubscriptionQueryPort(Protocol): """Chain、Workflow、Agent 和 Application 共用的订阅查询端口。""" @@ -483,8 +529,9 @@ class SubscriptionWritePort(Protocol): username: Optional[str] = None, after_commit: Optional[AfterCommitEffect] = None, notification: Optional[Mapping[str, JsonData]] = None, + occurrence_id: Optional[str] = None, ) -> tuple[int, str]: - """在独立同步事务中新增订阅。""" + """在独立同步事务中新增订阅；occurrence_id 标识本次创建事实。""" ... async def async_add( @@ -494,8 +541,9 @@ class SubscriptionWritePort(Protocol): username: Optional[str] = None, after_commit: Optional[AsyncAfterCommitEffect] = None, notification: Optional[Mapping[str, JsonData]] = None, + occurrence_id: Optional[str] = None, ) -> tuple[int, str]: - """在独立异步事务中新增订阅。""" + """在独立异步事务中新增订阅；occurrence_id 标识本次创建事实。""" ... class SubscriptionStagingPort(Protocol): diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index a6223e3e9..3e3067872 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -16,9 +16,10 @@ app/application/history.py 里整理历史的写入路径同构。 """ from collections.abc import Awaitable, Callable, Mapping, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional, Protocol, Tuple +from uuid import uuid4 from app.application.outbox import SUBSCRIBE_ADDED_TOPIC, OutboxIntent from app.application.subscription.contract import ( @@ -29,6 +30,9 @@ from app.application.subscription.contract import ( SubscriptionStagingPort, SubscriptionWritePort, SubscriptionWriteResult, + subscription_added_event_key, + subscription_added_notification_key, + subscription_added_report_key, ) from app.domain.context import MediaInfo, MusicInfo from app.schemas.common import JsonData @@ -99,6 +103,7 @@ class CreateSubscriptionCommand: username: Optional[str] = None, after_commit: Optional[AfterCommitEffect] = None, notification: Mapping[str, JsonData] | None = None, + occurrence_id: Optional[str] = None, ) -> Tuple[int, str]: """执行同步新增；事务失败回滚，提交后副作用失败不反向回滚。""" try: @@ -111,6 +116,7 @@ class CreateSubscriptionCommand: payload.to_payload(), username, notification, + occurrence_id=occurrence_id, ): self._outbox.stage(intent, now) self._unit_of_work.commit() @@ -143,6 +149,7 @@ class AsyncCreateSubscriptionCommand: username: Optional[str] = None, after_commit: Optional[AsyncAfterCommitEffect] = None, notification: Mapping[str, JsonData] | None = None, + occurrence_id: Optional[str] = None, ) -> Tuple[int, str]: """执行异步新增；事务失败回滚，提交后副作用失败不反向回滚。""" try: @@ -159,6 +166,7 @@ class AsyncCreateSubscriptionCommand: payload.to_payload(), username, notification, + occurrence_id=occurrence_id, ): await self._outbox.stage(intent, now) await self._unit_of_work.commit() @@ -179,6 +187,7 @@ class SubscriptionCreateRequest: username: Optional[str] = None notification: Mapping[str, JsonData] | None = None after_commit: Optional[AsyncAfterCommitEffect] = None + occurrence_id: str = field(default_factory=lambda: uuid4().hex) SubscriptionBatchAfterCommitEffect = Callable[ @@ -245,6 +254,7 @@ class AsyncCreateSubscriptionBatchCommand: request.payload.to_payload(), request.username, request.notification, + occurrence_id=request.occurrence_id, ): await self._outbox.stage(intent, now) if created_requests: @@ -264,9 +274,16 @@ def _subscribe_added_intents( payload: Mapping[str, JsonData], username: str | None, notification: Mapping[str, JsonData] | None = None, + *, + occurrence_id: Optional[str] = None, ) -> tuple[OutboxIntent, ...]: """构造订阅新增事件、通知与外部统计的同事务 durable intents。""" - event_key = subscription_added_event_key(subscribe_id, payload) + resolved_occurrence_id = occurrence_id or uuid4().hex + event_key = subscription_added_event_key( + subscribe_id, + payload, + occurrence_id=resolved_occurrence_id, + ) event_payload = { "subscribe_id": subscribe_id, "idempotency_key": event_key, @@ -283,12 +300,17 @@ def _subscribe_added_intents( if notification: intents.append( OutboxIntent( - event_key=subscription_added_notification_key(subscribe_id, payload), + event_key=subscription_added_notification_key( + subscribe_id, + payload, + occurrence_id=resolved_occurrence_id, + ), topic="subscribe.added.notification", payload={ "idempotency_key": subscription_added_notification_key( subscribe_id, payload, + occurrence_id=resolved_occurrence_id, ), "message": dict(notification), }, @@ -296,7 +318,11 @@ def _subscribe_added_intents( ) intents.append( OutboxIntent( - event_key=subscription_added_report_key(subscribe_id, payload), + event_key=subscription_added_report_key( + subscribe_id, + payload, + occurrence_id=resolved_occurrence_id, + ), topic="subscribe.added.report", payload={ "subscribe_info": { @@ -304,6 +330,7 @@ def _subscribe_added_intents( "idempotency_key": subscription_added_report_key( subscribe_id, payload, + occurrence_id=resolved_occurrence_id, ), } }, @@ -312,34 +339,6 @@ def _subscribe_added_intents( return tuple(intents) -def subscription_added_event_key( - subscribe_id: int, - payload: Mapping[str, JsonData], -) -> str: - """由订阅 ID 与媒体身份构造重试期间稳定、重建后不碰撞的幂等键。""" - return ( - f"subscribe.added:{subscribe_id}:" - f"{payload.get('media_source') or 'unknown'}:" - f"{payload.get('media_id') or 'unknown'}:v1" - ) - - -def subscription_added_report_key( - subscribe_id: int, - payload: Mapping[str, JsonData], -) -> str: - """返回与新增事件身份一致但可独立重试的统计幂等键。""" - return f"{subscription_added_event_key(subscribe_id, payload)}:report" - - -def subscription_added_notification_key( - subscribe_id: int, - payload: Mapping[str, JsonData], -) -> str: - """构造订阅新增通知的稳定幂等键。""" - return f"{subscription_added_event_key(subscribe_id, payload)}:notification" - - def _music_entity(mediainfo: MediaInfo | MusicInfo) -> Optional[str]: """ 取音乐实体类型；非音乐媒体一律为空。 @@ -407,6 +406,7 @@ def build_subscription_create_request( mediainfo: MediaInfo | MusicInfo, notification: Mapping[str, JsonData] | None = None, after_commit: Optional[AsyncAfterCommitEffect] = None, + occurrence_id: Optional[str] = None, **kwargs: JsonData, ) -> Optional[SubscriptionCreateRequest]: """把统一订阅字段映射冻结为一条可参与原子批量写入的请求。""" @@ -420,6 +420,7 @@ def build_subscription_create_request( username=username, notification=notification, after_commit=after_commit, + occurrence_id=occurrence_id or uuid4().hex, ) @@ -428,6 +429,7 @@ def add_subscribe( subscribe_oper: SubscriptionWritePort, after_commit: Optional[AfterCommitEffect] = None, notification: Mapping[str, JsonData] | None = None, + occurrence_id: Optional[str] = None, **kwargs: JsonData, ) -> Tuple[int, str]: """ @@ -436,6 +438,7 @@ def add_subscribe( :param mediainfo: 识别结果 :param subscribe_oper: 调用方显式注入的订阅写入端口 :param after_commit: 提交后副作用编排；返回 False 表示统计 intent 等待重试 + :param occurrence_id: 本次订阅创建事实的唯一标识，用于避免数据库主键复用造成键冲突 :param kwargs: 订阅设置；owner_scope 为真时按用户名限定查重范围 :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增 """ @@ -443,11 +446,12 @@ def add_subscribe( if translated is None: return INCOMPLETE_IDENTITY identity, payload, username = translated - if after_commit is None: + if occurrence_id is None: return subscribe_oper.add( identity=identity, payload=payload, username=username, + after_commit=after_commit, notification=notification, ) return subscribe_oper.add( @@ -456,6 +460,7 @@ def add_subscribe( username=username, after_commit=after_commit, notification=notification, + occurrence_id=occurrence_id, ) @@ -464,6 +469,7 @@ async def async_add_subscribe( subscribe_oper: SubscriptionWritePort, after_commit: Optional[AsyncAfterCommitEffect] = None, notification: Mapping[str, JsonData] | None = None, + occurrence_id: Optional[str] = None, **kwargs: JsonData, ) -> Tuple[int, str]: """ @@ -472,6 +478,7 @@ async def async_add_subscribe( :param mediainfo: 识别结果 :param subscribe_oper: 调用方显式注入的订阅写入端口 :param after_commit: 异步提交后副作用编排；返回 False 表示统计 intent 等待重试 + :param occurrence_id: 本次订阅创建事实的唯一标识，用于避免数据库主键复用造成键冲突 :param kwargs: 订阅设置；owner_scope 为真时按用户名限定查重范围 :return: (订阅 ID, 结果说明)；ID 为 0 表示未新增 """ @@ -479,11 +486,12 @@ async def async_add_subscribe( if translated is None: return INCOMPLETE_IDENTITY identity, payload, username = translated - if after_commit is None: + if occurrence_id is None: return await subscribe_oper.async_add( identity=identity, payload=payload, username=username, + after_commit=after_commit, notification=notification, ) return await subscribe_oper.async_add( @@ -492,6 +500,7 @@ async def async_add_subscribe( username=username, after_commit=after_commit, notification=notification, + occurrence_id=occurrence_id, ) diff --git a/app/chain/subscribe/context.py b/app/chain/subscribe/context.py index 322afaad8..981f2180c 100644 --- a/app/chain/subscribe/context.py +++ b/app/chain/subscribe/context.py @@ -1,7 +1,8 @@ """订阅创建与提交后阶段使用的不可变上下文。""" -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Dict, Optional +from uuid import uuid4 from app.domain.context import ( MediaInfo, @@ -31,6 +32,7 @@ class _SubscribePostCommitContext: username: Optional[str] message: bool notification: Optional[dict[str, Any]] = None + occurrence_id: str = field(default_factory=lambda: uuid4().hex) @dataclass(slots=True) diff --git a/app/chain/subscribe/create.py b/app/chain/subscribe/create.py index 936a88d73..8b0f567c6 100644 --- a/app/chain/subscribe/create.py +++ b/app/chain/subscribe/create.py @@ -420,6 +420,7 @@ class SubscribeCreateOwner(_SubscribeOwnerBase): username=context.username, after_commit=_after_commit, notification=post_commit_context.notification, + occurrence_id=post_commit_context.occurrence_id, **context.options, ) if not sid: @@ -454,6 +455,7 @@ class SubscribeCreateOwner(_SubscribeOwnerBase): username=context.username, after_commit=_after_commit, notification=post_commit_context.notification, + occurrence_id=post_commit_context.occurrence_id, **context.options, ) if not sid: @@ -659,6 +661,7 @@ class SubscribeCreateOwner(_SubscribeOwnerBase): context.mediainfo, notification=notification, after_commit=after_commit, + occurrence_id=post_commit_context.occurrence_id, season=context.season, username=context.username, **context.options, diff --git a/app/chain/subscribe/notify.py b/app/chain/subscribe/notify.py index ae9af2355..ef85d3912 100644 --- a/app/chain/subscribe/notify.py +++ b/app/chain/subscribe/notify.py @@ -10,6 +10,7 @@ from app.application.configuration import ( from app.application.messaging.message import MessageTemplateHelper from app.application.subscription.contract import ( SubscriptionSnapshot, + subscription_added_event_key, ) from app.chain.subscribe.context import _SubscribeCreateContext, _SubscribePostCommitContext from app.chain.subscribe.contract import _SubscribeOwnerBase @@ -129,7 +130,14 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase): EventType.SubscribeAdded, { "subscribe_id": subscribe_id, - "idempotency_key": (f"subscribe.added:{subscribe_id}:{context.media_source}:{context.media_id}:v1"), + "idempotency_key": subscription_added_event_key( + subscribe_id, + { + "media_source": str(context.media_source) if context.media_source else None, + "media_id": context.media_id, + }, + occurrence_id=context.occurrence_id, + ), "username": context.username, "mediainfo": context.mediainfo.to_dict(), }, @@ -157,7 +165,14 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase): EventType.SubscribeAdded, { "subscribe_id": subscribe_id, - "idempotency_key": (f"subscribe.added:{subscribe_id}:{context.media_source}:{context.media_id}:v1"), + "idempotency_key": subscription_added_event_key( + subscribe_id, + { + "media_source": str(context.media_source) if context.media_source else None, + "media_id": context.media_id, + }, + occurrence_id=context.occurrence_id, + ), "username": context.username, "mediainfo": context.mediainfo.to_dict(), }, diff --git a/app/db/adapters/subscription.py b/app/db/adapters/subscription.py index 9f5968bb6..f51e23806 100644 --- a/app/db/adapters/subscription.py +++ b/app/db/adapters/subscription.py @@ -7,6 +7,7 @@ from collections.abc import Awaitable, Callable, Mapping, Sequence from contextlib import AbstractAsyncContextManager from datetime import datetime, timedelta, timezone from typing import Optional, TypeVar, cast +from uuid import uuid4 from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -34,6 +35,9 @@ from app.application.subscription.contract import ( SubscriptionSnapshot, SubscriptionStagingPort, SubscriptionWriteResult, + subscription_added_event_key, + subscription_added_notification_key, + subscription_added_report_key, ) from app.application.subscription.write import ( AsyncCreateSubscriptionBatchCommand, @@ -42,9 +46,6 @@ from app.application.subscription.write import ( AsyncUnitOfWork, CreateSubscriptionCommand, SubscriptionCreateRequest, - subscription_added_event_key, - subscription_added_notification_key, - subscription_added_report_key, ) from app.db.adapters.outbox import ( SqlAlchemyAsyncOutboxDispatchStore, @@ -196,8 +197,10 @@ class _TransactionalSubscriptionWriter: username: str | None = None, after_commit: TypedAfterCommitEffect | None = None, notification: Mapping[str, JsonData] | None = None, + occurrence_id: str | None = None, ) -> tuple[int, str]: """在独占同步会话内执行一次完整订阅新增事务。""" + resolved_occurrence_id = occurrence_id or uuid4().hex write_payload = payload.to_payload() notification_payload = dict(notification) if notification else None session = self._sync_session() @@ -219,6 +222,7 @@ class _TransactionalSubscriptionWriter: write_payload, notification_payload, lambda: after_commit(subscribe_id), + occurrence_id=resolved_occurrence_id, ) return command.execute( @@ -227,6 +231,7 @@ class _TransactionalSubscriptionWriter: username, delivered, notification_payload, + occurrence_id=resolved_occurrence_id, ) finally: session.close() @@ -238,8 +243,10 @@ class _TransactionalSubscriptionWriter: username: str | None = None, after_commit: TypedAsyncAfterCommitEffect | None = None, notification: Mapping[str, JsonData] | None = None, + occurrence_id: str | None = None, ) -> tuple[int, str]: """在独占异步会话作用域内执行一次完整订阅新增事务。""" + resolved_occurrence_id = occurrence_id or uuid4().hex write_payload = payload.to_payload() notification_payload = dict(notification) if notification else None async with self._async_session() as session: @@ -260,6 +267,7 @@ class _TransactionalSubscriptionWriter: write_payload, notification_payload, lambda: after_commit(subscribe_id), + occurrence_id=resolved_occurrence_id, ) return await command.execute( @@ -268,6 +276,7 @@ class _TransactionalSubscriptionWriter: username, delivered, notification_payload, + occurrence_id=resolved_occurrence_id, ) @@ -878,6 +887,7 @@ class SessionSubscriptionBatchWriter: payload, notification, invoke, + occurrence_id=request.occurrence_id, ) return await self._command.execute(requests, after_commit=settle) @@ -887,12 +897,31 @@ def _added_effect_keys( subscribe_id: int, payload: dict[str, JsonData], notification: Optional[dict[str, JsonData]], + occurrence_id: str, ) -> tuple[str, ...]: """返回组合回调实际包含的独立 durable effect 键。""" - keys = [subscription_added_event_key(subscribe_id, payload)] + keys = [ + subscription_added_event_key( + subscribe_id, + payload, + occurrence_id=occurrence_id, + ) + ] if notification: - keys.append(subscription_added_notification_key(subscribe_id, payload)) - keys.append(subscription_added_report_key(subscribe_id, payload)) + keys.append( + subscription_added_notification_key( + subscribe_id, + payload, + occurrence_id=occurrence_id, + ) + ) + keys.append( + subscription_added_report_key( + subscribe_id, + payload, + occurrence_id=occurrence_id, + ) + ) return tuple(keys) @@ -929,12 +958,14 @@ def _deliver_added_effects( payload: dict[str, JsonData], notification: Optional[dict[str, JsonData]], effect: Callable[[], Optional[bool]], + *, + occurrence_id: str, ) -> None: """认领组合回调并按事件、通知、统计的确认结果分别结算。""" now = datetime.now(timezone.utc) claimed = _claim_added_effects( store, - _added_effect_keys(subscribe_id, payload, notification), + _added_effect_keys(subscribe_id, payload, notification, occurrence_id), now, ) if claimed is None: @@ -1001,12 +1032,14 @@ async def _deliver_added_effects_async( payload: dict[str, JsonData], notification: Optional[dict[str, JsonData]], effect: Callable[[], Awaitable[bool | None]], + *, + occurrence_id: str, ) -> None: """异步认领组合回调并按各 intent 的确认结果分别结算。""" now = datetime.now(timezone.utc) claimed = await _claim_added_effects_async( store, - _added_effect_keys(subscribe_id, payload, notification), + _added_effect_keys(subscribe_id, payload, notification, occurrence_id), now, ) if claimed is None: diff --git a/tests/test_outbox.py b/tests/test_outbox.py index bc3261d50..d2c3c6947 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -4,6 +4,7 @@ from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from threading import Barrier from unittest.mock import MagicMock +from uuid import UUID import pytest from sqlalchemy import create_engine, select @@ -63,7 +64,10 @@ def test_subscription_and_outbox_intent_commit_together() -> None: assert result == (42, "ok") assert calls == ["subscription", "outbox", "outbox", "commit"] intent = outbox.stage.call_args_list[0].args[0] - assert intent.event_key == "subscribe.added:42:unknown:unknown:v1" + key_parts = intent.event_key.split(":") + assert key_parts[:4] == ["subscribe.added", "42", "unknown", "unknown"] + assert UUID(key_parts[4]).hex == key_parts[4] + assert key_parts[5] == "v1" assert intent.payload["subscribe_id"] == 42 report_intent = outbox.stage.call_args_list[1].args[0] assert report_intent.topic == "subscribe.added.report" diff --git a/tests/test_subscribe_create_command.py b/tests/test_subscribe_create_command.py index a087d4967..8b585cd31 100644 --- a/tests/test_subscribe_create_command.py +++ b/tests/test_subscribe_create_command.py @@ -5,7 +5,7 @@ from dataclasses import replace from unittest.mock import AsyncMock, Mock import pytest -from sqlalchemy import select +from sqlalchemy import delete, select from app.application.subscription.contract import ( SubscriptionIdentity, @@ -374,6 +374,43 @@ def test_default_sync_writer_persists_once_and_reuses_duplicate(db) -> None: assert after_commit.call_args_list == [((first[0],), {})] +def test_default_sync_writer_allows_reused_subscribe_id_with_retained_outbox(db) -> None: + """删除订阅后复用数据库主键时，新增事件仍以本次创建事实区分。""" + db.watermark(Subscribe, OutboxMessage) + media = _media("arch-221-reused-id") + writer = _writer() + + first = add_subscribe( + mediainfo=media, + subscribe_oper=writer, + after_commit=lambda _subscribe_id: True, + ) + with SessionFactory() as session: + session.execute(delete(Subscribe).where(Subscribe.id == first[0])) + session.commit() + + second = add_subscribe( + mediainfo=media, + subscribe_oper=writer, + after_commit=lambda _subscribe_id: True, + ) + + assert second[0] == first[0] + intents = db.session.execute( + select(OutboxMessage) + .where(OutboxMessage.event_key.contains(media.media_id)) + .order_by(OutboxMessage.id) + ).scalars().all() + assert [intent.topic for intent in intents] == [ + "subscribe.added", + "subscribe.added.report", + "subscribe.added", + "subscribe.added.report", + ] + assert len({intent.event_key for intent in intents}) == 4 + assert all(intent.status == "completed" for intent in intents) + + def test_default_sync_writer_keeps_failed_report_pending_without_raising(db) -> None: """新增统计未确认时接口仍成功，事件 intent 收口而统计 intent 等待重试。""" db.watermark(Subscribe, OutboxMessage) diff --git a/tests/test_subscription_completion_command.py b/tests/test_subscription_completion_command.py index 65f46570a..540a99609 100644 --- a/tests/test_subscription_completion_command.py +++ b/tests/test_subscription_completion_command.py @@ -256,6 +256,33 @@ def test_completion_success_closes_event_then_report_intent(): assert calls[10][1]["idempotency_key"] == calls[3][1].event_key +def test_completion_keys_distinguish_reused_subscribe_id() -> None: + """同一订阅主键被复用时，完成事件与旧 Outbox 历史保持不同。""" + first_calls: list[tuple] = [] + first_command, first_notify, first_report = _command(first_calls) + first_command.execute( + 7, + {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2}, + {"title": "Test"}, + notify=first_notify, + report=first_report, + ) + + second_calls: list[tuple] = [] + second_command, second_notify, second_report = _command(second_calls) + second_command.execute( + 7, + {"id": 7, "media_source": "tmdb", "media_id": "123", "season": 2}, + {"title": "Test"}, + notify=second_notify, + report=second_report, + ) + + first_event = next(call[1] for call in first_calls if call[0] == "stage") + second_event = next(call[1] for call in second_calls if call[0] == "stage") + assert first_event.event_key != second_event.event_key + + def test_completion_stages_and_closes_notification_snapshot() -> None: """完成通知快照与业务事务同提交，成功即时投递后独立收口。""" calls = []