MeshBot/src/mesh/mesh.service.ts

import {
  Injectable,
  Logger,
  OnModuleDestroy,
  OnModuleInit,
  ServiceUnavailableException,
} from '@nestjs/common';
import * as net from 'node:net';
import {
  PORT_TEXT_MESSAGE_APP,
  buildTextToRadio,
  buildWantConfigToRadio,
  decodeFromRadio,
  extractFrames,
  frame,
} from './codec.js';

@Injectable()
export class MeshService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(MeshService.name);
  private readonly host = process.env.MESH_HOST ?? '192.168.1.4';
  private readonly port = parseInt(process.env.MESH_PORT ?? '4403', 10);
  // ponytail: fixed-interval retry, no backoff — fine for single-bridge use
  private readonly reconnectMs = parseInt(
    process.env.MESH_RECONNECT_MS ?? '5000',
    10,
  );
  // sends are pinned to the test channel ("testowy", index 2) on purpose — never the primary channel
  private readonly sendChannel = parseInt(
    process.env.MESH_SEND_CHANNEL ?? '2',
    10,
  );

  private socket: net.Socket | null = null;
  private buffer: Uint8Array = new Uint8Array(0);
  private connected = false;
  private myNodeNum: number | null = null;
  // channels streamed by the device during initial sync (before config_complete_id)
  private channelMap = new Map<
    number,
    import('@meshtastic/protobufs').Channel.Channel
  >();
  private reconnectTimer: NodeJS.Timeout | null = null;
  private shuttingDown = false;

  onModuleInit(): void {
    this.connect();
  }

  onModuleDestroy(): void {
    this.shuttingDown = true;
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
    this.socket?.destroy();
  }

  status() {
    return {
      connected: this.connected,
      myNodeNum: this.myNodeNum,
      host: this.host,
      port: this.port,
    };
  }

  // psk/keys are deliberately not included — never expose them over HTTP
  channels(): {
    index: number;
    name: string;
    uplinkEnabled: boolean;
    downlinkEnabled: boolean;
  }[] {
    return [...this.channelMap.values()]
      .sort((a, b) => a.index - b.index)
      .map(({ index, settings }) => ({
        index,
        name: settings?.name ?? '',
        uplinkEnabled: settings?.uplinkEnabled ?? false,
        downlinkEnabled: settings?.downlinkEnabled ?? false,
      }));
  }

  async sendText(to: number, text: string): Promise<number> {
    if (!this.connected || !this.socket)
      throw new ServiceUnavailableException('bridge not connected');
    const { bytes, id } = await buildTextToRadio(to, text, this.sendChannel);
    this.socket.write(bytes);
    return id;
  }

  private connect(): void {
    if (this.shuttingDown) return;
    const socket = net.createConnection({ host: this.host, port: this.port });
    this.socket = socket;
    socket.on('connect', () => {
      this.connected = true;
      // fresh sync on every (re)connect — drop stale channels from the old session
      this.channelMap = new Map();
      this.logger.log(`bridge connected ${this.host}:${this.port}`);
      const nonce = Math.floor(Math.random() * 0xffffffff) || 1;
      buildWantConfigToRadio(nonce)
        .then((bytes) => socket.write(frame(bytes)))
        .catch((err: unknown) =>
          this.logger.error(`want_config failed: ${String(err)}`),
        );
      this.logger.log(`requested initial sync (want_config_id=${nonce})`);
    });
    socket.on('data', (chunk) => {
      const merged = new Uint8Array(this.buffer.length + chunk.length);
      merged.set(this.buffer, 0);
      merged.set(chunk, this.buffer.length);
      this.buffer = merged;
      const { packets, rest } = extractFrames(this.buffer);
      this.buffer = rest;
      for (const p of packets) {
        decodeFromRadio(p)
          .then((fr) => this.handleFromRadio(fr))
          .catch((err: unknown) =>
            this.logger.error(`decode failed: ${String(err)}`),
          );
      }
    });
    const onDown = () => {
      this.connected = false;
      this.scheduleReconnect();
    };
    socket.on('error', (err) => {
      this.logger.warn(`bridge error: ${err.message}`);
      onDown();
    });
    socket.on('close', onDown);
  }

  private scheduleReconnect(): void {
    if (this.shuttingDown) return;
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
    this.reconnectTimer = setTimeout(() => this.connect(), this.reconnectMs);
  }

  private handleFromRadio(
    fr: import('@meshtastic/protobufs').Mesh.FromRadio,
  ): void {
    const v = fr.payloadVariant;
    switch (v?.case) {
      case 'myInfo':
        this.myNodeNum = v.value.myNodeNum;
        this.logger.log(`my node num: ${this.myNodeNum}`);
        break;
      case 'channel':
        // device streams one Channel per table slot during initial sync
        this.channelMap.set(v.value.index, v.value);
        break;
      case 'configCompleteId':
        this.logger.log(
          `bridge ready (config complete), ${this.channelMap.size} channels`,
        );
        break;
      case 'packet': {
        const pkt = v.value;
        const decoded =
          pkt.payloadVariant?.case === 'decoded'
            ? pkt.payloadVariant.value
            : undefined;
        if (decoded && decoded.portnum === PORT_TEXT_MESSAGE_APP) {
          this.logger.log(
            `[mesh] ${pkt.from} -> ${pkt.to}: "${new TextDecoder().decode(decoded.payload)}"`,
          );
        }
        break;
      }
    }
  }
}