Socket Service

Implementation reference for the socket. For its architectural role, see Socket Service Architecture.

A small Rust service built on Axum, socketioxide, and a Redis subscriber. It is the bridge between the broker and the browser: it relays alerts and per-building telemetry to connected clients over WebSockets. It has no database and no domain model. The code follows a functional core / imperative shell split, expressed as the directory layout: src/core/ (rooms.rs, auth.rs, relay.rs) is pure and carries every unit test; src/shell/ (handlers.rs, server.rs, metrics.rs) touches sockets, Redis, and process state; src/main.rs only binds the port. The split is enforced by tests/architecture.rs — see Socket Service Architecture.

Connections are authenticated: a socketioxide connect middleware decodes the mesh-verified claims header before the namespace connection is accepted, so the service knows which account and domains each socket belongs to.

graph TD
    MAIN["main.rs\nbind + shutdown signal"] --> SRV["shell/server.rs\nrouter, CORS, Redis loop"]
    SRV --> HND["shell/handlers.rs\nconnect middleware, room joins, emits"]
    SRV --> REL["core/relay.rs\npure: telemetry_delivery, notification_delivery"]
    HND --> AUTH["core/auth.rs\npure: authenticate_claims_header"]
    HND --> ROOMS["core/rooms.rs\npure: room_for_building, room_for_domain, building_id_from_channel"]
    REL --> ROOMS
    SRV --> METRICS["shell/metrics.rs\nregistry + counters"]

Configuration

The HTTP/WebSocket server listens on port 3000. A GET /health endpoint backs the Kubernetes probes.

Metrics

GET /metrics exposes the Prometheus text format. Every series is registered at startup (metrics::init(), called from serve), so a freshly started pod reports each one at 0 rather than omitting it — a counter that only appears after its first increment leaves gaps in rate() and makes “no traffic” indistinguishable from “not scraped”.

Authentication

A socketioxide connect middleware (on_connect.with(authenticate)) runs once per connection, before any event handler:

• It reads the x-gateway-claims header from the handshake request. Istio’s RequestAuthentication verifies the gateway JWT exactly once, at the mesh ingress — the browser authenticates via its authentication_token cookie on the handshake’s initial HTTP request, and Istio extracts the JWT from that cookie before injecting the verified payload as this header.
• authenticate_claims_header (in auth.rs, pure and unit-tested) base64-decodes the header and extracts an Identity: account_id, account_name, and the domains the account is a member of (from the payload’s memberships). There is no signature check here — socket trusts the header rather than re-verifying a JWT itself, the same trust model as every other service behind the edge (see Identity & Tenancy Architecture.
• A missing or malformed header rejects the namespace connection with unauthorized; the client sees a connect_error. On success the identity is stored in the socket’s extensions so the connection is server-authoritative — the client cannot claim rooms it has no right to.

Broker Subscriptions and Emissions

At startup the service connects a Redis subscriber and binds two channels to two Socket.IO events:

For the pattern subscription, the building id is recovered from the channel name via telemetry_schema::building_of_filtered_channel — the prefix is written by dashboard and read here, so it is defined once — and used to target the room. Each incoming message is JSON-parsed before emission; a message that fails to parse is logged and skipped, so one bad publish cannot take the relay down.

The subscriber runs in a reconnect loop: if the initial connection fails (Redis not up yet at pod start) or the subscription later drops (Redis restart, connection killed), it logs, waits one second, and reconnects — indefinitely. This is not optional polish. The Node implementation got it for free from node-redis, which auto-reconnects by default; a single-shot connect is a silent black hole, because /health keeps returning 200 while no telemetry reaches any browser.

Notifications are domain-scoped: if the payload carries a domainName, only that domain’s room receives it, so members of one tenant never see another’s alerts. A payload without a domainName (a system-wide message) is broadcast to every client. The payload must parse as notification_schema::Notification to be routed at all; anything else is skipped, so a renamed domainName can never turn a scoped alert into a broadcast — see Notification Schema.

Client Subscription Protocol

Domain rooms are not client-controlled: on connection the service reads the verified identity and joins the socket to a domain:{name} room for each of the account’s memberships. The client only chooses which buildings to follow:

sequenceDiagram
    participant UI as Vue client
    participant WS as socket
    participant TW as digital-twin
    participant R as Redis
    UI->>WS: connect (authentication_token cookie)
    Note over WS: Istio verifies the JWT, injects x-gateway-claims — WS decodes it, joins domain:{...} rooms
    UI->>WS: subscribe_building(B)
    WS->>TW: GET /domain/B (cached 60s)
    TW-->>WS: ["acme", ...]
    Note over WS: shares a domain? → socket joins room building:B
    WS-->>UI: ack {subscribed: true, buildingId: B}
    R-->>WS: telemetry:filtered:B message
    WS->>UI: emit "telemetry" (room building:B only)
    R-->>WS: notifications message (domainName=D)
    WS->>UI: emit "notification" (room domain:D only)

The join is not immediate — it waits on the digital-twin lookup — so subscribe_building is acknowledged. Anything published to building:{buildingId} before the join lands is emitted to a room the caller has not entered, and socket.io drops it silently: no buffer, no replay. A client that wants no gap should wait for the ack, then take its REST snapshot, then apply live events. The Vue client emits without waiting and is covered by its snapshot refetch (stores/sensorData.ts), which backfills anything missed in that window; a client that only listens has no such safety net.

A socket is dropped once it has been open longer than SOCKET_MAX_LIFETIME_SECS, counted as socket_sessions_expired_total. socket.io-client does not retry a server-forced disconnect, so services/socket.ts reconnects explicitly on reason io server disconnect and useSensorData’s connect handler re-emits subscribe_building; the re-handshake re-reads the session cookie without any user interaction. A building room is only ever joined after the caller’s memberships are checked against the building’s owning domains, so a client cannot follow another tenant’s building by guessing its id. Telemetry is room-scoped so each browser receives only the buildings it is displaying; notifications are scoped to the recipient’s domain so tenants stay isolated. Only a system-wide notification with no domainName is broadcast to everyone.

Scaling Considerations

The service runs as a single Socket.IO process. Two changes are required to run multiple replicas:

• Sticky sessions. The HTTP long-polling handshake must return to the same instance; the Kubernetes ingress sets cookie affinity for this (see Deployment & Kubernetes.
• Cross-instance rooms. With multiple replicas a room spans instances, so emissions must propagate via the Socket.IO Redis adapter. This is a known extension point, not yet in place.

For where this service sits in the end-to-end real-time path, see Communication & Data Flow.