import { Injectable, Logger, OnModuleDestroy, OnModuleInit, ServiceUnavailableException, } from '@nestjs/common'; import * as net from 'node:net'; import type * as Proto from '@meshtastic/protobufs'; import { BROADCAST_ADDR, 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 & Partial>; export interface NodeSummary { num: number; longName?: string; shortName?: string; /** meshtastic.HardwareModel enum name (e.g. "TBEAM"); omitted when unknown or UNSET */ hwModel?: string; /** 0-100 = battery percent, >100 = mains powered (device convention) */ batteryLevel?: number; /** epoch seconds of the last heard packet; null when never heard */ lastHeard: number | null; position?: { lat: number; lon: number; alt?: number }; } @Injectable() export class MeshService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(MeshService.name); private readonly host = process.env.MESH_HOST ?? '192.168.1.4'; private readonly port = parseInt(process.env.MESH_PORT ?? '4403', 10); // ponytail: fixed-interval retry, no backoff — fine for single-bridge use private readonly reconnectMs = parseInt( process.env.MESH_RECONNECT_MS ?? '5000', 10, ); // sends are pinned to the test channel ("test", index 3) on purpose — never the primary channel private readonly sendChannel = parseInt( process.env.MESH_SEND_CHANNEL ?? '3', 10, ); private socket: net.Socket | null = null; private buffer: Uint8Array = new Uint8Array(0); private connected = false; private myNodeNum: number | null = null; // channels streamed by the device during initial sync (before config_complete_id) private channelMap = new Map< number, import('@meshtastic/protobufs').Channel.Channel >(); // node database: NodeInfo frames from the initial sync + live POSITION_APP updates private nodeMap = new Map(); // message history (newest kept, capped at HISTORY_LIMIT) + pending tx records awaiting their routing ACK/NACK private history: MessageRecord[] = []; private pendingTx = new Map(); private reconnectTimer: NodeJS.Timeout | null = null; private shuttingDown = false; onModuleInit(): void { this.connect(); } onModuleDestroy(): void { this.shuttingDown = true; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); this.socket?.destroy(); } /** Bridge connection state: `{ connected, myNodeNum, host, port }`. */ status() { return { connected: this.connected, myNodeNum: this.myNodeNum, host: this.host, port: this.port, }; } // psk/keys are deliberately not included — never expose them over HTTP /** Device channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]` sorted by index; empty until the initial sync completes. */ channels(): { index: number; name: string; uplinkEnabled: boolean; downlinkEnabled: boolean; }[] { return [...this.channelMap.values()] .sort((a, b) => a.index - b.index) .map(({ index, settings }) => ({ index, name: settings?.name ?? '', uplinkEnabled: settings?.uplinkEnabled ?? false, downlinkEnabled: settings?.downlinkEnabled ?? false, })); } /** * Lean summaries of all known nodes, sorted by node number. * Names/hwModel/batteryLevel appear only when the device reported them; * `lastHeard` is epoch seconds (null = never heard); `position` only when both coordinates are known. */ async nodes(): Promise { const names = await loadHwModelNames(); return [...this.nodeMap.values()] .sort((a, b) => a.num - b.num) .map((n) => this.summarize(n, names)); } /** * Look up one node: exact node number first (numbers and numeric strings), * then case-insensitive EXACT long/short name, else the unique case-insensitive * SUBSTRING match across long/short names. Returns null when nothing or more * than one node matches. Reused by send-by-name lookups. */ async node(idOrName: string | number): Promise { const q = String(idOrName).trim().toLowerCase(); if (!q) return null; const names = await loadHwModelNames(); if (/^\d+$/.test(q)) { const hit = this.nodeMap.get(Number(q)); if (hit) return this.summarize(hit, names); } const all = [...this.nodeMap.values()]; // exact name match wins over substring; a string always contains itself, so an ambiguous exact set can never become unique below const exact = all.filter( (n) => n.user && (n.user.longName.toLowerCase() === q || n.user.shortName.toLowerCase() === q), ); if (exact.length === 1) return this.summarize(exact[0], names); const sub = all.filter( (n) => n.user && (n.user.longName.toLowerCase().includes(q) || n.user.shortName.toLowerCase().includes(q)), ); if (sub.length === 1) return this.summarize(sub[0], names); return null; } /** * Resolve a send target to a node number for sendText(). * undefined/"" and "!b"/"broadcast" (case-insensitive, trimmed) → BROADCAST_ADDR; * numbers pass through as-is (must be an integer in 1..0xffffffff); numeric strings are parsed the same way; * anything else is looked up via node() — throws `no node matching ''` when nothing or more than one matches. */ async resolveDest(to?: number | string): Promise { if (typeof to === 'number') { this.assertNodeNum(to); return to; } const s = String(to ?? '').trim(); if (!s) return BROADCAST_ADDR; // omitted or empty → broadcast if (s === '!b' || s.toLowerCase() === 'broadcast') return BROADCAST_ADDR; if (/^\d+$/.test(s)) { this.assertNodeNum(Number(s)); return Number(s); } const hit = await this.node(s); if (!hit) throw new Error(`no node matching '${to}'`); return hit.num; } private assertNodeNum(n: number): void { if (!Number.isInteger(n) || n < 1 || n > BROADCAST_ADDR) throw new Error(`invalid node number '${n}'`); } /** * 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 { 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): NodeSummary { const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null }; if (n.user?.longName) s.longName = n.user.longName; if (n.user?.shortName) s.shortName = n.user.shortName; // omit UNSET (enum value 0) rather than report "UNSET" const hwName = n.user?.hwModel ? names.get(n.user.hwModel) : undefined; if (hwName && hwName !== 'UNSET') s.hwModel = hwName; // optional proto field: undefined when the device never reported metrics const battery = n.deviceMetrics?.batteryLevel; if (battery !== undefined) s.batteryLevel = battery; const p = n.position; if (p?.latitudeI !== undefined && p.longitudeI !== undefined) { // alt included only when the device sent it — protobuf-es leaves absent optionals undefined, so an explicit 0 (sea level) is kept s.position = { lat: p.latitudeI * 1e-7, lon: p.longitudeI * 1e-7 }; if (p.altitude !== undefined) s.position.alt = p.altitude; } return s; } private connect(): void { if (this.shuttingDown) return; const socket = net.createConnection({ host: this.host, port: this.port }); this.socket = socket; socket.on('connect', () => { this.connected = true; // 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) .then((bytes) => socket.write(frame(bytes))) .catch((err: unknown) => this.logger.error(`want_config failed: ${String(err)}`), ); this.logger.log(`requested initial sync (want_config_id=${nonce})`); }); socket.on('data', (chunk) => { const merged = new Uint8Array(this.buffer.length + chunk.length); merged.set(this.buffer, 0); merged.set(chunk, this.buffer.length); this.buffer = merged; const { packets, rest } = extractFrames(this.buffer); this.buffer = rest; for (const p of packets) { decodeFromRadio(p) .then((fr) => this.handleFromRadio(fr)) .catch((err: unknown) => this.logger.error(`decode failed: ${String(err)}`), ); } }); const onDown = () => { this.connected = false; this.scheduleReconnect(); }; socket.on('error', (err) => { this.logger.warn(`bridge error: ${err.message}`); onDown(); }); socket.on('close', onDown); } private scheduleReconnect(): void { if (this.shuttingDown) return; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); this.reconnectTimer = setTimeout(() => this.connect(), this.reconnectMs); } private async handleFromRadio( fr: import('@meshtastic/protobufs').Mesh.FromRadio, ): Promise { const v = fr.payloadVariant; switch (v?.case) { case 'myInfo': this.myNodeNum = v.value.myNodeNum; this.logger.log(`my node num: ${this.myNodeNum}`); break; case 'channel': // device streams one Channel per table slot during initial sync this.channelMap.set(v.value.index, v.value); break; case 'nodeInfo': // device streams the whole node DB during initial sync this.nodeMap.set(v.value.num, v.value); break; case 'configCompleteId': this.logger.log( `bridge ready (config complete), ${this.channelMap.size} channels, ${this.nodeMap.size} nodes`, ); break; case 'packet': { const pkt = v.value; const decoded = pkt.payloadVariant?.case === 'decoded' ? pkt.payloadVariant.value : undefined; if (decoded && decoded.portnum === PORT_TEXT_MESSAGE_APP) { 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; try { pos = await decodePosition(decoded.payload); } catch { pos = undefined; // malformed payload — skip, keep whatever we had } if (pos?.latitudeI !== undefined && pos.longitudeI !== undefined) { const num = pkt.from; this.nodeMap.set(num, { ...this.nodeMap.get(num), num, 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; } } } }