Message history with delivery tracking (GET /api/messages) (#8)

This commit is contained in:
2026-09-24 16:19:32 +02:00
parent 5292303786
commit 5125d27a33
8 changed files with 325 additions and 6 deletions
+1 -1
View File
@@ -17,7 +17,7 @@
## API
1. **Live base URL** — `http://bot.dev.kkhost.pl` (Traefik → docker-compose stack on this host; container port 3000 via external network dev_net, no published ports).
2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: <node number | "!b" | "broadcast"> }`, returns `{ id }`), `GET /api/channels` (the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]`; psk/keys redacted), `GET /api/nodes` (known nodes sorted by num as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? } }[]`, from the initial sync plus live POSITION_APP updates; reset on reconnect) and `GET /api/nodes/:id` (lookup by exact node number/numeric string, then case-insensitive exact long/short name, else unique substring; 404 when ambiguous or missing). Scalar API docs are served at `/` (OpenAPI JSON at `/docs-json`). Keep this list in sync with the controller. All `POST /api/send` traffic targets channel index 2 (`testowy`) on purpose — never the primary channel; override via `MESH_SEND_CHANNEL` (default `2`).
2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: <node number | "!b" | "broadcast"> }`, returns `{ id }`), `GET /api/channels` (the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]`; psk/keys redacted), `GET /api/messages` (message history newest first: rx text with `from`, tx sends with `to` + `status` pending→delivered/failed via routing ACKs — broadcasts stay pending because firmware strips want_ack; last 100, in-memory, reset on reconnect), `GET /api/nodes` (known nodes sorted by num as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? } }[]`, from the initial sync plus live POSITION_APP updates; reset on reconnect) and `GET /api/nodes/:id` (lookup by exact node number/numeric string, then case-insensitive exact long/short name, else unique substring; 404 when ambiguous or missing). Scalar API docs are served at `/` (OpenAPI JSON at `/docs-json`). Keep this list in sync with the controller. All `POST /api/send` traffic targets channel index 2 (`testowy`) on purpose — never the primary channel; override via `MESH_SEND_CHANNEL` (default `2`).
## Other
+6
View File
@@ -1,5 +1,11 @@
# Changelog
## v0.3.0 — 2026-09-24 14:08
- Message history with delivery tracking, exposed as `GET /api/messages` (newest first): inbound TEXT_MESSAGE_APP packets (`dir:'rx'`, with `from`) and our sends (`dir:'tx'`, with `to` + `status`). tx records start `pending` and move to `delivered`/`failed` when the routing ACK/NACK arrives — ROUTING_APP packets carry a serialized `meshtastic.Routing` in `Data.payload` whose oneof variant is `errorReason` (`Routing_Error`: 0 NONE = delivered, anything else failed with a human-readable `ackError`, e.g. NO_ROUTE/GOT_NAK/TIMEOUT, fallback `reason <n>`), matched to our sends via `Data.requestId`. Broadcasts stay `pending` because firmware strips want_ack pre-TX (documented behavior, no timer). History is a bounded in-memory buffer of the last 100 records (newest kept) and resets on every bridge reconnect alongside channels/nodes.
- Codec: new `PORT_ROUTING_APP = 5` constant (verified against the installed `@meshtastic/protobufs` PortNum enum), `decodeRouting()` for ROUTING_APP payloads, and `describeRoutingError()` which maps a Routing.Error code to its name from the installed package's `Mesh.Routing_Error` enum.
- Swagger: `GET /api/messages` documented with the pending→delivered/failed lifecycle, the broadcast caveat, and an example matching the record shape; README + AGENTS.md endpoint lists updated.
## v0.2.0 — 2026-09-24 13:08
- Node database collected during the initial sync (`FromRadio.nodeInfo` frames, upserted by node number) and kept fresh by live `POSITION_APP` packets (the Position is serialized in `Data.payload`; unknown senders get a stub entry). The map resets on every bridge reconnect alongside channels.
+2
View File
@@ -28,6 +28,7 @@ All API endpoints live under the `/api` prefix:
- `GET /api/status` — `{ connected, myNodeNum, host, port }`
- `POST /api/send` — body `{ "text": "...", "to"?: <node number | "!b" | "broadcast"> }`, returns `{ id }`
- `GET /api/channels` — the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]` (psk/keys are redacted; empty until the initial sync completes)
- `GET /api/messages` — message history, newest first: inbound text (`dir:'rx'`, with `from`) and our sends (`dir:'tx'`, with `to` + `status`). tx records start `pending` and move to `delivered`/`failed` (with human-readable `ackError`) when the routing ACK/NACK arrives; broadcasts stay `pending` because firmware strips want_ack before TX. In-memory only, last 100, reset on every bridge reconnect
- `GET /api/nodes` — known mesh nodes as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? } }[]`, sorted by node number. Collected during the initial sync plus live position updates; names/hwModel/battery appear only when reported, `lastHeard` is epoch seconds (null = never heard), and `position` only when both coordinates are known
- `GET /api/nodes/:id` — one node looked up by exact node number (numbers or numeric strings), then case-insensitive exact long/short name, else the unique case-insensitive substring match; 404 when nothing or more than one node matches
@@ -39,6 +40,7 @@ curl http://localhost:3000/api/status
curl -X POST http://localhost:3000/api/send -H 'content-type: application/json' -d '{"text":"hi"}'
curl -X POST http://localhost:3000/api/send -H 'content-type: application/json' -d '{"to":123456789,"text":"hello"}'
curl http://localhost:3000/api/channels
curl http://localhost:3000/api/messages
curl http://localhost:3000/api/nodes
curl http://localhost:3000/api/nodes/kamil
```
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "mesh-bot",
"version": "0.2.0",
"version": "0.3.0",
"description": "",
"author": "",
"private": true,
+19
View File
@@ -12,6 +12,8 @@ export const BROADCAST_ADDR = 0xffffffff;
export const PORT_TEXT_MESSAGE_APP = 1;
// verified against installed @meshtastic/protobufs Portnums.PortNum.POSITION_APP
export const PORT_POSITION_APP = 3;
// verified against installed @meshtastic/protobufs Portnums.PortNum.ROUTING_APP
export const PORT_ROUTING_APP = 5;
const START1 = 0x94;
const START2 = 0xc3;
@@ -71,6 +73,23 @@ export async function decodePosition(
return fromBinary(P.Mesh.PositionSchema, bytes);
}
/** ROUTING_APP packets carry a serialized meshtastic.Routing in Data.payload (ACK/NACK for want_ack sends). */
export async function decodeRouting(
bytes: Uint8Array,
): Promise<import('@meshtastic/protobufs').Mesh.Routing> {
const P = await loadProto();
return fromBinary(P.Mesh.RoutingSchema, bytes);
}
/** Human-readable name for a Routing.Error code (from the installed @meshtastic/protobufs enum), `reason <n>` fallback. */
export async function describeRoutingError(code: number): Promise<string> {
const P = await loadProto();
const hit = Object.entries(P.Mesh.Routing_Error).find(
([, v]) => typeof v === 'number' && v === code,
);
return hit ? String(hit[0]) : `reason ${code}`;
}
let hwModelNames: Map<number, string> | null = null;
/** Reverse lookup of meshtastic.HardwareModel (enum value -> name), loaded once. */
export async function loadHwModelNames(): Promise<Map<number, string>> {
+33
View File
@@ -107,6 +107,39 @@ export class MeshController {
return this.mesh.channels();
}
@Get('messages')
@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.',
})
@ApiResponse({
status: 200,
description: 'Messages newest first',
example: [
{
id: 456789,
dir: 'tx',
to: 42,
text: 'hello testowy',
channel: 2,
at: 1758000123456,
status: 'delivered',
},
{
id: 123456,
dir: 'rx',
from: 987654321,
text: 'ping',
channel: 2,
at: 1758000000000,
},
],
})
messages() {
return this.mesh.messages();
}
@Get('nodes')
@ApiOperation({
summary: 'Known mesh nodes (lean summaries)',
+175
View File
@@ -81,6 +81,63 @@ function positionPacketFrame(
}
const KAMIL_NUM = 123456789;
// 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,
errorReason: P.Mesh.Routing_Error,
): Uint8Array {
const data = create(P.Mesh.DataSchema, {
portnum: P.Portnums.PortNum.ROUTING_APP,
requestId,
payload: toBinary(
P.Mesh.RoutingSchema,
create(P.Mesh.RoutingSchema, {
variant: { case: 'errorReason', value: errorReason },
}),
),
});
const pkt = create(P.Mesh.MeshPacketSchema, {
from: 1,
id: 2,
payloadVariant: { case: 'decoded', value: data },
});
return frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'packet', value: pkt },
}),
),
);
}
function kamilNodeInfo(): Partial<P.Mesh.NodeInfo> {
return {
user: {
@@ -437,3 +494,121 @@ describe('MeshService node database', () => {
}
});
});
describe('MeshService message history', () => {
it('captures inbound text as rx records', 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,
);
bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2));
await waitFor(() => svc.messages().length === 1);
expect(svc.messages()).toEqual([
{
id: 1,
dir: 'rx',
from: KAMIL_NUM,
text: 'ping',
channel: 2,
at: expect.any(Number),
},
]);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('tracks a tx send pending -> delivered on the routing ACK', 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 id = await svc.sendText(42, 'hello testowy');
expect(svc.messages()[0]).toMatchObject({
id,
dir: 'tx',
to: 42,
text: 'hello testowy',
channel: 2,
status: 'pending',
});
bridge.sockets[0].write(routingFrame(id, P.Mesh.Routing_Error.NONE));
await waitFor(() => svc.messages()[0]?.status === 'delivered');
expect(svc.messages()[0]).toMatchObject({
id,
dir: 'tx',
status: 'delivered',
});
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('marks a tx failed with the routing error reason on NACK', 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 id = await svc.sendText(42, 'will fail');
expect(svc.messages()[0]?.status).toBe('pending');
bridge.sockets[0].write(routingFrame(id, P.Mesh.Routing_Error.TIMEOUT));
await waitFor(() => svc.messages()[0]?.status === 'failed');
expect(svc.messages()[0]).toMatchObject({
id,
status: 'failed',
ackError: 'TIMEOUT', // human-readable Routing.Error name
});
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('caps history at 100 records, dropping the oldest', 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,
);
for (let i = 0; i < 105; i++)
bridge.sockets[0].write(textPacketFrame(42, KAMIL_NUM, `m-${i}`));
await waitFor(() => svc.messages().length === 100, 10000);
expect(svc.messages()[0]?.text).toBe('m-104'); // newest first
expect(svc.messages()[99]?.text).toBe('m-5'); // m-0..m-4 dropped
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
});
+88 -4
View File
@@ -9,16 +9,40 @@ import * as net from 'node:net';
import type * as Proto from '@meshtastic/protobufs';
import {
PORT_POSITION_APP,
PORT_ROUTING_APP,
PORT_TEXT_MESSAGE_APP,
buildTextToRadio,
buildWantConfigToRadio,
decodeFromRadio,
decodePosition,
decodeRouting,
describeRoutingError,
extractFrames,
frame,
loadHwModelNames,
} from './codec.js';
/** One entry in the message history: an inbound text (dir 'rx') or one of our sends (dir 'tx'). */
export interface MessageRecord {
id: number;
dir: 'rx' | 'tx';
/** sender node num — rx only */
from?: number;
/** destination node num (broadcast addr for broadcasts) — tx only */
to?: number;
text: string;
channel: number;
/** epoch ms */
at: number;
/** tx only: pending until the routing ACK/NACK arrives; broadcasts stay pending (firmware strips want_ack pre-TX) */
status?: 'pending' | 'delivered' | 'failed';
/** tx only, when failed: human-readable Routing.Error reason */
ackError?: string;
}
// bounded history: keep the newest N records, in-memory only
const HISTORY_LIMIT = 100;
/** 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'> &
Partial<Omit<Proto.Mesh.NodeInfo, 'num'>>;
@@ -63,6 +87,9 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
>();
// node database: NodeInfo frames from the initial sync + live POSITION_APP updates
private nodeMap = new Map<number, NodeEntry>();
// message history (newest kept, capped at HISTORY_LIMIT) + pending tx records awaiting their routing ACK/NACK
private history: MessageRecord[] = [];
private pendingTx = new Map<number, MessageRecord>();
private reconnectTimer: NodeJS.Timeout | null = null;
private shuttingDown = false;
@@ -149,15 +176,41 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
return null;
}
/**
* 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.
*/
messages(): MessageRecord[] {
return [...this.history].reverse();
}
/** Send `text` to node `to` on the pinned test channel; returns the packet id. Throws ServiceUnavailable when the bridge is down. */
async sendText(to: number, text: string): Promise<number> {
if (!this.connected || !this.socket)
throw new ServiceUnavailableException('bridge not connected');
const { bytes, id } = await buildTextToRadio(to, text, this.sendChannel);
const rec: MessageRecord = {
id,
dir: 'tx',
to,
text,
channel: this.sendChannel,
at: Date.now(),
status: 'pending',
};
this.pushHistory(rec);
this.pendingTx.set(id, rec);
this.socket.write(bytes);
return id;
}
private pushHistory(rec: MessageRecord): void {
this.history.push(rec);
// ponytail: shift on a ≤100 array is fine; true ring buffer only if this ever gets hot
if (this.history.length > HISTORY_LIMIT) this.history.shift();
}
private summarize(n: NodeEntry, names: Map<number, string>): NodeSummary {
const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null };
if (n.user?.longName) s.longName = n.user.longName;
@@ -183,9 +236,11 @@ 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 from the old session
// fresh sync on every (re)connect — drop stale channels/nodes/history from the old session
this.channelMap = new Map();
this.nodeMap = new Map();
this.history = [];
this.pendingTx.clear();
this.logger.log(`bridge connected ${this.host}:${this.port}`);
const nonce = Math.floor(Math.random() * 0xffffffff) || 1;
buildWantConfigToRadio(nonce)
@@ -256,9 +311,16 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
? pkt.payloadVariant.value
: undefined;
if (decoded && decoded.portnum === PORT_TEXT_MESSAGE_APP) {
this.logger.log(
`[mesh] ${pkt.from} -> ${pkt.to}: "${new TextDecoder().decode(decoded.payload)}"`,
);
const text = new TextDecoder().decode(decoded.payload);
this.logger.log(`[mesh] ${pkt.from} -> ${pkt.to}: "${text}"`);
this.pushHistory({
id: pkt.id,
dir: 'rx',
from: pkt.from,
text,
channel: pkt.channel ?? 0,
at: Date.now(),
});
} else if (decoded && decoded.portnum === PORT_POSITION_APP) {
// live position update: the Position is serialized in Data.payload
let pos: Proto.Mesh.Position | undefined;
@@ -275,6 +337,28 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
position: pos,
});
}
} else if (decoded && decoded.portnum === PORT_ROUTING_APP) {
// ACK/NACK for our want_ack sends: Routing serialized in Data.payload,
// Data.requestId = the original packet id we sent
let routing: Proto.Mesh.Routing | undefined;
try {
routing = await decodeRouting(decoded.payload);
} catch {
routing = undefined; // malformed payload — skip
}
if (routing?.variant.case === 'errorReason' && decoded.requestId) {
const rec = this.pendingTx.get(decoded.requestId);
if (rec) {
const reason = routing.variant.value;
if (reason === 0)
rec.status = 'delivered'; // Routing_Error.NONE
else {
rec.status = 'failed';
rec.ackError = await describeRoutingError(reason);
}
this.pendingTx.delete(decoded.requestId);
}
}
}
break;
}