MeshClient.ts

import { create, toBinary } from "@bufbuild/protobuf";
import * as Protobuf from "@meshtastic/protobufs";
import type { Logger } from "tslog";
import { Constants } from "../constants/index.ts";
import { EventBus } from "../event-bus/EventBus.ts";
import { PacketTooLargeError } from "../errors/MeshError.ts";
import { generatePacketId } from "../identifiers/PacketId.ts";
import { createLogger } from "../logging/logger.ts";
import { decodePacket } from "../packet-codec/decodePacket.ts";
import { Queue } from "../queue/Queue.ts";
import { createStore, type ReadonlySignal } from "../signals/createStore.ts";
import type { Transport } from "../transport/Transport.ts";
import { DeviceStatusEnum } from "../transport/Transport.ts";
import {
  ChannelNumber,
  type Destination,
  Emitter,
  type PacketMetadata,
} from "../types.ts";
import { Xmodem } from "../xmodem/Xmodem.ts";
import { ChatClient } from "../../features/chat/index.ts";
import { ChannelsClient } from "../../features/channels/index.ts";
import { ConfigClient } from "../../features/config/index.ts";
import { DeviceClient } from "../../features/device/index.ts";
import { FilesClient } from "../../features/files/index.ts";
import { NodesClient } from "../../features/nodes/index.ts";
import { PositionClient } from "../../features/position/index.ts";
import { TelemetryClient } from "../../features/telemetry/index.ts";
import type { TelemetryClientOptions } from "../../features/telemetry/index.ts";
import { TraceRouteClient } from "../../features/traceroute/index.ts";

import type { ChatClientOptions } from "../../features/chat/ChatClient.ts";
import type { NodesClientOptions } from "../../features/nodes/NodesClient.ts";

export interface MeshClientOptions {
  transport: Transport;
  configId?: number;
  logger?: Logger<unknown>;
  chat?: ChatClientOptions;
  nodes?: NodesClientOptions;
  telemetry?: TelemetryClientOptions;
}

/**
 * Per-section / per-event tally of what has streamed in since the most
 * recent `configure()` call. UI surfaces use this for live "received X
 * channels, Y nodes" feedback during the handshake.
 */
export interface ConnectionProgressCounters {
  config: number;
  modules: number;
  channels: number;
  nodes: number;
  myInfo: boolean;
  metadata: boolean;
}

/**
 * Connection progress signal shape. The state machine moves
 * `idle → configuring → configured` once `configure()` runs and the
 * device finishes streaming its config bundle.
 */
export type ConnectionProgress =
  | { phase: "idle" }
  | { phase: "configuring"; received: ConnectionProgressCounters }
  | { phase: "configured"; received: ConnectionProgressCounters };

const EMPTY_COUNTERS: ConnectionProgressCounters = {
  config: 0,
  modules: 0,
  channels: 0,
  nodes: 0,
  myInfo: false,
  metadata: false,
};

/**
 * Orchestrator for a single connected Meshtastic device.
 *
 * Owns the transport, event bus, queue, and xmodem instances. Exposes one
 * client per feature slice; slice clients consume events from the bus and
 * publish signal-backed state that UI layers subscribe to.
 */
export class MeshClient {
  public readonly log: Logger<unknown>;
  public readonly transport: Transport;
  public readonly events: EventBus;
  public readonly queue: Queue;
  public readonly xModem: Xmodem;
  public configId: number;

  public readonly device: DeviceClient;
  public readonly chat: ChatClient;
  public readonly nodes: NodesClient;
  public readonly channels: ChannelsClient;
  public readonly config: ConfigClient;
  public readonly telemetry: TelemetryClient;
  public readonly position: PositionClient;
  public readonly traceroute: TraceRouteClient;
  public readonly files: FilesClient;

  /**
   * Live connection-handshake progress. Resets to `configuring` with empty
   * counters when `configure()` is called; tallies each inbound config /
   * module / channel / node packet; flips to `configured` when the device
   * sends `configCompleteId`.
   */
  public readonly progress: ReadonlySignal<ConnectionProgress>;
  private readonly progressStore = createStore<ConnectionProgress>({
    phase: "idle",
  });

  private _heartbeatIntervalId: ReturnType<typeof setInterval> | undefined;
  private _fromDeviceAc: AbortController | undefined;
  private _fromDevicePipe: Promise<void> | undefined;

