From 16cd9f75ed387ea92303a04e581f4045a672baf8 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 29 Oct 2026 21:43:18 +0800 Subject: [PATCH] fix(apppush): deliver business notifications to all active users (#7235) --- app/application/security/user.py | 14 +++ app/chain/_messaging.py | 56 +++++++++-- app/db/adapters/user.py | 28 ++++++ app/modules/__init__.py | 6 +- app/modules/apppush/__init__.py | 36 ++++--- docs/mcp-api.md | 4 +- docs/testing.md | 5 + tests/test_apppush.py | 28 ++++++ tests/test_notification_routing_parity.py | 109 +++++++++++++++++++++- tests/test_user_repository.py | 29 ++++++ 10 files changed, 290 insertions(+), 25 deletions(-) diff --git a/app/application/security/user.py b/app/application/security/user.py index 77918f85b..901b5bd9d 100644 --- a/app/application/security/user.py +++ b/app/application/security/user.py @@ -258,6 +258,20 @@ class ChainUserRepository(Protocol): :return: 用户通知设置的只读快照；用户不存在时返回 None """ + def get_notification_broadcast_targets(self) -> Mapping[str, tuple[str, ...]]: + """ + 查询所有启用用户的逐账号通知目标，仅由所有用户范围使用。 + + :return: 渠道绑定键与不可变用户身份集合，不包含用户自行填写的目标或停用账号 + """ + + async def async_get_notification_broadcast_targets(self) -> Mapping[str, tuple[str, ...]]: + """ + 异步查询所有启用用户的逐账号通知目标。 + + :return: 与同步入口一致的渠道绑定键和不可变用户身份集合 + """ + def find_name_by_bindings( self, bindings: Mapping[str, object], diff --git a/app/chain/_messaging.py b/app/chain/_messaging.py index 851714d86..6578e43a6 100644 --- a/app/chain/_messaging.py +++ b/app/chain/_messaging.py @@ -31,11 +31,11 @@ class _NotificationRouteLookup: """ 通知路由状态机请求读取指定用户的渠道设置。 - :param username: 需要查询个人通知设置的 MoviePilot 用户名 + :param username: 需要查询个人通知设置的 MoviePilot 用户名；None 表示查询所有启用账号 :param log_message: 执行本次路由查询前记录的中文诊断说明 """ - username: str + username: Optional[str] log_message: str @@ -70,6 +70,32 @@ _NotificationRouteStep = Union[ ] +def _notification_broadcast_message( + message: Message, + broadcast_targets: Mapping[str, tuple[str, ...]], + delivered_targets: list[Mapping[str, object]], +) -> Message: + """ + 为所有用户消息补充逐账号渠道目标，保留原目标并排除已投递账号。 + + :param message: 尚未修改的原通知，不在本方法中原地变更 + :param broadcast_targets: 仓储确认的启用账号身份，其他渠道继续使用自身默认投递规则 + :param delivered_targets: 本次路由已发送的用户目标，避免组合范围重复投递 + :return: 携带明确目标及所有用户范围的独立消息 + """ + routed = copy.deepcopy(message) + targets = dict(routed.targets or {}) + for channel, owners in broadcast_targets.items(): + if owners and not targets.get(channel): + targets[channel] = tuple( + owner for owner in owners + if all(previous.get(channel) != owner for previous in delivered_targets) + ) + routed.targets = targets or None + routed.set_notification_route_scope("all") + return routed + + def _notification_route_steps( message: Message, *, @@ -86,9 +112,11 @@ def _notification_route_steps( 生成器只表达用户设置查询、日志和投递决策，调用方分别执行真实同步或 异步 I/O，避免两套外壳各自维护管理员回退和原消息投递规则。没有可用的 用户范围必须携带用户个人渠道目标；管理员范围允许在个人设置为空时由 - 各渠道使用自身配置的默认目标。 + 各渠道使用自身配置的默认目标。所有用户范围为逐账号渠道补充启用账号身份， + 其他渠道保持原有默认投递，不把缺少目标当作无条件广播。 """ admin_sent = False + delivered_targets: list[Mapping[str, object]] = [] send_original = action is None route_actions: tuple[str, ...] = () if action is not None: @@ -157,6 +185,15 @@ def _notification_route_steps( if username == superuser: admin_sent = True elif route_action == "all": + broadcast_targets = yield _NotificationRouteLookup( + username=None, + log_message=f"{routed_message.mtype} 的消息已设置发送给所有用户", + ) + message = _notification_broadcast_message( + message, + cast(Mapping[str, tuple[str, ...]], broadcast_targets or {}), + delivered_targets, + ) send_original = not admin_sent break else: @@ -172,6 +209,7 @@ def _notification_route_steps( admin_sent = True routed_message.targets = dict(settings) if settings is not None else None routed_message.set_notification_route_scope("admin") + delivered_targets.append(routed_message.targets or {}) yield _NotificationRouteDelivery( message=routed_message, immediately=False, @@ -446,8 +484,10 @@ class NotificationMixin: lookup_result = None if isinstance(step, _NotificationRouteLookup): logger.info(step.log_message) - lookup_result = self.user_repository.get_notification_settings( - step.username + lookup_result = ( + self.user_repository.get_notification_broadcast_targets() + if step.username is None + else self.user_repository.get_notification_settings(step.username) ) elif isinstance(step, _NotificationRouteLog): logger.info(step.message) @@ -529,9 +569,9 @@ class NotificationMixin: if isinstance(step, _NotificationRouteLookup): logger.info(step.log_message) lookup_result = ( - await self.user_repository.async_get_notification_settings( - step.username - ) + await self.user_repository.async_get_notification_broadcast_targets() + if step.username is None + else await self.user_repository.async_get_notification_settings(step.username) ) elif isinstance(step, _NotificationRouteLog): logger.info(step.message) diff --git a/app/db/adapters/user.py b/app/db/adapters/user.py index 20788557a..94bf535ef 100644 --- a/app/db/adapters/user.py +++ b/app/db/adapters/user.py @@ -389,6 +389,34 @@ class TransactionalUserRepository(ChainUserRepository): settings["apppush"] = user.notification_identity return settings + def get_notification_broadcast_targets(self) -> Mapping[str, tuple[str, ...]]: + """ + 在独立会话中读取启用账号的通知身份，不使用可编辑的个人设置。 + + :return: AppPush 的明确接收账号集合；没有可用账号时返回空字典 + """ + with self._sync_session() as session: + owners = tuple(session.execute( + select(User.notification_identity) + .where(User.is_active.is_(True), User.notification_identity != "") + .order_by(User.id) + ).scalars()) + return {"apppush": owners} if owners else {} + + async def async_get_notification_broadcast_targets(self) -> Mapping[str, tuple[str, ...]]: + """ + 在独占异步会话中读取与同步入口相同的启用账号集合。 + + :return: AppPush 的明确接收账号集合；没有可用账号时返回空字典 + """ + async with self._async_session() as session: + owners = tuple((await session.execute( + select(User.notification_identity) + .where(User.is_active.is_(True), User.notification_identity != "") + .order_by(User.id) + )).scalars()) + return {"apppush": owners} if owners else {} + def find_name_by_bindings( self, bindings: Mapping[str, object], diff --git a/app/modules/__init__.py b/app/modules/__init__.py index b55fa3dd5..fdbc22094 100644 --- a/app/modules/__init__.py +++ b/app/modules/__init__.py @@ -366,13 +366,13 @@ class _MessageBase(ServiceBase[TService, NotificationConf]): def get_notification_targets(self, message: Message) -> Optional[dict[str, Any]]: """ - 返回渠道可用的通知目标，管理员范围缺少个人目标时启用渠道默认目标。 + 返回渠道可用的通知目标，管理员和所有用户范围缺少本渠道目标时使用默认目标。 :param message: 已确定路由范围和用户绑定目标的通知 - :return: 渠道目标映射；管理员范围没有个人目标时返回 None，允许使用渠道默认目标 + :return: 渠道目标映射；允许默认投递的范围没有本渠道目标时返回 None """ if ( - message.notification_route_scope == "admin" + message.notification_route_scope in {"admin", "all"} and not self._has_notification_target(message) ): return None diff --git a/app/modules/apppush/__init__.py b/app/modules/apppush/__init__.py index d806356dd..1550fb293 100644 --- a/app/modules/apppush/__init__.py +++ b/app/modules/apppush/__init__.py @@ -211,6 +211,20 @@ class AppPushModule(_ModuleBase, _MessageBase[object]): except ValueError as error: return {"success": False, "message": str(error)} + def _notification_owners(self, message: Message) -> tuple[str, ...]: + """ + 解析明确单账号或所有用户路由生成的账号集合，缺少目标时不隐式广播。 + + :param message: 已由通知链确定范围和账号目标的消息 + :return: 本渠道需要逐个投递的不可变账号身份 + """ + target = (self.get_notification_targets(message) or {}).get("apppush") + if isinstance(target, str): + return (target,) if target else () + if message.notification_route_scope == "all" and isinstance(target, tuple): + return tuple(dict.fromkeys(target)) + return () + def post_message(self, message: Message, **kwargs: object) -> None: """ 仅向路由确定的不可变账号身份投递；缺少绑定绝不退回全设备广播。 @@ -218,18 +232,16 @@ class AppPushModule(_ModuleBase, _MessageBase[object]): for conf in self.get_configs().values(): if not self.check_message(message, conf.name): continue - owner = str((self.get_notification_targets(message) or {}).get("apppush") or "") - if not owner: - continue link = message.link or "/?shotcut=message" # 外部下载或媒体链接留在消息列表中，系统通知点击始终交给本实例。 route = link if link.startswith("/") and not link.startswith("//") and "\\" not in link else "/?shotcut=message" - try: - result = self._request(conf, "POST", "messages", { - "owner": owner, "title": (message.title or "MoviePilot")[:256], - "body": (message.text or "")[:1000], "route": route, - }) - if result.get("failed"): - logger.warning(f"AppPush 有 {len(result['failed'])} 个设备投递失败，请检查 Server 配置") - except ValueError as error: - logger.warning(f"AppPush 投递失败：{error}") + for owner in self._notification_owners(message): + try: + result = self._request(conf, "POST", "messages", { + "owner": owner, "title": (message.title or "MoviePilot")[:256], + "body": (message.text or "")[:1000], "route": route, + }) + if result.get("failed"): + logger.warning(f"AppPush 有 {len(result['failed'])} 个设备投递失败，请检查 Server 配置") + except ValueError as error: + logger.warning(f"AppPush 投递失败：{error}") diff --git a/docs/mcp-api.md b/docs/mcp-api.md index b430ca4b1..e44d84979 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -99,7 +99,9 @@ Web 会话沿用当前用户；无渠道调用不携带收件人，插件回执 AppPush 地址 `PUSH_SERVER_URL` 留空时使用 `MP_SERVER_HOST`；首次操作自动生成并先行持久保存256位实例凭据，再向中心幂等登记现有 `user_uid`。UID 只关联来源，不能据此获取、重置或认领已有凭据。内部 `AppPushCredentials` 按中心地址隔离，不进入通知渠道配置；中心只保存凭据摘要。已有 `PUSH_RELAY_KEY` 手工配置继续兼容。 `POST /api/v1/notification/app/instance` 仅容器超级管理员可调用，正文为 `{"action":"rotate","source":"已保存的渠道名称"}`。轮换保留设备，候选凭据先落库，响应丢失可在重启后接续；返回值不包含密钥。普通用户的 `/notification/app` 不接受该动作。 Apple/VAPID 私钥只放在中心 Server，不放入 App 或用户容器。 -消息沿用现有类型开关和管理员/用户路由；无明确用户目标时不广播。部署与免费自签说明见 +消息沿用现有类型开关和管理员/用户路由；“所有用户”范围会查询本实例所有启用账号的不可变通知身份， +逐账号投递到各自绑定的设备。停用账号和空身份不会进入接收集合，组合路由中已投递的账号不重复发送。 +未经过该范围解析且缺少明确用户目标时不广播，其他渠道继续使用原有默认目标。部署与免费自签说明见 [MoviePilot-Server](https://github.com/jxxghp/MoviePilot-Server#私有-app-消息推送)。 Discord 仅指定 `userid` 的定向通知使用私聊；无法私聊时返回失败，不转发公共频道。 diff --git a/docs/testing.md b/docs/testing.md index 410056f75..5a7bf38a7 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -62,6 +62,11 @@ focused 测试验证，不要求启动产品全套测试。 提交失败与取消后的重试，以及关闭超时保留资源。`tests/test_telegram.py` 覆盖海报失败后的文本 兜底、会话/话题/按钮保留、429 不追加请求，以及发送失败不能被 outbox 当作成功确认。 +AppPush 的所有用户范围由 `tests/test_notification_routing_parity.py` 验证：下载、入库和订阅消息 +经同步、异步与持久事件入口到达真实渠道；明确用户和管理员范围保持隔离，组合范围不重复发送。 +`tests/test_user_repository.py` 使用临时数据库验证接收集合只含启用账号的不可变身份， +`tests/test_apppush.py` 验证无明确范围时不广播，以及单账号投递失败不阻断其他账号；投递 HTTP 均使用替身。 + `tests/test_db_error_diagnostics.py` 在临时 SQLite 的同步及异步连接池中注入 `SQLITE_IOERR_SHORT_READ`：失败事务保留原始异常、不自动重放写入，后续独立事务 使用新连接恢复；锁争用和约束错误不触发额外的连接失效。 diff --git a/tests/test_apppush.py b/tests/test_apppush.py index 6b9a62a19..928d4e43b 100644 --- a/tests/test_apppush.py +++ b/tests/test_apppush.py @@ -80,6 +80,34 @@ def test_apppush_reuses_message_switches_and_never_broadcasts(monkeypatch): assert len(calls) == 1 +@pytest.mark.parametrize("scope", [None, "user", "admin", "all"]) +def test_apppush_account_sets_require_explicit_all_scope(monkeypatch, scope): + """ + 账号集合只在明确的所有用户范围生效，重复身份去重且单账号失败不阻断后续账号。 + """ + conf = NotificationConf(name="App", type="apppush", enabled=True, switchs=["资源下载"]) + module = AppPushModule.__new__(AppPushModule) + module._channel = NotificationChannel.AppPush + module._configs = {conf.name: conf} + monkeypatch.setattr(module, "get_configs", lambda: {conf.name: conf}) + owners = [] + + def request(_conf, _method, _path, data): + """ + 在中转边界记录账号并模拟首个接收账号投递失败。 + """ + owners.append(data["owner"]) + if len(owners) == 1: + raise ValueError("临时投递失败") + return {"total": 1, "accepted": 1} + + monkeypatch.setattr(module, "_request", request) + message = Message(mtype=MessageType.Download, targets={"apppush": ("a" * 32, "a" * 32, "b" * 32)}) + message.set_notification_route_scope(scope) + module.post_message(message) + assert owners == (["a" * 32, "b" * 32] if scope == "all" else []) + + def test_device_management_endpoint_reaches_private_relay_without_exposing_key(monkeypatch): """ 真实端点经渠道和 HTTP 适配器管理本账号设备，私有密钥只进入中转请求头。 diff --git a/tests/test_notification_routing_parity.py b/tests/test_notification_routing_parity.py index ce9520466..4a4a14d77 100644 --- a/tests/test_notification_routing_parity.py +++ b/tests/test_notification_routing_parity.py @@ -10,11 +10,14 @@ from unittest.mock import AsyncMock, Mock import pytest from app.chain._messaging import MessageProcessingMixin, NotificationMixin +from app.modules.apppush import AppPushModule +from app.modules.discord import DiscordModule +from app.modules.slack import SlackModule from app.modules.telegram.module import TelegramModule from app.modules.wechat import WechatModule from app.schemas.message import Message from app.schemas.system import NotificationConf -from app.schemas.types import MessageType +from app.schemas.types import MessageType, NotificationChannel class _NotificationSettingsRepository: @@ -58,6 +61,23 @@ class _NotificationSettingsRepository: raise RuntimeError(f"无法读取 {username} 的通知设置") return self.settings.get(username) + def get_notification_broadcast_targets(self) -> dict[str, tuple[str, ...]]: + """ + 只为需要逐账号投递的 AppPush 提供测试用户身份，不读取真实数据库。 + """ + owners = tuple( + str(settings["apppush"]) + for settings in self.settings.values() + if settings and settings.get("apppush") + ) + return {"apppush": owners} if owners else {} + + async def async_get_notification_broadcast_targets(self) -> dict[str, tuple[str, ...]]: + """ + 异步查询使用相同的测试账号集合。 + """ + return self.get_notification_broadcast_targets() + class _NotificationHarness(MessageProcessingMixin, NotificationMixin): """ @@ -171,6 +191,93 @@ def test_notification_routing_sync_async_parity( assert async_chain.messageoper.async_add.await_count == 1 +@pytest.mark.parametrize("mode", ["sync", "async", "strict"]) +@pytest.mark.parametrize("mtype", [MessageType.Download, MessageType.Organize, MessageType.Subscribe]) +@pytest.mark.parametrize( + ("action", "expected_owners"), + [ + ("all", ["a" * 32, "b" * 32]), + ("user", ["a" * 32]), + ("admin", ["b" * 32]), + ("user,admin", ["a" * 32, "b" * 32]), + ("user,all", ["a" * 32, "b" * 32]), + ], +) +def test_apppush_receives_business_notifications_through_real_routing( + monkeypatch, mode: str, mtype: MessageType, action: str, expected_owners: list[str], +) -> None: + """ + 业务消息经同步、异步和持久事件入口到达真实渠道，所有用户范围逐账号去重投递。 + """ + monkeypatch.setattr("app.chain._messaging.get_notification_switch", lambda _mtype: action) + repository = _NotificationSettingsRepository({ + "alice": {"apppush": "a" * 32}, + "admin": {"apppush": "b" * 32}, + }) + chain = _NotificationHarness(repository) + conf = NotificationConf(name="App", type="apppush", enabled=True, switchs=[mtype.value]) + module = AppPushModule.__new__(AppPushModule) + module._channel = NotificationChannel.AppPush + module._configs = {conf.name: conf} + monkeypatch.setattr(module, "get_configs", lambda: {conf.name: conf}) + requests = Mock(return_value={"total": 1, "accepted": 1}) + monkeypatch.setattr(module, "_request", requests) + + def deliver(_method: str, *, message: Message, **_kwargs: object) -> None: + """ + 仅在最终中转 HTTP 边界使用替身，保留链路和渠道的真实筛选。 + """ + module.post_message(message) + + chain.messagequeue.send_message.side_effect = deliver + chain.messagequeue.async_send_message.side_effect = deliver + chain.run_module_strict = Mock(side_effect=deliver) + message = Message(mtype=mtype, title="业务通知", username="alice", save_history=False) + if mode == "async": + asyncio.run(chain.async_post_message(message)) + elif mode == "strict": + chain.post_message_strict(message, event_key="download.added.notification:test") + else: + chain.post_message(message) + + assert [call.args[3]["owner"] for call in requests.call_args_list] == expected_owners + assert all(call.args[2] == "messages" for call in requests.call_args_list) + assert message.targets is None + assert message.notification_route_scope is None + + +@pytest.mark.parametrize( + ("module_type", "channel"), + [ + (TelegramModule, NotificationChannel.Telegram), + (SlackModule, NotificationChannel.Slack), + (DiscordModule, NotificationChannel.Discord), + (WechatModule, NotificationChannel.Wechat), + ], +) +def test_all_user_apppush_targets_preserve_other_channel_defaults(monkeypatch, module_type, channel) -> None: + """ + 补齐 AppPush 账号后，其他渠道仍只发送一次默认通知，不误判为缺少个人绑定。 + """ + monkeypatch.setattr("app.chain._messaging.get_notification_switch", lambda _mtype: "all") + chain = _NotificationHarness(_NotificationSettingsRepository({"admin": {"apppush": "a" * 32}})) + chain.post_message(Message(mtype=MessageType.Download, title="下载完成")) + message = chain.messagequeue.send_message.call_args.kwargs["message"] + conf = NotificationConf(name="channel", type="channel", enabled=True, switchs=["资源下载"]) + module = module_type.__new__(module_type) + module._channel = channel + module._configs = {conf.name: conf} + client = Mock() + module._instances = {conf.name: client} + monkeypatch.setattr(module, "get_configs", lambda: {conf.name: conf}) + monkeypatch.setattr(module, "get_instance", lambda _name: client) + + assert module.get_notification_targets(message) is None + module.post_message(message) + client.send_msg.assert_called_once() + assert client.send_msg.call_args.kwargs["userid"] is None + + def test_notification_routing_lookup_failure_sync_async_parity(monkeypatch) -> None: """ 用户设置查询失败时两种入口都应传播错误且不得产生错误路由投递。 diff --git a/tests/test_user_repository.py b/tests/test_user_repository.py index 88ebadfa7..816f561de 100644 --- a/tests/test_user_repository.py +++ b/tests/test_user_repository.py @@ -611,3 +611,32 @@ async def test_notification_targets_use_immutable_identity_and_ignore_user_setti session.commit() _insert_user(sync_factory, id=user_id, name="renamed") assert repository.get_notification_settings("renamed")["apppush"] != first + + +@pytest.mark.asyncio +async def test_notification_broadcast_targets_only_include_active_immutable_identities(user_repository): + """ + 所有用户目标由数据库身份生成，过滤停用和空身份账号，不接受个人设置注入。 + """ + repository, sync_factory = user_repository + first_id = _insert_user(sync_factory, settings={"apppush": "injected-owner"}) + second_id = _insert_user(sync_factory, name="bob", is_superuser=False) + _insert_user(sync_factory, name="disabled", is_active=False) + _insert_user(sync_factory, name="empty", notification_identity="") + with sync_factory() as session: + expected = tuple(session.get(User, user_id).notification_identity for user_id in (first_id, second_id)) + assert repository.get_notification_broadcast_targets() == {"apppush": expected} + assert await repository.async_get_notification_broadcast_targets() == {"apppush": expected} + assert "injected-owner" not in expected + + with sync_factory() as session: + session.get(User, first_id).name = "renamed" + session.get(User, second_id).is_active = False + session.commit() + assert repository.get_notification_broadcast_targets() == {"apppush": expected[:1]} + assert await repository.async_get_notification_broadcast_targets() == {"apppush": expected[:1]} + with sync_factory() as session: + session.get(User, first_id).is_active = False + session.commit() + assert repository.get_notification_broadcast_targets() == {} + assert await repository.async_get_notification_broadcast_targets() == {}