mesh: POST /api/respond + MCP respond tool — reply to a message by packet id

2026-09-25 15:32:34 +00:00
parent 88b617f8b7
commit c33f4613b5
6 changed files with 422 additions and 44 deletions
@@ -1,5 +1,6 @@
 import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
 import * as P from '@meshtastic/protobufs';
+import { BadRequestException, NotFoundException } from '@nestjs/common';
 import { DatabaseSync } from 'node:sqlite';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { BROADCAST_ADDR, extractFrames, frame } from './codec.js';
@@ -12,6 +13,7 @@ import {
  nodeInfoFrame,
  startBridge,
  telemetryPacketFrame,
+  textPacketFrame,
  waitFor,
 } from './fake-bridge.js';
 import { OpencodeService } from '../opencode/opencode.service.js';
@@ -74,34 +76,6 @@ function nodeInfoPacketFrame(
  );
 }

-// inbound text message: TEXT_MESSAGE_APP packet with UTF-8 in Data.payload
-function textPacketFrame(
-  from: number,
-  to: number,
-  text: string,
-  channel = 0,
-): Uint8Array {
-  const data = create(P.Mesh.DataSchema, {
-    portnum: P.Portnums.PortNum.TEXT_MESSAGE_APP,
-    payload: new TextEncoder().encode(text),
-  });
-  const pkt = create(P.Mesh.MeshPacketSchema, {
-    from,
-    to,
-    id: 1,
-    channel,
-    payloadVariant: { case: 'decoded', value: data },
-  });
-  return frame(
-    toBinary(
-      P.Mesh.FromRadioSchema,
-      create(P.Mesh.FromRadioSchema, {
-        payloadVariant: { case: 'packet', value: pkt },
-      }),
-    ),
-  );
-}
-
 // ACK/NACK for our sends: ROUTING_APP packet with Data.requestId = original id and a serialized Routing in the payload
 function routingFrame(
  requestId: number,
@@ -693,6 +667,228 @@ describe('MeshService message history', () => {
  });
 });

