import { create, fromBinary, toBinary } from '@bufbuild/protobuf'; import * as P from '@meshtastic/protobufs'; import * as net from 'node:net'; import type { AddressInfo } from 'node:net'; import { describe, expect, it } from 'vitest'; import { extractFrames, frame } from './codec.js'; import { MeshService } from './mesh.service.js'; function startBridge(onSocket?: (s: net.Socket) => void): Promise<{ port: number; sockets: net.Socket[]; close: () => void; }> { const sockets: net.Socket[] = []; const server = net.createServer((socket) => { onSocket?.(socket); // attached at accept time, before any data can arrive sockets.push(socket); }); return new Promise((resolve) => { server.listen(0, '127.0.0.1', () => { resolve({ port: (server.address() as AddressInfo).port, sockets, close: () => server.close(), }); }); }); } async function waitFor(cond: () => boolean, timeoutMs = 5000): Promise { const t0 = Date.now(); while (!cond()) { if (Date.now() - t0 > timeoutMs) throw new Error('timed out waiting'); await new Promise((r) => setTimeout(r, 10)); } } 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 2) 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: "testowy" (index 2) 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 testowy'); 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(2); // "testowy" — 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 testowy'); } finally { delete process.env.MESH_SEND_CHANNEL; svc.onModuleDestroy(); bridge.close(); } }); });