mesh: replace notify webhook with in-process opencode notifier; drop forwarder service

src/mesh/mesh.module.ts

@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { mkdirSync } from 'node:fs';
 import { dirname } from 'node:path';
 import { DatabaseSync } from 'node:sqlite';
+import { OpencodeService } from '../opencode/opencode.service.js';
 import { McpController } from './mcp.controller.js';
 import { MeshController } from './mesh.controller.js';
 import { initMessagesTable, MeshService } from './mesh.service.js';
@@ -18,9 +19,12 @@ function openMessageDb(): DatabaseSync {
 @Module({
   controllers: [MeshController, McpController],
   providers: [
+    OpencodeService,
     {
       provide: MeshService,
-      useFactory: () => new MeshService(openMessageDb()),
+      useFactory: (opencode: OpencodeService) =>
+        new MeshService(openMessageDb(), opencode),
+      inject: [OpencodeService],
     },
   ],
   exports: [MeshService],

src/mesh/mesh.service.spec.ts

@@ -1,5 +1,4 @@
 import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
-import { Logger } from '@nestjs/common';
 import * as P from '@meshtastic/protobufs';
 import { DatabaseSync } from 'node:sqlite';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
@@ -15,6 +14,7 @@ import {
   telemetryPacketFrame,
   waitFor,
 } from './fake-bridge.js';
+import { OpencodeService } from '../opencode/opencode.service.js';
 import { initMessagesTable, MeshService } from './mesh.service.js';
 
 // live position update: POSITION_APP packet with a serialized Position in Data.payload
@@ -692,12 +692,14 @@ describe('MeshService message history', () => {
   });
 });
 
-describe('MeshService notify webhook', () => {
-  const fetchMock = vi.fn(async (..._args: unknown[]) => undefined);
+describe('MeshService opencode notifier', () => {
   // our own node number for the self-echo filter specs (distinct from KAMIL_NUM, an external user)
   const MY_NODE = 987654321;
+  let notifyMock: ReturnType<typeof vi.fn>;
 
-  beforeEach(() => fetchMock.mockClear());
+  beforeEach(() => {
+    notifyMock = vi.fn(async (_text: string, _from?: number) => {});
+  });
 
   // myInfo frame as streamed during the initial sync — sets svc.myNodeNum
   function myInfoFrame(num: number): Uint8Array {
@@ -711,13 +713,20 @@ describe('MeshService notify webhook', () => {
     );
   }
 
-  it('POSTs {text, from} to MESH_NOTIFY_URL when an rx text lands on the notify channel (default 3)', async () => {
+  // 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('notifies via the opencode service 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();
+    delete process.env.MESH_NOTIFY_CHANNEL; // default 3
+    const svc = makeSvc();
     try {
       svc.onModuleInit();
       await waitFor(
@@ -726,36 +735,22 @@ describe('MeshService notify webhook', () => {
 
       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,
-      });
+      await waitFor(() => notifyMock.mock.calls.length === 1);
+      expect(notifyMock).toHaveBeenCalledTimes(1);
+      expect(notifyMock).toHaveBeenCalledWith('ping', KAMIL_NUM);
     } finally {
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
+      delete process.env.MESH_NOTIFY_CHANNEL;
       svc.onModuleDestroy();
       bridge.close();
     }
   });
 
-  it('does not call fetch for rx texts on other channels', async () => {
+  it('does not notify 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();
+    delete process.env.MESH_NOTIFY_CHANNEL; // default 3
+    const svc = makeSvc();
     try {
       svc.onModuleInit();
       await waitFor(
@@ -764,109 +759,48 @@ describe('MeshService notify webhook', () => {
 
       bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2));
 
-      // the message is captured in history — but no webhook fires off-channel
+      // the message is captured in history — but no notification fires off-channel
       await waitFor(() => svc.messages().length === 1);
-      expect(fetchMock).not.toHaveBeenCalled();
+      expect(notifyMock).not.toHaveBeenCalled();
     } finally {
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
+      delete process.env.MESH_NOTIFY_CHANNEL;
       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();
-    }
-  });
-
-  it('fires the webhook for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => {
-    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
+  it('fires the notifier for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => {
     process.env.MESH_NOTIFY_CHANNEL = '0';
-    vi.stubGlobal('fetch', fetchMock);
-    const svc = new MeshService();
+    const svc = makeSvc();
     try {
       svc.injectMessage('hi', 42);
 
-      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: 'hi', from: 42 });
+      expect(notifyMock).toHaveBeenCalledTimes(1);
+      expect(notifyMock).toHaveBeenCalledWith('hi', 42);
     } finally {
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
       delete process.env.MESH_NOTIFY_CHANNEL;
     }
   });
 
-  it('omits `from` in the webhook body when an injected message has no sender', async () => {
-    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
+  it('passes `from` through as undefined when an injected message has no sender', async () => {
     process.env.MESH_NOTIFY_CHANNEL = '0';
-    vi.stubGlobal('fetch', fetchMock);
-    const svc = new MeshService();
+    const svc = makeSvc();
     try {
       svc.injectMessage('hi');
 
-      expect(fetchMock).toHaveBeenCalledTimes(1);
-      const [, init] = fetchMock.mock.calls[0];
-      const opts = init as { body?: string };
-      expect(JSON.parse(String(opts.body))).toEqual({ text: 'hi' });
+      expect(notifyMock).toHaveBeenCalledTimes(1);
+      expect(notifyMock).toHaveBeenCalledWith('hi', undefined);
     } finally {
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
       delete process.env.MESH_NOTIFY_CHANNEL;
     }
   });
 
-  it('does not call fetch for rx texts from our own node on the notify channel', async () => {
+  it('does not notify for rx texts 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);
-    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
-    vi.stubGlobal('fetch', fetchMock);
-    const svc = new MeshService();
+    delete process.env.MESH_NOTIFY_CHANNEL; // default 3
+    const svc = makeSvc();
     try {
       svc.onModuleInit();
       await waitFor(
@@ -877,28 +811,25 @@ describe('MeshService notify webhook', () => {
       bridge.sockets[0].write(myInfoFrame(MY_NODE));
       await waitFor(() => svc.status().myNodeNum === MY_NODE);
 
-      // an rx text from ourselves on the notify channel must not fire the webhook
+      // an rx text from ourselves on the notify channel must not fire the notifier
       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(fetchMock).not.toHaveBeenCalled();
+      expect(notifyMock).not.toHaveBeenCalled();
     } finally {
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
+      delete process.env.MESH_NOTIFY_CHANNEL;
       svc.onModuleDestroy();
       bridge.close();
     }
   });
 
-  it('does not call fetch for an injected message from our own node', async () => {
+  it('does not notify for an injected message from our own node', 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';
     process.env.MESH_NOTIFY_CHANNEL = '0';
-    vi.stubGlobal('fetch', fetchMock);
-    const svc = new MeshService();
+    const svc = makeSvc();
     try {
       svc.onModuleInit();
       await waitFor(
@@ -910,109 +841,13 @@ describe('MeshService notify webhook', () => {
 
       svc.injectMessage('hi', MY_NODE);
 
-      expect(fetchMock).not.toHaveBeenCalled();
+      expect(notifyMock).not.toHaveBeenCalled();
     } finally {
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
       delete process.env.MESH_NOTIFY_CHANNEL;
       svc.onModuleDestroy();
       bridge.close();
     }
   });
-
-  it('retries a failed webhook POST with backoff and stops on success', async () => {
-    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
-    process.env.MESH_NOTIFY_CHANNEL = '0';
-    const fetchMock = vi
-      .fn()
-      .mockRejectedValueOnce(new Error('boom 1'))
-      .mockRejectedValueOnce(new Error('boom 2'))
-      .mockResolvedValueOnce(undefined);
-    vi.stubGlobal('fetch', fetchMock);
-    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
-    const sleep = vi.fn(async (_ms: number) => {});
-    const svc = new MeshService();
-    (svc as unknown as { sleep(ms: number): Promise<void> }).sleep = sleep;
-    try {
-      expect(() => svc.injectMessage('hi', 42)).not.toThrow();
-
-      await waitFor(() => fetchMock.mock.calls.length === 3);
-      // success on the third attempt: exactly 3 POSTs, no dropped log
-      expect(fetchMock).toHaveBeenCalledTimes(3);
-      expect(sleep).toHaveBeenNthCalledWith(1, 2_000);
-      expect(sleep).toHaveBeenNthCalledWith(2, 10_000);
-      const warns = warnSpy.mock.calls.map((c) => String(c[0]));
-      expect(warns.some((w) => w.includes('attempt 1/4'))).toBe(true);
-      expect(warns.some((w) => w.includes('attempt 2/4'))).toBe(true);
-      expect(warns.some((w) => w.includes('notify dropped'))).toBe(false);
-    } finally {
-      warnSpy.mockRestore();
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
-      delete process.env.MESH_NOTIFY_CHANNEL;
-    }
-  });
-
-  it('retries a webhook POST that answers with an HTTP error status and stops on success', async () => {
-    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
-    process.env.MESH_NOTIFY_CHANNEL = '0';
-    const fetchMock = vi
-      .fn()
-      .mockResolvedValueOnce({ ok: false, status: 502 })
-      .mockResolvedValueOnce({ ok: false, status: 502 })
-      .mockResolvedValueOnce({ ok: true, status: 200 });
-    vi.stubGlobal('fetch', fetchMock);
-    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
-    const sleep = vi.fn(async (_ms: number) => {});
-    const svc = new MeshService();
-    (svc as unknown as { sleep(ms: number): Promise<void> }).sleep = sleep;
-    try {
-      expect(() => svc.injectMessage('hi', 42)).not.toThrow();
-
-      await waitFor(() => fetchMock.mock.calls.length === 3);
-      // HTTP 502 twice, then 200: exactly 3 POSTs, no dropped log
-      expect(fetchMock).toHaveBeenCalledTimes(3);
-      const warns = warnSpy.mock.calls.map((c) => String(c[0]));
-      expect(warns.some((w) => w.includes('attempt 1/4'))).toBe(true);
-      expect(warns.some((w) => w.includes('HTTP 502'))).toBe(true);
-      expect(warns.some((w) => w.includes('notify dropped'))).toBe(false);
-    } finally {
-      warnSpy.mockRestore();
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
-      delete process.env.MESH_NOTIFY_CHANNEL;
-    }
-  });
-
-  it('gives up after 4 attempts and logs notify dropped', async () => {
-    process.env.MESH_NOTIFY_URL = 'http://notify.example/hook';
-    process.env.MESH_NOTIFY_CHANNEL = '0';
-    const fetchMock = vi.fn().mockRejectedValue(new Error('down'));
-    vi.stubGlobal('fetch', fetchMock);
-    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
-    const sleep = vi.fn(async (_ms: number) => {});
-    const svc = new MeshService();
-    (svc as unknown as { sleep(ms: number): Promise<void> }).sleep = sleep;
-    try {
-      expect(() => svc.injectMessage('hi', 42)).not.toThrow();
-
-      await waitFor(() =>
-        warnSpy.mock.calls.some((c) =>
-          String(c[0]).includes('notify dropped after 4 attempts'),
-        ),
-      );
-      // exactly 4 attempts, final WARN with the exact message, nothing thrown
-      expect(fetchMock).toHaveBeenCalledTimes(4);
-      const warns = warnSpy.mock.calls.map((c) => String(c[0]));
-      expect(warns).toContain('notify dropped after 4 attempts: down');
-      expect(sleep).toHaveBeenNthCalledWith(3, 30_000);
-    } finally {
-      warnSpy.mockRestore();
-      vi.unstubAllGlobals();
-      delete process.env.MESH_NOTIFY_URL;
-      delete process.env.MESH_NOTIFY_CHANNEL;
-    }
-  });
 });
 
 describe('MeshService injectMessage', () => {

src/mesh/mesh.service.ts

@@ -9,6 +9,7 @@ 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,
@@ -132,8 +133,6 @@ function jsonSafe<T>(value: T): T {
 
 // key material must never leave the process — replaced with a marker so callers can see the field exists
 const REDACTED = 'REDACTED';
-// ponytail: hardcoded notify-webhook backoff [2s, 10s, 30s] (~45 s window) — no env knob; tune here if outages run longer
-const NOTIFY_RETRY_DELAYS_MS = [2_000, 10_000, 30_000];
 /** Config sections that carry secrets: security (PKI keys) and network (wifi psk). */
 function redactSection(
   name: string,
@@ -161,8 +160,8 @@ 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 ?? '';
+  // 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,
@@ -172,10 +171,13 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
 
   /** 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) {
+  constructor(db?: DatabaseSync, opencode?: OpencodeService) {
     this.db = db ?? null;
+    this.opencode = opencode ?? null;
   }
 
   private socket: net.Socket | null = null;
@@ -430,54 +432,20 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
       at: Date.now(),
     };
     this.pushHistory(rec);
-    // injected records ride channel 0 — the webhook fires only when MESH_NOTIFY_CHANNEL is set to match
+    // injected records ride channel 0 — the notifier fires only when MESH_NOTIFY_CHANNEL is set to match
     this.notifyRx(0, text, from);
     return { id: rec.id };
   }
 
-  /** Fire-and-forget notify-webhook POST for an rx text on `channel` (no-op unless URL is set and channels match). */
+  /** Fire-and-forget opencode notifier push for an rx text on `channel` (no-op unless channels match and a notifier is wired). */
   private notifyRx(channel: number, text: string, from?: number): void {
-    if (this.notifyUrl && channel === this.notifyChannel) {
-      // 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;
-      }
-      const body: Record<string, unknown> = { text };
-      if (from !== undefined) body.from = from;
-      void this.postNotify(body);
+    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;
     }
-  }
-
-  /** POST the webhook with retries: up to NOTIFY_RETRY_DELAYS_MS.length + 1 attempts total. */
-  private async postNotify(body: Record<string, unknown>): Promise<void> {
-    const maxAttempts = NOTIFY_RETRY_DELAYS_MS.length + 1;
-    for (let attempt = 1; ; attempt++) {
-      try {
-        const res = await fetch(this.notifyUrl, {
-          method: 'POST',
-          headers: { 'content-type': 'application/json' },
-          body: JSON.stringify(body),
-        });
-        if (res && !res.ok) throw new Error('HTTP ' + res.status); // HTTP errors retry like network failures
-        return; // success on any attempt — same as today's single POST
-      } catch (err: unknown) {
-        const msg = (err as Error).message;
-        this.logger.warn(
-          `notify webhook failed (attempt ${attempt}/${maxAttempts}): ${msg}`,
-        );
-        if (attempt === maxAttempts) {
-          this.logger.warn(`notify dropped after ${attempt} attempts: ${msg}`);
-          return;
-        }
-        await this.sleep(NOTIFY_RETRY_DELAYS_MS[attempt - 1]);
-      }
-    }
-  }
-
-  /** Injectable so specs can skip the backoff delays. */
-  private sleep(ms: number): Promise<void> {
-    return new Promise((resolve) => setTimeout(resolve, ms));
+    void this.opencode.notify(text, from);
   }
 
   /**