Announcements/server/src/mqtt/index.js

const PLAYBACK_STATE_VERSION = 1;

// Commands that change durable state are delivered at least once. next and
// previous are not idempotent and carry no command ID, so a redelivered copy
// would silently skip a slide; they stay at QoS 0 deliberately.
const COMMANDS = Object.freeze([
  { name: "mode", suffix: "mode", qos: 1 },
  { name: "pause", suffix: "pause", qos: 1 },
  { name: "resume", suffix: "resume", qos: 1 },
  { name: "reload", suffix: "reload", qos: 1 },
  { name: "static", suffix: "static", qos: 1 },
  { name: "next", suffix: "next", qos: 0 },
  { name: "previous", suffix: "previous", qos: 0 },
]);

const UUID_PATTERN =
  /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

function transportOf(snapshot) {
  if (snapshot.staticTarget) return "static";
  if (snapshot.transport?.rotationPaused) return "paused";
  return snapshot.normalTarget ? "rotating" : "idle";
}

function compactState(snapshot) {
  return {
    version: PLAYBACK_STATE_VERSION,
    mode: snapshot.mode,
    transport: transportOf(snapshot),
    activePlaylistId: snapshot.activePlaylistId ?? null,
    currentEntryId: snapshot.currentEntryId ?? null,
    currentAssetId: snapshot.normalTarget?.assetId ?? null,
    staticAssetId: snapshot.staticTarget?.assetId ?? null,
    revision: snapshot.playbackRevision,
  };
}

