800 lines
32 KiB
TypeScript
800 lines
32 KiB
TypeScript
import {
|
|
BadRequestException,
|
|
Injectable,
|
|
Logger,
|
|
NotFoundException,
|
|
OnModuleDestroy,
|
|
OnModuleInit,
|
|
ServiceUnavailableException,
|
|
} from '@nestjs/common';
|
|
import * as net from 'node:net';
|
|
// type-only import: node:sqlite is only required at runtime by whoever opens a database (mesh.module.ts / tests)
|
|
import type { DatabaseSync } from 'node:sqlite';
|
|
import type * as Proto from '@meshtastic/protobufs';
|
|
import { OpencodeService } from '../opencode/opencode.service.js';
|
|
import {
|
|
BROADCAST_ADDR,
|
|
PORT_NODEINFO_APP,
|
|
PORT_POSITION_APP,
|
|
PORT_ROUTING_APP,
|
|
PORT_TELEMETRY_APP,
|
|
PORT_TEXT_MESSAGE_APP,
|
|
buildTextToRadio,
|
|
buildWantConfigToRadio,
|
|
decodeFromRadio,
|
|
decodeNodeInfo,
|
|
decodePosition,
|
|
decodeRouting,
|
|
decodeTelemetry,
|
|
describeRoutingError,
|
|
extractFrames,
|
|
frame,
|
|
loadHwModelNames,
|
|
randomPacketId,
|
|
} 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) — captured rx packets and all tx records */
|
|
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;
|
|
}
|
|
|
|
/** One entry in the forwarded ring: an rx text that passed the notify gate and was handed off to opencode. */
|
|
export interface ForwardedRecord {
|
|
/** packet id (real rx) or generated record id (injected) */
|
|
id: number;
|
|
/** sender node num — omitted when there is no sender */
|
|
from?: number;
|
|
text: string;
|
|
channel: number;
|
|
/** epoch ms */
|
|
at: number;
|
|
}
|
|
|
|
/** One entry in the telemetry history: a TELEMETRY_APP packet from some node. */
|
|
export interface TelemetryRecord {
|
|
/** sender node num */
|
|
from: number;
|
|
/** epoch ms */
|
|
at: number;
|
|
/** meshtastic.Telemetry oneof case name (e.g. "deviceMetrics", "environmentMetrics") */
|
|
kind: string;
|
|
/** the variant payload, JSON-safe (proto uint64 fields as decimal strings); unset optional fields omitted */
|
|
data: object;
|
|
}
|
|
|
|
// bounded in-memory buffer: keep the newest N records (SQLite keeps more — see MESSAGES_DDL)
|
|
const HISTORY_LIMIT = 100;
|
|
/** Rows kept in SQLite across restarts; older rows are pruned on every poller tick. */
|
|
const DB_RETENTION = 1000;
|
|
|
|
// one row per MessageRecord, keyed by packet id (tx ids and injected ids are random uint32, rx uses the wire id)
|
|
export const MESSAGES_DDL = `CREATE TABLE IF NOT EXISTS messages(
|
|
id INTEGER PRIMARY KEY,
|
|
dir TEXT NOT NULL,
|
|
to_node INTEGER,
|
|
from_node INTEGER,
|
|
text TEXT NOT NULL,
|
|
channel INTEGER NOT NULL,
|
|
at INTEGER NOT NULL,
|
|
status TEXT,
|
|
ack_error TEXT
|
|
)`;
|
|
|
|
/** Create the messages table if missing (idempotent — safe on every startup). */
|
|
export function initMessagesTable(db: DatabaseSync): void {
|
|
db.exec(MESSAGES_DDL);
|
|
}
|
|
|
|
/** Map a messages-table row to a MessageRecord (NULLs → omitted optionals). */
|
|
function recordFromRow(row: Record<string, unknown>): MessageRecord {
|
|
const rec: MessageRecord = {
|
|
id: Number(row.id),
|
|
dir: row.dir === 'tx' ? 'tx' : 'rx',
|
|
text: String(row.text),
|
|
channel: Number(row.channel),
|
|
at: Number(row.at),
|
|
};
|
|
if (row.to_node != null) rec.to = Number(row.to_node);
|
|
if (row.from_node != null) rec.from = Number(row.from_node);
|
|
if (row.status != null) rec.status = row.status as MessageRecord['status'];
|
|
if (row.ack_error != null) rec.ackError = String(row.ack_error);
|
|
return rec;
|
|
}
|
|
|
|
/** What we know about a node so far: full NodeInfo after sync, possibly just {num}+position for live-only nodes. */
|
|
type NodeEntry = Pick<Proto.Mesh.NodeInfo, 'num'> &
|
|
Partial<Omit<Proto.Mesh.NodeInfo, 'num'>>;
|
|
|
|
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 };
|
|
/** signal-to-noise ratio of the last received packet, dB (NodeInfo.snr) */
|
|
snr?: number;
|
|
/** hops from this node to the one that reported it (optional in proto) */
|
|
hopsAway?: number;
|
|
/** true when the node info arrived via MQTT rather than LoRa */
|
|
viaMqtt?: boolean;
|
|
}
|
|
|
|
/** Shape of deviceInfo() output: sync-captured device data, key material redacted. */
|
|
export interface DeviceInfo {
|
|
myNodeNum: number | null;
|
|
firmwareVersion?: string;
|
|
metadata?: Proto.Mesh.DeviceMetadata;
|
|
localConfig: Record<string, Record<string, unknown>>;
|
|
moduleConfig: Record<string, Record<string, unknown>>;
|
|
}
|
|
|
|
// protobuf-es decodes proto uint64 fields as native bigint, which JSON.stringify rejects;
|
|
// recursively convert so deviceInfo()/telemetry() output is always JSON-safe (bigint -> decimal string).
|
|
// $-prefixed keys are protobuf-es internals ($typeName/$unknown), not device data — dropped.
|
|
function jsonSafe<T>(value: T): T {
|
|
if (typeof value === 'bigint') return String(value) as T;
|
|
if (Array.isArray(value)) return value.map(jsonSafe) as T;
|
|
if (value !== null && typeof value === 'object') {
|
|
const out: Record<string, unknown> = {};
|
|
for (const [k, v] of Object.entries(value))
|
|
if (!k.startsWith('$')) out[k] = jsonSafe(v);
|
|
return out as T;
|
|
}
|
|
return value; // primitives + undefined pass through
|
|
}
|
|
|
|
// key material must never leave the process — replaced with a marker so callers can see the field exists
|
|
const REDACTED = 'REDACTED';
|
|
/** Config sections that carry secrets: security (PKI keys) and network (wifi psk). */
|
|
function redactSection(
|
|
name: string,
|
|
value: Record<string, unknown>,
|
|
): Record<string, unknown> {
|
|
if (name !== 'security' && name !== 'network') return value;
|
|
const out = { ...value };
|
|
for (const key of ['publicKey', 'privateKey', 'adminKey', 'wifiPsk'])
|
|
if (out[key] !== undefined) out[key] = REDACTED;
|
|
return out;
|
|
}
|
|
|
|
@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,
|
|
);
|
|
// outbound notification: fire-and-forget push to the opencode notifier when an rx text lands on the notify channel;
|
|
// disabled state (missing credentials) lives in OpencodeService itself
|
|
private readonly notifyChannel = parseInt(
|
|
process.env.MESH_NOTIFY_CHANNEL ?? '3',
|
|
10,
|
|
);
|
|
// how often the poller re-upserts the history buffer into SQLite (ms)
|
|
private readonly pollMs = parseInt(process.env.MESH_POLL_MS ?? '1000', 10);
|
|
|
|
/** SQLite database backing the message history; null = persistence disabled (tests construct without one). */
|
|
private db: DatabaseSync | null = null;
|
|
/** In-process opencode notifier for rx texts on the notify channel; null in tests that don't wire one. */
|
|
private readonly opencode: OpencodeService | null;
|
|
private pollTimer: NodeJS.Timeout | null = null;
|
|
|
|
constructor(db?: DatabaseSync, opencode?: OpencodeService) {
|
|
this.db = db ?? null;
|
|
this.opencode = opencode ?? null;
|
|
}
|
|
|
|
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<number, NodeEntry>();
|
|
// device metadata + per-section config/module-config streamed during the initial sync
|
|
private metadata: Proto.Mesh.DeviceMetadata | null = null;
|
|
private configSections = new Map<string, Record<string, unknown>>();
|
|
private moduleConfigSections = new Map<string, Record<string, unknown>>();
|
|
// message history (newest kept, capped at HISTORY_LIMIT) + pending tx records awaiting their routing ACK/NACK
|
|
private history: MessageRecord[] = [];
|
|
// rx texts handed off to the opencode notifier (newest first, capped at HISTORY_LIMIT); in-memory only
|
|
private forwardedHistory: ForwardedRecord[] = [];
|
|
private pendingTx = new Map<number, MessageRecord>();
|
|
// telemetry history (newest kept, capped at HISTORY_LIMIT)
|
|
private telemetryHistory: TelemetryRecord[] = [];
|
|
private reconnectTimer: NodeJS.Timeout | null = null;
|
|
private shuttingDown = false;
|
|
|
|
onModuleInit(): void {
|
|
if (this.db) {
|
|
initMessagesTable(this.db);
|
|
this.seedHistory(); // history survives process restarts
|
|
// ponytail: full re-upsert of the ≤100-row buffer every tick — trivial load, zero bookkeeping
|
|
this.pollTimer = setInterval(() => this.flushHistory(), this.pollMs);
|
|
}
|
|
this.connect();
|
|
}
|
|
|
|
onModuleDestroy(): void {
|
|
this.shuttingDown = true;
|
|
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
|
|
if (this.pollTimer) clearInterval(this.pollTimer);
|
|
this.socket?.destroy();
|
|
this.db?.close();
|
|
}
|
|
|
|
/** 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,
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Device info captured during the initial sync: this device's node number, firmware version and
|
|
* metadata (DeviceMetadata as-is — it carries no key material), plus per-section config and
|
|
* module-config messages keyed by section name. In-memory only; empty until the sync completes,
|
|
* reset on every bridge reconnect. Key material is redacted: security publicKey/privateKey/adminKey
|
|
* and network wifiPsk are replaced with "REDACTED" — never exposed over HTTP. Proto uint64 fields
|
|
* (e.g. power.powermonEnables) come back as decimal strings so the output always JSON-serializes.
|
|
*/
|
|
deviceInfo(): DeviceInfo {
|
|
const out: DeviceInfo = {
|
|
myNodeNum: this.myNodeNum,
|
|
localConfig: {},
|
|
moduleConfig: {},
|
|
};
|
|
if (this.metadata) {
|
|
out.firmwareVersion = this.metadata.firmwareVersion;
|
|
out.metadata = jsonSafe(this.metadata);
|
|
}
|
|
for (const [name, value] of this.configSections)
|
|
out.localConfig[name] = jsonSafe(redactSection(name, value));
|
|
for (const [name, value] of this.moduleConfigSections)
|
|
out.moduleConfig[name] = jsonSafe(value);
|
|
return out;
|
|
}
|
|
|
|
/**
|
|
* Lean summaries of all known nodes, sorted by node number.
|
|
* Names/hwModel/batteryLevel/snr/hopsAway/viaMqtt appear only when the device reported them;
|
|
* `lastHeard` is epoch seconds (null = never heard); `position` only when both coordinates are known.
|
|
*/
|
|
async nodes(): Promise<NodeSummary[]> {
|
|
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<NodeSummary | null> {
|
|
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 '<to>'` when nothing or more than one matches.
|
|
*/
|
|
async resolveDest(to?: number | string): Promise<number> {
|
|
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. Backed by SQLite: the buffer is
|
|
* seeded from the database on startup (newest 100) and re-upserted every MESH_POLL_MS, so history
|
|
* survives process restarts and bridge reconnects; the DB keeps the newest 1000 rows.
|
|
*/
|
|
messages(): MessageRecord[] {
|
|
return [...this.history].reverse();
|
|
}
|
|
|
|
/**
|
|
* The last HISTORY_LIMIT rx texts that passed the notify gate and were handed off to opencode, newest first:
|
|
* channel matched MESH_NOTIFY_CHANNEL, not a self-echo, and a notifier is wired. In-memory only — reset on restart.
|
|
*/
|
|
forwarded(): ForwardedRecord[] {
|
|
return [...this.forwardedHistory];
|
|
}
|
|
|
|
/**
|
|
* Re-upsert the entire in-memory history buffer into SQLite (INSERT OR REPLACE per row) and enforce
|
|
* retention: only the newest DB_RETENTION rows survive. Called by the poller every MESH_POLL_MS;
|
|
* public so tests can drive a tick without real timers. No-op when no database is configured.
|
|
*/
|
|
flushHistory(): void {
|
|
const db = this.db;
|
|
if (!db) return;
|
|
const upsert = db.prepare(
|
|
'INSERT OR REPLACE INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
|
|
);
|
|
for (const r of this.history) {
|
|
upsert.run(
|
|
r.id,
|
|
r.dir,
|
|
r.to ?? null,
|
|
r.from ?? null,
|
|
r.text,
|
|
r.channel,
|
|
r.at,
|
|
r.status ?? null,
|
|
r.ackError ?? null,
|
|
);
|
|
}
|
|
db.exec(
|
|
`DELETE FROM messages WHERE id NOT IN (SELECT id FROM messages ORDER BY at DESC LIMIT ${DB_RETENTION})`,
|
|
);
|
|
}
|
|
|
|
/** Load the newest HISTORY_LIMIT rows from SQLite into the buffer (oldest-first) so history survives restarts. */
|
|
private seedHistory(): void {
|
|
const rows = this.db!.prepare(
|
|
'SELECT id, dir, to_node, from_node, text, channel, at, status, ack_error FROM messages ORDER BY at DESC LIMIT ?',
|
|
).all(HISTORY_LIMIT) as Array<Record<string, unknown>>;
|
|
// rows come back newest first — the buffer is oldest-first
|
|
for (const row of [...rows].reverse())
|
|
this.history.push(recordFromRow(row));
|
|
}
|
|
|
|
/** Look up a history record by packet id: in-memory buffer first, then SQLite (newest 1000). */
|
|
private findRecord(id: number): MessageRecord | undefined {
|
|
const hit = this.history.find((r) => r.id === id);
|
|
if (hit) return hit;
|
|
const db = this.db;
|
|
if (!db) return undefined;
|
|
const row = db
|
|
.prepare(
|
|
'SELECT id, dir, to_node, from_node, text, channel, at, status, ack_error FROM messages WHERE id = ?',
|
|
)
|
|
.get(id) as Record<string, unknown> | undefined;
|
|
return row ? recordFromRow(row) : undefined;
|
|
}
|
|
|
|
/**
|
|
* Reply to an earlier message by packet id — the caller never thinks about where it goes:
|
|
* rx records are answered as a DM back to the original sender, tx records go to their destination;
|
|
* broadcast originals (and injected messages without a sender) get a broadcast. The reply always
|
|
* rides the record's channel and carries replyId = messageId, so the recipient app renders it as a
|
|
* threaded quote of that packet. Returns { id } with the new packet id. Throws BadRequest on invalid
|
|
* arguments, NotFound when no message has the given id (in memory or in SQLite), 503 when the bridge is down.
|
|
*/
|
|
async respond(messageId: number, content: string): Promise<{ id: number }> {
|
|
if (
|
|
!Number.isInteger(messageId) ||
|
|
messageId < 1 ||
|
|
messageId > BROADCAST_ADDR
|
|
)
|
|
throw new BadRequestException(
|
|
'messageId must be an integer in 1..0xffffffff',
|
|
);
|
|
if (typeof content !== 'string' || content.length === 0)
|
|
throw new BadRequestException('content must be a non-empty string');
|
|
const rec = this.findRecord(messageId);
|
|
if (!rec) throw new NotFoundException(`no message with id ${messageId}`);
|
|
// rx → DM back to the sender unless the original was a broadcast (then reply as a broadcast — that's what the stored destination is for);
|
|
// tx → its own destination; injected records without a sender → broadcast
|
|
const target =
|
|
rec.dir === 'rx'
|
|
? rec.to === BROADCAST_ADDR
|
|
? BROADCAST_ADDR
|
|
: rec.from
|
|
: rec.to;
|
|
const to = target ?? BROADCAST_ADDR;
|
|
return { id: await this.sendText(to, content, messageId, rec.channel) };
|
|
}
|
|
|
|
/**
|
|
* Inject a synthetic inbound (rx) record into the message history — as if received on the mesh.
|
|
* NOT transmitted over RF: nothing is written to the bridge or air; the record exists so external
|
|
* systems can push messages into the bot's inbox to trigger communication later. `from` is optional
|
|
* (integer in 1..0xffffffff when provided) and omitted from the record when absent. The record rides
|
|
* channel 0 like a primary-channel capture and appears in messages() newest first, subject to the
|
|
* same HISTORY_LIMIT cap as real captures. Returns { id } with a random uint32 packet id (same
|
|
* generation as tx sends). Throws on empty text or an invalid `from`.
|
|
*/
|
|
injectMessage(text: string, from?: number): { id: number } {
|
|
if (typeof text !== 'string' || text.length === 0)
|
|
throw new Error('text must be a non-empty string');
|
|
if (from !== undefined) this.assertNodeNum(from);
|
|
const rec: MessageRecord = {
|
|
id: randomPacketId(),
|
|
dir: 'rx',
|
|
...(from !== undefined ? { from } : {}),
|
|
text,
|
|
channel: 0,
|
|
at: Date.now(),
|
|
};
|
|
this.pushHistory(rec);
|
|
// injected records ride channel 0 — the notifier fires only when MESH_NOTIFY_CHANNEL is set to match
|
|
this.notifyRx(0, text, from, rec.id);
|
|
return { id: rec.id };
|
|
}
|
|
|
|
/** Fire-and-forget opencode notifier push for an rx text on `channel` (no-op unless channels match and a notifier is wired). Both call sites pass the packet id. */
|
|
private notifyRx(
|
|
channel: number,
|
|
text: string,
|
|
from?: number,
|
|
id?: number,
|
|
): void {
|
|
if (!this.opencode || channel !== this.notifyChannel) return;
|
|
// self-echo filter: our own node must never trigger a notification of itself
|
|
if (from !== undefined && from === this.myNodeNum) {
|
|
this.logger.debug(`notify skipped: rx from our own node ${from}`);
|
|
return;
|
|
}
|
|
// record the forward at hand-off — the ring holds exactly what a wired notifier receives (newest first).
|
|
// Pinned semantics: an unwired notifier records nothing (early return above); a notifier disabled internally
|
|
// (empty credentials) still counts — MeshService cannot see its enabled flag.
|
|
const entry: ForwardedRecord = { id: id!, text, channel, at: Date.now() };
|
|
if (from !== undefined) entry.from = from;
|
|
this.forwardedHistory.unshift(entry);
|
|
if (this.forwardedHistory.length > HISTORY_LIMIT)
|
|
this.forwardedHistory.pop(); // oldest evicted
|
|
void this.opencode.notify(id!, text, from); // both call sites always pass the packet id
|
|
}
|
|
|
|
/**
|
|
* Telemetry records, newest first: TELEMETRY_APP packets from any node — device metrics
|
|
* (battery/voltage/uptime/channel+air utilization) and environment sensor readings where nodes have them.
|
|
* `kind` is the meshtastic.Telemetry oneof case name; `data` is that variant's payload with proto uint64
|
|
* fields as decimal strings, unset optional fields omitted. In-memory only, last 100, reset on reconnect.
|
|
*/
|
|
telemetry(): TelemetryRecord[] {
|
|
return [...this.telemetryHistory].reverse();
|
|
}
|
|
|
|
/** Send `text` to node `to`; returns the packet id. Rides the pinned test channel unless an explicit `channel` index is given (e.g. battery alerts on "testowy"). Optional `replyId` marks the message as a reply to an earlier packet (the recipient app renders it as a threaded quote). Throws ServiceUnavailable when the bridge is down. */
|
|
async sendText(
|
|
to: number,
|
|
text: string,
|
|
replyId?: number,
|
|
channel?: number,
|
|
): Promise<number> {
|
|
if (!this.connected || !this.socket)
|
|
throw new ServiceUnavailableException('bridge not connected');
|
|
const ch = channel ?? this.sendChannel;
|
|
const { bytes, id } = await buildTextToRadio(to, text, ch, replyId);
|
|
const rec: MessageRecord = {
|
|
id,
|
|
dir: 'tx',
|
|
to,
|
|
text,
|
|
channel: ch,
|
|
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 pushTelemetry(rec: TelemetryRecord): void {
|
|
this.telemetryHistory.push(rec);
|
|
// ponytail: shift on a ≤100 array is fine; true ring buffer only if this ever gets hot
|
|
if (this.telemetryHistory.length > HISTORY_LIMIT)
|
|
this.telemetryHistory.shift();
|
|
}
|
|
|
|
private summarize(n: NodeEntry, names: Map<number, string>): NodeSummary {
|
|
const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null };
|
|
if (n.user?.longName) s.longName = n.user.longName;
|
|
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 fields: undefined when the device never reported them
|
|
const battery = n.deviceMetrics?.batteryLevel;
|
|
if (battery !== undefined) s.batteryLevel = battery;
|
|
// non-optional proto scalars decode with their defaults (snr=0, viaMqtt=false), so a default value means "not reported"
|
|
if (n.snr !== undefined && n.snr !== 0) s.snr = n.snr;
|
|
if (n.hopsAway !== undefined) s.hopsAway = n.hopsAway; // optional in proto → truly absent when not reported
|
|
if (n.viaMqtt === true) s.viaMqtt = true;
|
|
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/device info/telemetry from the old session;
|
|
// message history is NOT reset: it's an append-only log backed by SQLite and survives reconnects
|
|
this.channelMap = new Map();
|
|
this.nodeMap = new Map();
|
|
this.metadata = null;
|
|
this.configSections.clear();
|
|
this.moduleConfigSections.clear();
|
|
this.telemetryHistory = [];
|
|
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<void> {
|
|
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 'metadata':
|
|
// DeviceMetadata streamed once during initial sync (no key material in it)
|
|
this.metadata = v.value;
|
|
break;
|
|
case 'config': {
|
|
// device streams one Config per section during initial sync
|
|
const sec = v.value.payloadVariant;
|
|
if (sec?.case) this.configSections.set(sec.case, sec.value);
|
|
break;
|
|
}
|
|
case 'moduleConfig': {
|
|
const mod = v.value.payloadVariant;
|
|
if (mod?.case) this.moduleConfigSections.set(mod.case, mod.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,
|
|
to: pkt.to, // destination (broadcast addr for broadcasts) — lets a later reply tell DM vs broadcast apart
|
|
text,
|
|
channel: pkt.channel ?? 0,
|
|
at: Date.now(),
|
|
});
|
|
this.notifyRx(pkt.channel ?? 0, text, pkt.from, pkt.id);
|
|
} 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_NODEINFO_APP) {
|
|
// live node-info update: NodeInfo serialized in Data.payload; deviceMetrics carries
|
|
// battery_level so /api/nodes and the battery alerts see fresh values between syncs
|
|
let ni: Proto.Mesh.NodeInfo | undefined;
|
|
try {
|
|
ni = await decodeNodeInfo(decoded.payload);
|
|
} catch {
|
|
ni = undefined; // malformed payload — skip, keep whatever we had
|
|
}
|
|
if (ni?.deviceMetrics !== undefined) {
|
|
const num = pkt.from;
|
|
this.nodeMap.set(num, {
|
|
...this.nodeMap.get(num),
|
|
num,
|
|
deviceMetrics: ni.deviceMetrics,
|
|
});
|
|
}
|
|
} 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);
|
|
}
|
|
}
|
|
} else if (decoded && decoded.portnum === PORT_TELEMETRY_APP) {
|
|
// live telemetry: the Telemetry is serialized in Data.payload
|
|
let tel: Proto.Telemetry.Telemetry | undefined;
|
|
try {
|
|
tel = await decodeTelemetry(decoded.payload);
|
|
} catch {
|
|
tel = undefined; // malformed payload — skip
|
|
}
|
|
const variant = tel?.variant;
|
|
if (variant?.case) {
|
|
this.pushTelemetry({
|
|
from: pkt.from,
|
|
at: Date.now(),
|
|
kind: variant.case,
|
|
data: jsonSafe(variant.value),
|
|
});
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|