Files
MeshBot/src/mesh/mesh.service.spec.ts
T

1513 lines
48 KiB
TypeScript

import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
import * as P from '@meshtastic/protobufs';
import { DatabaseSync } from 'node:sqlite';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { BROADCAST_ADDR, extractFrames, frame } from './codec.js';
import {
KAMIL_NUM,
configSectionFrame,
kamilNodeInfo,
metadataFrame,
moduleConfigSectionFrame,
nodeInfoFrame,
startBridge,
telemetryPacketFrame,
waitFor,
} from './fake-bridge.js';
import { OpencodeService } from '../opencode/opencode.service.js';
import { initMessagesTable, MeshService } from './mesh.service.js';
// live position update: POSITION_APP packet with a serialized Position in Data.payload
function positionPacketFrame(
from: number,
pos: { latitudeI?: number; longitudeI?: number },
): Uint8Array {
const data = create(P.Mesh.DataSchema, {
portnum: P.Portnums.PortNum.POSITION_APP,
payload: toBinary(
P.Mesh.PositionSchema,
create(P.Mesh.PositionSchema, pos),
),
});
const pkt = create(P.Mesh.MeshPacketSchema, {
from,
id: 1,
payloadVariant: { case: 'decoded', value: data },
});
return frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'packet', value: pkt },
}),
),
);
}
// live node-info update: NODEINFO_APP packet with a serialized NodeInfo in Data.payload
function nodeInfoPacketFrame(
from: number,
fields: {
user?: { id?: string; longName?: string };
deviceMetrics?: { batteryLevel?: number };
},
): Uint8Array {
const data = create(P.Mesh.DataSchema, {
portnum: P.Portnums.PortNum.NODEINFO_APP,
payload: toBinary(
P.Mesh.NodeInfoSchema,
create(P.Mesh.NodeInfoSchema, { num: from, ...fields }),
),
});
const pkt = create(P.Mesh.MeshPacketSchema, {
from,
id: 1,
payloadVariant: { case: 'decoded', value: data },
});
return frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'packet', value: pkt },
}),
),
);
}
// 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 channelFrame(
index: number,
settings?: { name?: string; uplinkEnabled?: boolean },
): Uint8Array {
const ch = create(P.Channel.ChannelSchema, {
index,
...(settings
? {
// psk must never leak into channels() output
settings: {
name: settings.name ?? '',
psk: new Uint8Array(16).fill(7),
uplinkEnabled: settings.uplinkEnabled ?? false,
},
}
: {}),
});
const fr = create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'channel', value: ch },
});
return frame(toBinary(P.Mesh.FromRadioSchema, fr));
}
describe('MeshService channel pipeline', () => {
it('collects channels from FromRadio frames and redacts keys', 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 {
expect(svc.channels()).toEqual([]);
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(
channelFrame(2, { name: 'LongFast', uplinkEnabled: true }),
);
bridge.sockets[0].write(channelFrame(0)); // channel with no settings at all
await waitFor(() => svc.channels().length === 2);
expect(svc.channels()).toEqual([
{ index: 0, name: '', uplinkEnabled: false, downlinkEnabled: false },
{
index: 2,
name: 'LongFast',
uplinkEnabled: true,
downlinkEnabled: false,
},
]);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('resets channels on reconnect (no stale data)', 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 svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(channelFrame(1, { name: 'Stale' }));
await waitFor(() => svc.channels().length === 1);
// drop the connection; bot reconnects and must start with an empty table
bridge.sockets[0].destroy();
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
await waitFor(() => svc.channels().length === 0);
} finally {
delete process.env.MESH_RECONNECT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService connect', () => {
it('sends want_config_id on connect and applies the initial sync', async () => {
const out: Uint8Array[] = [];
const bridge = await startBridge((s) => s.on('data', (c) => out.push(c)));
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(() => svc.status().connected && out.length > 0);
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;
}
const { packets } = extractFrames(merged);
expect(packets.length).toBeGreaterThan(0);
const tr = fromBinary(P.Mesh.ToRadioSchema, packets[0]);
expect(tr.payloadVariant?.case).toBe('wantConfigId');
const nonce =
tr.payloadVariant && tr.payloadVariant.case === 'wantConfigId'
? tr.payloadVariant.value
: -1;
expect(nonce).toBeGreaterThan(0);
// feed back the initial sync: myInfo, then configCompleteId echoing the nonce
bridge.sockets[0].write(
frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'myInfo', value: { myNodeNum: 1234 } },
}),
),
),
);
bridge.sockets[0].write(
frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'configCompleteId', value: nonce },
}),
),
),
);
await waitFor(() => svc.status().myNodeNum === 1234);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService send', () => {
it('sends on the test channel (index 3) by default, never primary', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
delete process.env.MESH_SEND_CHANNEL; // default path: "test" (index 3)
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
const out: Uint8Array[] = [];
bridge.sockets[0].on('data', (c) => out.push(c));
const id = await svc.sendText(42, 'hello test');
expect(id).toBeGreaterThan(0);
const readOutbound = (): P.Mesh.MeshPacket | undefined => {
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;
}
const { packets } = extractFrames(merged);
// outbound now starts with the want_config_id frame — find the text packet
for (const p of packets) {
const tr = fromBinary(P.Mesh.ToRadioSchema, p);
if (tr.payloadVariant?.case === 'packet')
return tr.payloadVariant.value;
}
return undefined;
};
await waitFor(() => readOutbound() !== undefined);
const pkt = readOutbound()!;
expect(pkt.channel).toBe(3); // "test" — never the primary channel (0)
expect(pkt.to).toBe(42);
const decoded =
pkt.payloadVariant?.case === 'decoded'
? pkt.payloadVariant.value
: undefined;
expect(
new TextDecoder().decode(decoded?.payload ?? new Uint8Array()),
).toBe('hello test');
} finally {
delete process.env.MESH_SEND_CHANNEL;
svc.onModuleDestroy();
bridge.close();
}
});
it('sets Data.reply_id when replyId is given and leaves it unset otherwise', 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 out: Uint8Array[] = [];
bridge.sockets[0].on('data', (c) => out.push(c));
await svc.sendText(42, 'reply send', 42); // replyId = 42
await svc.sendText(43, 'plain send');
const readOutbound = (): 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;
}
const { packets } = extractFrames(merged);
// outbound starts with the want_config_id frame — collect only text packets, in order
return packets.flatMap((p) => {
const tr = fromBinary(P.Mesh.ToRadioSchema, p);
if (tr.payloadVariant?.case === 'packet')
return [tr.payloadVariant.value];
return [];
});
};
await waitFor(() => readOutbound().length === 2);
const [replyPkt, plainPkt] = readOutbound();
const replyData =
replyPkt.payloadVariant?.case === 'decoded'
? replyPkt.payloadVariant.value
: undefined;
expect(
new TextDecoder().decode(replyData?.payload ?? new Uint8Array()),
).toBe('reply send');
expect(replyData?.replyId).toBe(42); // threaded reply to packet 42
const plainData =
plainPkt.payloadVariant?.case === 'decoded'
? plainPkt.payloadVariant.value
: undefined;
expect(
new TextDecoder().decode(plainData?.payload ?? new Uint8Array()),
).toBe('plain send');
expect(plainData?.replyId).toBe(0); // omitted → proto3 default, not a reply
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService node database', () => {
it('collects NodeInfo during sync and summarizes nodes', 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(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
bridge.sockets[0].write(nodeInfoFrame(987654321)); // bare entry: no user/position/metrics yet
await waitFor(async () => (await svc.nodes()).length === 2);
expect(await svc.nodes()).toEqual([
{
num: KAMIL_NUM,
longName: 'Kamil SP8MK/P',
shortName: 'KK',
hwModel: 'TBEAM',
batteryLevel: 87,
lastHeard: 1758000000,
position: { lat: 521049637 * 1e-7, lon: 209833152 * 1e-7, alt: 123 },
},
{ num: 987654321, lastHeard: null }, // names/hwModel/battery/position omitted when absent
]);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('creates a stub entry for POSITION_APP packets from unknown nodes', 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(
positionPacketFrame(555, {
latitudeI: 10000000,
longitudeI: -20000000,
}),
);
await waitFor(async () => (await svc.nodes()).length === 1);
expect(await svc.nodes()).toEqual([
{ num: 555, lastHeard: null, position: { lat: 1, lon: -2 } }, // no alt sent -> omitted
]);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('updates batteryLevel from live NODEINFO_APP packets (and stubs unknown nodes)', 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,
);
// sync entry with a stale battery, then a live NODEINFO_APP update
bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
await waitFor(async () => (await svc.nodes()).length === 1);
expect((await svc.nodes())[0]?.batteryLevel).toBe(87);
bridge.sockets[0].write(
nodeInfoPacketFrame(KAMIL_NUM, { deviceMetrics: { batteryLevel: 9 } }),
);
await waitFor(async () => (await svc.nodes())[0]?.batteryLevel === 9);
// other fields of the sync entry survive the live update
expect(await svc.node(KAMIL_NUM)).toMatchObject({
longName: 'Kamil SP8MK/P',
shortName: 'KK',
batteryLevel: 9,
});
// unknown node: a NODEINFO_APP with deviceMetrics creates a stub entry
bridge.sockets[0].write(
nodeInfoPacketFrame(555, { deviceMetrics: { batteryLevel: 3 } }),
);
await waitFor(async () => (await svc.nodes()).length === 2);
expect((await svc.node(555))?.batteryLevel).toBe(3);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('looks up nodes by exact num, numeric string, exact name and unique substring', 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(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
bridge.sockets[0].write(
nodeInfoFrame(42, {
user: { id: '!ot', longName: 'Other Node', shortName: 'OT' },
}),
);
await waitFor(async () => (await svc.nodes()).length === 2);
expect((await svc.node(KAMIL_NUM))?.longName).toBe('Kamil SP8MK/P'); // exact num
expect((await svc.node(String(KAMIL_NUM)))?.num).toBe(KAMIL_NUM); // numeric string
expect((await svc.node('kamil sp8mk/p'))?.shortName).toBe('KK'); // case-insensitive exact name
expect((await svc.node('kamil'))?.num).toBe(KAMIL_NUM); // unique substring
expect(await svc.node('zzz-nope')).toBeNull(); // no match
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('returns null for ambiguous name matches', 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,
);
// two nodes whose names both contain "kamil" and share the short name "KK"
bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
bridge.sockets[0].write(
nodeInfoFrame(777, {
user: { id: '!kj', longName: 'Kamil Jr.', shortName: 'KK' },
}),
);
await waitFor(async () => (await svc.nodes()).length === 2);
expect(await svc.node('kamil')).toBeNull(); // substring matches two nodes
expect(await svc.node('kk')).toBeNull(); // exact name matches two nodes
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('resets the node database on reconnect (no stale data)', 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 svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
await waitFor(async () => (await svc.nodes()).length === 1);
// drop the connection; bot reconnects and must start with an empty node DB
bridge.sockets[0].destroy();
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
await waitFor(async () => (await svc.nodes()).length === 0);
} finally {
delete process.env.MESH_RECONNECT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
});
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,
to: 42, // destination persisted on rx records (broadcast addr for broadcasts)
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 test');
expect(svc.messages()[0]).toMatchObject({
id,
dir: 'tx',
to: 42,
text: 'hello test',
channel: 3,
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();
}
});
});
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;
let notifyMock: ReturnType<typeof vi.fn>;
beforeEach(() => {
notifyMock = vi.fn(
async (_id: number, _text: string, _from?: number) => {},
);
});
// myInfo frame as streamed during the initial sync — sets svc.myNodeNum
function myInfoFrame(num: number): Uint8Array {
return frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'myInfo', value: { myNodeNum: num } },
}),
),
);
}
// MeshService with a mocked OpencodeService wired in (structural cast — only notify() is used)
function makeSvc(): MeshService {
return new MeshService(undefined, {
notify: notifyMock,
} as unknown as OpencodeService);
}
it('notifies via the opencode service when an rx text lands on the notify channel (default 3)', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
delete process.env.MESH_NOTIFY_CHANNEL; // default 3
const svc = makeSvc();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3));
await waitFor(() => notifyMock.mock.calls.length === 1);
expect(notifyMock).toHaveBeenCalledTimes(1);
expect(notifyMock).toHaveBeenCalledWith(1, 'ping', KAMIL_NUM); // packet id threaded through
} finally {
delete process.env.MESH_NOTIFY_CHANNEL;
svc.onModuleDestroy();
bridge.close();
}
});
it('does not notify for rx texts on other channels', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
delete process.env.MESH_NOTIFY_CHANNEL; // default 3
const svc = makeSvc();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2));
// the message is captured in history — but no notification fires off-channel
await waitFor(() => svc.messages().length === 1);
expect(notifyMock).not.toHaveBeenCalled();
} finally {
delete process.env.MESH_NOTIFY_CHANNEL;
svc.onModuleDestroy();
bridge.close();
}
});
it('fires the notifier for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => {
process.env.MESH_NOTIFY_CHANNEL = '0';
const svc = makeSvc();
try {
const r = svc.injectMessage('hi', 42);
expect(notifyMock).toHaveBeenCalledTimes(1);
expect(notifyMock).toHaveBeenCalledWith(r.id, 'hi', 42); // generated record id threaded through
} finally {
delete process.env.MESH_NOTIFY_CHANNEL;
}
});
it('passes `from` through as undefined when an injected message has no sender', async () => {
process.env.MESH_NOTIFY_CHANNEL = '0';
const svc = makeSvc();
try {
const r = svc.injectMessage('hi');
expect(notifyMock).toHaveBeenCalledTimes(1);
expect(notifyMock).toHaveBeenCalledWith(r.id, 'hi', undefined);
} finally {
delete process.env.MESH_NOTIFY_CHANNEL;
}
});
it('does not notify for rx texts from our own node on the notify channel', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
delete process.env.MESH_NOTIFY_CHANNEL; // default 3
const svc = makeSvc();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
// initial sync: our own node number arrives via myInfo first
bridge.sockets[0].write(myInfoFrame(MY_NODE));
await waitFor(() => svc.status().myNodeNum === MY_NODE);
// an rx text from ourselves on the notify channel must not fire the notifier
bridge.sockets[0].write(textPacketFrame(MY_NODE, 42, 'ping', 3));
// history capture proves the packet was processed (notifyRx runs right after it)
await waitFor(() => svc.messages().length === 1);
expect(notifyMock).not.toHaveBeenCalled();
} finally {
delete process.env.MESH_NOTIFY_CHANNEL;
svc.onModuleDestroy();
bridge.close();
}
});
it('does not notify for an injected message from our own node', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_NOTIFY_CHANNEL = '0';
const svc = makeSvc();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(myInfoFrame(MY_NODE));
await waitFor(() => svc.status().myNodeNum === MY_NODE);
svc.injectMessage('hi', MY_NODE);
expect(notifyMock).not.toHaveBeenCalled();
} finally {
delete process.env.MESH_NOTIFY_CHANNEL;
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService injectMessage', () => {
it('injects rx records that appear newest-first in messages()', () => {
const svc = new MeshService();
expect(svc.messages()).toEqual([]);
const r1 = svc.injectMessage('first');
const r2 = svc.injectMessage('second', KAMIL_NUM);
expect(r1.id).toBeGreaterThan(0); // random non-zero uint32, like tx ids
expect(r2.id).toBeGreaterThan(0);
expect(svc.messages()).toEqual([
{
id: r2.id,
dir: 'rx',
from: KAMIL_NUM,
text: 'second',
channel: 0,
at: expect.any(Number),
},
{
id: r1.id,
dir: 'rx',
text: 'first',
channel: 0,
at: expect.any(Number),
},
]);
});
it('omits from when not provided and rejects empty text or invalid from', () => {
const svc = new MeshService();
const rec = svc.injectMessage('no sender');
expect(svc.messages()[0]).toEqual({
id: rec.id,
dir: 'rx',
text: 'no sender',
channel: 0,
at: expect.any(Number),
});
expect(svc.messages()[0]).not.toHaveProperty('from');
for (const bad of ['', undefined]) {
expect(() => svc.injectMessage(bad as unknown as string)).toThrow(
'text must be a non-empty string',
);
}
for (const bad of [0, -1, 42.5, BROADCAST_ADDR + 1]) {
expect(() => svc.injectMessage('x', bad)).toThrow(/invalid node number/);
}
expect(svc.messages()).toHaveLength(1); // rejected calls add nothing
});
it('caps history at 100 records, dropping the oldest', () => {
const svc = new MeshService();
for (let i = 0; i < 105; i++) svc.injectMessage(`m-${i}`);
expect(svc.messages()).toHaveLength(100);
expect(svc.messages()[0]?.text).toBe('m-104'); // newest first
expect(svc.messages()[99]?.text).toBe('m-5'); // m-0..m-4 dropped
});
it('shares the history buffer with captured packets', 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,
);
svc.injectMessage('injected');
bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'real', 2));
await waitFor(() => svc.messages().length === 2);
expect(svc.messages()[0]).toMatchObject({ dir: 'rx', text: 'real' }); // newest first
expect(svc.messages()[1]).toMatchObject({
dir: 'rx',
text: 'injected',
channel: 0,
});
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService telemetry', () => {
it('captures device metrics and environment variants with from/kind/data', 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,
);
expect(svc.telemetry()).toEqual([]); // nothing before the first packet
bridge.sockets[0].write(
telemetryPacketFrame(KAMIL_NUM, {
case: 'deviceMetrics',
value: { batteryLevel: 87, voltage: 4.1, uptimeSeconds: 3600 },
}),
);
bridge.sockets[0].write(
telemetryPacketFrame(987654321, {
case: 'environmentMetrics',
value: { temperature: 21.5, relativeHumidity: 48 },
}),
);
await waitFor(() => svc.telemetry().length === 2);
expect(svc.telemetry()).toEqual([
{
from: 987654321, // newest first
at: expect.any(Number),
kind: 'environmentMetrics',
// the deprecated repeated one_wire_temperature field decodes as an empty list even when unset
data: {
temperature: 21.5,
relativeHumidity: 48,
oneWireTemperature: [],
},
},
{
from: KAMIL_NUM,
at: expect.any(Number),
kind: 'deviceMetrics',
// proto float is IEEE-754 single precision — 4.1 widens to this exact double
data: {
batteryLevel: 87,
voltage: 4.099999904632568,
uptimeSeconds: 3600,
},
},
]);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('caps telemetry 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(
telemetryPacketFrame(42, {
case: 'deviceMetrics',
value: { batteryLevel: i },
}),
);
await waitFor(() => svc.telemetry().length === 100, 10000);
expect(svc.telemetry()[0]?.data).toEqual({ batteryLevel: 104 }); // newest first
expect(svc.telemetry()[99]?.data).toEqual({ batteryLevel: 5 }); // i=0..4 dropped
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('resets telemetry on reconnect (no stale data)', 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 svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(
telemetryPacketFrame(KAMIL_NUM, { case: 'deviceMetrics', value: {} }),
);
await waitFor(() => svc.telemetry().length === 1);
// drop the connection; bot reconnects and must start with an empty history
bridge.sockets[0].destroy();
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
await waitFor(() => svc.telemetry().length === 0);
} finally {
delete process.env.MESH_RECONNECT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService resolveDest', () => {
it('resolves broadcast sentinels, numbers and names to node numbers', 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(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
await waitFor(async () => (await svc.nodes()).length === 1);
expect(await svc.resolveDest(undefined)).toBe(BROADCAST_ADDR); // omitted → broadcast
expect(await svc.resolveDest('')).toBe(BROADCAST_ADDR); // empty → broadcast
expect(await svc.resolveDest('!b')).toBe(BROADCAST_ADDR);
expect(await svc.resolveDest('BROADCAST')).toBe(BROADCAST_ADDR); // case-insensitive
expect(await svc.resolveDest(42)).toBe(42); // number passthrough
expect(await svc.resolveDest(String(KAMIL_NUM))).toBe(KAMIL_NUM); // numeric string
expect(await svc.resolveDest('Kamil SP8MK/P')).toBe(KAMIL_NUM); // exact longName
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('throws for unknown names and invalid node numbers', 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(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
await waitFor(async () => (await svc.nodes()).length === 1);
await expect(svc.resolveDest('zzz-nope')).rejects.toThrow(
"no node matching 'zzz-nope'",
);
for (const bad of [0, -1, 42.5, BROADCAST_ADDR + 1]) {
await expect(svc.resolveDest(bad)).rejects.toThrow(
/invalid node number/,
);
}
await expect(svc.resolveDest('0')).rejects.toThrow(/invalid node number/);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService device info', () => {
it('captures DeviceMetadata and exposes firmwareVersion', 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,
);
// before the sync: only the node number (null until myInfo arrives)
expect(svc.deviceInfo()).toEqual({
myNodeNum: null,
localConfig: {},
moduleConfig: {},
});
bridge.sockets[0].write(
metadataFrame({ firmwareVersion: '2.7.0.1', hasWifi: false, role: 3 }),
);
await waitFor(() => svc.deviceInfo().firmwareVersion === '2.7.0.1');
const info = svc.deviceInfo();
expect(info.firmwareVersion).toBe('2.7.0.1');
expect(info.metadata?.firmwareVersion).toBe('2.7.0.1'); // metadata stored as-is
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('captures config/module-config sections and redacts key material', 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 secret = 'super-secret-wifi-psk';
bridge.sockets[0].write(
configSectionFrame('security', {
publicKey: new Uint8Array([1, 2, 3]),
privateKey: new Uint8Array([4, 5, 6]),
adminKey: [new Uint8Array([7])],
isManaged: true,
}),
);
bridge.sockets[0].write(
configSectionFrame('network', {
wifiEnabled: false,
wifiSsid: 'home',
wifiPsk: secret,
}),
);
bridge.sockets[0].write(configSectionFrame('device', { role: 3 }));
bridge.sockets[0].write(
moduleConfigSectionFrame('telemetry', {
environmentUpdateInterval: 900,
powerMeasurementEnabled: true,
}),
);
await waitFor(
() =>
Object.keys(svc.deviceInfo().localConfig).length === 3 &&
Object.keys(svc.deviceInfo().moduleConfig).length === 1,
);
const info = svc.deviceInfo();
expect(Object.keys(info.localConfig).sort()).toEqual([
'device',
'network',
'security',
]);
expect(info.moduleConfig.telemetry).toMatchObject({
environmentUpdateInterval: 900,
powerMeasurementEnabled: true,
});
// key material is replaced with a marker — never exposed
expect(info.localConfig.security).toMatchObject({
publicKey: 'REDACTED',
privateKey: 'REDACTED',
adminKey: 'REDACTED',
isManaged: true, // non-secret fields pass through
});
expect(info.localConfig.network?.wifiPsk).toBe('REDACTED');
const json = JSON.stringify(info);
expect(json).not.toContain(secret);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('serializes proto uint64 fields as strings (JSON-safe)', 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,
);
// powermonEnables is a proto uint64 — protobuf-es decodes it as native bigint
bridge.sockets[0].write(
configSectionFrame('power', { powermonEnables: 123n }),
);
await waitFor(() => svc.deviceInfo().localConfig.power !== undefined);
const info = svc.deviceInfo();
expect(info.localConfig.power?.powermonEnables).toBe('123'); // bigint -> decimal string
expect(() => JSON.stringify(info)).not.toThrow(); // no "Do not know how to serialize a BigInt"
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('summarizes NodeInfo snr/hopsAway/viaMqtt when present', 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(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
bridge.sockets[0].write(
nodeInfoFrame(42, {
user: { id: '!ot', longName: 'Other Node' },
snr: -2.5,
hopsAway: 1,
viaMqtt: true,
}),
);
await waitFor(async () => (await svc.nodes()).length === 2);
const nodes = await svc.nodes();
expect(nodes.find((n) => n.num === 42)).toEqual({
num: 42,
longName: 'Other Node',
lastHeard: null,
snr: -2.5,
hopsAway: 1,
viaMqtt: true,
});
// absent in the proto → omitted from the summary
expect(nodes.find((n) => n.num === KAMIL_NUM)).not.toHaveProperty('snr');
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
});
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('persists the destination (to) of captured rx packets on poller 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,
);
bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3));
// 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 id = ?')
.get(1) as Record<string, unknown>;
expect(row.dir).toBe('rx');
expect(row.from_node).toBe(KAMIL_NUM);
expect(row.to_node).toBe(42); // rx rows carry the destination now (no schema change — to_node existed for tx)
} 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();
}
});
});