fix(webpush): add WNS cache policy for Windows Edge push (#6034)

WNS rejects pywebpush default ttl=0 with 400 Bad Request unless X-WNS-Cache-Policy matches TTL; iOS/APNs endpoints are unaffected.

Co-authored-by: Cursor <cursoragent@cursor.com>

app/api/endpoints/message.py

@@ -17,7 +17,7 @@ from app.db.message_oper import MessageOper
 from app.db.systemconfig_oper import SystemConfigOper
 from app.db.user_oper import get_current_active_superuser
 from app.helper.service import ServiceConfigHelper
-from app.helper.webpush import is_webpush_subscription_gone
+from app.helper.webpush import is_webpush_subscription_gone, webpush_options_for_endpoint
 from app.log import logger
 from app.modules.wechat.WXBizMsgCrypt3 import WXBizMsgCrypt
 from app.schemas.types import MessageChannel, SystemConfigKey
@@ -316,6 +316,7 @@ def send_notification(
                 data=json.dumps(payload.model_dump()),
                 vapid_private_key=settings.VAPID.get("privateKey"),
                 vapid_claims={"sub": settings.VAPID.get("subject")},
+                **webpush_options_for_endpoint(sub.get("endpoint")),
             )
         except WebPushException as err:
             logger.error(f"WebPush发送失败: {str(err)}")

app/helper/webpush.py

@@ -2,6 +2,9 @@ from typing import Any
 
 from pywebpush import WebPushException
 
+# WNS 默认 TTL（秒）；ttl>0 时需配 X-WNS-Cache-Policy: cache
+_WNS_DEFAULT_TTL = 86400
+
 
 def is_webpush_subscription_gone(error: WebPushException) -> bool:
     """
@@ -10,3 +13,25 @@ def is_webpush_subscription_gone(error: WebPushException) -> bool:
     response: Any = getattr(error, "response", None)
     status_code = getattr(response, "status_code", None) or getattr(response, "status", None)
     return status_code in {404, 410}
+
+
+def is_wns_endpoint(endpoint: str | None) -> bool:
+    """
+    判断是否为 Microsoft WNS（Edge/Windows）推送端点。
+    """
+    return bool(endpoint and "notify.windows.com" in endpoint)
+
+
+def webpush_options_for_endpoint(endpoint: str | None) -> dict[str, Any]:
+    """
+    按推送服务返回 pywebpush 额外参数。
+
+    WNS 要求 TTL 与 X-WNS-Cache-Policy 一致，否则返回 400。
+    见 https://github.com/web-push-libs/pywebpush/issues/162
+    """
+    if not is_wns_endpoint(endpoint):
+        return {}
+    return {
+        "ttl": _WNS_DEFAULT_TTL,
+        "headers": {"X-WNS-Cache-Policy": "cache"},
+    }

app/modules/webpush/__init__.py

@@ -4,7 +4,7 @@ from typing import Union, Tuple
 from pywebpush import webpush, WebPushException
 
 from app.core.config import global_vars, settings
-from app.helper.webpush import is_webpush_subscription_gone
+from app.helper.webpush import is_webpush_subscription_gone, webpush_options_for_endpoint
 from app.log import logger
 from app.modules import _ModuleBase, _MessageBase
 from app.schemas import Notification
@@ -95,6 +95,7 @@ class WebPushModule(_ModuleBase, _MessageBase):
                             vapid_claims={
                                 "sub": settings.VAPID.get("subject")
                             },
+                            **webpush_options_for_endpoint(sub.get("endpoint")),
                         )
                     except WebPushException as err:
                         logger.error(f"WebPush发送失败: {str(err)}")

tests/test_webpush_helper.py

@@ -0,0 +1,43 @@
+from app.helper.webpush import (
+    is_webpush_subscription_gone,
+    is_wns_endpoint,
+    webpush_options_for_endpoint,
+)
+from pywebpush import WebPushException
+
+
+class _FakeResponse:
+    def __init__(self, status_code: int):
+        self.status_code = status_code
+
+
+def test_is_webpush_subscription_gone_for_expired_status_codes() -> None:
+    for status_code in (404, 410):
+        err = WebPushException("gone")
+        err.response = _FakeResponse(status_code)
+        assert is_webpush_subscription_gone(err)
+
+
+def test_is_webpush_subscription_gone_for_other_errors() -> None:
+    err = WebPushException("bad request")
+    err.response = _FakeResponse(400)
+    assert not is_webpush_subscription_gone(err)
+
+
+def test_is_wns_endpoint_detects_windows_push_url() -> None:
+    assert is_wns_endpoint("https://wns2-sg2p.notify.windows.com/w/?token=abc")
+    assert not is_wns_endpoint("https://web.push.apple.com/abc")
+    assert not is_wns_endpoint(None)
+    assert not is_wns_endpoint("")
+
+
+def test_webpush_options_for_wns_endpoint() -> None:
+    options = webpush_options_for_endpoint("https://wns2-pn1p.notify.windows.com/x")
+    assert options == {
+        "ttl": 86400,
+        "headers": {"X-WNS-Cache-Policy": "cache"},
+    }
+
+
+def test_webpush_options_for_non_wns_endpoint() -> None:
+    assert webpush_options_for_endpoint("https://fcm.googleapis.com/fcm/send/abc") == {}