ThingsBoard Integration Runbook

Scope

This runbook covers the Task 6 observability baseline for monitoring stream integration:

reconnect attempts and bounded reconnect delay
backpressure depth baseline
heartbeat hooks from upstream stream events

Metrics Baseline

MonitoringObservabilityService tracks per-stream-scope metrics:

reconnectAttempts: increments when upstream disconnect hook fires
reconnectDelayMs: bounded exponential delay using constants:
- MONITORING_RECONNECT_BASE_DELAY_MS
- MONITORING_RECONNECT_MAX_DELAY_MS
heartbeatAt: updated on payload receive and heartbeat hook
backpressureDepth: current listener depth baseline for each stream scope
lastEventLatencyMs / eventCount: ingest-to-broadcast latency and a
cumulative event counter (see “Latency and Throughput Metrics” below)

Reconnect Behavior

Reconnect delay formula:

delay = min(BASE_DELAY * 2^attempt, MAX_DELAY)
attempt is zero-based for delay calculation

Operational effect:

first disconnect records base reconnect delay metric
subsequent disconnects record exponentially increased delay metric
recorded delay never exceeds configured max cap

ThingsboardIntegrationAdapterService uses the same backoff formula to drive an
actual reconnect, not just the metric: on WebSocket close, it schedules a
setTimeout using the computed delay, then fetches a fresh JWT, opens a new
socket authenticated via that JWT in the connection URL (see “WebSocket Wire
Protocol” below), and resends a tsSubCmds subscribe command for every scope
that was still active (i.e. not explicitly unsubscribed). If the reconnect’s
JWT fetch fails, the error is logged and another reconnect attempt is
scheduled with the next backoff step — since authentication now happens
before any socket is opened, a failed JWT refresh never leaves a dangling,
unauthenticated socket to tear down; there is simply nothing to open yet.

The attempt counter deliberately does not reset on the socket’s open
event: reaching open proves the connection was accepted, but not yet that
ThingsBoard is actually pushing data. The counter only resets once an actual
message arrives on the socket — the first point a reconnect is proven to be
genuinely receiving data again — after which a stable connection returns to
the base delay on its next disconnect.

onModuleDestroy clears activeSubscriptions before closing the socket. This
matters because ws.close() only fires the close handler asynchronously
once the close handshake completes — without clearing subscriptions first,
that handler would still see them as active and schedule a reconnect after
the provider has already been torn down (e.g. during app shutdown or test
teardown), leaking a socket/timer past the module’s lifecycle.

Fixed: concurrent connects and stale close events

A Codex review flagged two socket-identity races, both made materially more
likely by the per-device fan-out redesign above (N subscribeToScope calls
now fire synchronously for a single multi-device scope, instead of at most
one):

Duplicate sockets during JWT fetch. subscribeToScope’s “no socket
yet” branch used to call getJwtToken().then(openSocket) independently for
every caller. getJwtToken() dedupes the login request itself, but each
attached .then() still ran openSocket, so N concurrent subscribers
opened N WebSockets that each resent tsSubCmds for every active
subscription — duplicate telemetry delivery, and N-1 leaked sockets (only
the last one survives in this.ws). Fixed by adding a private connect()
helper that gates the JWT-fetch-then-open sequence behind a single
in-flight connectPromise; both subscribeToScope and reconnectSocket
now go through it, so only one socket is ever created for concurrent
callers.
Stale close events discarding a live replacement socket. If a socket
emits error, the handler closes it and immediately nulls this.ws so a
new subscription can open a replacement right away — but the errored
socket’s own close event still fires later (asynchronously). That stale
close handler used to unconditionally null this.ws and call
scheduleReconnect(), which — if a fresh replacement socket had since been
opened — wiped out the live socket reference and scheduled an unnecessary
extra reconnect. Fixed by capturing the specific ws instance in each
handler’s closure and only acting on close when this.ws === ws (i.e.
the event’s socket is still the current one); a stale event from an
already-replaced socket is now a no-op.

Fixed: stale reconnect timer duplicating an already-restored connection

A scheduled reconnect timer could still open a redundant, leaked second
socket even after the two races above were fixed. Sequence: a disconnect
schedules a reconnect after the current backoff delay (this.ws is null
during that window); before the timer fires, a brand-new subscribeToScope
call arrives (e.g. a client opens the monitoring page for another device)
and — since no socket exists yet — immediately re-establishes the
connection via connect(). reconnectSocket() (the timer’s callback) had no
way to know this had happened: it unconditionally called connect() again
once its delay elapsed, and since connect() only dedupes in-flight
attempts (via connectPromise, already resolved by the earlier subscribe),
it opened a second, redundant WebSocket alongside the one already live,
duplicating telemetry delivery for every active scope. Fixed by having
reconnectSocket() bail out early if this.ws is already set, since that
means the connection was already restored through another path and this
scheduled attempt is now stale.

Fixed: WebSocket errors now trigger a reconnect, not just a teardown

The socket’s error handler closed the errored socket and cleared this.ws,
but never scheduled a reconnect itself — it relied on the subsequent close
event to run that logic. The close handler’s stale-socket guard
(this.ws !== ws, added for the race above) defeated this: since the error
handler had already nulled this.ws by the time close eventually fired for
the same socket, close always treated it as a stale event from an
already-replaced instance and returned early, even though no replacement had
actually been created. Existing subscriptions were then left with no socket
and no scheduled reconnect until an unrelated new subscribeToScope call
happened to arrive — a common transient error (e.g. ECONNRESET) silently
stopped telemetry forever for already-connected clients. Fixed by extracting
the disconnect-hooks-plus-reconnect logic into a shared
handleSocketTerminated() helper and calling it directly from the error
handler (guarded by this.ws === ws, so it’s skipped if this socket was
already superseded), instead of waiting on an unreliable close event.

Stream Hook Integration

MonitoringStreamService connects observability to subscription lifecycle:

on attach: initialize backpressure depth baseline
on upstream payload: mark heartbeat and refresh backpressure depth
on upstream disconnect: increment reconnect metric and compute capped delay
on detach: update backpressure depth

Fixed: per-device fan-out no longer amplifies reconnect/error/heartbeat counters

ThingsboardIntegrationAdapterService’s onDisconnect/onHeartbeat/onError
hooks report on the single shared ThingsBoard WebSocket connection itself
(fired from connection-wide disconnectHooks/heartbeatHooks/errorHooks
Sets), not on any one device’s own per-cmdId subscription. After the
per-device fan-out redesign, MonitoringStreamService.openDeviceSubscriptions
registered these hooks once per device in a multi-device scope, so a single
real disconnect/error event fired all of them — a 3-device organization’s
reconnectAttempts/errorCount jumped straight from 0 to 3 on one outage,
immediately crossing MONITORING_RECONNECT_ALERT_THRESHOLD/
MONITORING_ERROR_ALERT_THRESHOLD (3) after a single event instead of after
genuinely repeated ones. Fixed by registering the hooks only once per scope
(on the first device with a ThingsBoard mapping) — since the hooks are
connection-wide regardless of which device’s subscription call carries them,
this reports the exact same signal without amplification.

Fixed: one org’s telemetry could reset another org’s degraded status via a shared heartbeat

