fix(mqtt): skip offline status telemetry samples

backend/src/db/index.ts

@@ -709,12 +709,13 @@ export async function insertNodeStatusSample(sample: {
   channelUtilization?: number | null;
   airUtilTx?: number | null;
   stats?: Record<string, unknown> | null;
+  status?: string | null;
 }): Promise<void> {
   await pool.query(
     `WITH sample_insert AS (
        INSERT INTO node_status_samples
-       (time, node_id, network, battery_mv, uptime_secs, tx_air_secs, rx_air_secs, channel_utilization, air_util_tx, stats)
-       VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9)
+       (time, node_id, network, battery_mv, uptime_secs, tx_air_secs, rx_air_secs, channel_utilization, air_util_tx, stats, status)
+       VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
        RETURNING time
      )
      UPDATE nodes
@@ -732,6 +733,7 @@ export async function insertNodeStatusSample(sample: {
       sample.channelUtilization ?? null,
       sample.airUtilTx ?? null,
       sample.stats ? JSON.stringify(sample.stats) : null,
+      sample.status ?? null,
     ],
   );
 }

backend/src/db/migrations/052_node_status_sample_status.sql

@@ -0,0 +1,3 @@
+-- Preserve the MQTT status envelope alongside extracted node telemetry.
+ALTER TABLE node_status_samples
+  ADD COLUMN IF NOT EXISTS status TEXT;

backend/src/db/schema/base.sql

@@ -229,7 +229,8 @@ CREATE TABLE IF NOT EXISTS node_status_samples (
   rx_air_secs          BIGINT,
   channel_utilization  DOUBLE PRECISION,
   air_util_tx          DOUBLE PRECISION,
-  stats                JSONB
+  stats                JSONB,
+  status               TEXT
 );
 
 SELECT create_hypertable('node_status_samples', 'time', if_not_exists => TRUE);

backend/src/mqtt/client.ts

@@ -14,6 +14,7 @@ import { shouldDiscardUnverifiedTxAdvert, statusEnvelopeTargetsObserver } from '
 import { extractNeighborNodes } from './neighborPayload.js';
 import { parseMqttTopic } from './topic.js';
 import { createDurableMqttHandleMessage } from './durableHandler.js';
+import { persistStatusTelemetrySample } from './statusTelemetry.js';
 import {
   boundedNetworkMetricLabel,
   mqttIngestActive,
@@ -194,112 +195,6 @@ function isEmptyPacketEnvelope(json: Record<string, unknown>, rawHex: string, pa
     && (payloadLen ?? 0) <= 0;
 }
 
-function toRecord(value: unknown): Record<string, unknown> | undefined {
-  return value && typeof value === 'object' && !Array.isArray(value)
-    ? value as Record<string, unknown>
-    : undefined;
-}
-
-function readNum(obj: Record<string, unknown> | undefined, ...keys: string[]): number | undefined {
-  if (!obj) return undefined;
-  for (const key of keys) {
-    if (!(key in obj)) continue;
-    const n = toNum(obj[key]);
-    if (n != null) return n;
-  }
-  return undefined;
-}
-
-type StatusTelemetrySample = {
-  batteryMv?: number;
-  uptimeSecs?: number;
-  txAirSecs?: number;
-  rxAirSecs?: number;
-  channelUtilization?: number;
-  airUtilTx?: number;
-  stats?: Record<string, unknown>;
-};
-
-const OWNER_STATUS_STATS_KEYS = new Set([
-  'battery_mv', 'solar_mv', 'board_temp_c', 'wifi_rssi', 'wifi_ssid', 'wifi_uptime_ms',
-  'ntp_synced', 'ntp_sync_age_ms', 'boot_count', 'reset_reason', 'max_loop_ms',
-  'max_loop_at_ms', 'nodes_heard_24h', 'channel_utilization', 'air_util_tx', 'air_util_rx',
-  'last_rx_rssi', 'last_rx_snr', 'tx_power_dbm', 'config_version', 'config_crc32',
-  'fs_free_bytes', 'fs_total_bytes', 'nvs_free_entries', 'channel_id', 'git_commit', 'boot_epoch',
-]);
-
-function extractStatusTelemetry(
-  json: Record<string, unknown>,
-  options?: { allowRawStatsOnly?: boolean },
-): StatusTelemetrySample | null {
-  const stats = toRecord(json['stats']);
-  const hasRawStats = Boolean(stats && Object.keys(stats).length > 0);
-  const hasOwnerStats = Boolean(
-    stats && (
-      Object.keys(stats).some((key) => OWNER_STATUS_STATS_KEYS.has(key))
-      || toRecord(stats['mqtt'])
-    ),
-  );
-  const batteryMv = readNum(stats, 'battery_mv', 'batteryMv');
-  const uptimeSecs = (() => {
-    const direct = readNum(stats, 'uptime_secs', 'uptimeSecs');
-    if (direct != null) return direct;
-    const uptimeMs = readNum(stats, 'uptime_ms', 'uptimeMs');
-    if (uptimeMs == null) return undefined;
-    return Math.floor(uptimeMs / 1000);
-  })();
-  const txAirSecs = readNum(stats, 'tx_air_secs', 'txAirSecs');
-  const rxAirSecs = readNum(stats, 'rx_air_secs', 'rxAirSecs');
-  const channelUtilization = readNum(
-    stats,
-    'channel_utilization',
-    'channel_utilization_pct',
-    'channel_util',
-    'channelUtil',
-    'channelUtilization',
-  );
-  const airUtilTx = readNum(
-    stats,
-    'air_util_tx',
-    'air_util_tx_pct',
-    'tx_air_util',
-    'tx_air_utilization',
-    'airUtilTx',
-  );
-
-  if (
-    batteryMv == null
-    && uptimeSecs == null
-    && txAirSecs == null
-    && rxAirSecs == null
-    && channelUtilization == null
-    && airUtilTx == null
-  ) {
-    if ((options?.allowRawStatsOnly || hasOwnerStats) && hasRawStats) {
-      return {
-        batteryMv,
-        uptimeSecs,
-        txAirSecs,
-        rxAirSecs,
-        channelUtilization,
-        airUtilTx,
-        stats,
-      };
-    }
-    return null;
-  }
-
-  return {
-    batteryMv,
-    uptimeSecs,
-    txAirSecs,
-    rxAirSecs,
-    channelUtilization,
-    airUtilTx,
-    stats,
-  };
-}
-
 /** In-flight pre-resolve tracking — prevents duplicate concurrent resolutions for the same hash/network. */
 const preResolveInFlight = new Set<string>();
 const preResolveLastStarted = new Map<string, number>();
@@ -552,22 +447,11 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise<void> {
       allowTestOverride: network === 'test' && nodeId === observerKey,
       mqttObserver: true,
     })];
-    const telemetry = extractStatusTelemetry(json, {
-      allowRawStatsOnly: network === 'test',
-    });
-    if (telemetry) {
-      writes.push(insertNodeStatusSample({
-        nodeId,
-        network,
-        batteryMv: telemetry.batteryMv,
-        uptimeSecs: telemetry.uptimeSecs,
-        txAirSecs: telemetry.txAirSecs,
-        rxAirSecs: telemetry.rxAirSecs,
-        channelUtilization: telemetry.channelUtilization,
-        airUtilTx: telemetry.airUtilTx,
-        stats: telemetry.stats,
-      }));
-    }
+    writes.push(persistStatusTelemetrySample(
+      json,
+      { nodeId, network, allowRawStatsOnly: network === 'test' },
+      insertNodeStatusSample,
+    ));
     const writeResults = await Promise.allSettled(writes);
     const writeErrors: unknown[] = [];
     for (const result of writeResults) {

backend/src/mqtt/statusTelemetry.test.ts

@@ -0,0 +1,41 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { persistStatusTelemetrySample, type NodeStatusSampleInput } from './statusTelemetry.js';
+
+const NODE_ID = 'A'.repeat(64);
+
+test('offline Will telemetry is skipped, including a retained replay with boot-fresh stats', async () => {
+  const writes: NodeStatusSampleInput[] = [];
+  const written = await persistStatusTelemetrySample(
+    {
+      status: 'offline',
+      stats: { battery_mv: '4200', uptime_secs: '0', boot_count: '1' },
+    },
+    { nodeId: NODE_ID, network: 'ukmesh' },
+    async (sample) => { writes.push(sample); },
+  );
+
+  assert.equal(written, false);
+  assert.deepEqual(writes, []);
+});
+
+test('real telemetry is stored with its top-level status', async () => {
+  const writes: NodeStatusSampleInput[] = [];
+  const written = await persistStatusTelemetrySample(
+    {
+      status: 'online',
+      stats: { battery_mv: '4100', uptime_secs: '7200' },
+    },
+    { nodeId: NODE_ID, network: 'ukmesh' },
+    async (sample) => { writes.push(sample); },
+  );
+
+  assert.equal(written, true);
+  assert.equal(writes.length, 1);
+  assert.equal(writes[0]?.status, 'online');
+  assert.equal(writes[0]?.nodeId, NODE_ID);
+  assert.equal(writes[0]?.network, 'ukmesh');
+  assert.equal(writes[0]?.batteryMv, 4100);
+  assert.equal(writes[0]?.uptimeSecs, 7200);
+  assert.deepEqual(writes[0]?.stats, { battery_mv: '4100', uptime_secs: '7200' });
+});

backend/src/mqtt/statusTelemetry.ts

@@ -0,0 +1,159 @@
+export type StatusTelemetrySample = {
+  status?: string;
+  batteryMv?: number;
+  uptimeSecs?: number;
+  txAirSecs?: number;
+  rxAirSecs?: number;
+  channelUtilization?: number;
+  airUtilTx?: number;
+  stats?: Record<string, unknown>;
+};
+
+export type NodeStatusSampleInput = StatusTelemetrySample & {
+  nodeId: string;
+  network: string;
+};
+
+type StatusSampleOptions = {
+  nodeId: string;
+  network: string;
+  allowRawStatsOnly?: boolean;
+};
+
+type StatusSampleWriter = (sample: NodeStatusSampleInput) => Promise<void>;
+
+const OWNER_STATUS_STATS_KEYS = new Set([
+  'battery_mv', 'solar_mv', 'board_temp_c', 'wifi_rssi', 'wifi_ssid', 'wifi_uptime_ms',
+  'ntp_synced', 'ntp_sync_age_ms', 'boot_count', 'reset_reason', 'max_loop_ms',
+  'max_loop_at_ms', 'nodes_heard_24h', 'channel_utilization', 'air_util_tx', 'air_util_rx',
+  'last_rx_rssi', 'last_rx_snr', 'tx_power_dbm', 'config_version', 'config_crc32',
+  'fs_free_bytes', 'fs_total_bytes', 'nvs_free_entries', 'channel_id', 'git_commit', 'boot_epoch',
+]);
+
+function toRecord(value: unknown): Record<string, unknown> | undefined {
+  return value && typeof value === 'object' && !Array.isArray(value)
+    ? value as Record<string, unknown>
+    : undefined;
+}
+
+function toNum(value: unknown): number | undefined {
+  if (typeof value === 'number') return Number.isFinite(value) ? value : undefined;
+  if (typeof value === 'string' && value.trim() !== '') {
+    const number = Number(value);
+    return Number.isFinite(number) ? number : undefined;
+  }
+  return undefined;
+}
+
+function readNum(obj: Record<string, unknown> | undefined, ...keys: string[]): number | undefined {
+  if (!obj) return undefined;
+  for (const key of keys) {
+    if (!(key in obj)) continue;
+    const number = toNum(obj[key]);
+    if (number != null) return number;
+  }
+  return undefined;
+}
+
+function envelopeStatus(json: Record<string, unknown>): string | undefined {
+  if (typeof json['status'] !== 'string') return undefined;
+  const status = json['status'].trim();
+  return status || undefined;
+}
+
+export function extractStatusTelemetry(
+  json: Record<string, unknown>,
+  options?: { allowRawStatsOnly?: boolean },
+): StatusTelemetrySample | null {
+  const status = envelopeStatus(json);
+  // The LWT carries boot-fresh statistics. Ignore it on the same path whether
+  // it arrives live after a session drop or as a retained replay on subscribe.
+  if (status?.toLowerCase() === 'offline') return null;
+
+  const stats = toRecord(json['stats']);
+  const hasRawStats = Boolean(stats && Object.keys(stats).length > 0);
+  const hasOwnerStats = Boolean(
+    stats && (
+      Object.keys(stats).some((key) => OWNER_STATUS_STATS_KEYS.has(key))
+      || toRecord(stats['mqtt'])
+    ),
+  );
+  const batteryMv = readNum(stats, 'battery_mv', 'batteryMv');
+  const uptimeSecs = (() => {
+    const direct = readNum(stats, 'uptime_secs', 'uptimeSecs');
+    if (direct != null) return direct;
+    const uptimeMs = readNum(stats, 'uptime_ms', 'uptimeMs');
+    if (uptimeMs == null) return undefined;
+    return Math.floor(uptimeMs / 1000);
+  })();
+  const txAirSecs = readNum(stats, 'tx_air_secs', 'txAirSecs');
+  const rxAirSecs = readNum(stats, 'rx_air_secs', 'rxAirSecs');
+  const channelUtilization = readNum(
+    stats,
+    'channel_utilization',
+    'channel_utilization_pct',
+    'channel_util',
+    'channelUtil',
+    'channelUtilization',
+  );
+  const airUtilTx = readNum(
+    stats,
+    'air_util_tx',
+    'air_util_tx_pct',
+    'tx_air_util',
+    'tx_air_utilization',
+    'airUtilTx',
+  );
+
+  if (
+    batteryMv == null
+    && uptimeSecs == null
+    && txAirSecs == null
+    && rxAirSecs == null
+    && channelUtilization == null
+    && airUtilTx == null
+  ) {
+    if ((options?.allowRawStatsOnly || hasOwnerStats) && hasRawStats) {
+      return {
+        status,
+        batteryMv,
+        uptimeSecs,
+        txAirSecs,
+        rxAirSecs,
+        channelUtilization,
+        airUtilTx,
+        stats,
+      };
+    }
+    return null;
+  }
+
+  return {
+    status,
+    batteryMv,
+    uptimeSecs,
+    txAirSecs,
+    rxAirSecs,
+    channelUtilization,
+    airUtilTx,
+    stats,
+  };
+}
+
+export async function persistStatusTelemetrySample(
+  json: Record<string, unknown>,
+  options: StatusSampleOptions,
+  writeSample: StatusSampleWriter,
+): Promise<boolean> {
+  const telemetry = extractStatusTelemetry(json, {
+    allowRawStatsOnly: options.allowRawStatsOnly,
+  });
+  if (!telemetry) return false;
+
+  await writeSample({
+    nodeId: options.nodeId,
+    network: options.network,
+    ...telemetry,
+  });
+  return true;
+}