+// outbound text packets written to the bridge, in order (want_config frames skipped)
+function outboundPackets(out: Uint8Array[]): P.Mesh.MeshPacket[] {
+  const merged = new Uint8Array(out.reduce((n, c) => n + c.length, 0));
+  let off = 0;
+  for (const c of out) {
+    merged.set(c, off);
+    off += c.length;
+  }
+  return extractFrames(merged).packets.flatMap((p) => {
+    const tr = fromBinary(P.Mesh.ToRadioSchema, p);
+    if (tr.payloadVariant?.case === 'packet') return [tr.payloadVariant.value];
+    return [];
+  });
+}
+
+describe('MeshService respond', () => {
+  it('replies to an rx DM as a directed send back to the sender on the original channel with replyId = messageId', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    const svc = new MeshService();
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      // rx DM from KAMIL to our node on channel 2 (packet id 1)
+      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2));
+      await waitFor(() => svc.messages().length === 1);
+
+      const out: Uint8Array[] = [];
+      bridge.sockets[0].on('data', (c) => out.push(c));
+      const { id } = await svc.respond(1, 'pong');
+
+      expect(id).toBeGreaterThan(0);
+      // DM back to the original sender on the original channel
+      expect(svc.messages()[0]).toMatchObject({
+        id,
+        dir: 'tx',
+        to: KAMIL_NUM,
+        text: 'pong',
+        channel: 2,
+        status: 'pending',
+      });
+
+      await waitFor(() => outboundPackets(out).length === 1);
+      const pkt = outboundPackets(out)[0]!;
+      expect(pkt.to).toBe(KAMIL_NUM);
+      expect(pkt.channel).toBe(2);
+      const data =
+        pkt.payloadVariant?.case === 'decoded'
+          ? pkt.payloadVariant.value
+          : undefined;
+      expect(data?.replyId).toBe(1); // threaded quote of the original packet
+    } finally {
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('replies to a broadcast original as a broadcast on the original channel with replyId = messageId', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    const svc = new MeshService();
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      // rx broadcast from KAMIL on channel 2 (packet id 1)
+      bridge.sockets[0].write(
+        textPacketFrame(KAMIL_NUM, BROADCAST_ADDR, 'hello all', 2),
+      );
+      await waitFor(() => svc.messages().length === 1);
+
+      const out: Uint8Array[] = [];
+      bridge.sockets[0].on('data', (c) => out.push(c));
+      const { id } = await svc.respond(1, 'pong');
+
+      expect(id).toBeGreaterThan(0);
+      expect(svc.messages()[0]).toMatchObject({
+        dir: 'tx',
+        to: BROADCAST_ADDR, // broadcast original → broadcast reply
+        text: 'pong',
+        channel: 2,
+      });
+
+      await waitFor(() => outboundPackets(out).length === 1);
+      const pkt = outboundPackets(out)[0]!;
+      expect(pkt.to).toBe(BROADCAST_ADDR);
+      expect(pkt.channel).toBe(2);
+      const data =
+        pkt.payloadVariant?.case === 'decoded'
+          ? pkt.payloadVariant.value
+          : undefined;
+      expect(data?.replyId).toBe(1);
+    } finally {
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('replies to a tx record targeting its destination on its channel with replyId = messageId', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    const svc = new MeshService();
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      const origId = await svc.sendText(42, 'orig'); // tx record: to=42, default channel 3
+      const out: Uint8Array[] = [];
+      bridge.sockets[0].on('data', (c) => out.push(c));
+      const { id } = await svc.respond(origId, 'reply');
+
+      expect(id).toBeGreaterThan(0);
+      expect(svc.messages()[0]).toMatchObject({
+        dir: 'tx',
+        to: 42, // the tx record's own destination
+        text: 'reply',
+        channel: 3,
+      });
+
+      // the orig send may share a chunk with the reply — take the newest packet
+      await waitFor(() => outboundPackets(out).length >= 1);
+      const pkts = outboundPackets(out);
+      const pkt = pkts[pkts.length - 1]!;
+      expect(pkt.to).toBe(42);
+      const data =
+        pkt.payloadVariant?.case === 'decoded'
+          ? pkt.payloadVariant.value
+          : undefined;
+      expect(data?.replyId).toBe(origId);
+    } finally {
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('throws NotFoundException when no message has the given id (memory or SQLite)', async () => {
+    const svc = new MeshService();
+    await expect(svc.respond(999, 'hi')).rejects.toBeInstanceOf(
+      NotFoundException,
+    );
+    await expect(svc.respond(999, 'hi')).rejects.toThrow(
+      'no message with id 999',
+    );
+
+    const db = new DatabaseSync(':memory:'); // empty table — the DB lookup path also misses
+    initMessagesTable(db);
+    const svc2 = new MeshService(db);
+    await expect(svc2.respond(999, 'hi')).rejects.toBeInstanceOf(
+      NotFoundException,
+    );
+  });
+
+  it('falls back to SQLite when the record is not in the in-memory buffer', async () => {
+    const db = new DatabaseSync(':memory:');
+    initMessagesTable(db);
+    // 150 rows — onModuleInit seeds only the newest 100 into the buffer (ids 2050..2149)
+    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(
+        2000 + i,
+        'rx',
+        42,
+        KAMIL_NUM,
+        `old-${i}`,
+        2,
+        1_700_000_000_000 + i * 1000,
+        null,
+        null,
+      );
+
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    const svc = new MeshService(db);
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      // id 2000 is in SQLite but outside the seeded buffer
+      const { id } = await svc.respond(2000, 'pong');
+      expect(id).toBeGreaterThan(0);
+      expect(svc.messages()[0]).toMatchObject({
+        dir: 'tx',
+        to: KAMIL_NUM, // DM back to the sender of the old rx record
+        text: 'pong',
+        channel: 2, // the record's channel
+      });
+    } finally {
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('rejects invalid messageId and empty content with BadRequestException', async () => {
+    const svc = new MeshService();
+    for (const bad of [0, BROADCAST_ADDR + 1, 42.5, 'x' as unknown as number])
+      await expect(svc.respond(bad, 'hi')).rejects.toBeInstanceOf(
+        BadRequestException,
+      );
+    await expect(svc.respond(1, '')).rejects.toThrow(
+      'content must be a non-empty string',
+    );
+    await expect(svc.respond(1, undefined)).rejects.toBeInstanceOf(
+      BadRequestException,
+    );
+  });
+});
+
 describe('MeshService opencode notifier', () => {
  // our own node number for the self-echo filter specs (distinct from KAMIL_NUM, an external user)
  const MY_NODE = 987654321;