Integrate upstream-compatible read-only MeshCentral transport

implementation/src/upstream-transport.ts

@@ -0,0 +1,124 @@
+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();
+  }
+}