The admin-facing live event stream (GET /api/admin/realtime/stream) and its catalogue (GET /api/admin/realtime/catalogue).
The allowlist of outbox events that may reach an admin toast (REALTIME_EVENT_MAP), and the permission each one requires.
Publishing a relayed outbox row to every connected API instance over Redis pub/sub.
The durable notification record of every published event, and the per-user notification_read join table.
Reconnect replay: a bounded history buffer in Redis, read on connect and filtered by the caller's permissions.
The control channel (realtime:control): instructions about the connections themselves, today only close-user, which ends every open stream of one person on every API instance. The staff role edit publishes it when it removes a role.
The attach re-check: a stream whose session's active role changed between authentication and registration is ended as soon as it registers.
The personal notification stream (GET /api/mobile/notifications/stream), the notification centre list, unread counts, or push devices — those belong to the notification module. See /docs/developer/notification/backend.
The operational admin feed that lists persisted notification rows for browsing (GET /api/notifications) — that belongs to the separate notification-feed module, which reads the same table this module writes.
The outbox itself (the transactional write, the dispatcher, retry and dead-letter handling) — owned by the outbox module. This module is a consumer that is called after a batch has already been dispatched.
Email, SMS, or push delivery. This module never sends anything to a device; it only tells an already-open admin tab what happened.
Composes the leaf and re-exports the shared transport so the outbox dispatcher can publish into it.
RealtimeAdminModule
Leaf
admin/realtime-admin.module.ts
RealtimeAdminController
None (imports RealtimeSharedModule, RoleModule)
None
Owns the only controller: catalogue and stream routes.
RealtimeSharedModule
Leaf
shared/realtime-shared.module.ts
None
RealtimePublisherService, RealtimeStreamService
Both services
The transport: publish and consume, shared by the admin controller and the outbox dispatcher.
RealtimeControlModule
Leaf
shared/realtime-control.module.ts
None
RealtimeControlPublisher
RealtimeControlPublisher
The publishing half of the control channel and nothing else. Imported by PeopleModule.
app.module.ts composes RealtimeAdminModule directly into ADMIN_MODULES — not the RealtimeModule aggregate — because Swagger's module include does not recurse and the aggregate declares no controllers of its own.
RealtimeControlModule is separate from RealtimeSharedModule on purpose. Importing the shared module instantiates RealtimeStreamService, which opens a Redis subscriber connection. PeopleModule only needs to publish close-user, and the data-transfer worker imports PeopleModule; importing the control module instead keeps a stream subscriber out of anything that merely publishes.
Not exposed by this module — no public read route for individual rows lives here.
outbox_event_id
integer
No
—
UNIQUE, not a FK
outbox row (by convention, not FK)
Deliberately not a foreign key: dispatched outbox rows are purged separately, and a cascade would erase notification history with them. The unique constraint is the entire at-least-once dedupe story.
kind
text
No
—
indexed via permission composite reads
N/A
The contract's stable event name, e.g. feedback.submitted.
permission
text
No
—
notification_permission_idx
N/A
The permission a viewer must hold to see the row; not a foreign key into the permission catalogue.
aggregate_id
text
No
—
none
N/A
Public id of the thing that changed.
summary
text
No
—
none
N/A
Pre-built, already-safe toast text. Never raw payload.
sound_class
text
No
—
none
N/A
One of REALTIME_SOUND_CLASSES, stored as text rather than an enum.
occurred_at
timestamptz
No
—
notification_occurred_at_idx (desc)
N/A
When the underlying thing happened, not when this row was written.
Outbox dispatcher, after its own transaction commits
REALTIME_EVENT_MAP
notification rows, Redis history list, Redis sequence key
Redis PUBLISH on REALTIME_CHANNEL per event
Swallows per-row errors; logs at error and continues the batch.
buildEvent(row) (private)
publishRelayed
REALTIME_EVENT_MAP via realtimeEventKey
None
Allocates a monotonic id via redis.incr(REALTIME_SEQUENCE_KEY)
Returns null for an unmapped event or a classify() result of null.
persist(outboxEventId, event) (private)
buildEvent's caller (publishRelayed)
None
One notification row, onConflictDoNothing on outbox_event_id
None
A failed insert is caught by the caller's try/catch and logged; the publish still proceeds.
dispatch(event) (private)
publishRelayed
None
Redis RPUSH + LTRIM on REALTIME_HISTORY_KEY (bounded to 500), then PUBLISH
Fans the event to every subscribed API instance
None thrown directly; failures propagate to the caller's catch.
Explain:
Transaction boundaries. None of this runs inside the outbox's own transaction. It runs after commit, so a failure here can never roll back a dispatch that already enqueued a job.
Idempotency. The unique outbox_event_id constraint on notification makes a redelivered outbox row a no-op insert. The Redis publish itself is not deduplicated — a redelivery still fans out a second live frame, which the client de-duplicates on id where possible, or simply sees as a second toast for the rare redelivery case.
Ordering. History is written before the pub/sub publish, deliberately: a client that reconnects in the gap between the two would otherwise be told (via a Last-Event-ID it saw live) about an id it can no longer replay. Reversing this order can at worst replay an event the client already received live, never lose one.
Fail-open. Every failure here is caught, logged, and does not propagate — publishing a notification is an enhancement, never allowed to fail the underlying business write it rides on.
Opens a dedicated Redis subscriber connection, subscribes to REALTIME_CHANNEL and REALTIME_CONTROL_CHANNEL, and dispatches each message by channel
A malformed frame on either channel, or an unknown control message, is logged at error and dropped, not thrown.
canReceive(subscription, event)
deliver(), replay(), the per-connection listener
In-memory subscription state
None
None
Pure.
replay(subscription)
attach()
Redis history list
None
None
A Redis error propagates to the caller, which logs it and still opens the live stream.
attach(response, subscription)
RealtimeAdminController.stream()
Redis (via replay)
None
Writes SSE headers, ready, replay frames, event frames, heartbeat comments; registers a listener and a closer, the closer also under subscription.userId; returns that closer
Guards on shuttingDown (returns a no-op closer); closes the connection on a slow-consumer or transport error.
closeUserStreams(userId)
The control-channel handler, on close-user
The per-user closer index
None
Ends every open stream of that user on this instance; returns how many
None.
currentActiveRoleId(sessionId)
RealtimeAdminController.stream(), right after attach
sessions.active_role_id for the session
None
None
A database error propagates to the caller, which closes the stream.
deliver(event)
The event-channel handler
In-memory listener set
None
Calls every registered listener synchronously
None.
beforeApplicationShutdown()
Nest lifecycle
In-memory closer set
None
Ends every open response before server.close() waits on them
None.
RealtimeSubscription carries permissions, bypassPermissions, an optional lastEventId, and — both required — the subscriber's userId (what close-user matches) and sessionId (what the attach re-check reads).
Explain:
Why a dedicated Redis connection. ioredis puts a client into subscriber mode on SUBSCRIBE, and a subscriber-mode client cannot run ordinary commands. Sharing the app's main client would silently break every cache read process-wide — a failure mode indistinguishable from "Redis is down." replay() still reads through the shared client; only the message listener uses the dedicated one.
Fan-out granularity. One Redis subscription per API instance serves every connected admin on that instance; the per-connection cost is a permission filter and a listener callback, not a second subscription.
Backpressure.response.writableLength (bytes queued, not yet flushed to the socket) is checked after every write. Above REALTIME_MAX_BUFFERED_BYTES (1 MiB) the connection is closed — the connection is expendable, the process is not.
Response object, not @Sse(). Nest's @Sse() decorator serialises every emission as a data: frame (so a heartbeat would look like an event to the client) and delivers through an rxjs Subject, which is synchronous and defeats any buffered-bytes check. Writing the response directly also removes this route from the global interceptors' per-response logging, so a long-lived connection does not log a line per frame.
Shutdown ordering.app.close() runs beforeApplicationShutdown before server.close(). server.close() waits for every open connection to finish, and an SSE stream never finishes on its own — so ending every response here is what lets a deploy's SIGTERM complete instead of requiring a force-kill.
Why a stream is closed when its user loses a role. Permissions are resolved once, at connect. A stream left open after its person lost a role would keep delivering what that role could read until it happened to reconnect. Ending it makes EventSource reconnect after retry: 3000, and the reconnect resolves the permissions the person holds now — or is refused if they hold none in that session. A role-change close is logged at log, not warn: it is routine.
Registered before the re-read.attach files the stream under its userId synchronously, before returning the closer. The controller then reads the session's active role again. A role removal committed before that read is seen by it; one committed after it finds the registered stream through the control channel. Registering after the read would leave a window neither covers.
Opens its own Redis connection with createRedisClient: no offline queue, the configured connect timeout and the configured command timeout
Throws if either timeout is missing from config, which the validated config never allows (both default).
closeUserStreams(userId)
StaffService after a role edit that removed a role commits
None
None
PUBLISH of { kind: "close-user", userId } on REALTIME_CONTROL_CHANNEL
Never throws. A failure — Redis unreachable, reconnecting, or slow past the timeout — is logged at warn naming the user and what the loss leaves behind.
onApplicationShutdown()
Nest lifecycle
None
None
Quits its connection
None.
Its own connection, failing fast. The shared client queues commands while Redis is unreachable and retries without limit, so a publish through it during an outage would hold the request that triggered it until Redis returned. This connection refuses instead. The caller's write has already committed; the close is the part that may be lost.
Both bounds come from config. They are declared in config/env.validation.ts with defaults and listed in apps/api/.env.example; see the table below.
A lost message is accepted. Pub/sub carries no delivery guarantee. A close-user published while Redis is down, or while an instance's subscriber is reconnecting, is lost; that instance's streams for the person keep their connect-time permissions until they next reconnect. While Redis is down no events flow on the stream at all, so the gap matters only once it recovers.
Publishes, never subscribes. It is provided by RealtimeControlModule, which registers nothing that subscribes.
Variable
Type
Default
What it bounds
REALTIME_CONTROL_CONNECT_TIMEOUT_MS
Positive integer, milliseconds
2000
How long one connection attempt to Redis may take before it is abandoned and retried.
REALTIME_CONTROL_PUBLISH_TIMEOUT_MS
Positive integer, milliseconds
2000
How long one PUBLISH may take before it fails and is logged at warn.
Both are optional. A value that is not a positive integer fails validation at boot.
Covered as flow 5.1 in the Features and Flows doc; the code path is RealtimeAdminController.stream() → RealtimeStreamService.attach() → replay(), then currentActiveRoleId().
None. This module has no queue, job, or scheduler of its own. It is invoked synchronously (in-process, not via a queue) by the outbox dispatcher's relay step, after that dispatcher's own retry and dead-letter handling has already run.
RealtimeEvent with kind: "feedback.submitted", soundClass: "soft"
Admin panel stream listener
At-least-once from the outbox; deduplicated at persistence, not at publish.
exam:exam.published
Exam module, via the outbox
Every admin holding Exams_READ
soundClass: "neutral"
Admin panel
Same as above.
exam:exam.result_published
Exam module, via the outbox
Every admin holding ExamResults_READ
soundClass: "positive"
Admin panel
Same as above.
exam_marks:exam.marks_submitted
Exam module, via the outbox
Every admin holding ExamMarksVerify_READ
soundClass: "soft"
Admin panel
Same as above; the permission is the verifier's read code, not the marks read code, because the toast exists to route work to whoever can act on it.
One control message travels beside these events and is never delivered to a subscriber:
Message
Producer
Target
Payload
Consumer
Reliability Notes
close-user on realtime:control
StaffService, after a role edit that removed a role commits
Every open stream of that user, on every API instance
{ kind: "close-user", userId }
RealtimeStreamService
At-most-once pub/sub, not the outbox. A lost message leaves the streams on their connect-time permissions until they reconnect; the loss is logged.
The map also carries dormant infrastructure for a classify()-driven inventory reconciliation event (RECONCILIATION_REASONS, LOSS_REASONS, SILENT_REASONS, REALTIME_RECONCILIATION_URGENT_DELTA), but no inventory module exists in this repository today, so no outbox key currently triggers it.
Guards.JwtAuthGuard and RoleGuard, both global-by-default and explicitly listed on RealtimeAdminController.
Permissions. The catalogue route requires System_READ. The stream route declares no @Permissions() — it only requires an authenticated admin with an active role — because the per-event filter in RealtimeStreamService.canReceive is the actual boundary, applied per event rather than gating the whole feature behind the widest permission it carries.
Rate limiting. The catalogue route carries IpThrottlerGuard with RATE_LIMITS.ADMIN_READ. The stream route carries no IP throttle — it is one long-lived connection per session, not a request rate to bound.
Fail-closed on missing identity. A request reaching stream() with no activeRole throws 403 PERMISSION_USER_CONTEXT_MISSING before any Redis or listener work happens.
A stream does not outlive a removed role. The attach re-check closes a stream whose session's active role changed after authentication, and a staff role removal closes the person's open streams through close-user. Both make the browser reconnect, and the reconnect is authorised afresh.
No credential in the URL. The route is documented and designed to be opened server-to-server (by the admin panel's Next.js BFF, which holds the session cookie) using an ordinary Authorization: Bearer header — never a token in a query string, which would otherwise land in access logs and Referer headers.
Audit.ActivityAuditInterceptor is global and covers mutating requests; the stream and catalogue routes are reads and are not separately audited by this module.
No other module-specific error code exists. JwtAuthGuard/RoleGuard produce their own standard 401/403 responses for missing or invalid auth, and a RoleGuard permission failure on the catalogue route produces the platform's standard permission-denied response.
The only channel for a swallowed publish failure — nothing else can see it.
Log (error)
RealtimeStreamService Redis message handler
An unparseable frame on either channel, or an unknown control message.
Log (error)
RealtimeStreamService.attach replay .catch
A failed history read on connect.
Log (warn)
RealtimeStreamServiceclose()
A connection closed for a reason other than a normal client disconnect, shutdown or role change (chiefly the slow-consumer bound, a transport error, or a failed attach re-check).
Log (log)
RealtimeStreamServiceclose()
A connection closed because its user's roles changed — by close-user or by the attach re-check.
Log (warn)
RealtimeControlPublisher.closeUserStreams
A close-user that could not be published; names the user whose streams keep their old permissions until they reconnect.
Log (log)
RealtimeStreamService.beforeApplicationShutdown
How many streams were ended for a deploy shutdown.
Metric
RealtimeStreamService.connectionCount getter
Live connections on this instance; exposed for tests and the health surface, not wired to an external metrics sink in this module.
Event map shape, classify() behavior for the inventory reconciliation thresholds, realtimeEventKey.
Unit
shared/realtime-stream.service.spec.ts
canReceive, replay filtering and slicing, backpressure closing, heartbeat, shutdown draining, frame formatting, closing one user's streams and nobody else's, the closer attach returns.
Integration
shared/realtime-publisher.int.spec.ts
End-to-end publishRelayed against a real database and Redis: persistence, dedupe on redelivery, history trimming, publish.
Integration
shared/realtime-control.int.spec.ts
Against a real Redis and Postgres: close-user ends both streams of one user and leaves another user's open; a malformed control message is ignored and the channel keeps working; the stream route keeps a stream whose session still acts as the resolved role, and ends one whose active role was removed or switched before it registered.
Integration
people/staff/staff.service.spec.ts
A staff role removal closes the person's open stream through real Redis and leaves a bystander's open; an edit that only adds a role publishes nothing; a failed publish does not fail the edit.
Validation commands: pnpm --filter @skoolsewa/api test, or narrow to this module with pnpm --filter @skoolsewa/api test -- realtime. The *.int.spec.ts files run with pnpm --filter @skoolsewa/api test:int.
Allowlist (REALTIME_EVENT_MAP) instead of relaying every outbox event
The outbox carries the platform's entire internal event traffic
Only mapped events reach the stream
Relay everything and filter downstream
Prevents payload leakage and unreviewed notifications
Every new event kind needs a code change
A new domain module lands and wants a toast
One notification row per event, fan-out on read
Avoids a row-per-recipient write multiplying by staff count
Filter by permission at read/connect time
Insert one row per eligible recipient at publish time
No write amplification; permission changes apply retroactively to old rows on next read
A permission grant made after the event still shows it on the next fetch, which is accepted as correct here
Never — documented as the intended behavior
Resolve permissions once per connection
Avoids a role lookup on every fan-out for every connected admin
Cache the permission set for the connection's lifetime, and force a reconnect when a staff role is removed (close-user)
Re-resolve per event
Cheap fan-out
A change to a role's own permissions on the Roles screen, a role's deletion, or a pupil's graduate access still takes effect only on the next reconnect
If one of those paths needs instantaneous revocation mid-stream
close-user over plain pub/sub, not the outbox
A role removal must reach every instance's open streams
Best-effort Redis publish after commit, from a fail-fast connection
An outbox event per removal; closing local streams as a fallback on publish failure
One path for every close, and a write that never waits on Redis
A close published during an outage or a subscriber reconnect is lost, and the streams keep their old permissions until they reconnect
If a missed close must be recovered rather than logged