Unlike onDisconnect/onError (genuinely connection-wide events — the whole
shared WebSocket going down or erroring really does affect every currently
subscribed org), a heartbeat is precisely attributable: every incoming
message carries a subscriptionId that identifies exactly one cmdId’s
subscription. Despite this, onHeartbeat was previously stored in the same
kind of connection-wide Set as the other two hooks and fired for every
registered scope on any parsed message, regardless of which subscription it
actually belonged to. Since MonitoringObservabilityService.markHeartbeat
resets reconnectAttempts/errorCount back to zero, this meant one
organization’s telemetry (message flowing normally) silently reset a
completely different, genuinely degraded organization’s counters back to
healthy purely because both share the one physical connection — a real
masking risk in any multi-tenant deployment with regular cross-tenant
traffic. Fixed by keying heartbeat hooks per-cmdId
(heartbeatHooksByCmdId: Map<number, () => void>) instead of a shared Set,
and firing only the hook for the specific cmdId a message’s subscriptionId
identifies (inside handleIncomingMessage, once the subscription is
confirmed to still have a registered handler). onDisconnect/onError
remain intentionally connection-wide, since there is no better attribution
available for those events. This also means a malformed frame still never
counts as a heartbeat (the lookup only runs after successful JSON.parse
and subscriptionId/handler resolution), preserving the existing error-alert
threshold behavior.

Fixed: device soft-delete now tears down its ThingsBoard fan-out subscription

devices.service.ts’s soft-delete (remove()) previously had no wiring into
the monitoring module at all: an already-open per-device ThingsBoard WS
subscription (opened lazily the first time a monitoring client resolved this
device into its scope) kept delivering live telemetry for the device
indefinitely after it was deleted, leaking an adapter-level cmdId
subscription forever. The frontend-side symptom (a deleted device’s stale row
reappearing) was fixed earlier by filtering live events against
scopedDeviceIds, but the upstream subscription itself kept running.

Fixed by adding MonitoringStreamService.evictDevice(internalDeviceId),
called from DevicesService.remove() right after the soft-delete commits.
It looks up every currently-active scope (there is normally at most one per
organization) for a subscription matching that device id, tears down just
that device’s adapter-level subscription, and — if it was the last device
still subscribed in that scope — also tears down the scope’s shared
connection hooks and clears its observability metrics, mirroring the normal
all-listeners-disconnected teardown path. A multi-device organization scope’s
other devices are completely unaffected.

Implementing eviction required a small refactor first: previously,
onDisconnect/onError were registered by tying them to whichever device’s
subscribeToScope call happened to run first in a scope’s fan-out loop (see
“Fixed: per-device fan-out no longer amplifies…” above). That meant evicting
that specific device — plausible, since eviction has no reason to prefer
one device over another — would have silently killed the whole scope’s
disconnect/error observability even though its other devices’ subscriptions
kept running. Since onDisconnect/onError are genuinely connection-wide
(not attributable to any one device), they’re now registered independently of
any device subscription via a new
ThingsboardIntegrationAdapterService.registerConnectionHooks() method,
called once per scope in attachClient regardless of device count.
onHeartbeat remains registered per-device (it’s attributable per-cmdId,
and every device’s own real traffic is valid independent proof the scope is
alive), so it now fires once per device with data flowing rather than once
per scope — a deliberate, harmless change (heartbeat resets are idempotent).

Fixed: shutdown during an in-flight initial JWT fetch could leak a socket

connect() (the shared JWT-fetch-then-openSocket sequence used by both
subscribeToScope and reconnectSocket, see “Concurrent Connects” below) had
no awareness of provider shutdown: if onModuleDestroy() ran while the very
first getJwtToken() call for a scope was still in flight (e.g. app shutdown,
or a Nest testing module torn down mid-test, during a slow first
subscribeToScope() login), onModuleDestroy() would set destroyed = true
and close this.ws — but this.ws was still null at that point, since no
socket had opened yet. Once the pending login resolved afterward, connect()'s
.then((token) => this.openSocket(token)) continuation still ran unconditionally
and opened a brand-new WebSocket that nothing would ever close, leaking a
connection and potentially keeping the process (or a test’s open-handle
detection) alive past shutdown. Fixed by checking this.destroyed inside that
continuation before calling openSocket(), so a login that resolves after
shutdown is a no-op instead of opening an orphaned socket.

Frontend Reconnect Behavior

MonitoringPage merges realtime monitoring:event pushes into its visible
telemetry table (keyed by deviceId:metric) on top of the initial
GET /monitoring/snapshot fetch. A dropped WebSocket connection means any
telemetry emitted during the outage is never pushed to that client, so the
table would otherwise silently go stale until a manual page reload.

To close that gap, MonitoringPage tracks isConnected transitions and
calls refetch() on the snapshot query whenever the stream reconnects after
having been connected before (i.e. a real drop-and-recover, not the initial
mount). This backfills whatever telemetry was missed during the outage
without needing server-side event buffering/replay. The very first
connection is intentionally excluded from this refetch since the initial
snapshot fetch already covers it.

MonitoringPage’s per-key merge also guards against out-of-order delivery:
before overwriting the row for a deviceId:metric key, it compares the
incoming event’s ts against the currently-shown row’s ts and drops the
incoming event if it is older. Realtime transports can redeliver an older
message after a newer one (e.g. socket.io buffering across a reconnect, or
network-level reordering), and without this guard the table would silently
regress to a stale value — this closes issue #107’s “thứ tự thông điệp”
(message ordering) acceptance criterion for the frontend display layer.

The same ordering guard also applies to the reconnect-triggered snapshot
refetch above: when the refetch resolves, MonitoringPage merges each
returned row into the table per-key by ts instead of unconditionally
replacing the whole table. Without this, a snapshot response that was
already stale by the time it arrived (e.g. it raced a monitoring:event
push that landed first, right after reconnect) could silently overwrite a
newer value already shown on screen.

useMonitoringStream also surfaces a monitoring:error push from
MonitoringStreamGateway (FORBIDDEN_SCOPE when the resolved scope is
rejected, SUBSCRIBE_FAILED on any other subscribe error) as a streamError
field. MonitoringPage renders it as a dismissible-by-recovery Alert
banner above the existing telemetry table (which keeps showing whatever data
it already has), and the banner clears automatically once a subsequent
monitoring:event arrives. Previously these gateway-pushed errors were
silently dropped — the socket stayed marked “Connected” with no telemetry
ever arriving and no indication why, leaving issue #109’s “UI xử lý được …
error state” acceptance criterion only partially covered (empty/disconnected
states were handled, but not an explicit stream-level error).

streamError is also cleared the moment a fresh monitoring:subscribe
attempt is made on the socket’s connect handler (not only once a new
monitoring:event arrives). The gateway sends no explicit success
acknowledgment for a subscribe — success is implicit — so without this, a
stale error banner from a prior failed attempt (e.g. FORBIDDEN_SCOPE)
could keep showing after a reconnect resolved the issue, simply because no
new telemetry had arrived yet to clear it.

A SUBSCRIBE_FAILED error is a different failure mode than a dropped
connection: the WebSocket transport itself stays connected (only the
application-level monitoring:subscribe attempt failed, e.g. a transient
error in MonitoringScopeResolverService), so no connect/disconnect
event will ever fire to give the client another chance to retry. Without a
dedicated retry, a transient failure here left the user stuck on the error
banner indefinitely with no live telemetry until a full page reload.
useMonitoringStream now re-emits monitoring:subscribe after a
SUBSCRIBE_FAILED error using a bounded exponential backoff (3s, 6s, 12s,
24s, capped at 30s, cancelled if a monitoring:event or a fresh connect
happens first), giving up after 5 consecutive failures with no intervening
success. The retry budget resets to zero the moment a connect or
monitoring:event proves the stream healthy again. Earlier this was a
fixed 3-second interval with no cap, so a persistent (non-transient) scope
resolver failure — as opposed to the transient case this retry is meant for
— would retry forever, indefinitely hammering the server with no backoff.
FORBIDDEN_SCOPE is deliberately excluded from this retry: it reflects a
permission/tenant-scope decision that a retry cannot change, so retrying
would just repeat the same rejection.

