Notifications

pleate_backend.notifications is one function, notify(), that reaches people where they want to be reached: the in app inbox (with realtime updates over Server-Sent Events), email and Web Push. Users choose the channels per kind of notification.

Sending

from pleate_backend.notifications.services import notify

notify(
    user,
    "order.shipped",                       # kind: free form, dotted by convention
    "Your order is on its way",            # title
    body="Two bags of oat flour, arriving Thursday.",
    url="/orders/1042",                     # relative URLs are joined to FRONTEND_URL in emails
    data={"order": "1042"},
    channels=None,                          # None: follow the user's preferences
)

notify returns the stored Notification, or None when the in app channel is not used. Channels are, in order of precedence: the channels argument, the user's preference for this kind, their "*" preference, then DEFAULT_CHANNELS.

• in_app: stored, then published on the realtime channel user:<id> as a notification event followed by unread_count.
• email: the notification template (title, body, action_url, action_label), in the user's language.
• push: Web Push through pywebpush with the VAPID keys (pleate-backend[push]), sent from a Celery task so notify() does not wait for it. Skipped when the keys are not set; subscriptions that answer 404 or 410 are deleted. Only endpoints on PUSH_ALLOWED_HOSTS are accepted and contacted.

Endpoints

The push subscription routes need a signed in session (403 api_key_not_allowed with an API key). The endpoint must be on a known Web Push service (400 unknown_push_service). Posting an endpoint that another account holds answers 409 push_subscription_taken, unless the keys are the same: that is the same browser signed in with another account, and the subscription moves to it.

The stream

/notifications/stream sends the unread count on connect, then a notification and an unread_count whenever something changes. A comment line (: ping) keeps idle connections open every HEARTBEAT_SECONDS. The server ends the stream after MAX_STREAM_SECONDS (300) so a worker is not held forever, and retry: 3000 tells clients how fast to reconnect (useNotifications does it on its own).

retry: 3000

event: unread_count
data: {"count": 3}

event: notification
data: {"id": "6720…", "kind": "order.shipped", "title": "Your order is on its way", "read": false, …}

event: unread_count
data: {"count": 4}

: ping

useNotifications reads it over fetch (so it works with any headers), merges new items into the cached first page, updates the count and reconnects with exponential backoff up to 30 seconds. With a plain EventSource, pass { withCredentials: true }.

On the frontend

const inbox = useNotifications({ pageSize: 20 });   // live: true by default
<NotificationBell {...inbox.notificationCenterProps} icons={icons} />

// Preferences, with the plain client
const client = usePleateClient();
await client.notifications.updatePreferences({ "*": ["in_app"], "billing.payment_failed": ["in_app", "email"] });

Web Push

Generate VAPID keys once, set them in the settings, register a service worker that shows the push, and subscribe:

const push = usePushSubscription();
const registration = await navigator.serviceWorker.register("/sw.js");
push.mutate(registration);   // asks permission, subscribes with the backend's VAPID key, saves the subscription

Settings

PLEATE["NOTIFICATIONS"]:

Realtime: PLEATE["REALTIME"] with BACKEND (MemoryBroker), HEARTBEAT_SECONDS (20) and MAX_STREAM_SECONDS (300, 0 keeps streams open). Notifications, preferences and push subscriptions are part of the account export.

Your own realtime channels

from pleate_backend.core.realtime import broker, channel_stream, sse_response

broker().publish(f"order:{order.pk}", {"type": "status", "data": {"status": "shipped"}})

class OrderStream(APIView):
    def get(self, request, pk):
        order = get_object_or_404(Order, pk=pk, user=request.user)
        return sse_response(channel_stream(f"order:{order.pk}"))