meshcentral-openclaw-bridge/implementation/src/upstream-transport.ts

import { WebSocket } from 'ws';
import { AuthCredentials, MeshCtrlResponse, CONTROL_PATH, encodeAuthHeader, MAX_RESPONSE_SIZE, parseResponse } from './meshctrl-client';

/** Minimal upstream-compatible, read-only MeshCentral control transport. */
export class UpstreamMeshTransport {
  private ws: WebSocket | null = null;
  private connectionPromise: Promise<WebSocket> | null = null;
  private credentials: AuthCredentials | null = null;
  private readonly url: string;

  constructor(url: string) {
    const normalized = url.replace(/^https:/, 'wss:').replace(/^http:/, 'ws:');
    this.url = normalized.endsWith(CONTROL_PATH) ? normalized : `${normalized.replace(/\/$/, '')}${CONTROL_PATH}`;
  }

  async connect(credentials: AuthCredentials): Promise<void> {
    this.credentials = credentials;
    await this.open(credentials);
  }

  async sendRequest(request: Record<string, unknown>): Promise<MeshCtrlResponse> {
    if (!this.credentials) throw new Error('Not connected');
    const ws = await this.open(this.credentials);
    const responseId = typeof request.responseid === 'string' ? request.responseid : `openclaw-${Date.now()}`;
    const message = { ...request, responseid: responseId };

    return new Promise<MeshCtrlResponse>((resolve, reject) => {
      const timeout = setTimeout(() => {
        ws.removeListener('message', onMessage);
        reject(new Error('MeshCentral request timed out'));
      }, 30_000);

      const finish = (fn: () => void): void => {
        clearTimeout(timeout);
        ws.removeListener('message', onMessage);
        fn();
      };

      const onMessage = (raw: Buffer | string): void => {
        const text = Buffer.isBuffer(raw) ? raw.toString() : raw;
        if (Buffer.byteLength(text, 'utf8') > MAX_RESPONSE_SIZE) return finish(() => reject(new Error('MeshCentral response exceeds size limit')));
        let parsed: MeshCtrlResponse;
        try {
          parsed = parseResponse(text);
        } catch {
          return;
        }

        // Bootstrap and event messages are unsolicited. MeshCentral's
        // legacy meshes response also omits responseid, so correlate it by
        // action while this transport has one active request.
        if (parsed.responseid !== responseId && !(request.action === 'meshes' && parsed.action === 'meshes')) return;
        finish(() => resolve(parsed));
      };

      ws.on('message', onMessage);
      ws.send(JSON.stringify(message));
    });
  }

  async request(credentials: AuthCredentials, request: Record<string, unknown>): Promise<MeshCtrlResponse> {
    await this.connect(credentials);
    return this.sendRequest(request);
  }

  private async open(credentials: AuthCredentials): Promise<WebSocket> {
    if (this.ws?.readyState === WebSocket.OPEN) return this.ws;
    if (this.connectionPromise) return this.connectionPromise;

    this.connectionPromise = new Promise<WebSocket>((resolve, reject) => {
      const ws = new WebSocket(this.url, {
        rejectUnauthorized: true,
        perMessageDeflate: false,
        headers: { 'x-meshauth': encodeAuthHeader(credentials) },
      });
      this.ws = ws;
      const timeout = setTimeout(() => {
        ws.terminate();
        reject(new Error('MeshCentral connection timed out'));
      }, 30_000);
      ws.once('open', async () => {
        clearTimeout(timeout);
        // Match the upstream client's startup behavior before the first command.
        await new Promise((settled) => setTimeout(settled, 750));
        resolve(ws);
      });
      ws.once('error', (error: Error) => {
        clearTimeout(timeout);
        reject(error);
      });
      ws.once('close', () => {
        if (this.ws === ws) this.ws = null;
      });
    }).finally(() => {
      this.connectionPromise = null;
    });

    return this.connectionPromise;
  }

  async close(): Promise<void> {
    const ws = this.ws;
    this.ws = null;
    this.connectionPromise = null;
    this.credentials = null;
    if (!ws) return;
    if (ws.readyState === WebSocket.CLOSED) return;
    await new Promise<void>((resolve) => {
      const timeout = setTimeout(() => {
        ws.terminate();
        resolve();
      }, 2_000);
      ws.once('close', () => {
        clearTimeout(timeout);
        resolve();
      });
      ws.close(1000, 'bridge-close');
    });
  }

  async disconnect(): Promise<void> {
    await this.close();
  }
}

/** Parse and normalize MeshCentral's native grouped node response. */
export function normalizeGroupedNodes(response: MeshCtrlResponse): Array<{
  id: string;
  name: string;
  meshid: string;
  online: boolean;
  device?: { os?: string; ip?: string[]; uptime?: number; lastseen?: number; agentver?: string };
}> {
  const grouped = (response.nodes ?? {}) as Record<string, Array<{
    id?: string;
    _id?: string;
    name: string;
    meshid?: string;
    online?: boolean;
    conn?: number;
    osdesc?: string;
    ip?: string[];
    device?: { os?: string; ip?: string[]; uptime?: number; lastseen?: number; agentver?: string };
  }>>;

  return Object.entries(grouped).flatMap(([groupId, nodes]) => nodes.map((node) => ({
    id: node.id ?? node._id ?? '',
    name: node.name,
    meshid: node.meshid ?? groupId,
    online: node.online ?? node.conn === 1,
    device: node.device ?? { os: node.osdesc, ip: node.ip },
  }))).filter((node) => node.id !== '');
}

/** Normalize MeshCentral's native mesh identifiers and description field. */
export function normalizeMeshes(response: MeshCtrlResponse): Array<{
  id: string;
  name: string;
  description?: string;
}> {
  return ((response.meshes ?? []) as Array<{
    id?: string;
    _id?: string;
    name: string;
    description?: string;
    desc?: string;
  }>).map((mesh) => ({
    id: mesh.id ?? mesh._id ?? '',
    name: mesh.name,
    description: mesh.description ?? mesh.desc,
  })).filter((mesh) => mesh.id !== '');
}