The snapshot-driven merge effect also preserves rows that are absent from a
given GET /monitoring/snapshot response instead of dropping them.
MonitoringQueryService.getSnapshot fans out one telemetry fetch per device
via Promise.allSettled and silently omits any device whose fetch fails
(logged as a warning server-side, with no per-device error surfaced in the
response body) — so a device missing from data[] does not necessarily mean
it left the organization’s scope, only that this particular fetch didn’t
return data for it. Merging into the existing map (updating/adding only the
keys present in the new snapshot, keeping every other key untouched) means a
transient per-device fetch failure — most likely to be hit right after the
reconnect-triggered refetch above — can no longer erase an already-live-
tracked telemetry row from the table.

Preserving every row unconditionally traded one bug for another: a device
that genuinely left scope (deleted, unmapped from ThingsBoard, or moved out
of the organization) would keep showing its last known telemetry forever,
since nothing ever told the frontend the difference between “still in scope,
this fetch failed” and “actually gone.” Closed by having
MonitoringQueryService.getSnapshot additionally return
scopedDeviceIds — every non-deleted device the resolver currently maps for
the caller’s scope, independent of whether that device’s own telemetry fetch
in this response succeeded (see MonitoringSnapshotDto). MonitoringPage
uses this as the authoritative eviction signal: after merging in whatever
data[] contains (still never dropping a row just because it’s merely
absent from data[]), it removes any row whose deviceId is not in
scopedDeviceIds. A transient per-device fetch failure never affects
scopedDeviceIds (the device is still resolved, only its telemetry read
failed), so it still can’t evict a row — only a device the resolver/repository
no longer returns at all can.

Fixed: a stale live push could resurrect a row already evicted by scopedDeviceIds

scopedDeviceIds-driven eviction (above) only runs inside the snapshot-merge
effect, which fires on mount and on each GET /monitoring/snapshot refetch —
but at the time this fix landed, the live WebSocket subscription behind an
already-connected client was never torn down just because a device left scope
after that client subscribed: devices.service.ts’s soft-delete (remove())
only set deletedAt and had no dependency on the monitoring module, so a real
ThingsBoard device that still reported telemetry could keep pushing
monitoring:events to an already-open client indefinitely for a device the
backend had otherwise evicted everywhere else. That backend gap was closed
afterward — see “Fixed: device soft-delete now tears down its ThingsBoard
fan-out subscription” above, which adds MonitoringStreamService.evictDevice.
This frontend guard remains necessary regardless: evictDevice’s unsubscribe
races the WS delivery path, so a push ThingsBoard sent just before the
teardown took effect can still arrive here after eviction, and
MonitoringScopeResolverService’s deletedAt IS NULL filter (see
“Soft-Deleted Device Exclusion”) only guards new scope resolutions —
initial subscribe, RPC dispatch, REST snapshot/history — not a stale message
already in flight on an existing subscription.

Before this fix, MonitoringPage’s live-event merge effect applied no scope
check at all, so one of these stale pushes would silently re-add the exact
row the snapshot-merge effect had just evicted, undoing the eviction
guarantee the moment the next stray event arrived. Fixed by having
MonitoringPage cache the most recently seen scopedDeviceIds (as a Set,
via scopedDeviceIdSetRef) from the snapshot-merge effect and having the
live-event merge effect reject lastEvent.deviceIds that are outside it,
mirroring the snapshot-driven eviction guarantee against live pushes too.
This is safe against false negatives: a client can only ever receive a
lastEvent for a device that was in scope when this connection’s per-device
fan-out subscriptions were opened, and scopedDeviceIds only stops listing
that device once it has genuinely left scope — so the filter can never
suppress a legitimately still-in-scope device’s telemetry, including a
brand-new device (which the backend was never subscribed to on this
connection in the first place, so it can’t emit a lastEvent here at all
until the client resubscribes and a fresh snapshot picks it up).

This does not close the underlying architectural gap — the per-device
upstream ThingsBoard subscription for an evicted device stays open
server-side, wasting a small, bounded amount of adapter/observability
resources until the client disconnects or resubscribes — only the
user-visible symptom (a deleted device’s stale telemetry reappearing in the
UI). Properly closing the server-side leak needs devices.service.ts to
notify MonitoringStreamService on delete/unmap so it can detach just that
device’s fan-out subscription without disturbing the rest of the org’s
scope; tracked as a Known Gap below rather than attempted here, since it
requires wiring a new cross-module dependency and no other server-side
correctness or tenant-isolation issue depends on it.

Verification Checklist

Run targeted specs:
- make backend:test FILE=backend/src/monitoring/services/monitoring-observability.service.spec.ts
- make backend:test FILE=backend/src/monitoring/services/monitoring-stream.service.spec.ts
- make backend:test FILE=backend/src/monitoring/services/thingsboard-integration-adapter.service.spec.ts
Ensure tests validate reconnect increment after upstream disconnect hook and stream lifecycle hook wiring.
Ensure the adapter spec’s backoff test validates that the socket is not
recreated before the computed delay elapses, and that active scopes are
resubscribed once it reconnects.

REST Retry Behavior

Idempotent ThingsBoard REST calls (JWT login, getLatestTelemetry,
getTelemetryHistory) retry transient failures — network errors and 5xx
responses — with exponential backoff before giving up:

delay = min(MONITORING_REST_RETRY_BASE_DELAY_MS * 2^attempt, MONITORING_REST_RETRY_MAX_DELAY_MS)
up to MONITORING_REST_RETRY_MAX_ATTEMPTS attempts total, each bounded by the
existing 10s per-attempt timeout
4xx responses are treated as non-retryable (client/auth errors) and fail on
the first attempt

sendDeviceRpc (device RPC commands) is deliberately excluded from this
transport-level retry policy: a command may have already reached the device
even if the response is lost, so blind retry inside the adapter risks
double-executing a non-idempotent command.

Application-level safety is instead provided via an optional
idempotencyKey on the RPC command envelope (MonitoringCommandDto): when a
caller (e.g. a UI retrying after a timed-out response) resubmits the same
idempotencyKey for the same device with the same method and params,
MonitoringCommandService looks up the prior monitoring.command.sent audit
record (matched via AuditService.findByMetadataMatch against
idempotencyKey + deviceId + method + params together) and returns
that recorded response — tagged deduplicated: true — instead of
re-dispatching to ThingsBoard. Matching on the full command (not just the key

device) is deliberate: a stale or accidentally-reused key attached to a
genuinely different command must not short-circuit and silently skip
execution. Omitting idempotencyKey preserves the original at-most-once,
no-dedup behavior.

findByMetadataMatch compares each requested field with exact equality
(->>'key' = value for scalars, ->'key' = value::jsonb for object-valued
fields like params) rather than Postgres JSONB containment (@>).
Containment is a subset check: a stored params value like
{ "enabled": true, "duration": 10 } would satisfy a containment query for
{ "enabled": true }, so an idempotencyKey reused with a smaller params
object would have incorrectly matched an unrelated prior command and skipped
dispatch entirely. Exact per-field equality closes that gap.

