fix(mqtt): skip offline status telemetry samples (MQTT Will) (#98)

* fix(mqtt): skip offline status telemetry samples

* docs: add MQTT Will telemetry handoff

* chore(db): renumber MQTT status migration to 057

052 is already taken on main by owner packet sharing; 055 predates main (unmerged branch) and is left untouched; 057 is the next free number after 056. Also appends the migration to the fresh-database expectation in packetBatch.integration.test.ts in sorted order.

---------

Co-authored-by: gadgethd <111318106+gadgethd@users.noreply.github.com>
Co-authored-by: hermes-gadget <284726080+hermes-gadget@users.noreply.github.com>

HANDOFF.md

@@ -1,29 +1,44 @@
-# B3 packet-sharing handoff
+# Handoff — Mission B2 · MQTT Will fake telemetry (#332)
 
 ## Status
 
-Backend packet-sharing routes, contracts, persistence, forwarding, and tests are committed locally in `fbf9803` (`feat(owner): add owner-controlled packet sharing`). No migration was applied, no service was restarted, and nothing was pushed.
+Done. Offline Will payloads are skipped for status samples, ordinary telemetry stores its top-level status, and the schema migration is included but unapplied.
 
-## WIP inventory
+## Branch and commits
 
-The packet-sharing slice rescued from the main checkout contains migrations 052–054 for destination/rule/delivery tables, destination metadata, and default-off rules; owner GET/POST routes and contracts; broker-secret encryption; destination configuration and delivery-queue services; MQTT forwarding and worker wiring; and route/config/forwarder tests. The main checkout also has mixed public-feed, tagger, owner-portal, and unrelated backend work.
+- Branch: `fix/mqtt-will-telemetry`
+- Base: `d44ef5e` (`origin/main`)
+- Implementation: `ad473779ce0f1ff250f6a9494d8abe9abba7a9f1` — `fix(mqtt): skip offline status telemetry samples`
+- Push state: not pushed; coordinator relays this branch.
 
-The main checkout and its WIP remain untouched. The packet-sharing backend copy there becomes redundant after this change is merged and deployed. The mixed owner-portal UI edits were not included in the backend scope in this brief and need separate disposition if they are not handled elsewhere.
+## Files changed
 
-## Migration decision
+- `backend/src/mqtt/client.ts`
+- `backend/src/mqtt/statusTelemetry.ts`
+- `backend/src/mqtt/statusTelemetry.test.ts`
+- `backend/src/db/index.ts`
+- `backend/src/db/schema/base.sql`
+- `backend/src/db/migrations/057_node_status_sample_status.sql`
 
-Kept the supplied `052_owner_packet_sharing.sql`, `053_owner_packet_share_destination_metadata.sql`, and `054_owner_packet_share_default_off.sql` names. No duplicate prefix exists for these files in the target worktree, so no renumbering was needed. Prefix 055 is reserved by `055_message_tags.sql` on `fix/health-latency-bundle`; the separate untracked `055_public_feed.sql` in the main checkout was not copied.
+## Checks
 
-## Tests
+Commands were run from `backend/`:
 
-- `cd backend && npm test` — passed, 341 tests.
-- `cd backend && npm run typecheck` — passed.
-- `git diff --check` — passed.
+- `node --import tsx --test src/mqtt/statusTelemetry.test.ts` — passed, 2 tests.
+- `npm test` — passed, 326 tests, 0 failures. The script excludes `*.integration.test.ts`.
+- `npm run typecheck` — passed.
 
-The worktree had no installed dependencies. Tests reused the existing main-checkout `backend/node_modules` through a temporary symlink in this worktree; the symlink was removed afterward. No tests wrote to a live database.
+## Gated items for Ben
 
-## Gated items and open questions
+- Apply migration `057_node_status_sample_status.sql` through the normal deployment process, then deploy the ingest change. No live DB writes, migration application, or service restarts were performed in this worktree session.
+- Historical rows and retained broker messages were not modified; clearing retained messages is out of scope for #332.
 
-- Ben owns migration application and deployment.
-- The target worktree does not contain the `OWNER_PACKET_SHARE_*` Compose passthroughs or `owner_packet_share_deliveries` in the retention target list described by the brief. No `.env` or config files were changed under the VPS constraint. Before deployment, Ben must confirm the runtime passes `OWNER_PACKET_SHARE_ENCRYPTION_KEY` to the backend and includes `owner_packet_share_deliveries` in lifecycle retention targets. Poll interval and batch size have code defaults.
-- Confirm whether the packet-sharing owner-portal UI edits in the main checkout are tracked in another change; they were left untouched and are not in `fbf9803`.
+## Open questions
+
+None.
+
+## Evidence paths
+
+- `BRIEF.md` — mission and constraints.
+- `backend/src/mqtt/statusTelemetry.test.ts` — offline Will skip, ordinary sample storage, and top-level status coverage.
+- `backend/src/db/migrations/057_node_status_sample_status.sql` — additive status column migration.

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/057_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/packetBatch.integration.test.ts

@@ -107,6 +107,7 @@ test('base schema and every migration apply to a brand-new database', {
     '053_owner_packet_share_destination_metadata.sql',
     '054_owner_packet_share_default_off.sql',
     '056_async_privacy_rematerialization.sql',
+    '057_node_status_sample_status.sql',
   ]);
   assert.deepEqual(await runMigrations(freshPool), []);
 

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;
+}