import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
import * as P from '@meshtastic/protobufs';
import { BadRequestException, NotFoundException } from '@nestjs/common';
import * as net from 'node:net';
import { DatabaseSync } from 'node:sqlite';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { BROADCAST_ADDR, extractFrames, frame } from './codec.js';
import {
  KAMIL_NUM,
  configSectionFrame,
  kamilNodeInfo,
  metadataFrame,
  moduleConfigSectionFrame,
  nodeInfoFrame,
  outboundPackets,
  startBridge,
  telemetryPacketFrame,
  textPacketFrame,
  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
function positionPacketFrame(
  from: number,
  pos: { latitudeI?: number; longitudeI?: number },
): Uint8Array {
  const data = create(P.Mesh.DataSchema, {
    portnum: P.Portnums.PortNum.POSITION_APP,
    payload: toBinary(
      P.Mesh.PositionSchema,
      create(P.Mesh.PositionSchema, pos),
    ),
  });
  const pkt = create(P.Mesh.MeshPacketSchema, {
    from,
    id: 1,
    payloadVariant: { case: 'decoded', value: data },
  });
  return frame(
    toBinary(
      P.Mesh.FromRadioSchema,
      create(P.Mesh.FromRadioSchema, {
        payloadVariant: { case: 'packet', value: pkt },
      }),
    ),
  );
}

// live node-info update: NODEINFO_APP packet with a serialized NodeInfo in Data.payload
function nodeInfoPacketFrame(
  from: number,
  fields: {
    user?: { id?: string; longName?: string };
    deviceMetrics?: { batteryLevel?: number };
  },
): Uint8Array {
  const data = create(P.Mesh.DataSchema, {
    portnum: P.Portnums.PortNum.NODEINFO_APP,
    payload: toBinary(
      P.Mesh.NodeInfoSchema,
      create(P.Mesh.NodeInfoSchema, { num: from, ...fields }),
    ),
  });
  const pkt = create(P.Mesh.MeshPacketSchema, {
    from,
    id: 1,
    payloadVariant: { case: 'decoded', value: data },
  });
  return frame(
    toBinary(
      P.Mesh.FromRadioSchema,
      create(P.Mesh.FromRadioSchema, {
        payloadVariant: { case: 'packet', value: pkt },
      }),
    ),
  );
}

// ACK/NACK for our sends: ROUTING_APP packet with Data.requestId = original id and a serialized Routing in the payload
function routingFrame(
  requestId: number,
  errorReason: P.Mesh.Routing_Error,
): Uint8Array {
  const data = create(P.Mesh.DataSchema, {
    portnum: P.Portnums.PortNum.ROUTING_APP,
    requestId,
    payload: toBinary(
      P.Mesh.RoutingSchema,
      create(P.Mesh.RoutingSchema, {
        variant: { case: 'errorReason', value: errorReason },
      }),
    ),
  });
  const pkt = create(P.Mesh.MeshPacketSchema, {
    from: 1,
    id: 2,
    payloadVariant: { case: 'decoded', value: data },
  });
  return frame(
    toBinary(
      P.Mesh.FromRadioSchema,
      create(P.Mesh.FromRadioSchema, {
        payloadVariant: { case: 'packet', value: pkt },
      }),
    ),
  );
}

function channelFrame(
  index: number,
  settings?: { name?: string; uplinkEnabled?: boolean },
): Uint8Array {
  const ch = create(P.Channel.ChannelSchema, {
    index,
    ...(settings
      ? {
          // psk must never leak into channels() output
          settings: {
            name: settings.name ?? '',
            psk: new Uint8Array(16).fill(7),
            uplinkEnabled: settings.uplinkEnabled ?? false,
          },
        }
      : {}),
  });
  const fr = create(P.Mesh.FromRadioSchema, {
    payloadVariant: { case: 'channel', value: ch },
  });
  return frame(toBinary(P.Mesh.FromRadioSchema, fr));
}

describe('MeshService channel pipeline', () => {
  it('collects channels from FromRadio frames and redacts keys', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      expect(svc.channels()).toEqual([]);

      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(
        channelFrame(2, { name: 'LongFast', uplinkEnabled: true }),
      );
      bridge.sockets[0].write(channelFrame(0)); // channel with no settings at all

      await waitFor(() => svc.channels().length === 2);
      expect(svc.channels()).toEqual([
        { index: 0, name: '', uplinkEnabled: false, downlinkEnabled: false },
        {
          index: 2,
          name: 'LongFast',
          uplinkEnabled: true,
          downlinkEnabled: false,
        },
      ]);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('resets channels on reconnect (no stale data)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50';
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(channelFrame(1, { name: 'Stale' }));
      await waitFor(() => svc.channels().length === 1);

      // drop the connection; bot reconnects and must start with an empty table
      bridge.sockets[0].destroy();
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
      await waitFor(() => svc.channels().length === 0);
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService connect', () => {
  it('sends want_config_id on connect and applies the initial sync', async () => {
    const out: Uint8Array[] = [];
    const bridge = await startBridge((s) => s.on('data', (c) => out.push(c)));
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(() => svc.status().connected && out.length > 0);

      const merged = new Uint8Array(out.reduce((n, c) => n + c.length, 0));
      let off = 0;
      for (const c of out) {
        merged.set(c, off);
        off += c.length;
      }
      const { packets } = extractFrames(merged);
      expect(packets.length).toBeGreaterThan(0);
      const tr = fromBinary(P.Mesh.ToRadioSchema, packets[0]);
      expect(tr.payloadVariant?.case).toBe('wantConfigId');
      const nonce =
        tr.payloadVariant && tr.payloadVariant.case === 'wantConfigId'
          ? tr.payloadVariant.value
          : -1;
      expect(nonce).toBeGreaterThan(0);

      // feed back the initial sync: myInfo, then configCompleteId echoing the nonce
      bridge.sockets[0].write(
        frame(
          toBinary(
            P.Mesh.FromRadioSchema,
            create(P.Mesh.FromRadioSchema, {
              payloadVariant: { case: 'myInfo', value: { myNodeNum: 1234 } },
            }),
          ),
        ),
      );
      bridge.sockets[0].write(
        frame(
          toBinary(
            P.Mesh.FromRadioSchema,
            create(P.Mesh.FromRadioSchema, {
              payloadVariant: { case: 'configCompleteId', value: nonce },
            }),
          ),
        ),
      );

      await waitFor(() => svc.status().myNodeNum === 1234);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

describe('MeshService idle watchdog', () => {
  it('forces a reconnect when an established connection goes silent past MESH_IDLE_TIMEOUT_MS', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
    process.env.MESH_IDLE_TIMEOUT_MS = '100'; // override — the default (600000) would never fire in a test
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // zombie case: connected but no inbound traffic at all — wait past the timeout, then drive a watchdog tick
      await sleep(250);
      svc.checkIdle();

      // socket torn down → existing connect/retry loop re-establishes with a fresh session
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      delete process.env.MESH_IDLE_TIMEOUT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('raw non-frame bytes do NOT keep the watchdog alive — a garbage-only stream forces a reconnect', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
    process.env.MESH_IDLE_TIMEOUT_MS = '100'; // override — the default (600000) would never fire in a test
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // today's incident: the peer keeps dribbling small non-frame bytes (keepalive noise / desynced stream) — raw chunks must not reset the idle clock
      const garbage = Buffer.from('zz-no-frames-here'); // pure ASCII — never forms a 0x94 0xc3 frame header
      for (let i = 0; i < 6; i++) {
        await sleep(50); // ~300 ms wall time, well past the timeout
        bridge.sockets[0].write(garbage);
      }

      svc.checkIdle();
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      delete process.env.MESH_IDLE_TIMEOUT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('forces a reconnect when the undecoded frame buffer exceeds MESH_MAX_FRAME_BUFFER', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
    process.env.MESH_MAX_FRAME_BUFFER = '100'; // tiny cap — Node delivers socket data in ~64 KB events, far below the 256_000 default
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // non-frame bytes (0x42 is not part of any frame header) — a desynced stream that never forms frames; the first ~64 KB data event alone exceeds the cap
      bridge.sockets[0].write(Buffer.alloc(300_000, 0x42));

      // overflow forced a reconnect — fresh session
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);

      // the new socket starts with a clean buffer and decodes valid frames normally (the myInfo frame is well under the cap)
      const fr = create(P.Mesh.FromRadioSchema, {
        payloadVariant: { case: 'myInfo', value: { myNodeNum: 777 } },
      });
      bridge.sockets[1].write(frame(toBinary(P.Mesh.FromRadioSchema, fr)));
      await waitFor(() => svc.status().myNodeNum === 777);
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      delete process.env.MESH_MAX_FRAME_BUFFER;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('inbound traffic keeps resetting the idle clock — no false trigger while packets arrive', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_IDLE_TIMEOUT_MS = '100';
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // steady drip of inbound frames for well over the timeout in wall time — each chunk resets lastPacketAt
      for (let i = 0; i < 5; i++) {
        await sleep(60);
        bridge.sockets[0].write(nodeInfoFrame(42, kamilNodeInfo()));
      }

      svc.checkIdle(); // wall time is past the timeout, but traffic kept resetting it
      await sleep(100); // enough for a destroy to surface as 'close' (connected would flip false)
      expect(svc.status().connected).toBe(true);
      expect(bridge.sockets.length).toBe(1);
    } finally {
      delete process.env.MESH_IDLE_TIMEOUT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('does not trigger under the default timeout (MESH_IDLE_TIMEOUT_MS unset)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    delete process.env.MESH_IDLE_TIMEOUT_MS; // default 600000 — a test can't wait that long, so silence must NOT fire
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      await sleep(250); // silent, but far under the default budget
      svc.checkIdle();

      expect(svc.status().connected).toBe(true);
      expect(bridge.sockets.length).toBe(1);
    } finally {
      delete process.env.MESH_IDLE_TIMEOUT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('enables TCP keepalive on the bridge socket', async () => {
    const setKeepAlive = vi.spyOn(net.Socket.prototype, 'setKeepAlive');
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(() => svc.status().connected);
      expect(setKeepAlive).toHaveBeenCalledWith(true, 30_000);
    } finally {
      setKeepAlive.mockRestore();
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService send', () => {
  it('sends on the test channel (index 3) by default, never primary', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    delete process.env.MESH_SEND_CHANNEL; // default path: "test" (index 3)
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const out: Uint8Array[] = [];
      bridge.sockets[0].on('data', (c) => out.push(c));
      const id = await svc.sendText(42, 'hello test');
      expect(id).toBeGreaterThan(0);

      const readOutbound = (): P.Mesh.MeshPacket | undefined => {
        const merged = new Uint8Array(out.reduce((n, c) => n + c.length, 0));
        let off = 0;
        for (const c of out) {
          merged.set(c, off);
          off += c.length;
        }
        const { packets } = extractFrames(merged);
        // outbound now starts with the want_config_id frame — find the text packet
        for (const p of packets) {
          const tr = fromBinary(P.Mesh.ToRadioSchema, p);
          if (tr.payloadVariant?.case === 'packet')
            return tr.payloadVariant.value;
        }
        return undefined;
      };

      await waitFor(() => readOutbound() !== undefined);
      const pkt = readOutbound()!;
      expect(pkt.channel).toBe(3); // "test" — never the primary channel (0)
      expect(pkt.to).toBe(42);
      const decoded =
        pkt.payloadVariant?.case === 'decoded'
          ? pkt.payloadVariant.value
          : undefined;
      expect(
        new TextDecoder().decode(decoded?.payload ?? new Uint8Array()),
      ).toBe('hello test');
    } finally {
      delete process.env.MESH_SEND_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('sets Data.reply_id when replyId is given and leaves it unset otherwise', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const out: Uint8Array[] = [];
      bridge.sockets[0].on('data', (c) => out.push(c));
      await svc.sendText(42, 'reply send', 42); // replyId = 42
      await svc.sendText(43, 'plain send');

      const readOutbound = (): P.Mesh.MeshPacket[] => {
        const merged = new Uint8Array(out.reduce((n, c) => n + c.length, 0));
        let off = 0;
        for (const c of out) {
          merged.set(c, off);
          off += c.length;
        }
        const { packets } = extractFrames(merged);
        // outbound starts with the want_config_id frame — collect only text packets, in order
        return packets.flatMap((p) => {
          const tr = fromBinary(P.Mesh.ToRadioSchema, p);
          if (tr.payloadVariant?.case === 'packet')
            return [tr.payloadVariant.value];
          return [];
        });
      };

      await waitFor(() => readOutbound().length === 2);
      const [replyPkt, plainPkt] = readOutbound();
      const replyData =
        replyPkt.payloadVariant?.case === 'decoded'
          ? replyPkt.payloadVariant.value
          : undefined;
      expect(
        new TextDecoder().decode(replyData?.payload ?? new Uint8Array()),
      ).toBe('reply send');
      expect(replyData?.replyId).toBe(42); // threaded reply to packet 42

      const plainData =
        plainPkt.payloadVariant?.case === 'decoded'
          ? plainPkt.payloadVariant.value
          : undefined;
      expect(
        new TextDecoder().decode(plainData?.payload ?? new Uint8Array()),
      ).toBe('plain send');
      expect(plainData?.replyId).toBe(0); // omitted → proto3 default, not a reply
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService node database', () => {
  it('collects NodeInfo during sync and summarizes nodes', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      bridge.sockets[0].write(nodeInfoFrame(987654321)); // bare entry: no user/position/metrics yet

      await waitFor(async () => (await svc.nodes()).length === 2);
      expect(await svc.nodes()).toEqual([
        {
          num: KAMIL_NUM,
          longName: 'Kamil SP8MK/P',
          shortName: 'KK',
          hwModel: 'TBEAM',
          batteryLevel: 87,
          lastHeard: 1758000000,
          position: { lat: 521049637 * 1e-7, lon: 209833152 * 1e-7, alt: 123 },
        },
        { num: 987654321, lastHeard: null }, // names/hwModel/battery/position omitted when absent
      ]);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('creates a stub entry for POSITION_APP packets from unknown nodes', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(
        positionPacketFrame(555, {
          latitudeI: 10000000,
          longitudeI: -20000000,
        }),
      );

      await waitFor(async () => (await svc.nodes()).length === 1);
      expect(await svc.nodes()).toEqual([
        { num: 555, lastHeard: null, position: { lat: 1, lon: -2 } }, // no alt sent -> omitted
      ]);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('updates batteryLevel from live NODEINFO_APP packets (and stubs unknown nodes)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // sync entry with a stale battery, then a live NODEINFO_APP update
      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      await waitFor(async () => (await svc.nodes()).length === 1);
      expect((await svc.nodes())[0]?.batteryLevel).toBe(87);

      bridge.sockets[0].write(
        nodeInfoPacketFrame(KAMIL_NUM, { deviceMetrics: { batteryLevel: 9 } }),
      );

      await waitFor(async () => (await svc.nodes())[0]?.batteryLevel === 9);
      // other fields of the sync entry survive the live update
      expect(await svc.node(KAMIL_NUM)).toMatchObject({
        longName: 'Kamil SP8MK/P',
        shortName: 'KK',
        batteryLevel: 9,
      });

      // unknown node: a NODEINFO_APP with deviceMetrics creates a stub entry
      bridge.sockets[0].write(
        nodeInfoPacketFrame(555, { deviceMetrics: { batteryLevel: 3 } }),
      );

      await waitFor(async () => (await svc.nodes()).length === 2);
      expect((await svc.node(555))?.batteryLevel).toBe(3);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('looks up nodes by exact num, numeric string, exact name and unique substring', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      bridge.sockets[0].write(
        nodeInfoFrame(42, {
          user: { id: '!ot', longName: 'Other Node', shortName: 'OT' },
        }),
      );
      await waitFor(async () => (await svc.nodes()).length === 2);

      expect((await svc.node(KAMIL_NUM))?.longName).toBe('Kamil SP8MK/P'); // exact num
      expect((await svc.node(String(KAMIL_NUM)))?.num).toBe(KAMIL_NUM); // numeric string
      expect((await svc.node('kamil sp8mk/p'))?.shortName).toBe('KK'); // case-insensitive exact name
      expect((await svc.node('kamil'))?.num).toBe(KAMIL_NUM); // unique substring
      expect(await svc.node('zzz-nope')).toBeNull(); // no match
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('returns null for ambiguous name matches', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // two nodes whose names both contain "kamil" and share the short name "KK"
      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      bridge.sockets[0].write(
        nodeInfoFrame(777, {
          user: { id: '!kj', longName: 'Kamil Jr.', shortName: 'KK' },
        }),
      );
      await waitFor(async () => (await svc.nodes()).length === 2);

      expect(await svc.node('kamil')).toBeNull(); // substring matches two nodes
      expect(await svc.node('kk')).toBeNull(); // exact name matches two nodes
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('resets the node database on reconnect (no stale data)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50';
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      await waitFor(async () => (await svc.nodes()).length === 1);

      // drop the connection; bot reconnects and must start with an empty node DB
      bridge.sockets[0].destroy();
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
      await waitFor(async () => (await svc.nodes()).length === 0);
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService message history', () => {
  it('captures inbound text as rx records', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    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));

      await waitFor(() => svc.messages().length === 1);
      expect(svc.messages()).toEqual([
        {
          id: 1,
          dir: 'rx',
          from: KAMIL_NUM,
          to: 42, // destination persisted on rx records (broadcast addr for broadcasts)
          text: 'ping',
          channel: 2,
          at: expect.any(Number),
        },
      ]);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('tracks a tx send pending -> delivered on the routing ACK', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const id = await svc.sendText(42, 'hello test');
      expect(svc.messages()[0]).toMatchObject({
        id,
        dir: 'tx',
        to: 42,
        text: 'hello test',
        channel: 3,
        status: 'pending',
      });

      bridge.sockets[0].write(routingFrame(id, P.Mesh.Routing_Error.NONE));

      await waitFor(() => svc.messages()[0]?.status === 'delivered');
      expect(svc.messages()[0]).toMatchObject({
        id,
        dir: 'tx',
        status: 'delivered',
      });
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('marks a tx failed with the routing error reason on NACK', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const id = await svc.sendText(42, 'will fail');
      expect(svc.messages()[0]?.status).toBe('pending');

      bridge.sockets[0].write(routingFrame(id, P.Mesh.Routing_Error.TIMEOUT));

      await waitFor(() => svc.messages()[0]?.status === 'failed');
      expect(svc.messages()[0]).toMatchObject({
        id,
        status: 'failed',
        ackError: 'TIMEOUT', // human-readable Routing.Error name
      });
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('caps history at 100 records, dropping the oldest', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      for (let i = 0; i < 105; i++)
        bridge.sockets[0].write(textPacketFrame(42, KAMIL_NUM, `m-${i}`));

      await waitFor(() => svc.messages().length === 100, 10000);
      expect(svc.messages()[0]?.text).toBe('m-104'); // newest first
      expect(svc.messages()[99]?.text).toBe('m-5'); // m-0..m-4 dropped
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService respond', () => {
  it('replies to an rx DM as a directed send back to the sender on the original channel with replyId = messageId', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // rx DM from KAMIL to our node on channel 2 (packet id 1)
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2));
      await waitFor(() => svc.messages().length === 1);

      const out: Uint8Array[] = [];
      bridge.sockets[0].on('data', (c) => out.push(c));
      const { id } = await svc.respond(1, 'pong');

      expect(id).toBeGreaterThan(0);
      // DM back to the original sender on the original channel
      expect(svc.messages()[0]).toMatchObject({
        id,
        dir: 'tx',
        to: KAMIL_NUM,
        text: 'pong',
        channel: 2,
        status: 'pending',
      });

      await waitFor(() => outboundPackets(out).length === 1);
      const pkt = outboundPackets(out)[0]!;
      expect(pkt.to).toBe(KAMIL_NUM);
      expect(pkt.channel).toBe(2);
      const data =
        pkt.payloadVariant?.case === 'decoded'
          ? pkt.payloadVariant.value
          : undefined;
      expect(data?.replyId).toBe(1); // threaded quote of the original packet
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('replies to a broadcast original as a broadcast on the original channel with replyId = messageId', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // rx broadcast from KAMIL on channel 2 (packet id 1)
      bridge.sockets[0].write(
        textPacketFrame(KAMIL_NUM, BROADCAST_ADDR, 'hello all', 2),
      );
      await waitFor(() => svc.messages().length === 1);

      const out: Uint8Array[] = [];
      bridge.sockets[0].on('data', (c) => out.push(c));
      const { id } = await svc.respond(1, 'pong');

      expect(id).toBeGreaterThan(0);
      expect(svc.messages()[0]).toMatchObject({
        dir: 'tx',
        to: BROADCAST_ADDR, // broadcast original → broadcast reply
        text: 'pong',
        channel: 2,
      });

      await waitFor(() => outboundPackets(out).length === 1);
      const pkt = outboundPackets(out)[0]!;
      expect(pkt.to).toBe(BROADCAST_ADDR);
      expect(pkt.channel).toBe(2);
      const data =
        pkt.payloadVariant?.case === 'decoded'
          ? pkt.payloadVariant.value
          : undefined;
      expect(data?.replyId).toBe(1);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('replies to a tx record targeting its destination on its channel with replyId = messageId', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const origId = await svc.sendText(42, 'orig'); // tx record: to=42, default channel 3
      const out: Uint8Array[] = [];
      bridge.sockets[0].on('data', (c) => out.push(c));
      const { id } = await svc.respond(origId, 'reply');

      expect(id).toBeGreaterThan(0);
      expect(svc.messages()[0]).toMatchObject({
        dir: 'tx',
        to: 42, // the tx record's own destination
        text: 'reply',
        channel: 3,
      });

      // the orig send may share a chunk with the reply — take the newest packet
      await waitFor(() => outboundPackets(out).length >= 1);
      const pkts = outboundPackets(out);
      const pkt = pkts[pkts.length - 1]!;
      expect(pkt.to).toBe(42);
      const data =
        pkt.payloadVariant?.case === 'decoded'
          ? pkt.payloadVariant.value
          : undefined;
      expect(data?.replyId).toBe(origId);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('throws NotFoundException when no message has the given id (memory or SQLite)', async () => {
    const svc = new MeshService();
    await expect(svc.respond(999, 'hi')).rejects.toBeInstanceOf(
      NotFoundException,
    );
    await expect(svc.respond(999, 'hi')).rejects.toThrow(
      'no message with id 999',
    );

    const db = new DatabaseSync(':memory:'); // empty table — the DB lookup path also misses
    initMessagesTable(db);
    const svc2 = new MeshService(db);
    await expect(svc2.respond(999, 'hi')).rejects.toBeInstanceOf(
      NotFoundException,
    );
  });

  it('falls back to SQLite when the record is not in the in-memory buffer', async () => {
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    // 150 rows — onModuleInit seeds only the newest 100 into the buffer (ids 2050..2149)
    const ins = db.prepare(
      'INSERT INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
    );
    for (let i = 0; i < 150; i++)
      ins.run(
        2000 + i,
        'rx',
        42,
        KAMIL_NUM,
        `old-${i}`,
        2,
        1_700_000_000_000 + i * 1000,
        null,
        null,
      );

    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService(db);
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // id 2000 is in SQLite but outside the seeded buffer
      const { id } = await svc.respond(2000, 'pong');
      expect(id).toBeGreaterThan(0);
      expect(svc.messages()[0]).toMatchObject({
        dir: 'tx',
        to: KAMIL_NUM, // DM back to the sender of the old rx record
        text: 'pong',
        channel: 2, // the record's channel
      });
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('rejects invalid messageId and empty content with BadRequestException', async () => {
    const svc = new MeshService();
    for (const bad of [0, BROADCAST_ADDR + 1, 42.5, 'x' as unknown as number])
      await expect(svc.respond(bad, 'hi')).rejects.toBeInstanceOf(
        BadRequestException,
      );
    await expect(svc.respond(1, '')).rejects.toThrow(
      'content must be a non-empty string',
    );
    await expect(svc.respond(1, undefined)).rejects.toBeInstanceOf(
      BadRequestException,
    );
  });
});

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(() => {
    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('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);
    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(notifyMock).toHaveBeenCalledTimes(1);
      expect(notifyMock).toHaveBeenCalledWith(1, 'ping', KAMIL_NUM); // packet id threaded through
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  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);
    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 no notification fires off-channel
      await waitFor(() => svc.messages().length === 1);
      expect(notifyMock).not.toHaveBeenCalled();
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('fires the notifier for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => {
    process.env.MESH_NOTIFY_CHANNEL = '0';
    const svc = makeSvc();
    try {
      const r = svc.injectMessage('hi', 42);

      expect(notifyMock).toHaveBeenCalledTimes(1);
      expect(notifyMock).toHaveBeenCalledWith(r.id, 'hi', 42); // generated record id threaded through
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
    }
  });

  it('passes `from` through as undefined when an injected message has no sender', async () => {
    process.env.MESH_NOTIFY_CHANNEL = '0';
    const svc = makeSvc();
    try {
      const r = svc.injectMessage('hi');

      expect(notifyMock).toHaveBeenCalledTimes(1);
      expect(notifyMock).toHaveBeenCalledWith(r.id, 'hi', undefined);
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
    }
  });

  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);
    delete process.env.MESH_NOTIFY_CHANNEL; // default 3
    const svc = makeSvc();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // initial sync: our own node number arrives via myInfo first
      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 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(notifyMock).not.toHaveBeenCalled();
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  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_CHANNEL = '0';
    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);

      svc.injectMessage('hi', MY_NODE);

      expect(notifyMock).not.toHaveBeenCalled();
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('notifies for rx texts on any channel in a comma-separated MESH_NOTIFY_CHANNEL list, and not for others', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_NOTIFY_CHANNEL = '2,5';
    const svc = makeSvc();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // channel 5 is in the list — notified
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'on-list', 5));

      await waitFor(() => notifyMock.mock.calls.length === 1);
      expect(notifyMock).toHaveBeenCalledWith(1, 'on-list', KAMIL_NUM);

      // channel 3 is not in the list — captured in history but no notification
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'off-list', 3));

      await waitFor(() => svc.messages().length === 2);
      expect(notifyMock).toHaveBeenCalledTimes(1);
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('ignores ch0 broadcasts even when 0 is in the list, but notifies for directed DMs', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_NOTIFY_CHANNEL = '3,0';
    const svc = makeSvc();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // ch0 broadcast: shared default channel noise — no notification
      bridge.sockets[0].write(
        textPacketFrame(KAMIL_NUM, BROADCAST_ADDR, 'hello all', 0),
      );

      await waitFor(() => svc.messages().length === 1);
      expect(notifyMock).not.toHaveBeenCalled();

      // ch0 directed DM: both sender and recipient present — notified
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'dm', 0));

      await waitFor(() => notifyMock.mock.calls.length === 1);
      expect(notifyMock).toHaveBeenCalledWith(1, 'dm', KAMIL_NUM);
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('notifies for a ch0 injection when 0 is in the list (multi-list)', () => {
    process.env.MESH_NOTIFY_CHANNEL = '3,0';
    const svc = makeSvc();
    try {
      const r = svc.injectMessage('hi', 42);

      // injected records are exempt from the ch0 DM-only rule
      expect(notifyMock).toHaveBeenCalledTimes(1);
      expect(notifyMock).toHaveBeenCalledWith(r.id, 'hi', 42);
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
    }
  });

  it('ignores empty and invalid entries in MESH_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_CHANNEL = 'abc, ,5,-2,x7';
    const svc = makeSvc();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // only the valid entry (5) survives parsing — notified
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 5));

      await waitFor(() => notifyMock.mock.calls.length === 1);
      expect(notifyMock).toHaveBeenCalledWith(1, 'ping', KAMIL_NUM);

      // channel 7 only appears inside the invalid entry "x7" — must not match
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'nope', 7));

      await waitFor(() => svc.messages().length === 2);
      expect(notifyMock).toHaveBeenCalledTimes(1);
    } finally {
      delete process.env.MESH_NOTIFY_CHANNEL;
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

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();
    expect(svc.messages()).toEqual([]);

    const r1 = svc.injectMessage('first');
    const r2 = svc.injectMessage('second', KAMIL_NUM);
    expect(r1.id).toBeGreaterThan(0); // random non-zero uint32, like tx ids
    expect(r2.id).toBeGreaterThan(0);

    expect(svc.messages()).toEqual([
      {
        id: r2.id,
        dir: 'rx',
        from: KAMIL_NUM,
        text: 'second',
        channel: 0,
        at: expect.any(Number),
      },
      {
        id: r1.id,
        dir: 'rx',
        text: 'first',
        channel: 0,
        at: expect.any(Number),
      },
    ]);
  });

  it('omits from when not provided and rejects empty text or invalid from', () => {
    const svc = new MeshService();
    const rec = svc.injectMessage('no sender');
    expect(svc.messages()[0]).toEqual({
      id: rec.id,
      dir: 'rx',
      text: 'no sender',
      channel: 0,
      at: expect.any(Number),
    });
    expect(svc.messages()[0]).not.toHaveProperty('from');

    for (const bad of ['', undefined]) {
      expect(() => svc.injectMessage(bad as unknown as string)).toThrow(
        'text must be a non-empty string',
      );
    }
    for (const bad of [0, -1, 42.5, BROADCAST_ADDR + 1]) {
      expect(() => svc.injectMessage('x', bad)).toThrow(/invalid node number/);
    }
    expect(svc.messages()).toHaveLength(1); // rejected calls add nothing
  });

  it('caps history at 100 records, dropping the oldest', () => {
    const svc = new MeshService();
    for (let i = 0; i < 105; i++) svc.injectMessage(`m-${i}`);
    expect(svc.messages()).toHaveLength(100);
    expect(svc.messages()[0]?.text).toBe('m-104'); // newest first
    expect(svc.messages()[99]?.text).toBe('m-5'); // m-0..m-4 dropped
  });

  it('shares the history buffer with captured packets', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      svc.injectMessage('injected');
      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'real', 2));

      await waitFor(() => svc.messages().length === 2);
      expect(svc.messages()[0]).toMatchObject({ dir: 'rx', text: 'real' }); // newest first
      expect(svc.messages()[1]).toMatchObject({
        dir: 'rx',
        text: 'injected',
        channel: 0,
      });
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService telemetry', () => {
  it('captures device metrics and environment variants with from/kind/data', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      expect(svc.telemetry()).toEqual([]); // nothing before the first packet

      bridge.sockets[0].write(
        telemetryPacketFrame(KAMIL_NUM, {
          case: 'deviceMetrics',
          value: { batteryLevel: 87, voltage: 4.1, uptimeSeconds: 3600 },
        }),
      );
      bridge.sockets[0].write(
        telemetryPacketFrame(987654321, {
          case: 'environmentMetrics',
          value: { temperature: 21.5, relativeHumidity: 48 },
        }),
      );

      await waitFor(() => svc.telemetry().length === 2);
      expect(svc.telemetry()).toEqual([
        {
          from: 987654321, // newest first
          at: expect.any(Number),
          kind: 'environmentMetrics',
          // the deprecated repeated one_wire_temperature field decodes as an empty list even when unset
          data: {
            temperature: 21.5,
            relativeHumidity: 48,
            oneWireTemperature: [],
          },
        },
        {
          from: KAMIL_NUM,
          at: expect.any(Number),
          kind: 'deviceMetrics',
          // proto float is IEEE-754 single precision — 4.1 widens to this exact double
          data: {
            batteryLevel: 87,
            voltage: 4.099999904632568,
            uptimeSeconds: 3600,
          },
        },
      ]);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('caps telemetry at 100 records, dropping the oldest', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      for (let i = 0; i < 105; i++)
        bridge.sockets[0].write(
          telemetryPacketFrame(42, {
            case: 'deviceMetrics',
            value: { batteryLevel: i },
          }),
        );

      await waitFor(() => svc.telemetry().length === 100, 10000);
      expect(svc.telemetry()[0]?.data).toEqual({ batteryLevel: 104 }); // newest first
      expect(svc.telemetry()[99]?.data).toEqual({ batteryLevel: 5 }); // i=0..4 dropped
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('resets telemetry on reconnect (no stale data)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50';
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(
        telemetryPacketFrame(KAMIL_NUM, { case: 'deviceMetrics', value: {} }),
      );
      await waitFor(() => svc.telemetry().length === 1);

      // drop the connection; bot reconnects and must start with an empty history
      bridge.sockets[0].destroy();
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
      await waitFor(() => svc.telemetry().length === 0);
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService resolveDest', () => {
  it('resolves broadcast sentinels, numbers and names to node numbers', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      await waitFor(async () => (await svc.nodes()).length === 1);

      expect(await svc.resolveDest(undefined)).toBe(BROADCAST_ADDR); // omitted → broadcast
      expect(await svc.resolveDest('')).toBe(BROADCAST_ADDR); // empty → broadcast
      expect(await svc.resolveDest('!b')).toBe(BROADCAST_ADDR);
      expect(await svc.resolveDest('BROADCAST')).toBe(BROADCAST_ADDR); // case-insensitive
      expect(await svc.resolveDest(42)).toBe(42); // number passthrough
      expect(await svc.resolveDest(String(KAMIL_NUM))).toBe(KAMIL_NUM); // numeric string
      expect(await svc.resolveDest('Kamil SP8MK/P')).toBe(KAMIL_NUM); // exact longName
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('throws for unknown names and invalid node numbers', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      await waitFor(async () => (await svc.nodes()).length === 1);

      await expect(svc.resolveDest('zzz-nope')).rejects.toThrow(
        "no node matching 'zzz-nope'",
      );
      for (const bad of [0, -1, 42.5, BROADCAST_ADDR + 1]) {
        await expect(svc.resolveDest(bad)).rejects.toThrow(
          /invalid node number/,
        );
      }
      await expect(svc.resolveDest('0')).rejects.toThrow(/invalid node number/);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService device info', () => {
  it('captures DeviceMetadata and exposes firmwareVersion', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // before the sync: only the node number (null until myInfo arrives)
      expect(svc.deviceInfo()).toEqual({
        myNodeNum: null,
        localConfig: {},
        moduleConfig: {},
      });

      bridge.sockets[0].write(
        metadataFrame({ firmwareVersion: '2.7.0.1', hasWifi: false, role: 3 }),
      );

      await waitFor(() => svc.deviceInfo().firmwareVersion === '2.7.0.1');
      const info = svc.deviceInfo();
      expect(info.firmwareVersion).toBe('2.7.0.1');
      expect(info.metadata?.firmwareVersion).toBe('2.7.0.1'); // metadata stored as-is
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('captures config/module-config sections and redacts key material', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const secret = 'super-secret-wifi-psk';
      bridge.sockets[0].write(
        configSectionFrame('security', {
          publicKey: new Uint8Array([1, 2, 3]),
          privateKey: new Uint8Array([4, 5, 6]),
          adminKey: [new Uint8Array([7])],
          isManaged: true,
        }),
      );
      bridge.sockets[0].write(
        configSectionFrame('network', {
          wifiEnabled: false,
          wifiSsid: 'home',
          wifiPsk: secret,
        }),
      );
      bridge.sockets[0].write(configSectionFrame('device', { role: 3 }));
      bridge.sockets[0].write(
        moduleConfigSectionFrame('telemetry', {
          environmentUpdateInterval: 900,
          powerMeasurementEnabled: true,
        }),
      );

      await waitFor(
        () =>
          Object.keys(svc.deviceInfo().localConfig).length === 3 &&
          Object.keys(svc.deviceInfo().moduleConfig).length === 1,
      );
      const info = svc.deviceInfo();
      expect(Object.keys(info.localConfig).sort()).toEqual([
        'device',
        'network',
        'security',
      ]);
      expect(info.moduleConfig.telemetry).toMatchObject({
        environmentUpdateInterval: 900,
        powerMeasurementEnabled: true,
      });

      // key material is replaced with a marker — never exposed
      expect(info.localConfig.security).toMatchObject({
        publicKey: 'REDACTED',
        privateKey: 'REDACTED',
        adminKey: 'REDACTED',
        isManaged: true, // non-secret fields pass through
      });
      expect(info.localConfig.network?.wifiPsk).toBe('REDACTED');
      const json = JSON.stringify(info);
      expect(json).not.toContain(secret);
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('serializes proto uint64 fields as strings (JSON-safe)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      // powermonEnables is a proto uint64 — protobuf-es decodes it as native bigint
      bridge.sockets[0].write(
        configSectionFrame('power', { powermonEnables: 123n }),
      );

      await waitFor(() => svc.deviceInfo().localConfig.power !== undefined);
      const info = svc.deviceInfo();
      expect(info.localConfig.power?.powermonEnables).toBe('123'); // bigint -> decimal string
      expect(() => JSON.stringify(info)).not.toThrow(); // no "Do not know how to serialize a BigInt"
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('summarizes NodeInfo snr/hopsAway/viaMqtt when present', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    const svc = new MeshService();
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(nodeInfoFrame(KAMIL_NUM, kamilNodeInfo()));
      bridge.sockets[0].write(
        nodeInfoFrame(42, {
          user: { id: '!ot', longName: 'Other Node' },
          snr: -2.5,
          hopsAway: 1,
          viaMqtt: true,
        }),
      );

      await waitFor(async () => (await svc.nodes()).length === 2);
      const nodes = await svc.nodes();
      expect(nodes.find((n) => n.num === 42)).toEqual({
        num: 42,
        longName: 'Other Node',
        lastHeard: null,
        snr: -2.5,
        hopsAway: 1,
        viaMqtt: true,
      });
      // absent in the proto → omitted from the summary
      expect(nodes.find((n) => n.num === KAMIL_NUM)).not.toHaveProperty('snr');
    } finally {
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});

describe('MeshService SQLite persistence', () => {
  it('persists an injected message row on poller tick (no manual flush)', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_POLL_MS = '50';
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    const svc = new MeshService(db);
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      svc.injectMessage('hello db', KAMIL_NUM);

      // the poller tick must pick it up — no flushHistory() call here
      await waitFor(
        () =>
          (
            db.prepare('SELECT COUNT(*) AS n FROM messages').get() as {
              n: number;
            }
          ).n === 1,
      );
      const row = db
        .prepare('SELECT * FROM messages WHERE text = ?')
        .get('hello db') as Record<string, unknown>;
      expect(row.dir).toBe('rx');
      expect(row.from_node).toBe(KAMIL_NUM);
      expect(row.to_node).toBeNull(); // rx rows carry no destination
      expect(row.channel).toBe(0);
      expect(row.status).toBeNull();
    } finally {
      delete process.env.MESH_POLL_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('persists the destination (to) of captured rx packets on poller tick', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_POLL_MS = '50';
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    const svc = new MeshService(db);
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3));

      // the poller tick must pick it up — no flushHistory() call here
      await waitFor(
        () =>
          (
            db.prepare('SELECT COUNT(*) AS n FROM messages').get() as {
              n: number;
            }
          ).n === 1,
      );
      const row = db
        .prepare('SELECT * FROM messages WHERE id = ?')
        .get(1) as Record<string, unknown>;
      expect(row.dir).toBe('rx');
      expect(row.from_node).toBe(KAMIL_NUM);
      expect(row.to_node).toBe(42); // rx rows carry the destination now (no schema change — to_node existed for tx)
    } finally {
      delete process.env.MESH_POLL_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('reflects the ACK status update in the database after the next tick', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_POLL_MS = '50';
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    const svc = new MeshService(db);
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      const id = await svc.sendText(42, 'ack me');
      expect(svc.messages()[0]?.status).toBe('pending');

      bridge.sockets[0].write(routingFrame(id, P.Mesh.Routing_Error.NONE));
      await waitFor(() => svc.messages()[0]?.status === 'delivered');

      // the in-place status mutation must reach SQLite on the next poller tick
      // (the row may not exist before the first tick — waitFor re-checks, so stay null-safe)
      await waitFor(
        () =>
          (
            db.prepare('SELECT status FROM messages WHERE id = ?').get(id) as
              | {
                  status: string | null;
                }
              | undefined
          )?.status === 'delivered',
      );
    } finally {
      delete process.env.MESH_POLL_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });

  it('seeds the history buffer from a pre-populated database on startup', () => {
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    const ins = db.prepare(
      'INSERT INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
    );
    for (let i = 0; i < 150; i++)
      ins.run(
        1000 + i,
        'rx',
        null,
        42,
        `seed-${i}`,
        0,
        1_700_000_000_000 + i * 1000,
        null,
        null,
      );

    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = '1'; // closed port — no real bridge needed for seeding
    const svc = new MeshService(db);
    try {
      svc.onModuleInit();
      expect(svc.messages()).toHaveLength(100); // newest 100 of the 150 rows
      expect(svc.messages()[0]).toMatchObject({
        id: 1149,
        dir: 'rx',
        from: 42,
        text: 'seed-149',
        channel: 0,
      }); // newest first
      expect(svc.messages()[99]).toMatchObject({ id: 1050, text: 'seed-50' });
      expect(svc.messages()[0]).not.toHaveProperty('status'); // rx rows carry no status
    } finally {
      delete process.env.MESH_HOST;
      delete process.env.MESH_PORT;
      svc.onModuleDestroy();
    }
  });

  it('prunes database rows beyond the retention limit (newest 1000)', () => {
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    const ins = db.prepare(
      'INSERT INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
    );
    for (let i = 0; i < 1200; i++)
      ins.run(
        5000 + i,
        'rx',
        null,
        42,
        `old-${i}`,
        0,
        1_700_000_000_000 + i * 1000,
        null,
        null,
      );

    const svc = new MeshService(db); // no onModuleInit — flushHistory() is the unit under test
    try {
      svc.flushHistory(); // empty buffer: upserts nothing, retention still applies
      expect(
        (
          db.prepare('SELECT COUNT(*) AS n FROM messages').get() as {
            n: number;
          }
        ).n,
      ).toBe(1000);
      expect(
        db.prepare('SELECT text FROM messages WHERE id = ?').get(6199),
      ).toMatchObject({ text: 'old-1199' }); // newest kept
      expect(
        db.prepare('SELECT text FROM messages WHERE id = ?').get(5199),
      ).toBeUndefined(); // oldest 200 pruned (i=0..199)
    } finally {
      svc.onModuleDestroy();
    }
  });

  it('keeps the message history across a bridge reconnect', async () => {
    const bridge = await startBridge();
    process.env.MESH_HOST = '127.0.0.1';
    process.env.MESH_PORT = String(bridge.port);
    process.env.MESH_RECONNECT_MS = '50';
    const db = new DatabaseSync(':memory:');
    initMessagesTable(db);
    const svc = new MeshService(db);
    try {
      svc.onModuleInit();
      await waitFor(
        () => bridge.sockets.length === 1 && svc.status().connected,
      );

      svc.injectMessage('survives reconnect', KAMIL_NUM);
      expect(svc.messages()).toHaveLength(1);

      // drop the connection; bot reconnects and must keep the history (unlike channels/nodes/telemetry)
      bridge.sockets[0].destroy();
      await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
      expect(svc.messages()).toHaveLength(1);
      expect(svc.messages()[0]).toMatchObject({
        dir: 'rx',
        text: 'survives reconnect',
      });
    } finally {
      delete process.env.MESH_RECONNECT_MS;
      svc.onModuleDestroy();
      bridge.close();
    }
  });
});