From e62f831ee5f7eb9afff6628bde2c3141a23caf71 Mon Sep 17 00:00:00 2001 From: Kamil Krzywda Date: Fri, 25 Sep 2026 15:48:09 +0000 Subject: [PATCH] mesh: in-memory ring of the last 100 messages forwarded to opencode --- src/mesh/mesh.service.spec.ts | 155 ++++++++++++++++++++++++++++++++++ src/mesh/mesh.service.ts | 30 +++++++ 2 files changed, 185 insertions(+) diff --git a/src/mesh/mesh.service.spec.ts b/src/mesh/mesh.service.spec.ts index c415a0a..e0467d3 100644 --- a/src/mesh/mesh.service.spec.ts +++ b/src/mesh/mesh.service.spec.ts @@ -1048,6 +1048,161 @@ describe('MeshService opencode notifier', () => { }); }); +describe('MeshService forwarded ring', () => { + const MY_NODE = 987654321; + let notifyMock: ReturnType; + + beforeEach(() => { + notifyMock = vi.fn( + async (_id: number, _text: string, _from?: number) => {}, + ); + }); + + // 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 } }, + }), + ), + ); + } + + // MeshService with a mocked OpencodeService wired in (structural cast — only notify() is used) + function makeSvc(): MeshService { + return new MeshService(undefined, { + notify: notifyMock, + } as unknown as OpencodeService); + } + + it('records an entry when an rx text passes the gate and is handed off (channel match)', 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_CHANNEL; // default 3 + const svc = makeSvc(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3)); + + await waitFor(() => notifyMock.mock.calls.length === 1); + expect(svc.forwarded()).toEqual([ + { + id: 1, + from: KAMIL_NUM, + text: 'ping', + channel: 3, + at: expect.any(Number), + }, + ]); + } finally { + delete process.env.MESH_NOTIFY_CHANNEL; + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('does not record on channel mismatch', 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_CHANNEL; // default 3 + const svc = makeSvc(); + 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 off-channel it never reaches the ring + await waitFor(() => svc.messages().length === 1); + expect(notifyMock).not.toHaveBeenCalled(); + expect(svc.forwarded()).toEqual([]); + } finally { + delete process.env.MESH_NOTIFY_CHANNEL; + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('does not record a self-echo (rx 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); + delete process.env.MESH_NOTIFY_CHANNEL; // default 3 + const svc = makeSvc(); + 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); + + 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(notifyMock).not.toHaveBeenCalled(); + expect(svc.forwarded()).toEqual([]); + } finally { + delete process.env.MESH_NOTIFY_CHANNEL; + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('does not record when no notifier is wired', () => { + process.env.MESH_NOTIFY_CHANNEL = '0'; // injected records ride channel 0 — the gate would pass + const svc = new MeshService(); // unwired — pinned semantics: nothing forwarded, nothing recorded + try { + svc.injectMessage('hi'); + + expect(svc.forwarded()).toEqual([]); + } finally { + delete process.env.MESH_NOTIFY_CHANNEL; + } + }); + + it('counts injected messages that pass the gate and omits `from` when there is no sender', () => { + process.env.MESH_NOTIFY_CHANNEL = '0'; + const svc = makeSvc(); + try { + const r = svc.injectMessage('hi'); // no from + + expect(svc.forwarded()).toEqual([ + { id: r.id, text: 'hi', channel: 0, at: expect.any(Number) }, + ]); + expect('from' in svc.forwarded()[0]!).toBe(false); + } finally { + delete process.env.MESH_NOTIFY_CHANNEL; + } + }); + + it('caps at 100 (oldest evicted), newest first', () => { + process.env.MESH_NOTIFY_CHANNEL = '0'; + const svc = makeSvc(); + try { + for (let i = 0; i < 101; i++) svc.injectMessage(`m-${i}`); + + expect(svc.forwarded()).toHaveLength(100); + expect(svc.forwarded()[0]!.text).toBe('m-100'); // newest first + expect(svc.forwarded()[99]!.text).toBe('m-1'); // m-0 evicted + } finally { + delete process.env.MESH_NOTIFY_CHANNEL; + } + }); +}); + describe('MeshService injectMessage', () => { it('injects rx records that appear newest-first in messages()', () => { const svc = new MeshService(); diff --git a/src/mesh/mesh.service.ts b/src/mesh/mesh.service.ts index a5bd8fe..cf3e0be 100644 --- a/src/mesh/mesh.service.ts +++ b/src/mesh/mesh.service.ts @@ -51,6 +51,18 @@ export interface MessageRecord { 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 */ @@ -215,6 +227,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { private moduleConfigSections = new Map>(); // 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(); // telemetry history (newest kept, capped at HISTORY_LIMIT) private telemetryHistory: TelemetryRecord[] = []; @@ -376,6 +390,14 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { 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; @@ -502,6 +524,14 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { 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 }