The dedupe-check-then-dispatch-then-record sequence for a given
idempotencyKey runs inside a single DB transaction guarded by a Postgres
advisory lock (pg_advisory_xact_lock, keyed by a hash of
organizationId:deviceId:idempotencyKey, held until commit/rollback). Without
this, two concurrent requests carrying the same key (e.g. a client retry
racing the still-in-flight original call) could both read “no prior record”
before either audit write landed, defeating the dedup guarantee and
double-executing the RPC against ThingsBoard. Commands without an
idempotencyKey skip the lock/transaction entirely, since there is nothing to
serialize against.

REST Auth and Credential Rotation

All ThingsBoard REST calls (sendDeviceRpc, getLatestTelemetry,
getTelemetryHistory) authenticate with X-Authorization: Bearer <jwt>,
using the same service-account JWT (THINGSBOARD_USERNAME /
THINGSBOARD_PASSWORD login) as the realtime WebSocket connection — this
closes the “service-to-service auth” gap flagged for issue #106. There is a
single shared, in-memory JWT cache (getJwtToken()); every REST caller goes
through getRestAuthHeaders() to fetch it.

Rotation handling: if a REST call receives 401 Unauthorized (e.g. the
service-account password was rotated, or ThingsBoard invalidated the
session), the adapter invalidates its cached JWT and retries the call once
with a freshly fetched token — no restart is required to pick up rotated
credentials. This retry is safe even for sendDeviceRpc: a 401 means
ThingsBoard rejected the request before it reached the device, so retrying
cannot double-execute the command.

Correlation IDs

ThingsboardIntegrationAdapterService generates a correlationId (UUID) for
each REST call and logs it on every attempt, retry, success, and failure, so
an operator can grep one ID to see a call’s full retry history:

JWT login, getLatestTelemetry, and getTelemetryHistory each generate
their own correlation ID internally (there is no caller-facing request to
thread it from).
sendDeviceRpc accepts an optional caller-supplied correlationId and
returns it in its response. MonitoringCommandService.sendCommand generates
one per command and passes it through, then attaches the same ID to both the
monitoring.command.sent and monitoring.command.failed audit records —
so a support engineer can start from an audit entry and find the matching
adapter log lines for that exact incident, and vice versa.

This is scoped to the ThingsBoard integration flow only; there is no
app-wide correlation-ID/tracing middleware in this repo to plug into.

Integration Health Endpoint

GET /api/v1/monitoring/health (requires monitoring:read:all) exposes the
current org-level integration health as JSON:

{
  "status": "healthy",
  "connected": true,
  "reconnectAttempts": 0,
  "reconnectDelayMs": 1000,
  "heartbeatAt": 1710000000000,
  "backpressureDepth": 2,
  "lastEventLatencyMs": 340,
  "eventCount": 128
}

connected: whether the shared ThingsBoard WebSocket adapter currently has
an open socket.
The remaining fields come from MonitoringObservabilityService under the
same scope key the caller’s default subscription would actually stream
under — computed via the shared resolveMonitoringStreamScopeKey helper,
the same one MonitoringStreamService uses for its own fan-out/metrics
recording. This matters because that key is not always
organization:{organizationId}: an org that currently has exactly one
device resolves to device:{thingsboardDeviceId} instead (see
MonitoringScopeResolverService.toResolvedScope). Previously this endpoint
always hardcoded the organization:{organizationId} key, so any
single-device org’s /monitoring/health silently reported default
healthy/zero metrics no matter how degraded that device’s actual stream
was — the real metrics were being recorded under the device-scoped key and
never read. Fixed by resolving the caller’s default scope first and
deriving the metrics key from it, exactly like the stream service does.
Per-device scope metrics for multi-device orgs are still not included in
this first pass.
status is "degraded" once reconnectAttempts for the scope reaches
MONITORING_RECONNECT_ALERT_THRESHOLD (3); MonitoringObservabilityService
also logs a [ALERT] ... warning the moment a scope crosses that threshold,
so ops can alert on that log line even without a metrics backend.
reconnectAttempts (and the degraded status it drives) resets to zero the
next time a heartbeat is recorded for that scope, so a scope that recovers
and keeps receiving telemetry reports healthy again instead of staying
degraded forever from a past run of disconnects.
This is a minimal building block for a real dashboard/alerting setup (poll
the endpoint, or grep logs for [ALERT]), not a Prometheus/Grafana
integration — this repo has no existing dashboard/alerting convention to
build on top of.

Latency and Throughput Metrics

MonitoringStreamService measures the ingest-to-broadcast delay for every
telemetry event it fans out — Date.now() at delivery time minus the
event’s own ts — and records it via
MonitoringObservabilityService.markEvent(scope, latencyMs):

lastEventLatencyMs: the most recently observed delay for the scope (null
until the first event arrives, or if a payload has an unparsable ts).
eventCount: a cumulative per-scope counter of delivered events, usable as
a basic throughput signal (events per unit time, once sampled over an
interval by a caller/ops tool).

Both fields are included in GET /monitoring/health’s response, alongside
the existing reconnect/backpressure metrics.

status becomes "degraded" (and a [ALERT] line is logged) not only when
reconnectAttempts crosses its threshold, but also when
lastEventLatencyMs reaches MONITORING_LATENCY_ALERT_THRESHOLD_MS
(currently 5000ms). This is a provisional MVP target for issue #107’s
“Latency đạt ngưỡng mục tiêu đã thống nhất” acceptance criterion — the
measurement and alerting plumbing now exists, but the actual number should
be revisited once product/ops formally sign off on a target SLA. Unlike
reconnect degradation (which requires a heartbeat to clear), latency
degradation reflects only the most recent event, so it recovers as soon as a
subsequent event reports an acceptable delay.

Error Rate

MonitoringObservabilityService.markError(scope) tracks a cumulative
per-scope errorCount, closing the last of the four metrics named in issue
#108’s scope (throughput, latency, error rate, reconnect rate). An error is
recorded whenever the ThingsBoard adapter observes a fault it can’t already
attribute to a disconnect: a raw WebSocket error event, a malformed/
unparsable WebSocket message, or a failed JWT re-authentication after a
reconnect. errorCount is exposed on GET /monitoring/health, and status
becomes "degraded" (with a [ALERT] log line) once it reaches
MONITORING_ERROR_ALERT_THRESHOLD (currently 3). Like reconnect
degradation, errorCount resets to zero on the next heartbeat, since a
heartbeat proves the connection is alive and healthy again. REST-call
failures are intentionally excluded here — those are already visible via the
retry/backoff warn/error logs described above and via the caller’s own
exception (BadGatewayException), so counting them again in this metric
would double up on an already-observable signal without a clear scope to
attribute them to (a REST call is per-device, not per-stream-scope).

A malformed WebSocket message only counts as an error, never as a
heartbeat: heartbeatHooks (which reset errorCount back to zero) now fire
only after a message is successfully JSON.parsed, not before. Previously
the heartbeat hooks ran unconditionally at the top of the 'message'
handler, so every malformed frame reset errorCount to zero right before
notifyErrorHooks incremented it from zero — a persistent stream of bad
frames could never reach MONITORING_ERROR_ALERT_THRESHOLD, silently
masking a broken stream from the health/alert endpoint.

Contract Versioning

Closing issue #105’s “Có contract versioned cho telemetry và RPC envelope”
acceptance criterion: both wire contracts now carry an explicit
schemaVersion field.

