142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
import asyncio
|
|
import json
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Awaitable, Callable
|
|
|
|
from src.push_subscription_store import PushSubscriptionStore
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PushConfiguration:
|
|
public_key: str
|
|
private_key: str
|
|
subject: str
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self.public_key and self.private_key and self.subject)
|
|
|
|
|
|
async def send_web_push(
|
|
subscription: dict, payload: str, configuration: PushConfiguration
|
|
) -> None:
|
|
from pywebpush import webpush
|
|
|
|
await asyncio.to_thread(
|
|
webpush,
|
|
subscription_info=subscription,
|
|
data=payload,
|
|
vapid_private_key=configuration.private_key,
|
|
vapid_claims={"sub": configuration.subject},
|
|
ttl=300,
|
|
)
|
|
|
|
|
|
async def dispatch_unread_updates(
|
|
store: PushSubscriptionStore,
|
|
configuration: PushConfiguration,
|
|
unread: Callable[[], Awaitable[dict]],
|
|
send: Callable[[dict, str], Awaitable[None]] | None = None,
|
|
*,
|
|
session_active: Callable[[str], Awaitable[bool]] | None = None,
|
|
lease_seconds: float = 60.0,
|
|
send_timeout_seconds: float = 10.0,
|
|
max_concurrency: int = 8,
|
|
) -> int:
|
|
if not configuration.enabled:
|
|
return 0
|
|
owner = secrets.token_urlsafe(18)
|
|
acquired = await asyncio.to_thread(
|
|
store.acquire_dispatch_lease,
|
|
owner,
|
|
now=time.time(),
|
|
lease_seconds=lease_seconds,
|
|
)
|
|
if not acquired:
|
|
return 0
|
|
try:
|
|
page = await unread()
|
|
thread_ids = {
|
|
int(item["notification_id"])
|
|
for item in page.get("items", [])
|
|
if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit()
|
|
}
|
|
deliveries = await asyncio.to_thread(store.claim_unseen, thread_ids)
|
|
if session_active is not None:
|
|
try:
|
|
authorized = [await session_active(item.session_id) for item in deliveries]
|
|
except Exception:
|
|
# Authorization state is mandatory for delivery. Preserve subscriptions
|
|
# so a temporary registry failure can be retried safely.
|
|
return 0
|
|
for delivery, active in zip(deliveries, authorized):
|
|
if not active:
|
|
await asyncio.to_thread(store.delete_session, delivery.session_id)
|
|
deliveries = [
|
|
delivery
|
|
for delivery, active in zip(deliveries, authorized)
|
|
if active
|
|
]
|
|
semaphore = asyncio.Semaphore(max(1, max_concurrency))
|
|
ownership_lost = asyncio.Event()
|
|
|
|
async def dispatch_device(delivery) -> int:
|
|
async with semaphore:
|
|
if ownership_lost.is_set():
|
|
return 0
|
|
count = 0
|
|
for thread_id in delivery.thread_ids:
|
|
still_owner = await asyncio.to_thread(
|
|
store.acquire_dispatch_lease,
|
|
owner,
|
|
now=time.time(),
|
|
lease_seconds=lease_seconds,
|
|
)
|
|
if not still_owner:
|
|
ownership_lost.set()
|
|
return count
|
|
payload = json.dumps(
|
|
{
|
|
"title": "New work update",
|
|
"body": "Tap to review it in Stackchain.",
|
|
"route": f"#/my-work/update/{thread_id}",
|
|
"tag": f"stackchain-update-{thread_id}",
|
|
"notification_id": thread_id,
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
try:
|
|
if send is None:
|
|
operation = send_web_push(
|
|
delivery.subscription, payload, configuration
|
|
)
|
|
else:
|
|
operation = send(delivery.subscription, payload)
|
|
await asyncio.wait_for(operation, timeout=send_timeout_seconds)
|
|
except Exception as error:
|
|
status = getattr(
|
|
getattr(error, "response", None), "status_code", None
|
|
)
|
|
if status in {404, 410}:
|
|
await asyncio.to_thread(
|
|
store.delete_session, delivery.session_id
|
|
)
|
|
break
|
|
# Leave this device's transient failures unseen for a later
|
|
# poll instead of paying the endpoint deadline repeatedly.
|
|
break
|
|
await asyncio.to_thread(
|
|
store.mark_delivered, delivery.session_id, (thread_id,)
|
|
)
|
|
count += 1
|
|
return count
|
|
|
|
counts = await asyncio.gather(
|
|
*(dispatch_device(delivery) for delivery in deliveries)
|
|
)
|
|
return sum(counts)
|
|
finally:
|
|
await asyncio.to_thread(store.release_dispatch_lease, owner)
|