allotment/packages/server/mqtt/MqttTtnIngest.js

import mqtt from "mqtt";
import { loadMqttConfig, resolveBrokerPassword } from "./loadConfig.js";
import { applyTtnUplink } from "../services/devices.js";
import { getOrCreateWaterTank } from "../services/waterTank.js";
import {
  formatPowerStateForClient,
  getOrCreatePowerState,
} from "../services/powerState.js";
import { getLogger } from "../logger.js";

const logger = getLogger("MQTT");

function extractTtnIds(topic, payloadObj) {
  let applicationId = payloadObj?.end_device_ids?.application_ids?.application_id;
  let deviceId = payloadObj?.end_device_ids?.device_id;
  if (!applicationId || !deviceId) {
    const parts = topic.split("/");
    const v3 = parts[0] === "v3";
    if (v3 && parts.length >= 4) {
      applicationId = applicationId || parts[1];
      const devIdx = parts.indexOf("devices");
      if (devIdx >= 0 && parts[devIdx + 1]) {
        deviceId = deviceId || parts[devIdx + 1];
      }
    }
  }
  return { applicationId, deviceId };
}

export class MqttTtnIngest {
  constructor(socketManager) {
    this.socketManager = socketManager;
    this.client = null;
    this.reconnectDelayMs = 3000;
    this.maxReconnectDelayMs = 60000;
    this._timer = null;
  }

  start() {
    const config = loadMqttConfig();
    if (!config?.broker?.host || !config?.subscriptions?.length) {
      logger.warn(
        "No config/mqtt.json (or missing broker/subscriptions) — MQTT ingest disabled.",
      );
      return;
    }
    this.config = config;
    this._connect();
  }

  _connect() {
    const { broker, subscriptions, clientId } = this.config;
    const password = resolveBrokerPassword(this.config);
    const protocol = broker.protocol === "mqtts" ? "mqtts" : "mqtt";
    const url = `${protocol}://${broker.host}:${broker.port || 1883}`;
    const opts = {
      username: broker.username || undefined,
      password: password || undefined,
      clientId: clientId || `toms-allotment-${process.pid}`,
      reconnectPeriod: 0,
      ...(broker.protocol === "mqtts" && {
        rejectUnauthorized: broker.rejectUnauthorized !== false,
      }),
    };

    this.client = mqtt.connect(url, opts);

    this.client.on("connect", () => {
      logger.info("Connected to", url);
      this.reconnectDelayMs = 3000;
      for (const sub of subscriptions) {
        this.client.subscribe(sub, (err) => {
          if (err) logger.error("Subscribe error:", sub, err);
          else logger.info("Subscribed:", sub);
        });
      }
    });

    this.client.on("message", async (topic, message) => {
      try {
        const text = message.toString("utf-8");
        const payloadObj = JSON.parse(text);
        const decoded =
          payloadObj.uplink_message?.decoded_payload ??
          payloadObj.decoded_payload;

        const { applicationId, deviceId } = extractTtnIds(topic, payloadObj);
        if (!applicationId || !deviceId) {
          logger.warn("Could not determine application/device id");
          return;
        }

        const result = await applyTtnUplink({
          applicationId,
          deviceId,
          decodedPayload: decoded,
        });

        if (result?.skipped) return;

        if (result?.device) {
          for (const patch of result.device.patches || []) {
            this.socketManager.emitPatchUpdated(patch);
          }
        }
        if (result?.tankLevel) {
          const tank = await getOrCreateWaterTank();
          this.socketManager.emitWaterTankUpdated({
            level: tank.level,
            updatedAt: tank.updatedAt,
          });
        }
        if (result?.power) {
          const pow = await getOrCreatePowerState();
          this.socketManager.emitPowerUpdated(formatPowerStateForClient(pow));
        }
      } catch (e) {
        logger.error("Message handler error:", e.message);
      }
    });

    this.client.on("error", (err) => {
      logger.error("Error:", err.message);
    });

    this.client.on("close", () => {
      this._scheduleReconnect();
    });
  }

  _scheduleReconnect() {
    if (this._timer) return;
    logger.info(`Reconnecting in ${this.reconnectDelayMs / 1000}s...`);
    this._timer = setTimeout(() => {
      this._timer = null;
      this.reconnectDelayMs = Math.min(
        this.maxReconnectDelayMs,
        Math.floor(this.reconnectDelayMs * 1.5),
      );
      if (this.client) {
        try {
          this.client.end(true);
        } catch (_) {}
      }
      this._connect();
    }, this.reconnectDelayMs);
  }
}