Telemetry (MonitoringEventDto, used by GET /monitoring/snapshot’s
data[] and every monitoring:event WebSocket push): schemaVersion is
always set by the server (MONITORING_TELEMETRY_SCHEMA_VERSION, currently
1) and validated against MONITORING_SUPPORTED_TELEMETRY_SCHEMA_VERSIONS.
RPC command envelope (MonitoringCommandDto/MonitoringCommandResponseDto,
POST /monitoring/commands): schemaVersion on the request is optional and
defaults to MONITORING_COMMAND_SCHEMA_VERSION (currently 1) so existing
callers built against the unversioned envelope keep working unchanged; the
response always echoes back the version the server actually processed.
Error model: sending an explicit but unsupported schemaVersion on a
command is rejected with 400 Bad Request by the global ValidationPipe
(@IsIn against the supported-versions list) rather than being silently
misprocessed — a client can tell “you’re on an old/wrong contract version”
apart from “your payload is otherwise malformed”.
Evolution rule: bump the current version (and add it to the
MONITORING_SUPPORTED_*_SCHEMA_VERSIONS list) only for breaking changes to
the wire shape. Purely additive optional fields don’t need a version bump.
Keep superseded versions in the supported list for as long as in-flight
clients built against them still need to be accepted during rollout.

WebSocket Stream RBAC Gate

MonitoringStreamGateway.handleConnection previously authenticated the
/monitoring socket.io namespace by JWT validity and organization membership
only — it never checked whether the connecting user actually holds the
monitoring:read:all permission that the equivalent REST endpoints
(GET /monitoring/snapshot, /history, /health) already enforce via
PermissionsGuard. Any authenticated user in an organization (e.g. Owner or
Staff roles, which are deliberately excluded from monitoring:read per the
RBAC migration) could connect directly with a socket.io client and receive
their org’s full realtime telemetry stream, bypassing RBAC entirely — the
frontend nav/route gating only hides the UI affordance, it doesn’t protect
the transport. This closed a real gap in issue #106’s “RBAC pass cho các use
case chính” acceptance criterion: handleConnection now decodes
permissions from the JWT payload and calls the same
AuthorizationPolicyService.hasPermission check used by PermissionsGuard,
disconnecting the socket if the required permission is missing.

handleConnection’s rejection paths (missing token, invalid/expired token,
missing organizationId, missing monitoring:read:all) previously all just
called client.disconnect() with no explanation. This interacted badly with
socket.io-client’s reconnection semantics: a server-initiated disconnect
is reported to the client with reason io server disconnect, which
socket.io-client explicitly excludes from its automatic reconnection — unlike
a transport-level drop, the client will never retry on its own. Since access
tokens are short-lived and this page is meant to be left open for long
operator sessions, and the frontend has no proactive token-refresh timer (it
only refreshes reactively off a REST 401 via the axios interceptor in
frontend/src/lib/api.ts), any WebSocket (re)connect attempt made with an
expired token would previously leave the client permanently stuck showing
“Disconnected” with zero explanation and zero recovery until a full page
reload — a token refresh triggered by unrelated REST traffic elsewhere in the
app was the only thing that could accidentally fix it.

handleConnection now emits a monitoring:error before disconnecting,
distinguishing two cases: UNAUTHORIZED for a missing/invalid/expired token
or a token missing organizationId (worth retrying with a fresh token), and
FORBIDDEN_PERMISSION for a valid token that simply lacks
monitoring:read:all (retrying can’t fix a missing permission, so no retry is
attempted — same non-retry rationale as FORBIDDEN_SCOPE). On the frontend,
useMonitoringStream reacts to UNAUTHORIZED by calling
useAuthStore.getState().refreshToken(); a successful refresh updates
accessToken, which changes connect()'s identity and lets the existing
mount effect tear down the dead socket and open a fresh one with the new
token — no new socket-management code was needed, this reuses the same
reactive plumbing that already handles token rotation. A failed refresh
clears accessToken, which the hook’s existing !accessToken effect turns
into a clean disconnect (consistent with the rest of the app’s session-expiry
behavior).

Super Admin Nav Visibility