function createMqttService({
  config,
  playbackService,
  connect,
  logger = console,
} = {}) {
  if (!playbackService) throw new TypeError("playbackService is required");
  const settings = config?.mqtt || {};
  const prefix = settings.topicPrefix || "announcements";
  const topic = (suffix) => `${prefix}/${suffix}`;
  const availabilityTopic = topic("availability");
  const modeStateTopic = topic("mode/state");
  const playbackStateTopic = topic("playback/state");

  const configured = Boolean(settings.url);
  let client = null;
  let connected = false;
  let connectionCount = 0;
  let lastConnectedAt = null;
  let lastErrorAt = null;
  let lastErrorMessage = null;
  let commandCount = 0;
  let stopping = false;

  function warn(error, message, details) {
    try {
      logger.warn?.(`${message}${details ? ` ${JSON.stringify(details)}` : ""}`, error);
    } catch {
      // Logging failures must never propagate into the broker callbacks.
    }
  }

  function publish(topicName, payload, options) {
    if (!client || !connected) return false;
    try {
      client.publish(topicName, payload, options, (error) => {
        if (error) warn(error, "MQTT publish failed", { topic: topicName });
      });
      return true;
    } catch (error) {
      warn(error, "MQTT publish threw", { topic: topicName });
      return false;
    }
  }

  // Retained authoritative state is only ever published on dedicated state
  // topics so a subscriber can never mistake an echo for a new command.
  function publishState(snapshot) {
    if (!snapshot) return;
    publish(modeStateTopic, snapshot.mode, { qos: 1, retain: true });
    publish(playbackStateTopic, JSON.stringify(compactState(snapshot)), {
      qos: 1,
      retain: true,
    });
  }

  function publishCurrentState() {
    try {
      publishState(playbackService.getSnapshot());
    } catch (error) {
      warn(error, "MQTT could not read the authoritative playback snapshot");
    }
  }

  function handleMode(payload) {
    const value = payload.trim().toLowerCase();
    if (value !== "on" && value !== "off") {
      warn(null, "Ignoring MQTT mode command with an unsupported payload", {
        payload: value.slice(0, 40),
      });
      return false;
    }
    playbackService.setMode(value);
    return true;
  }

  function handleStatic(payload) {
    const assetId = payload.trim();
    if (!UUID_PATTERN.test(assetId)) {
      warn(null, "Ignoring MQTT static command that is not a LibraryAsset UUID", {
        payload: assetId.slice(0, 60),
      });
      return false;
    }
    playbackService.enterStatic(assetId);
    return true;
  }

  const handlers = Object.freeze({
    mode: handleMode,
    static: handleStatic,
    next: () => (playbackService.next(), true),
    previous: () => (playbackService.previous(), true),
    pause: () => (playbackService.pause(), true),
    resume: () => (playbackService.resume(), true),
    reload: () => (playbackService.reload(), true),
  });

  const byTopic = new Map(
    COMMANDS.map((command) => [topic(command.suffix), command]),
  );

  function handleMessage(topicName, buffer) {
    const command = byTopic.get(topicName);
    if (!command) {
      warn(null, "Ignoring MQTT message on an unknown topic", { topic: topicName });
      return;
    }
    let payload = "";
    try {
      payload = buffer ? buffer.toString("utf8") : "";
    } catch (error) {
      warn(error, "Ignoring MQTT message with an undecodable payload", {
        topic: topicName,
      });
      return;
    }
    try {
      // Every command runs through the one authoritative playback service.
      // A rejected command leaves playback exactly as it was.
      if (handlers[command.name](payload)) commandCount += 1;
    } catch (error) {
      warn(error, "MQTT command was rejected by the playback service", {
        topic: topicName,
        code: error?.code,
      });
    }
  }

  function subscribeAll() {
    if (!client) return;
    for (const command of COMMANDS) {
      client.subscribe(topic(command.suffix), { qos: command.qos }, (error) => {
        if (error) warn(error, "MQTT subscribe failed", { topic: topic(command.suffix) });
      });
    }
  }

  function handleConnect() {
    connected = true;
    connectionCount += 1;
    lastConnectedAt = new Date().toISOString();
    logger.info?.(`MQTT connected to the configured broker as ${prefix}`);
    // Reconnect must restore both the subscriptions and the retained state a
    // broker restart would otherwise have dropped.
    publish(availabilityTopic, "online", { qos: 1, retain: true });
    subscribeAll();
    publishCurrentState();
  }

  function start() {
    if (!configured) {
      logger.info?.("MQTT is not configured; the integration stays disabled");
      return null;
    }
    if (client) return client;
    const factory = connect || require("mqtt").connect;
    client = factory(settings.url, {
      username: settings.username || undefined,
      password: settings.password || undefined,
      reconnectPeriod: settings.reconnectPeriodMs ?? 5_000,
      connectTimeout: settings.connectTimeoutMs ?? 10_000,
      clean: true,
      // The broker publishes this for us if the API disappears without a
      // clean disconnect, which is exactly the case a client cannot report.
      will: {
        topic: availabilityTopic,
        payload: "offline",
        qos: 1,
        retain: true,
      },
    });

    client.on("connect", handleConnect);
    client.on("message", handleMessage);
    client.on("reconnect", () => {
      connected = false;
    });
    client.on("close", () => {
      connected = false;
    });
    client.on("offline", () => {
      connected = false;
    });
    client.on("error", (error) => {
      connected = false;
      lastErrorAt = new Date().toISOString();
      lastErrorMessage = error?.message || "MQTT error";
      // A broker outage degrades the integration only. Playback, HTTP, and
      // the display protocol continue untouched and the client reconnects.
      warn(error, "MQTT connection error; the integration is degraded");
    });
    return client;
  }

  async function stop() {
    stopping = true;
    if (!client) return;
    const active = client;
    client = null;
    try {
      if (connected) {
        await new Promise((resolve) => {
          let settled = false;
          const finish = () => {
            if (settled) return;
            settled = true;
            resolve();
          };
          const timer = setTimeout(finish, 1_000);
          timer.unref?.();
          // A clean shutdown retracts availability itself; the will is only
          // for the unexpected case.
          active.publish(availabilityTopic, "offline", { qos: 1, retain: true }, () => {
            clearTimeout(timer);
            finish();
          });
        });
        for (const command of COMMANDS) {
          try {
            active.unsubscribe(topic(command.suffix));
          } catch {
            // Shutdown continues even if the broker already went away.
          }
        }
      }
    } catch (error) {
      warn(error, "MQTT shutdown publication failed");
    }
    connected = false;
    try {
      await new Promise((resolve) => {
        const timer = setTimeout(resolve, 1_000);
        timer.unref?.();
        active.end(false, {}, () => {
          clearTimeout(timer);
          resolve();
        });
      });
    } catch (error) {
      warn(error, "MQTT client did not close cleanly");
    }
  }

  function status() {
    return {
      configured,
      status: !configured ? "disabled" : connected ? "connected" : "disconnected",
      connected,
      topicPrefix: prefix,
      connectionCount,
      commandCount,
      lastConnectedAt,
      lastErrorAt,
      lastErrorMessage,
      stopping,
    };
  }

  return {
    availabilityTopic,
    commandTopics: COMMANDS.map((command) => topic(command.suffix)),
    compactState,
    modeStateTopic,
    playbackStateTopic,
    publishState,
    start,
    status,
    stop,
  };
}

module.exports = {
  COMMANDS,
  PLAYBACK_STATE_VERSION,
  createMqttService,
};