Notification Service

Implementation reference for the notification. For its architectural role, see Notification Service Architecture; for the domain model, see Alerting.

A Rust / Axum service in Ports & Adapters shape. It receives candidate alert events, debounces them, resolves recipients, and delivers through two channels: an in-app event over the Redis broker and a Web Push to registered devices.

Source Layout

graph TD
    HTTP["adapters/driving/http_api"] --> ALERTS["service/alerts.rs"]
    HTTP --> PREFS["service/preferences.rs"]
    LISTEN["adapters/driving/alert_listener.rs"] --> ALERTS
    ALERTS --> PUSH["service/push.rs"]
    ALERTS --> PORTS["service/ports.rs"]
    PREFS --> PORTS
    PUSH --> PORTS
    PORTS --> MONGO["adapters/driven/persistence"]
    PORTS --> REDIS["adapters/driven/redis_bus.rs"]
    PORTS --> WEBPUSH["adapters/driven/push.rs"]
    PORTS --> TWIN["adapters/driven/twin.rs"]

Configuration

Data Model

Two collections, kept separate by design (see Alerting.

// webpushsubscriptions — "where can I reach this account?"
struct WebPushSubscription {
    account_name: String,   // indexed
    endpoint: String,       // unique
    keys: SubscriptionKeys, // p256dh, auth
}
// compound unique index { accountName, endpoint }

// notificationsubscriptions — "what does this account want?"
struct AccountPreferences {
    account_name: String,
    domain_name: String,
    preferences: Vec<Preference>, // notificationType, isSubscribed
    created_at: String,
}
// compound unique index { accountName, domainName }

Collection names and field names are the mongoose ones — the store is shared with the pre-Rust data. Notification type is a metric key from telemetry_schema::ALERTABLE_METRICS: "temperature", "airQuality" or "peopleCount".

On startup, switch_on_new_metrics_once switches each alertable metric on for every account that has temperature on — once per metric, recorded as a switched-on:<metric> marker in the migrations collection, and never over a choice already stored. Those accounts subscribed before the metric existed; /subscribe now switches every metric on by itself.

A single account may own several WebPushSubscription records (one per device) but exactly one preferences document per domain.

HTTP API

Request bodies accept the canonical domainName field and the deprecated alias domainId. The account is always taken from the verified claims, never from a body accountName/userId or the URL param — this closed an IDOR on GET /preferences/:accountName. Preferences may be supplied as a single { type, enabled }, a types[] array with one enabled, or a granular preferences[] array — every accepted and rejected form is pinned in schemas/fixtures/notification-preferences.json.

Domain membership

Authentication is not enough: every route that names a domain also checks that the caller is a member of it, from memberships in the claims. Without this, any authenticated user of any tenant could subscribe themselves to another tenant’s alert feed, or push arbitrary text to every opted-in user of a domain they have nothing to do with.

Audience::of decides what a caller may reach: Unrestricted for a system: subject or a global admin, Domains(...) for everyone else. That mirrors policy.cedar’s ReadWithAdminBypass — domain membership or maxRoleWeight >= 100 — so alerting is never narrower than digital-twin’s own read rule, where an admin may read any building.

GET /preferences needs no check — it reads by authenticated account and cannot name a domain at all.

The Alert Throttle

The telemetry fires on every reading (~every 10 s). To avoid a storm of identical alerts, Alerts debounces per room and per breached field using a Redis key with a fixed TTL. It is also what absorbs the duplicates that the alerts topic’s at-least-once delivery allows.

sequenceDiagram
    autonumber
    participant SS as telemetry
    participant NS as notification
    participant R as Redis
    participant TW as digital-twin
    SS->>NS: AlertEvent on the alerts topic { type, field, buildingId, roomId, ... }
    NS->>R: GET alert:{type}:{field}:{buildingId}:{roomId}
    alt cooldown active
        R-->>NS: "1"
        Note over NS: suppressed, record settled
    else first alert / expired
        NS->>TW: GET /twin/domain/:buildingId
        TW-->>NS: domain names
        Note over NS: publish "notifications" + Web Push to opted-in recipients
        NS->>R: SET alert:{type}:{field}:{buildingId}:{roomId} "1" EX 300
    end

• The key is alert:{metric}:{field}:{buildingId}:{roomId}; the cooldown is 300 seconds. After it expires, the next breach starts a new cycle.
• One key per breached field, so a CO2 alert never silences a temperature or AQI alert in the same room.

Recipient Resolution and Delivery

service/push.rs separates who to notify from how to reach them.

• to_domain(payload, domain_name, type) queries the preferences collection for records in the domain whose preferences contain an enabled entry for type ($elemMatch), reduces to a unique set of account names, and delegates to to_accounts. An absent type matches every subscriber of the domain.
• A breach publishes one bell message per domain carrying metric, and pushes only to that metric’s subscribers. The browser hides a breach whose metric the account switched off — the same rule, applied client-side, so the broker and socket stay one message per domain. The frontend loads the preferences before its socket connects and again whenever the window regains focus (App.vue), and filters in services/socket.ts through isVisible (src/utils/notification.ts). The switches are one chip per alertable metric on every domain and subdomain card of the administration page (components/buttons/AlertSwitches.vue); NotificationType must list exactly ALERTABLE_METRICS, asserted against the preferences fixture.
• to_accounts(payload, account_names) loads every WebPushSubscription for those accounts and sends the encrypted payload to each endpoint concurrently — one dead endpoint never aborts the batch.
• The in-app channel carries the same message: NotificationBus::publish posts the notification_schema::Notification { id, type, title, message, timestamp, domainName?, icon? } to the Redis notifications channel, which the socket relays to browsers. The Web Push payload is those same bytes — see Notification Schema.

Preference upserts

PreferenceStore::set is idempotent per type: it first $pulls any existing entry for that notificationType, then $pushes the new { notificationType, isSubscribed }, using $setOnInsert to create the document on first write. Two writes, not one.

Stale-subscription cleanup

When the push endpoint answers 410 Gone or 403 Forbidden, it is permanently invalid and the subscription is deleted automatically. All other errors (network, 5xx) are logged but leave the subscription intact, as they may be transient.

Error Handling

DomainError maps to the same { type, message } body the Node service returned. Missing accountName/domainName and malformed push payloads are Validation Error (400); a failed twin lookup surfaces as Internal Server Error (500). The broker path never returns an error — a malformed alert or a failed lookup is logged and the cooldown is still armed.