pipeline/backend-node/src/fleet/mqttBroker.ts

/**
 * 내장 MQTT 브로커 (aedes)
 * - Pipeline이 MQTT 브로커 역할까지 수행
 * - aedes 내부 이벤트로 직접 publish 감지 (내부 client 불필요)
 *
 * 토픽 규칙 (vexplor_fleet 호환):
 *   vexplor/devices/{deviceId}/status    - 디바이스 → 서버
 *   vexplor/devices/{deviceId}/metrics   - 디바이스 → 서버
 *   vexplor/devices/{deviceId}/commands  - 서버 → 디바이스
 *   vexplor/devices/{deviceId}/responses - 디바이스 → 서버
 *   vexplor/devices/{deviceId}/data      - 디바이스 → 서버
 */

import Aedes from "aedes";
import { createServer } from "aedes-server-factory";
import { logger } from "../utils/logger";

const MQTT_PORT = parseInt(process.env.MQTT_PORT || "1883", 10);
const MQTT_WS_PORT = parseInt(process.env.MQTT_WS_PORT || "8083", 10);

type MessageHandler = (topic: string, payload: Buffer) => void | Promise<void>;

export class FleetMqttBroker {
  private aedes: any;
  private tcpServer: any;
  private wsServer: any;
  private started = false;

  private messageHandlers = new Map<string, MessageHandler[]>();

  constructor() {
    this.aedes = (Aedes as any)({ id: "pipeline-mqtt-broker" });

    this.aedes.on("client", (client) => {
      logger.info(`[MQTT] 클라이언트 연결: ${client.id}`);
    });

    this.aedes.on("clientDisconnect", (client) => {
      logger.info(`[MQTT] 클라이언트 연결 해제: ${client.id}`);
    });

    this.aedes.on("subscribe", (subscriptions, client) => {
      logger.debug(
        `[MQTT] 구독: ${subscriptions.map((s) => s.topic).join(", ")} (by ${client?.id})`,
      );
    });

    // 모든 publish 감지
    this.aedes.on("publish", (packet, client) => {
      if (!packet.topic || packet.topic.startsWith("$SYS")) return;
      // 내부에서 aedes.publish()로 보낸 것도 이 이벤트로 잡힘 - client가 null이면 내부 발행
      if (client) {
        this.dispatchMessage(packet.topic, packet.payload as Buffer);
      }
    });
  }

  /**
   * 브로커 시작 (TCP + WebSocket)
   */
  async start(): Promise<void> {
    if (this.started) return;

    await new Promise<void>((resolve, reject) => {
      this.tcpServer = createServer(this.aedes as any);
      this.tcpServer.once("error", reject);
      this.tcpServer.listen(MQTT_PORT, "0.0.0.0", () => {
        logger.info(`[MQTT] TCP 브로커 시작: mqtt://0.0.0.0:${MQTT_PORT}`);
        resolve();
      });
    });

    await new Promise<void>((resolve, reject) => {
      this.wsServer = createServer(this.aedes as any, { ws: true });
      this.wsServer.once("error", reject);
      this.wsServer.listen(MQTT_WS_PORT, "0.0.0.0", () => {
        logger.info(`[MQTT] WebSocket 브로커 시작: ws://0.0.0.0:${MQTT_WS_PORT}`);
        resolve();
      });
    });

    this.started = true;
    logger.info("[MQTT] 브로커 완전 기동 (내부 publish 가능)");
  }

  /**
   * 토픽 패턴 등록 (MQTT 와일드카드 지원: +, #)
   */
  on(topicPattern: string, handler: MessageHandler): void {
    if (!this.messageHandlers.has(topicPattern)) {
      this.messageHandlers.set(topicPattern, []);
    }
    this.messageHandlers.get(topicPattern)!.push(handler);
  }

  private dispatchMessage(topic: string, payload: Buffer): void {
    for (const [pattern, handlers] of this.messageHandlers) {
      if (this.topicMatches(pattern, topic)) {
        for (const handler of handlers) {
          Promise.resolve(handler(topic, payload)).catch((e) =>
            logger.error(`[MQTT] 핸들러 에러 (${pattern}):`, e),
          );
        }
      }
    }
  }

  private topicMatches(pattern: string, topic: string): boolean {
    const pParts = pattern.split("/");
    const tParts = topic.split("/");

    for (let i = 0; i < pParts.length; i++) {
      if (pParts[i] === "#") return true;
      if (i >= tParts.length) return false;
      if (pParts[i] === "+") continue;
      if (pParts[i] !== tParts[i]) return false;
    }
    return pParts.length === tParts.length;
  }

  /**
   * 서버 → 디바이스 메시지 발행 (aedes.publish 직접 사용)
   */
  publish(topic: string, message: string | object, qos: 0 | 1 | 2 = 1): Promise<void> {
    return new Promise((resolve, reject) => {
      if (!this.started) return reject(new Error("MQTT 브로커가 시작되지 않았습니다."));
      const payload =
        typeof message === "string" ? message : JSON.stringify(message);
      (this.aedes as any).publish(
        {
          cmd: "publish",
          qos,
          topic,
          payload: Buffer.from(payload),
          retain: false,
        },
        (err: Error | null) => {
          if (err) reject(err);
          else resolve();
        },
      );
    });
  }

  /**
   * 디바이스에 커맨드 발행
   */
  sendCommandToDevice(deviceId: string, command: object): Promise<void> {
    const topic = `vexplor/devices/${deviceId}/commands`;
    return this.publish(topic, command, 1);
  }

  /**
   * 토픽에서 deviceId 추출 (vexplor/devices/{deviceId}/...)
   */
  static extractDeviceId(topic: string): string | null {
    const m = topic.match(/^vexplor\/devices\/([^/]+)\//);
    return m ? m[1] : null;
  }

  async stop(): Promise<void> {
    if (this.tcpServer) this.tcpServer.close();
    if (this.wsServer) this.wsServer.close();
    this.aedes.close();
    this.started = false;
  }
}

let brokerInstance: FleetMqttBroker | null = null;

export function getFleetMqttBroker(): FleetMqttBroker {
  if (!brokerInstance) {
    brokerInstance = new FleetMqttBroker();
  }
  return brokerInstance;
}