  constructor(options: MeshClientOptions) {
    this.log = options.logger ?? createLogger("MeshClient");
    this.transport = options.transport;
    this.events = new EventBus();
    this.queue = new Queue();
    this.xModem = new Xmodem(this.sendRaw.bind(this));
    this.configId = options.configId ?? generatePacketId();

    this.device = new DeviceClient(this);
    this.chat = new ChatClient(this, options.chat);
    this.nodes = new NodesClient(this, options.nodes);
    this.channels = new ChannelsClient(this);
    this.config = new ConfigClient(this);
    this.telemetry = new TelemetryClient(this, options.telemetry);
    this.position = new PositionClient(this);
    this.traceroute = new TraceRouteClient(this);
    this.files = new FilesClient(this);

    this.progress = this.progressStore.read;
    this.wireProgressTracking();

    this.events.onDeviceStatus.subscribe((status) => {
      if (status === DeviceStatusEnum.DeviceDisconnected) {
        if (this._heartbeatIntervalId !== undefined) {
          clearInterval(this._heartbeatIntervalId);
        }
        this.complete();
      }
    });

    this._fromDeviceAc = new AbortController();
    this._fromDevicePipe = this.transport.fromDevice.pipeTo(decodePacket(this), {
      signal: this._fromDeviceAc.signal,
    });
    // Swallow abort/cancel rejection so an unhandled rejection does not
    // surface, but log unexpected transport/stream failures.
    void this._fromDevicePipe.catch((err) => {
      // Abort rejections may be DOMException, so do not require instanceof Error.
      if ((err as { name?: string } | null)?.name !== "AbortError") {
        const message = err instanceof Error ? err.message : String(err);
        this.log.error(
          Emitter[Emitter.ConnectionStatus],
          `Device decoding pipe failed: ${message}`,
        );
      }
    });
  }

  public get myNodeNum(): number {
    return this.device.myNodeNum.value ?? 0;
  }

  /**
   * Begin the wantConfigId → config-complete handshake. Resolves when the
   * device has ack'd the wantConfigId packet (status changes to
   * DeviceConfigured when the device finishes sending its configuration).
   */
  public async connect(): Promise<void> {
    this.updateDeviceStatus(DeviceStatusEnum.DeviceConnecting);
    await this.configure();
  }

  public configure(): Promise<number> {
    this.log.debug(
      Emitter[Emitter.Configure],
      "⚙️ Requesting device configuration",
    );
    this.updateDeviceStatus(DeviceStatusEnum.DeviceConfiguring);
    this.progressStore.write.value = {
      phase: "configuring",
      received: { ...EMPTY_COUNTERS },
    };

    const toRadio = create(Protobuf.Mesh.ToRadioSchema, {
      payloadVariant: { case: "wantConfigId", value: this.configId },
    });

    return this.sendRaw(toBinary(Protobuf.Mesh.ToRadioSchema, toRadio)).catch(
      (e) => {
        if (this.device.status.value === DeviceStatusEnum.DeviceDisconnected) {
          throw new Error("Device connection lost");
        }
        throw e;
      },
    );
  }

  /**
   * Subscribe progress counters to the relevant inbound events. Each
   * subscriber bumps the matching field on the current `configuring`
   * snapshot; `onConfigComplete` flips the phase to `configured`.
   *
   * Outside of an active handshake (phase=idle) inbound packets are
   * ignored — they belong to a later session that re-runs configure().
   */
  private wireProgressTracking(): void {
    const bump = (field: keyof ConnectionProgressCounters): void => {
      const cur = this.progressStore.read.value;
      if (cur.phase !== "configuring") return;
      const next: ConnectionProgressCounters = {
        ...cur.received,
        [field]:
          typeof cur.received[field] === "number"
            ? cur.received[field] + 1
            : true,
      } as ConnectionProgressCounters;
      this.progressStore.write.value = { phase: "configuring", received: next };
    };

    this.events.onConfigPacket.subscribe(() => bump("config"));
    this.events.onModuleConfigPacket.subscribe(() => bump("modules"));
    this.events.onChannelPacket.subscribe(() => bump("channels"));
    this.events.onNodeInfoPacket.subscribe(() => bump("nodes"));
    this.events.onMyNodeInfo.subscribe(() => bump("myInfo"));
    this.events.onDeviceMetadataPacket.subscribe(() => bump("metadata"));

    this.events.onConfigComplete.subscribe(() => {
      const cur = this.progressStore.read.value;
      const received =
        cur.phase === "configuring" ? cur.received : { ...EMPTY_COUNTERS };
      this.progressStore.write.value = { phase: "configured", received };
    });
  }

