mesh: in-memory ring of the last 100 messages forwarded to opencode
This commit is contained in:
@@ -1048,6 +1048,161 @@ describe('MeshService opencode notifier', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('MeshService forwarded ring', () => {
|
||||
const MY_NODE = 987654321;
|
||||
let notifyMock: ReturnType<typeof vi.fn>;
|
||||
|
||||
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();
|
||||
|
||||
Reference in New Issue
Block a user