Persist message history to SQLite with 1s poller (#15) (#19)

src/mesh/mcp.controller.ts

@@ -72,7 +72,7 @@ export function createMeshMcpServer(mesh: MeshService): McpServer {
   server.registerTool(
     'get_messages',
     {
-      description: `Message history, newest first (last 100): inbound text (dir "rx", with from) and our sends (dir "tx", with to + status pending→delivered/failed; broadcasts stay pending because firmware strips want_ack). In-memory only, reset on every bridge reconnect.${CHANNEL_NOTE}`,
+      description: `Message history, newest first (last 100): inbound text (dir "rx", with from) and our sends (dir "tx", with to + status pending→delivered/failed; broadcasts stay pending because firmware strips want_ack). Backed by SQLite — seeded on startup and re-upserted every second, so it survives restarts and bridge reconnects (the database keeps the newest 1000 rows).${CHANNEL_NOTE}`,
     },
     async () => jsonResult(mesh.messages()),
   );

src/mesh/mesh.controller.ts

@@ -166,7 +166,7 @@ export class MeshController {
   @ApiOperation({
     summary: 'Message history (last 100)',
     description:
-      'Inbound text messages and our sends, newest first. tx records start "pending" and move to "delivered"/"failed" when the routing ACK/NACK arrives; broadcasts stay "pending" because firmware strips want_ack before TX. History is in-memory only (last 100) and resets on every bridge reconnect.',
+      'Inbound text messages and our sends, newest first. tx records start "pending" and move to "delivered"/"failed" when the routing ACK/NACK arrives; broadcasts stay "pending" because firmware strips want_ack before TX. History is backed by SQLite: the in-memory buffer (last 100) is seeded from the database on startup and re-upserted every second, so it survives process restarts and bridge reconnects; the database keeps the newest 1000 rows.',
   })
   @ApiResponse({
     status: 200,

src/mesh/mesh.module.ts

@@ -1,10 +1,27 @@
 import { Module } from '@nestjs/common';
+import { mkdirSync } from 'node:fs';
+import { dirname } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
 import { McpController } from './mcp.controller.js';
 import { MeshController } from './mesh.controller.js';
-import { MeshService } from './mesh.service.js';
+import { initMessagesTable, MeshService } from './mesh.service.js';
+
+// SQLite-backed message history: file path from MESH_DB_PATH (default ./data/mesh-bot.db), parent dir created if missing.
+function openMessageDb(): DatabaseSync {
+  const path = process.env.MESH_DB_PATH ?? './data/mesh-bot.db';
+  mkdirSync(dirname(path), { recursive: true });
+  const db = new DatabaseSync(path);
+  initMessagesTable(db);
+  return db;
+}
 
 @Module({
   controllers: [MeshController, McpController],
-  providers: [MeshService],
+  providers: [
+    {
+      provide: MeshService,
+      useFactory: () => new MeshService(openMessageDb()),
+    },
+  ],
 })
 export class MeshModule {}

src/mesh/mesh.service.spec.ts

@@ -1,5 +1,6 @@
 import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
 import * as P from '@meshtastic/protobufs';
+import { DatabaseSync } from 'node:sqlite';
 import { describe, expect, it } from 'vitest';
 import { BROADCAST_ADDR, extractFrames, frame } from './codec.js';
 import {
@@ -13,7 +14,7 @@ import {
   telemetryPacketFrame,
   waitFor,
 } from './fake-bridge.js';
-import { MeshService } from './mesh.service.js';
+import { initMessagesTable, MeshService } from './mesh.service.js';
 
 // live position update: POSITION_APP packet with a serialized Position in Data.payload
 function positionPacketFrame(
@@ -987,3 +988,197 @@ describe('MeshService device info', () => {
     }
   });
 });
+
+describe('MeshService SQLite persistence', () => {
+  it('persists an injected message row on poller tick (no manual flush)', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    process.env.MESH_POLL_MS = '50';
+    const db = new DatabaseSync(':memory:');
+    initMessagesTable(db);
+    const svc = new MeshService(db);
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      svc.injectMessage('hello db', KAMIL_NUM);
+
+      // the poller tick must pick it up — no flushHistory() call here
+      await waitFor(
+        () =>
+          (
+            db.prepare('SELECT COUNT(*) AS n FROM messages').get() as {
+              n: number;
+            }
+          ).n === 1,
+      );
+      const row = db
+        .prepare('SELECT * FROM messages WHERE text = ?')
+        .get('hello db') as Record<string, unknown>;
+      expect(row.dir).toBe('rx');
+      expect(row.from_node).toBe(KAMIL_NUM);
+      expect(row.to_node).toBeNull(); // rx rows carry no destination
+      expect(row.channel).toBe(0);
+      expect(row.status).toBeNull();
+    } finally {
+      delete process.env.MESH_POLL_MS;
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('reflects the ACK status update in the database after the next tick', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    process.env.MESH_POLL_MS = '50';
+    const db = new DatabaseSync(':memory:');
+    initMessagesTable(db);
+    const svc = new MeshService(db);
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      const id = await svc.sendText(42, 'ack me');
+      expect(svc.messages()[0]?.status).toBe('pending');
+
+      bridge.sockets[0].write(routingFrame(id, P.Mesh.Routing_Error.NONE));
+      await waitFor(() => svc.messages()[0]?.status === 'delivered');
+
+      // the in-place status mutation must reach SQLite on the next poller tick
+      // (the row may not exist before the first tick — waitFor re-checks, so stay null-safe)
+      await waitFor(
+        () =>
+          (
+            db.prepare('SELECT status FROM messages WHERE id = ?').get(id) as
+              | {
+                  status: string | null;
+                }
+              | undefined
+          )?.status === 'delivered',
+      );
+    } finally {
+      delete process.env.MESH_POLL_MS;
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('seeds the history buffer from a pre-populated database on startup', () => {
+    const db = new DatabaseSync(':memory:');
+    initMessagesTable(db);
+    const ins = db.prepare(
+      'INSERT INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
+    );
+    for (let i = 0; i < 150; i++)
+      ins.run(
+        1000 + i,
+        'rx',
+        null,
+        42,
+        `seed-${i}`,
+        0,
+        1_700_000_000_000 + i * 1000,
+        null,
+        null,
+      );
+
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = '1'; // closed port — no real bridge needed for seeding
+    const svc = new MeshService(db);
+    try {
+      svc.onModuleInit();
+      expect(svc.messages()).toHaveLength(100); // newest 100 of the 150 rows
+      expect(svc.messages()[0]).toMatchObject({
+        id: 1149,
+        dir: 'rx',
+        from: 42,
+        text: 'seed-149',
+        channel: 0,
+      }); // newest first
+      expect(svc.messages()[99]).toMatchObject({ id: 1050, text: 'seed-50' });
+      expect(svc.messages()[0]).not.toHaveProperty('status'); // rx rows carry no status
+    } finally {
+      delete process.env.MESH_HOST;
+      delete process.env.MESH_PORT;
+      svc.onModuleDestroy();
+    }
+  });
+
+  it('prunes database rows beyond the retention limit (newest 1000)', () => {
+    const db = new DatabaseSync(':memory:');
+    initMessagesTable(db);
+    const ins = db.prepare(
+      'INSERT INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
+    );
+    for (let i = 0; i < 1200; i++)
+      ins.run(
+        5000 + i,
+        'rx',
+        null,
+        42,
+        `old-${i}`,
+        0,
+        1_700_000_000_000 + i * 1000,
+        null,
+        null,
+      );
+
+    const svc = new MeshService(db); // no onModuleInit — flushHistory() is the unit under test
+    try {
+      svc.flushHistory(); // empty buffer: upserts nothing, retention still applies
+      expect(
+        (
+          db.prepare('SELECT COUNT(*) AS n FROM messages').get() as {
+            n: number;
+          }
+        ).n,
+      ).toBe(1000);
+      expect(
+        db.prepare('SELECT text FROM messages WHERE id = ?').get(6199),
+      ).toMatchObject({ text: 'old-1199' }); // newest kept
+      expect(
+        db.prepare('SELECT text FROM messages WHERE id = ?').get(5199),
+      ).toBeUndefined(); // oldest 200 pruned (i=0..199)
+    } finally {
+      svc.onModuleDestroy();
+    }
+  });
+
+  it('keeps the message history across a bridge reconnect', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    process.env.MESH_RECONNECT_MS = '50';
+    const db = new DatabaseSync(':memory:');
+    initMessagesTable(db);
+    const svc = new MeshService(db);
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      svc.injectMessage('survives reconnect', KAMIL_NUM);
+      expect(svc.messages()).toHaveLength(1);
+
+      // drop the connection; bot reconnects and must keep the history (unlike channels/nodes/telemetry)
+      bridge.sockets[0].destroy();
+      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
+      expect(svc.messages()).toHaveLength(1);
+      expect(svc.messages()[0]).toMatchObject({
+        dir: 'rx',
+        text: 'survives reconnect',
+      });
+    } finally {
+      delete process.env.MESH_RECONNECT_MS;
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+});

src/mesh/mesh.service.ts

@@ -6,6 +6,8 @@ import {
   ServiceUnavailableException,
 } from '@nestjs/common';
 import * as net from 'node:net';
+// type-only import: node:sqlite is only required at runtime by whoever opens a database (mesh.module.ts / tests)
+import type { DatabaseSync } from 'node:sqlite';
 import type * as Proto from '@meshtastic/protobufs';
 import {
   BROADCAST_ADDR,
@@ -56,8 +58,28 @@ export interface TelemetryRecord {
   data: object;
 }
 
-// bounded history: keep the newest N records, in-memory only
+// bounded in-memory buffer: keep the newest N records (SQLite keeps more — see MESSAGES_DDL)
 const HISTORY_LIMIT = 100;
+/** Rows kept in SQLite across restarts; older rows are pruned on every poller tick. */
+const DB_RETENTION = 1000;
+
+// one row per MessageRecord, keyed by packet id (tx ids and injected ids are random uint32, rx uses the wire id)
+export const MESSAGES_DDL = `CREATE TABLE IF NOT EXISTS messages(
+  id INTEGER PRIMARY KEY,
+  dir TEXT NOT NULL,
+  to_node INTEGER,
+  from_node INTEGER,
+  text TEXT NOT NULL,
+  channel INTEGER NOT NULL,
+  at INTEGER NOT NULL,
+  status TEXT,
+  ack_error TEXT
+)`;
+
+/** Create the messages table if missing (idempotent — safe on every startup). */
+export function initMessagesTable(db: DatabaseSync): void {
+  db.exec(MESSAGES_DDL);
+}
 
 /** What we know about a node so far: full NodeInfo after sync, possibly just {num}+position for live-only nodes. */
 type NodeEntry = Pick<Proto.Mesh.NodeInfo, 'num'> &
@@ -135,6 +157,16 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
     process.env.MESH_SEND_CHANNEL ?? '3',
     10,
   );
+  // how often the poller re-upserts the history buffer into SQLite (ms)
+  private readonly pollMs = parseInt(process.env.MESH_POLL_MS ?? '1000', 10);
+
+  /** SQLite database backing the message history; null = persistence disabled (tests construct without one). */
+  private db: DatabaseSync | null = null;
+  private pollTimer: NodeJS.Timeout | null = null;
+
+  constructor(db?: DatabaseSync) {
+    this.db = db ?? null;
+  }
 
   private socket: net.Socket | null = null;
   private buffer: Uint8Array = new Uint8Array(0);
@@ -160,13 +192,21 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
   private shuttingDown = false;
 
   onModuleInit(): void {
+    if (this.db) {
+      initMessagesTable(this.db);
+      this.seedHistory(); // history survives process restarts
+      // ponytail: full re-upsert of the ≤100-row buffer every tick — trivial load, zero bookkeeping
+      this.pollTimer = setInterval(() => this.flushHistory(), this.pollMs);
+    }
     this.connect();
   }
 
   onModuleDestroy(): void {
     this.shuttingDown = true;
     if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
+    if (this.pollTimer) clearInterval(this.pollTimer);
     this.socket?.destroy();
+    this.db?.close();
   }
 
   /** Bridge connection state: `{ connected, myNodeNum, host, port }`. */
@@ -298,12 +338,66 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
   /**
    * Message history, newest first: inbound TEXT_MESSAGE_APP packets (dir 'rx') and our sends (dir 'tx').
    * tx records start 'pending' and move to 'delivered'/'failed' when the routing ACK/NACK arrives;
-   * broadcasts stay 'pending' because firmware strips want_ack before TX. In-memory only, last 100, reset on reconnect.
+   * broadcasts stay 'pending' because firmware strips want_ack before TX. Backed by SQLite: the buffer is
+   * seeded from the database on startup (newest 100) and re-upserted every MESH_POLL_MS, so history
+   * survives process restarts and bridge reconnects; the DB keeps the newest 1000 rows.
    */
   messages(): MessageRecord[] {
     return [...this.history].reverse();
   }
 
+  /**
+   * Re-upsert the entire in-memory history buffer into SQLite (INSERT OR REPLACE per row) and enforce
+   * retention: only the newest DB_RETENTION rows survive. Called by the poller every MESH_POLL_MS;
+   * public so tests can drive a tick without real timers. No-op when no database is configured.
+   */
+  flushHistory(): void {
+    const db = this.db;
+    if (!db) return;
+    const upsert = db.prepare(
+      'INSERT OR REPLACE INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
+    );
+    for (const r of this.history) {
+      upsert.run(
+        r.id,
+        r.dir,
+        r.to ?? null,
+        r.from ?? null,
+        r.text,
+        r.channel,
+        r.at,
+        r.status ?? null,
+        r.ackError ?? null,
+      );
+    }
+    db.exec(
+      `DELETE FROM messages WHERE id NOT IN (SELECT id FROM messages ORDER BY at DESC LIMIT ${DB_RETENTION})`,
+    );
+  }
+
+  /** Load the newest HISTORY_LIMIT rows from SQLite into the buffer (oldest-first) so history survives restarts. */
+  private seedHistory(): void {
+    const rows = this.db!.prepare(
+      'SELECT id, dir, to_node, from_node, text, channel, at, status, ack_error FROM messages ORDER BY at DESC LIMIT ?',
+    ).all(HISTORY_LIMIT) as Array<Record<string, unknown>>;
+    // rows come back newest first — the buffer is oldest-first
+    for (const row of [...rows].reverse()) {
+      const rec: MessageRecord = {
+        id: Number(row.id),
+        dir: row.dir === 'tx' ? 'tx' : 'rx',
+        text: String(row.text),
+        channel: Number(row.channel),
+        at: Number(row.at),
+      };
+      if (row.to_node != null) rec.to = Number(row.to_node);
+      if (row.from_node != null) rec.from = Number(row.from_node);
+      if (row.status != null)
+        rec.status = row.status as MessageRecord['status'];
+      if (row.ack_error != null) rec.ackError = String(row.ack_error);
+      this.history.push(rec);
+    }
+  }
+
   /**
    * Inject a synthetic inbound (rx) record into the message history — as if received on the mesh.
    * NOT transmitted over RF: nothing is written to the bridge or air; the record exists so external
@@ -401,13 +495,13 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
     this.socket = socket;
     socket.on('connect', () => {
       this.connected = true;
-      // fresh sync on every (re)connect — drop stale channels/nodes/device info/history from the old session
+      // fresh sync on every (re)connect — drop stale channels/nodes/device info/telemetry from the old session;
+      // message history is NOT reset: it's an append-only log backed by SQLite and survives reconnects
       this.channelMap = new Map();
       this.nodeMap = new Map();
       this.metadata = null;
       this.configSections.clear();
       this.moduleConfigSections.clear();
-      this.history = [];
       this.telemetryHistory = [];
       this.pendingTx.clear();
       this.logger.log(`bridge connected ${this.host}:${this.port}`);