Notifications, escalation and tenant switch

Notifications

Model

notifications (one row per recipient): tenant_id (nullable - a security notice about the user themself belongs to no tenant), user_id (the user's id as a string, no FK, like Membership), type (dotted key such as ticket.escalated, ticket.assigned, security.new_device), title, body, data (JSONB context), link (dashboard-relative path), read_at, created_at (clock_timestamp() so ordering is stable inside a transaction) and digested_at. Index (user_id, read_at, created_at) serves the feed and unread count.

Sending: notify

from app.services import notifications

await notifications.notify(
    db,
    [user_id],
    "security.new_device",
    "New sign-in",
    "Chrome on macOS",
    link="/settings/security",
    tenant_id=None,
    channels=None,
)

Any service can call it (e.g. new-device and MFA alerts). It commits, never raises, and returns the stored rows. Channels come from user_preferences.notification_channels (subset of in_app, email, sse):

Channel	Effect	If preferences are unset
in_app	row stored, visible in the feed	on
sse	pushed to the user's open stream	on
email	immediate mail via send_notification_email (best-effort) and eligibility for the digest	off (opt-in)

The optional channels argument narrows delivery further (intersection with the user's own channels); the escalation matrix uses it to honour each entry's notify_channels.

Endpoints (/api/v1/notifications, any authenticated user, own rows only, no X-Tenant-ID)

Method and path	Notes
GET /notifications?page&limit&unread	newest first; {results, meta:{page,limit,total,unread_count}}
GET /notifications/unread-count	{unread_count}
POST /notifications/{id}/read	the updated notification; 404 for someone else's
POST /notifications/read-all	{updated}
DELETE /notifications/{id}	204; 404 for someone else's
GET /notifications/stream	Server-Sent Events, below

SSE stream

GET /notifications/stream (text/event-stream) authenticates with the session cookie, so a browser EventSource works unchanged. Frames:

retry: 3000

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

id: <notification uuid>
event: notification
data: {"id":"...","type":"ticket.escalated","title":"...","body":"...","data":{...},"link":"/tickets/...","read_at":null,"created_at":"..."}

: ping

No Redis. Within a process an in-memory hub (NotificationHub) fans a freshly stored notification out to that user's open streams. Because several API workers may run, each stream also polls the table every few seconds for rows newer than the last one seen (with a small overlap and an id de-duplication set), so a notification created by another worker still arrives. Last-Event-ID (sent automatically by EventSource on reconnect) replays everything newer than that notification. The request's DB connection is released before streaming starts. A user with sse enabled but in_app disabled receives transient events with id: null that are not stored.

Settings: NOTIFICATIONS_SSE_POLL_SECONDS (3), NOTIFICATIONS_SSE_HEARTBEAT_SECONDS (15), NOTIFICATIONS_SSE_MAX_SECONDS (0 = unlimited; set to bound a stream, used by tests).

Email digest

notifications.send_digests mails each user whose channels include email one message listing their unread, not-yet-digested notifications from the last 24 hours, then stamps digested_at. Run it from the scheduler (NOTIFICATIONS_DIGEST_ENABLED=true, daily at NOTIFICATIONS_DIGEST_HOUR_UTC, default 7) or with uv run python scripts/send_notification_digests.py. Off by default.

Escalation

Matrix

escalation_matrix (per tenant): level (L1..L5), user_id (staff user, nullable), contact_email (external mailbox, nullable; one of the two is required), notify_channels (in_app/email, default both), escalate_after_minutes (default 60), is_active, notes. Unique on (tenant_id, level, user_id).

Endpoint (/api/v1/escalation, needs X-Tenant-ID)	Access
GET /matrix?level&active	escalation:read
POST /matrix, PATCH /matrix/{id}, DELETE /matrix/{id}	escalation:write (409 on duplicate user at a level, 404 for an unknown user)
GET /levels	escalation:read - the five tiers with staff_count (users whose escalation_level matches) and matrix_entries

Escalating a ticket

tickets gains escalation_level, escalated_at, escalated_by (also returned by TicketRead). A ticket with no level is implicitly at L1. POST /tickets/{id}/escalate {level?, reason} (tickets:escalate, normal ticket visibility rules) moves it to the next level - the first escalation lands on L2 - or to an explicit level, which must be strictly higher (400 otherwise, also for resolved/closed tickets and when already at L5). It writes an audit row (domain ticket, action escalate, payload from/to/reason/automatic - the escalation history, kept out of the client-visible comment thread), then notifies every active matrix entry for the new level: users through notify (type ticket.escalated, link to the ticket) and bare contact emails by email.

Adding a user to a ticket (POST /tickets/{id}/members) notifies them (ticket.assigned) unless they were already a member or added themself.

SLA auto-escalation

escalation.escalate_overdue_tickets(db) escalates every open/in-progress ticket that has sat at its level longer than the smallest escalate_after_minutes among that tenant's active entries for the level, measured from escalated_at (or creation). Levels without entries never time out. Actor is system. It runs from the scheduler every ESCALATION_AUTO_INTERVAL_MINUTES (5) only when ESCALATION_AUTO_ENABLED=true (default off).

Tenant switch

POST /api/v1/tenants/switch {tenant_id}: a platform role may switch to any tenant, other users need a tenant-scope membership (403 otherwise, 404 for an unknown tenant). It writes an audit row (domain tenant, action switch, in the target tenant) and returns the tenant (TenantRead). Tokens are not re-issued - the tenant is still chosen per request through X-Tenant-ID; the dashboard calls this when the user picks a tenant so the switch is explicit and audited.