Outbound notification webhook (MESH_NOTIFY_URL) on RX text — issue #17 (#20)

src/mesh/mesh.service.spec.ts

@@ -1,7 +1,7 @@
 import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
 import * as P from '@meshtastic/protobufs';
 import { DatabaseSync } from 'node:sqlite';
-import { describe, expect, it } from 'vitest';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { BROADCAST_ADDR, extractFrames, frame } from './codec.js';
 import {
   KAMIL_NUM,
@@ -562,6 +562,116 @@ describe('MeshService message history', () => {
   });
 });
 
+describe('MeshService notify webhook', () => {
+  const fetchMock = vi.fn(async (..._args: unknown[]) => undefined);
+
+  beforeEach(() => fetchMock.mockClear());
+
+  it('POSTs {text, from} to MESH_NOTIFY_URL when an rx text lands on the notify channel (default 3)', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
+    vi.stubGlobal('fetch', fetchMock);
+    const svc = new MeshService();
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3));
+
+      await waitFor(() => fetchMock.mock.calls.length === 1);
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+      const [url, init] = fetchMock.mock.calls[0];
+      expect(url).toBe('http://notify.example/hook');
+      const opts = init as {
+        method?: string;
+        headers?: Record<string, string>;
+        body?: string;
+      };
+      expect(opts.method).toBe('POST');
+      expect(opts.headers?.['content-type']).toBe('application/json');
+      expect(JSON.parse(String(opts.body))).toEqual({
+        text: 'ping',
+        from: KAMIL_NUM,
+      });
+    } finally {
+      vi.unstubAllGlobals();
+      delete process.env.MESH_NOTIFY_URL;
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('does not call fetch for rx texts on other channels', async () => {
+    const bridge = await startBridge();
+    process.env.MESH_HOST = '127.0.0.1';
+    process.env.MESH_PORT = String(bridge.port);
+    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
+    vi.stubGlobal('fetch', fetchMock);
+    const svc = new MeshService();
+    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 no webhook fires off-channel
+      await waitFor(() => svc.messages().length === 1);
+      expect(fetchMock).not.toHaveBeenCalled();
+    } finally {
+      vi.unstubAllGlobals();
+      delete process.env.MESH_NOTIFY_URL;
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+
+  it('never calls fetch when MESH_NOTIFY_URL is unset or empty', 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_URL; // unset → disabled
+    vi.stubGlobal('fetch', fetchMock);
+    const svc = new MeshService();
+    try {
+      svc.onModuleInit();
+      await waitFor(
+        () => bridge.sockets.length === 1 && svc.status().connected,
+      );
+
+      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3));
+      await waitFor(() => svc.messages().length === 1);
+      expect(fetchMock).not.toHaveBeenCalled();
+
+      // empty string → also disabled (a fresh service re-reads the env)
+      process.env.MESH_NOTIFY_URL = '';
+      const svc2 = new MeshService();
+      try {
+        svc2.onModuleInit();
+        await waitFor(
+          () => bridge.sockets.length === 2 && svc2.status().connected,
+        );
+
+        bridge.sockets[1].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3));
+        await waitFor(() => svc2.messages().length === 1);
+      } finally {
+        delete process.env.MESH_NOTIFY_URL;
+        svc2.onModuleDestroy();
+      }
+      expect(fetchMock).not.toHaveBeenCalled();
+    } finally {
+      vi.unstubAllGlobals();
+      svc.onModuleDestroy();
+      bridge.close();
+    }
+  });
+});
+
 describe('MeshService injectMessage', () => {
   it('injects rx records that appear newest-first in messages()', () => {
     const svc = new MeshService();

src/mesh/mesh.service.ts

@@ -157,6 +157,12 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
     process.env.MESH_SEND_CHANNEL ?? '3',
     10,
   );
+  // outbound notification webhook: fire-and-forget POST when an rx text lands on the notify channel; empty URL = disabled
+  private readonly notifyUrl = process.env.MESH_NOTIFY_URL ?? '';
+  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);
 
@@ -599,6 +605,19 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
             channel: pkt.channel ?? 0,
             at: Date.now(),
           });
+          if (this.notifyUrl && (pkt.channel ?? 0) === this.notifyChannel) {
+            const body: Record<string, unknown> = { text };
+            if (pkt.from !== undefined) body.from = pkt.from;
+            void fetch(this.notifyUrl, {
+              method: 'POST',
+              headers: { 'content-type': 'application/json' },
+              body: JSON.stringify(body),
+            }).catch((err: unknown) =>
+              this.logger.warn(
+                'notify webhook failed: ' + (err as Error).message,
+              ),
+            );
+          }
         } else if (decoded && decoded.portnum === PORT_POSITION_APP) {
           // live position update: the Position is serialized in Data.payload
           let pos: Proto.Mesh.Position | undefined;