Message history with delivery tracking (GET /api/messages) (#8)
CI / build + unit (push) Successful in 32s
CI / build + unit (push) Successful in 32s
This commit was merged in pull request #8.
This commit is contained in:
@@ -9,16 +9,40 @@ import * as net from 'node:net';
|
||||
import type * as Proto from '@meshtastic/protobufs';
|
||||
import {
|
||||
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<Proto.Mesh.NodeInfo, 'num'> &
|
||||
Partial<Omit<Proto.Mesh.NodeInfo, 'num'>>;
|
||||
@@ -63,6 +87,9 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
>();
|
||||
// node database: NodeInfo frames from the initial sync + live POSITION_APP updates
|
||||
private nodeMap = new Map<number, NodeEntry>();
|
||||
// message history (newest kept, capped at HISTORY_LIMIT) + pending tx records awaiting their routing ACK/NACK
|
||||
private history: MessageRecord[] = [];
|
||||
private pendingTx = new Map<number, MessageRecord>();
|
||||
private reconnectTimer: NodeJS.Timeout | null = null;
|
||||
private shuttingDown = false;
|
||||
|
||||
@@ -149,15 +176,41 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<number> {
|
||||
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<number, string>): NodeSummary {
|
||||
const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null };
|
||||
if (n.user?.longName) s.longName = n.user.longName;
|
||||
@@ -183,9 +236,11 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
this.socket = socket;
|
||||
socket.on('connect', () => {
|
||||
this.connected = true;
|
||||
// fresh sync on every (re)connect — drop stale channels/nodes from the old session
|
||||
// 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)
|
||||
@@ -256,9 +311,16 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
? pkt.payloadVariant.value
|
||||
: undefined;
|
||||
if (decoded && decoded.portnum === PORT_TEXT_MESSAGE_APP) {
|
||||
this.logger.log(
|
||||
`[mesh] ${pkt.from} -> ${pkt.to}: "${new TextDecoder().decode(decoded.payload)}"`,
|
||||
);
|
||||
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;
|
||||
@@ -275,6 +337,28 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user