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 { 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 }, }), ), ); } // 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(); } }); }); 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('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, 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 notify webhook', () => { const fetchMock = vi.fn(async (..._args: unknown[]) => undefined); // our own node number for the self-echo filter specs (distinct from KAMIL_NUM, an external user) const MY_NODE = 987654321; beforeEach(() => fetchMock.mockClear()); // 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 } }, }), ), ); } it('POSTs {text, from} to MESH_NOTIFY_URL 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); process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; vi.stubGlobal('fetch', fetchMock); 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', 3)); await waitFor(() => fetchMock.mock.calls.length === 1); expect(fetchMock).toHaveBeenCalledTimes(1); const [url, init] = fetchMock.mock.calls[0]; expect(url).toBe('http://notify.example/hook'); const opts = init as { method?: string; headers?: Record; body?: string; }; expect(opts.method).toBe('POST'); expect(opts.headers?.['content-type']).toBe('application/json'); expect(JSON.parse(String(opts.body))).toEqual({ text: 'ping', from: KAMIL_NUM, }); } finally { vi.unstubAllGlobals(); delete process.env.MESH_NOTIFY_URL; svc.onModuleDestroy(); bridge.close(); } }); it('does not call fetch 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); process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; vi.stubGlobal('fetch', fetchMock); 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)); // the message is captured in history — but no webhook fires off-channel await waitFor(() => svc.messages().length === 1); expect(fetchMock).not.toHaveBeenCalled(); } finally { vi.unstubAllGlobals(); delete process.env.MESH_NOTIFY_URL; svc.onModuleDestroy(); bridge.close(); } }); it('never calls fetch when MESH_NOTIFY_URL is unset or empty', 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_URL; // unset → disabled vi.stubGlobal('fetch', fetchMock); 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', 3)); await waitFor(() => svc.messages().length === 1); expect(fetchMock).not.toHaveBeenCalled(); // empty string → also disabled (a fresh service re-reads the env) process.env.MESH_NOTIFY_URL = ''; const svc2 = new MeshService(); try { svc2.onModuleInit(); await waitFor( () => bridge.sockets.length === 2 && svc2.status().connected, ); bridge.sockets[1].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3)); await waitFor(() => svc2.messages().length === 1); } finally { delete process.env.MESH_NOTIFY_URL; svc2.onModuleDestroy(); } expect(fetchMock).not.toHaveBeenCalled(); } finally { vi.unstubAllGlobals(); svc.onModuleDestroy(); bridge.close(); } }); it('fires the webhook for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => { process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; process.env.MESH_NOTIFY_CHANNEL = '0'; vi.stubGlobal('fetch', fetchMock); const svc = new MeshService(); try { svc.injectMessage('hi', 42); expect(fetchMock).toHaveBeenCalledTimes(1); const [url, init] = fetchMock.mock.calls[0]; expect(url).toBe('http://notify.example/hook'); const opts = init as { method?: string; headers?: Record; body?: string; }; expect(opts.method).toBe('POST'); expect(opts.headers?.['content-type']).toBe('application/json'); expect(JSON.parse(String(opts.body))).toEqual({ text: 'hi', from: 42 }); } finally { vi.unstubAllGlobals(); delete process.env.MESH_NOTIFY_URL; delete process.env.MESH_NOTIFY_CHANNEL; } }); it('omits `from` in the webhook body when an injected message has no sender', async () => { process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; process.env.MESH_NOTIFY_CHANNEL = '0'; vi.stubGlobal('fetch', fetchMock); const svc = new MeshService(); try { svc.injectMessage('hi'); expect(fetchMock).toHaveBeenCalledTimes(1); const [, init] = fetchMock.mock.calls[0]; const opts = init as { body?: string }; expect(JSON.parse(String(opts.body))).toEqual({ text: 'hi' }); } finally { vi.unstubAllGlobals(); delete process.env.MESH_NOTIFY_URL; delete process.env.MESH_NOTIFY_CHANNEL; } }); it('does not call fetch 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); process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; vi.stubGlobal('fetch', fetchMock); const svc = new MeshService(); 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 webhook 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(fetchMock).not.toHaveBeenCalled(); } finally { vi.unstubAllGlobals(); delete process.env.MESH_NOTIFY_URL; svc.onModuleDestroy(); bridge.close(); } }); it('does not call fetch 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_URL = 'http://notify.example/hook'; process.env.MESH_NOTIFY_CHANNEL = '0'; vi.stubGlobal('fetch', fetchMock); const svc = new MeshService(); 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(fetchMock).not.toHaveBeenCalled(); } finally { vi.unstubAllGlobals(); delete process.env.MESH_NOTIFY_URL; 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; 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(); } }); });