The super-admin.menu.tsx “Quan trắc” nav entry (added when the monitoring
page was first surfaced in “Tổng quan”) was removed: SUPER_ADMIN is a
system-scoped role with organizationId always null, but every
GET /api/v1/monitoring/* and POST /monitoring/commands endpoint requires
a non-null organizationId and responds 400 organizationId is required
otherwise. The nav item was visible (SUPER_ADMIN holds monitoring:read:all
via its wildcard permission) but every request the page made would fail
immediately. Re-add the entry once the monitoring page supports an explicit
organization picker for system-scoped users, or the backend adds a
system-scoped aggregate view.

Device Scope Input Validation

MonitoringScopeResolverService.resolveDeviceScope’s device:<id> scope
branch previously passed a client-supplied string straight into a
UUID-typed TypeORM where: { id } lookup with no format check. Every other
invalid-scope path in this method (missing scope, cross-tenant
organization: scope, unsupported scope prefix, device belonging to a
foreign org) rejects cleanly with a ForbiddenException, but a malformed
device: id (e.g. device:not-a-uuid, whether from a buggy client or a
probing request) instead reached Postgres and raised an unhandled
QueryFailedError (“invalid input syntax for type uuid”), which surfaced as
an opaque 500/SUBSCRIBE_FAILED with an error-level stack trace logged
for what is really just bad input at a system boundary, instead of the same
quiet warn-level ForbiddenException handling every other rejection in
this method already gets. resolveDeviceScope now validates the id with
class-validator’s isUUID before querying, short-circuiting to the same
ForbiddenException used for the foreign-device case without touching the
database.

Soft-Deleted Device Exclusion

Devices are soft-deleted (DevicesService.remove sets deletedAt, keeps the
row and its organizationId/thingsboardDeviceId); Device.deletedAt is a
plain @Column, not TypeORM’s @DeleteDateColumn, so nothing excludes
soft-deleted rows automatically — every query site in devices.service.ts
(11 of them) explicitly filters deletedAt: IsNull(). The monitoring module
never did: MonitoringScopeResolverService.resolveDeviceScope (both the
org-wide device list and the device:<id> lookup) queried by organizationId
or id alone with no deletedAt filter, and the direct device re-fetches in
MonitoringQueryService.getSnapshot/getHistory and
MonitoringCommandService’s RPC dispatch had the same gap. Since this
resolver gates every monitoring entry point (REST snapshot/history, the
WebSocket gateway’s subscribe, the RPC command bridge, and the health
endpoint), a decommissioned/removed device kept showing up in an org’s
realtime telemetry and — more seriously — remained reachable via
POST /monitoring/commands, letting a client send RPC commands (e.g.
setPump) to equipment that had already been marked deleted. All five query
sites now filter deletedAt: IsNull(), matching the convention already used
everywhere else in devices.service.ts.

MonitoringStreamGateway.handleConnection is a NestJS gateway lifecycle hook,
not a socket.io namespace middleware (io.use()) — it runs as a normal async
event listener after the socket’s connection has already been accepted, and
does not block the client from sending other messages while it’s in flight.
The frontend’s useMonitoringStream emits monitoring:subscribe the instant
it sees its own connect event, which can arrive at the server milliseconds
before handleConnection’s await this.jwtService.verifyAsync(token) call
resolves. Previously, handleSubscribe read client.data.user?.organizationId
directly; if it ran before handleConnection finished populating
client.data.user, it silently returned with no monitoring:error and no
attach — a stuck-forever failure mode indistinguishable from a healthy,
still-loading connection (no error banner, no retry, isConnected: true
forever with zero telemetry), matching the exact failure class already fixed
for SUBSCRIBE_FAILED (iteration 27) and UNAUTHORIZED (iteration 28) in
this epic. handleConnection now stores its own in-flight auth promise in a
WeakMap keyed by the client socket, and handleSubscribe awaits that same
promise before reading client.data.user — closing the race with no new
socket-level plumbing (once handleConnection has already resolved, awaiting
its stored promise is a no-op). Switching to a full io.use() namespace
middleware was considered but rejected: rejecting at the middleware layer
would replace the existing monitoring:error + clean disconnect UX with a
bare connect_error, losing the UNAUTHORIZED/FORBIDDEN_PERMISSION code
distinction the frontend already relies on for its token-refresh logic.

MonitoringStreamGateway.handleSubscribe is invoked once per
monitoring:subscribe message a client emits, and a client can legitimately
send more than one in quick succession on the same socket (e.g. the frontend’s
initial subscribe on connect overlapping a scope prop change, or a
SUBSCRIBE_FAILED resubscribe retry landing right as a fresh subscribe goes
out). Each call independently awaits MonitoringScopeResolverService .resolveDeviceScope, an async DB lookup whose latency isn’t guaranteed to
match send order. Previously, whichever call’s resolveDeviceScope happened
to resolve last unconditionally overwrote clientDetachMap and tore down
whatever the other call had already attached via its own priorDetach?.()
call — so an older, already-superseded subscribe request resolving after a
newer one would silently detach the newer (correct) subscription and
leave the client watching the older, stale scope, with no error emitted to
indicate anything went wrong.

Fixed by tagging each handleSubscribe call with a per-client generation
number (clientSubscribeGenerationMap, a WeakMap<socket, number>)
incremented synchronously before the async scope resolution starts. Before
acting on the resolver’s result (success or failure), the call checks
whether its own generation is still the client’s latest; if a newer subscribe
request has since started, the stale call returns without touching
clientDetachMap or emitting an error, so only the most recently sent
request ever wins regardless of resolution order.

handleDisconnect fires exactly once per socket and, until this fix, only
detached whatever subscription was already recorded in clientDetachMap at
that moment. If a client disconnected (tab closed, network drop) while an
earlier monitoring:subscribe call’s resolveDeviceScope lookup was still
pending, handleDisconnect ran first with nothing to detach yet — then the
in-flight handleSubscribe call later resolved, passed its (unrelated)
generation-staleness check from the “Subscribe Requests Resolving Out Of Send
Order” fix above, and called streamService.attachClient for a socket that
was already gone. Since handleDisconnect never fires a second time for that
socket, the resulting upstream ThingsBoard subscription and listener leaked
for the lifetime of the process.

Fixed by having handleDisconnect bump the same per-client
clientSubscribeGenerationMap counter used for the out-of-order-subscribe
fix, reusing the existing staleness check instead of adding new state: any
handleSubscribe call still awaiting resolveDeviceScope at disconnect time
now sees itself as stale once its lookup resolves and returns without
attaching.

Disconnect During The Connection Auth Check (Before A Generation Was Ever Captured)

A variant of the race above survived the fix above: the generation snapshot
in handleSubscribe was captured after await this.connectionReadyMap.get(client)
(the still-pending JWT verification for a client that emits
monitoring:subscribe the instant it sees its own connect event). If that
same client disconnected during that specific await — i.e. before
handleSubscribe had captured any generation for itself at all —
handleDisconnect’s bump had nothing to invalidate yet. Once auth finished
regardless (the server-side verification promise isn’t cancelled by the
socket disconnecting) and handleSubscribe resumed, it captured a fresh
generation reading the post-bump counter value, which by definition matched
the counter it had just written — every later staleness check in the method
trivially passed, and the call proceeded to attach an upstream subscription
for a socket that was already gone and would never be cleaned up.

Fixed by moving the generation capture to the very top of handleSubscribe,
synchronous and before the connectionReadyMap await, so a disconnect during
that wait is correctly observed by this call’s already-taken snapshot once
it resumes. This intentionally does not add a new staleness check
immediately after that await: two monitoring:subscribe calls that arrive in
the same tick (e.g. an initial subscribe followed immediately by a scope
change) must still both be allowed to reach resolveDeviceScope — bailing
out the older one immediately post-await, before ever calling the resolver,
would break the “keeps the most recently requested scope active…” guarantee
from “Subscribe Requests Resolving Out Of Send Order” above, which depends on
both requests reaching the resolver so the result — not just the
resolution order — can be compared. The existing staleness checks (in the
resolveDeviceScope catch block, and just before attachClient) are
sufficient once the generation is captured early enough.

Idempotency Key Retry After an Uncertain RPC Outcome

The idempotency-key dedup lookup in MonitoringCommandService.dispatchIdempotent
only ever matched prior monitoring.command.sent audit records. If a first
attempt’s RPC call to ThingsBoard itself failed (timeout, network error,
non-2xx response) after the request was actually sent — an outcome that is
inherently uncertain, since the physical device may already have received
and executed it — the resulting monitoring.command.failed record was
invisible to the dedup check. A client retrying with the exact same
idempotencyKey would pass the check and dispatch again, risking
double-executing a non-idempotent device command — defeating the very
guarantee the idempotency key exists to provide.

Every monitoring.command.failed audit record now carries a
dispatchAttempted boolean, true only when the failure occurred after
adapter.sendDeviceRpc was actually invoked (wrapped internally in
RpcDispatchOutcomeUncertainError, unwrapped again before rethrowing to
callers so exception types like BadGatewayException are unaffected).
dispatchIdempotent now also checks for a prior failed record with
dispatchAttempted: true matching the same idempotencyKey+device+method+
params; if found, it throws a ConflictException instead of redispatching,
requiring the caller to mint a fresh idempotencyKey once the device’s
actual state has been confirmed. Failures that happen before the RPC call
(scope rejection, unmapped device) still have dispatchAttempted: false and
remain safely retryable with the same key, since nothing was ever sent to
ThingsBoard.

Idempotency-Key Audit Write Must Not Fail Silently

AuditService.record swallows and logs repository failures by default — the
right behavior for ordinary audit logging, which must never block or fail the
business operation it’s describing. The idempotent RPC command path is an
exception: the monitoring.command.sent record it writes is the only
durable marker dispatchIdempotent’s dedupe check can find on a later retry.
If that save silently failed, MonitoringCommandService.sendCommand would
still return success to the caller even though no trace of the dispatch was
ever persisted — a retry with the same idempotencyKey would then find no
prior record and redispatch the RPC, exactly the double-execution the key
exists to prevent.

record now accepts an optional third { throwOnFailure: true } argument.
The idempotent dispatch path passes it for the SENT record only; a save
failure there is caught and re-wrapped as RpcDispatchOutcomeUncertainError
(the RPC really was sent to ThingsBoard, so the resulting
monitoring.command.failed record still correctly carries
dispatchAttempted: true, keeping the uncertain-outcome ConflictException
guard from the previous section intact). The non-idempotent path and the
outer failure-record write are unaffected and keep the original
swallow-and-log behavior, since neither makes an idempotency promise that a
lost write would violate.

A related Codex review finding on the same query (“compare boolean metadata
as JSON instead of text, since the query fails at runtime”) was investigated
and could not be reproduced: a direct TypeORM query-builder test against a
real local Postgres instance, using the exact metadata ->> 'dispatchAttempted' = :param shape with a bound JS boolean, executed successfully and matched
correctly both when a matching row existed and when it didn’t. Node-postgres
serializes bound booleans as the text 'true'/'false' before sending them,
and Postgres’s implicit parameter-type inference resolves the placeholder to
text from the ->> operator’s context, so no type-mismatch error occurs in
practice. Not changed, to avoid trading a working query for an unverified one.

The identical claim resurfaced in a later Codex review pass against the other
findByMetadataMatch call site in dispatchIdempotent (the uncertain-failure
lookup, which also binds dispatchAttempted: true as a boolean). Re-verified
by inserting a real audit row with that exact metadata shape into the local
Postgres instance and running the equivalent metadata ->> 'dispatchAttempted' = $1 query with a bound boolean parameter via the pg driver directly — it
matched correctly. Both call sites use the same query mechanism, so this is
the same already-debunked finding recurring, not a new bug.

The claim recurred a third time (13th PR review overall) against the
dispatchIdempotent SENT-record dedupe lookup specifically. Re-verified a
fourth time via TypeORM’s own QueryBuilder (not just the raw pg driver)
against a real local Postgres instance, reproducing the exact multi-clause
shape used in production (idempotencyKey/deviceId/method via ->>,
params via -> ::jsonb, and dispatchAttempted: true via ->> with a
bound JS boolean) end-to-end, including the query’s generated SQL. It
executed without error and matched the inserted row. Not changed.

Uncertain-Outcome FAILED Marker Must Be Written Inside The Advisory-Lock Transaction

dispatchIdempotent runs inside a dataSource.transaction(...) callback that
holds the pg_advisory_xact_lock acquired at its start for the transaction’s
whole lifetime — the lock is released automatically the instant the
transaction commits or rolls back. Previously, when dispatch() failed with
an uncertain outcome (the RPC may have already reached the physical device),
dispatchIdempotent simply let RpcDispatchOutcomeUncertainError propagate
out of the transaction callback. TypeORM responds to a thrown callback by
rolling back the transaction — which released the advisory lock — and only
afterwards did sendCommand’s outer catch write the durable
monitoring.command.failed (dispatchAttempted: true) marker, using a plain
(non-transactional) auditService.record call. This left a real race window:
a concurrent retry carrying the same idempotencyKey, blocked waiting on the
lock, could acquire it the moment the transaction rolled back — before the
outer catch’s marker write ever happened — find neither a SENT nor a
FAILED record, and redispatch the same non-idempotent RPC a second time.

Fixed by moving the marker write inside dispatchIdempotent itself: on an
uncertain-outcome failure, it now writes the monitoring.command.failed
record using the same transaction-scoped manager and then returns a
{ ok: false, cause } result instead of throwing, so the transaction commits
normally with the marker included, and only releases the lock once that
marker is durably visible to any other transaction. sendCommand inspects
the returned result and rethrows the original cause via a
CommandAlreadyAuditedError marker so its own generic outer-catch logic
doesn’t write a second, duplicate FAILED record for the same failure.
Failures that occur before the RPC is ever attempted (e.g. an unmapped
device) still propagate and roll back as before, since there’s no
double-execution risk and therefore nothing that must survive the rollback.

The same claim (“cast scalar metadata values before comparing… generates
text = boolean and Postgres raises an operator error”) resurfaced a third
time in a later review pass, again against findByMetadataMatch’s scalar
branch. Re-verified a third time by inserting a real audit row with a
dispatchAttempted: true boolean field into the local Postgres instance and
running the exact metadata ->> 'dispatchAttempted' = $1 query with a bound
JS true via the pg driver directly — it matched correctly with no
operator error. This confirms (for the third time, across three separate
review passes) that the claim does not reproduce against a real server; not
changed, to avoid trading a working query for an unverified one.

The FAILED Marker Write Itself Must Not Fail Silently

The fix above (writing the monitoring.command.failed marker inside the
advisory-lock transaction) still had the same class of gap the SENT-record
fix closed earlier: the write used a plain auditService.record(input, manager) call with no throwOnFailure option, so a swallowed save failure
(e.g. a dropped connection) made dispatchIdempotent return { ok: false, cause } as if the durable marker had been written, when in fact nothing was
persisted. sendCommand trusts that ok: false result completely — its
CommandAlreadyAuditedError path rethrows the original cause without any
further audit write, on the assumption that dispatchIdempotent already
recorded it. The net effect: a command that really was dispatched to
ThingsBoard could fail with no audit trail at all, and a later retry with the
same idempotencyKey would find neither a SENT nor a FAILED record and
redispatch the non-idempotent RPC a second time — the exact double-execution
the idempotency key exists to prevent.

Fixed by passing { throwOnFailure: true } on this write too. If it still
throws (the same broken connection that caused the original RPC failure
typically fails every write in the same transaction), dispatchIdempotent
rethrows the original RpcDispatchOutcomeUncertainError — not the save
error — so the transaction rolls back cleanly (there was no durable insert to
lose) and sendCommand’s outer catch makes a best-effort fallback write
outside the transaction/lock via its existing (unaffected)
RpcDispatchOutcomeUncertainError handling, still tagged
dispatchAttempted: true. The final error surfaced to the caller is
unchanged (the original RPC/save cause), so this is purely a reliability fix,
not a behavior change for callers.

WebSocket Wire Protocol (verified against a live ThingsBoard CE 4.2.1 server)

Every test in this module (unit, e2e) mocks the ws client and/or overrides
ThingsboardIntegrationAdapterService/ThingsBoardClientService at the Nest
DI level — for the first 40 iterations of this epic, no ThingsBoard instance
had ever actually run in any environment this work executed in, so 100% test
coverage only proved the adapter was internally self-consistent, never that
its WebSocket message shapes matched a real ThingsBoard server.

This was finally verified directly: a real ThingsBoard CE 4.2.1 instance was
run locally (docker run thingsboard/tb-postgres), a tenant device was
created, and the adapter’s exact wire messages were replayed against it with
a raw ws client. The previous implementation was confirmed completely
non-functional against a real server, and has been rewritten and re-verified
end-to-end. Concretely:

Auth was sent the wrong way entirely. The old adapter opened the socket
first, then sent {"authCmd":{"cmdId":0,"token":"..."}} as an in-band
message. A real ThingsBoard 4.2.1 server rejects this outright — the
connection closes immediately with WS close code 1007 (“The given string
value cannot be transformed to Json object”), before any other message is
even read. ThingsBoard authenticates the WebSocket via a ?token=<jwt>
query parameter on the connection URL itself, not an in-band command. The
adapter now fetches the JWT before opening the socket and connects to
<THINGSBOARD_WS_URL>?token=<jwt> — auth failures are therefore resolved
before any socket is ever created (see “Reconnect Behavior” above).
The subscribe command used the wrong field/shape. The old adapter sent
{"cmds":[{"cmdId":1,"type":"ENTITY_DATA","entityType":"DEVICE","entityId":"...","scope":"LATEST_TELEMETRY"}]}.
Real ThingsBoard’s TelemetryPluginCmdsWrapper has no cmds field at all;
the correct field is tsSubCmds, and its entries drop the type key:
{"tsSubCmds":[{"cmdId":1,"entityType":"DEVICE","entityId":"...","scope":"LATEST_TELEMETRY"}]}.
Unsubscribe is {"tsSubCmds":[{"cmdId":1,"unsubscribe":true}]}. Verified
live: the old shape produces no response and no error at all (the server
ignores an unrecognized field); the new shape gets an immediate telemetry
snapshot back.
The message handler assumed the wrong response shape. The old handler
expected message.data to already be a single flattened, pre-normalized
event ({ deviceId, metric, value, ts, quality }). A real ThingsBoard
tsSubCmds push looks like
{"subscriptionId":1,"errorCode":0,"errorMsg":null,"data":{"<metricKey>":[[ts,value],...]}}
— a per-metric map of [timestamp, value] samples, with no deviceId
field at all (the client already knows which entity a subscriptionId
maps to, since it chose that mapping when subscribing). The adapter now
parses this shape directly: one event is emitted per metric key using that
key’s most recent [ts, value] sample, with deviceId populated from the
ThingsBoard entity id tracked for that subscriptionId in
activeSubscriptions. MonitoringStreamService then translates that
ThingsBoard entity id back to our internal Device.id for the single-device
scope case (see internalDeviceId in attachClient), since
MonitoringEventDto.deviceId must always be our internal id.

All three fixes were re-verified end-to-end against the live server after
implementation: connecting with ?token=<jwt>, sending the new tsSubCmds
subscribe message, receiving a real telemetry snapshot, receiving a live delta
push after posting new telemetry mid-subscription, and sending the new
unsubscribe message, all behaved exactly as the adapter now expects.

Fixed: organization-wide (multi-device) scope now fans out per device

resolveMonitoringStreamScopeKey’s organization:<id> branch (the default
path whenever a caller subscribes without an explicit device scope — i.e.
what MonitoringPage uses today for any org with more than one device) used
to map to buildTsSubCmd’s entityType: 'TENANT', entityId: <organizationId>.
This was verified live to return zero telemetry: ThingsBoard’s tsSubCmds
API is fundamentally per-entity, and a TENANT-scoped subscription only
returns telemetry attached directly to the Tenant entity itself, never an
aggregate of every device belonging to it (confirmed: subscribing at TENANT
scope on a tenant with live device telemetry returns {"data":{}}, empty).
There is also no tenant/customer-per-organization provisioning anywhere in
this codebase (ThingsBoardClientService only creates/deletes individual
devices) — <our internal organizationId> was never a real ThingsBoard entity
in the first place.

This has been fixed by fanning out one device:<tbId> upstream subscription
per device instead of a single tenant-wide one:

ResolvedMonitoringScope now carries thingsboardDeviceIds, a ThingsBoard
device id per entry in deviceIds (same index), populated by
MonitoringScopeResolverService for every scope, not just single-device
ones.
MonitoringStreamService.attachClient opens one
adapter.subscribeToScope('device:<tbId>', ...) call per device that has a
known ThingsBoard mapping (devices without one are skipped — there is
nothing to subscribe to), all reporting into the same logical
organization:<id> bucket used for listener fan-out and observability
metrics. A single-device scope is simply the N=1 case of the same code
path, which also removed the old special-cased internalDeviceId
translation logic — each per-device subscription’s callback already knows
exactly which internal device id its events belong to.
Detaching the last listener for an organization scope tears down every
per-device upstream subscription it opened, same as the old single
subscription did.

This does mean an N-device organization now opens N ThingsBoard WebSocket
subscriptions (N cmdIds over the one shared connection) instead of one —
there is no cheaper way to get real per-device telemetry given ThingsBoard’s
per-entity tsSubCmds API, and REST-based snapshot/history fetching
(MonitoringQueryService) already fans out one HTTP call per device the same
way. internalDeviceId and resolveMonitoringStreamScopeKey’s single-device
thingsboardDeviceId field remain in place unchanged; the observability
bucket name and stream-level attachClient contract are unchanged, so
neither the frontend nor the WebSocket gateway needed any changes.

Frontend: A Burst Of Metrics Could Collapse To Only The Last One

ThingsBoard pushes a multi-metric WS message as several separate
monitoring:event server-to-client emissions – one per metric key, emitted
synchronously in a loop (see MonitoringStreamService.openDeviceSubscriptions
and ThingsboardIntegrationAdapterService.handleIncomingMessage). When
socket.io-client receives and processes several of these within the same JS
task (common when the underlying transport delivers more than one frame per
read), useMonitoringStream used to track only a single lastEvent state
slot, updated via setLastEvent(event). React batches multiple state updates
from the same task into one re-render, and a plain (non-functional) update
like this one always overwrites the pending value rather than queuing it, so
only the last of the several calls survived – MonitoringPage’s merge
effect, keyed on [lastEvent], only ever saw that final metric. The earlier
metrics from the same burst were silently dropped from the live telemetry
table until a later, unrelated event happened to update the same key.

Fixed by adding eventBatch: MonitoringEvent[] to useMonitoringStream’s
return value: every incoming event is appended via a functional state
update (setEventBatch((current) => [...current, event])), which – unlike
overwriting a single slot – is queued and applied in order by React even
when several fire in the same task, so no event in a burst is ever lost.
MonitoringPage’s merge effect now depends on [eventBatch, clearEvents],
merges every queued event (in arrival order, respecting the existing
out-of-order and scoped-device-eviction guards per event), then calls the
hook’s new clearEvents() to drain the batch. lastEvent is unchanged and
still used only for the “Latest event” stat card, where losing an
intermediate value from the same burst is inconsequential.

Known Gaps (tracked under issue #108)

No metrics-export endpoint (Prometheus/OpenTelemetry) or real dashboard/alert
channel exists — GET /monitoring/health and the [ALERT] log line above
are the current minimal substitute. A proper dashboard/alerting integration
(e.g. wiring the health endpoint into an existing ops tool, or exporting to
Prometheus) needs a product decision on which tool to standardize on before
more code should be added here.
sendDeviceRpc still has no automatic transport-level retry — this remains
intentional (see “REST Retry Behavior” above). Callers that want safe
retry semantics should supply idempotencyKey on the command envelope
instead of relying on adapter-side retry.
The WebSocket wire protocol (both single-device and organization-wide
multi-device scopes) is now verified/fixed against a live ThingsBoard CE
4.2.1 server (see “WebSocket Wire Protocol” and “Fixed: organization-wide
(multi-device) scope now fans out per device” above). Per-device scope
metrics for multi-device orgs are still bucketed together under one
organization:<id> observability key rather than broken out per device —
acceptable for now since the health endpoint’s purpose is scope-level
connectivity, not per-device diagnostics.
ThingsboardIntegrationAdapterService’s onHeartbeat hooks are now keyed
per-cmdId and only fire for the specific subscription a message’s
subscriptionId identifies — see “Fixed: one org’s telemetry could reset
another org’s degraded status via a shared heartbeat” above. onDisconnect
and onError, by contrast, remain intentionally connection-wide
(disconnectHooks/errorHooks Sets fired for every registered scope):
a shared-WebSocket-level disconnect or error genuinely does affect every
currently subscribed organization at once (there is only one physical
connection for the whole app), and a malformed/unparsable frame’s error
has no subscriptionId to attribute it to a narrower scope in the first
place, so firing conservatively for all open scopes is correct there, not
a gap.
devices.service.ts’s soft-delete now evicts a deleted device’s upstream
ThingsBoard subscription — see “Fixed: device soft-delete now tears down its
ThingsBoard fan-out subscription” below. Closed.