krz/hutch-notify

 1from __future__ import annotations
 2
 3import json
 4import time
 5from pathlib import Path
 6
 7import httpx
 8import jwt
 9
10from app.config import settings
11
12
13class APNSClient:
14    def __init__(self) -> None:
15        self._key_text = Path(settings.apns_private_key_path).read_text()
16        self._token_cache: tuple[str, int] | None = None
17        self._client = httpx.AsyncClient(http2=True, timeout=10.0)
18
19    async def aclose(self) -> None:
20        await self._client.aclose()
21
22    def _bearer_token(self) -> str:
23        now = int(time.time())
24        if self._token_cache and now - self._token_cache[1] < 3000:
25            return self._token_cache[0]
26
27        token = jwt.encode(
28            {"iss": settings.apns_team_id, "iat": now},
29            self._key_text,
30            algorithm="ES256",
31            headers={"kid": settings.apns_key_id},
32        )
33        self._token_cache = (token, now)
34        return token
35
36    async def send_alert(
37        self,
38        *,
39        device_token: str,
40        apns_env: str,
41        topic: str,
42        title: str,
43        body: str,
44        payload: dict,
45    ) -> tuple[bool, str | None, str | None]:
46        host = "https://api.push.apple.com" if apns_env == "production" else "https://api.sandbox.push.apple.com"
47        url = f"{host}/3/device/{device_token}"
48        merged_payload = {
49            "aps": {
50                "alert": {"title": title, "body": body},
51                "sound": "default",
52            },
53            **payload,
54        }
55        response = await self._client.post(
56            url,
57            headers={
58                "authorization": f"bearer {self._bearer_token()}",
59                "apns-topic": topic,
60                "apns-push-type": "alert",
61                "apns-priority": "10",
62            },
63            content=json.dumps(merged_payload).encode("utf-8"),
64        )
65        apns_id = response.headers.get("apns-id")
66        if 200 <= response.status_code < 300:
67            return True, apns_id, None
68        error = None
69        try:
70            error = response.json().get("reason")
71        except Exception:
72            error = response.text
73        return False, apns_id, error