  public heartbeat(): Promise<number> {
    this.log.debug(Emitter[Emitter.Ping], "❤️ Send heartbeat ping to radio");
    const toRadio = create(Protobuf.Mesh.ToRadioSchema, {
      payloadVariant: { case: "heartbeat", value: {} },
    });
    return this.sendRaw(toBinary(Protobuf.Mesh.ToRadioSchema, toRadio));
  }

  public setHeartbeatInterval(interval: number): void {
    if (this._heartbeatIntervalId !== undefined) {
      clearInterval(this._heartbeatIntervalId);
    }
    this._heartbeatIntervalId = setInterval(() => {
      this.heartbeat().catch((err: Error) => {
        this.log.error(
          Emitter[Emitter.Ping],
          `⚠️ Unable to send heartbeat: ${err.message}`,
        );
      });
    }, interval);
  }

  public updateDeviceStatus(status: DeviceStatusEnum): void {
    if (status !== this.device.status.value) {
      this.events.onDeviceStatus.dispatch(status);
    }
  }

  /**
   * Low-level send: wraps an arbitrary payload in a MeshPacket → ToRadio and
   * returns the ack promise from the queue. Feature slices delegate here.
   */
  public async sendPacket(
    byteData: Uint8Array,
    portNum: Protobuf.Portnums.PortNum,
    destination: Destination,
    channel: ChannelNumber = ChannelNumber.Primary,
    wantAck = true,
    wantResponse = true,
    echoResponse = false,
    replyId?: number,
    emoji?: number,
    packetId?: number,
  ): Promise<number> {
    this.log.trace(
      Emitter[Emitter.SendPacket],
      `📤 Sending ${Protobuf.Portnums.PortNum[portNum]} to ${destination}`,
    );

    const myNum = this.myNodeNum;
    const meshPacket = create(Protobuf.Mesh.MeshPacketSchema, {
      payloadVariant: {
        case: "decoded",
        value: {
          payload: byteData,
          portnum: portNum,
          wantResponse,
          emoji,
          replyId,
          dest: 0,
          requestId: 0,
          source: 0,
        },
      },
      from: myNum,
      to:
        destination === "broadcast"
          ? Constants.broadcastNum
          : destination === "self"
            ? myNum
            : destination,
      id: packetId ?? generatePacketId(),
      wantAck,
      channel,
    });

    const toRadioMessage = create(Protobuf.Mesh.ToRadioSchema, {
      payloadVariant: { case: "packet", value: meshPacket },
    });

    if (echoResponse) {
      meshPacket.rxTime = Math.trunc(Date.now() / 1000);
      this.events.onMeshPacket.dispatch(meshPacket);
    }
    return await this.sendRaw(
      toBinary(Protobuf.Mesh.ToRadioSchema, toRadioMessage),
      meshPacket.id,
    );
  }

  public async sendRaw(
    toRadio: Uint8Array,
    id: number = generatePacketId(),
  ): Promise<number> {
    if (toRadio.length > 512) {
      throw new PacketTooLargeError(toRadio.length);
    }
    this.queue.push({ id, data: toRadio });
    await this.queue.processQueue(this.transport.toDevice);
    return this.queue.wait(id);
  }

  /**
   * Dispatch a `PacketMetadata` echo for locally-composed messages.
   */
  public echoLocalMessage<T>(
    portnum: Protobuf.Portnums.PortNum,
    data: T,
    metadata: Omit<PacketMetadata<T>, "data">,
  ): void {
    // Reserved for use-cases that need to optimistically reflect outbound into stores.
    // Slice use-cases may call bus dispatchers directly; provided here for symmetry.
    void portnum;
    void data;
    void metadata;
  }

  public complete(): void {
    this.queue.clear();
  }

  public async disconnect(): Promise<void> {
    this.log.debug(Emitter[Emitter.Disconnect], "🔌 Disconnecting from device");
    if (this._heartbeatIntervalId !== undefined) {
      clearInterval(this._heartbeatIntervalId);
    }
    this.complete();

    // Signal the inbound pipe abort before blocking on IO so teardown still
    // completes when the transport is already gone (e.g. unplugged serial).
    this._fromDeviceAc?.abort();

    try {
      await this.transport.toDevice.close();
    } catch (err) {
      this.log.debug(
        Emitter[Emitter.Disconnect],
        `Writable stream already closed or errored: ${err instanceof Error ? err.message : String(err)}`,
      );
    }

    if (this._fromDevicePipe) {
      try {
        await this._fromDevicePipe;
      } catch {
        // Expected when the pipe is aborted during disconnect.
      }
    }

    await this.transport.disconnect();
    this.updateDeviceStatus(DeviceStatusEnum.DeviceDisconnected);
  }
}