Announcements/server/test/mqtt-broker.test.js

const test = require("node:test");
const assert = require("node:assert/strict");
const mqtt = require("mqtt");
const { loadConfig } = require("../src/config");
const { createRuntime } = require("../src/runtime");

// Protocol behaviour that a hand-written fake cannot prove — retained
// messages, subscription delivery, and a broker-published last will — is
// verified against a real broker. The isolated verification stack sets
// MQTT_TEST_URL; without it these checks are skipped rather than faked.
const BROKER_URL = process.env.MQTT_TEST_URL;
const silentLogger = { info() {}, warn() {}, error() {} };

function testConfig(overrides = {}) {
  return loadConfig({
    NODE_ENV: "test",
    HOST: "127.0.0.1",
    PORT: "0",
    DATABASE_PATH: ":memory:",
    APP_TIMEZONE: "America/Chicago",
    SHUTDOWN_TIMEOUT_MS: "1000",
    S3_ENDPOINT: "http://seaweedfs:8333",
    S3_BUCKET: "announcements-test",
    MQTT_URL: BROKER_URL,
    ...overrides,
  });
}

function testStorage() {
  return {
    async start() {},
    stop() {},
    health: () => ({ status: "healthy", checkedAt: null, lastSuccessAt: null }),
  };
}

function connectObserver() {
  return new Promise((resolve, reject) => {
    const client = mqtt.connect(BROKER_URL, { reconnectPeriod: 0 });
    client.once("connect", () => resolve(client));
    client.once("error", reject);
  });
}

function collect(client, topic) {
  const received = [];
  client.on("message", (messageTopic, payload) => {
    if (messageTopic === topic) received.push(payload.toString("utf8"));
  });
  return received;
}

function waitFor(predicate, { timeoutMs = 5_000, label = "condition" } = {}) {
  return new Promise((resolve, reject) => {
    const deadline = Date.now() + timeoutMs;
    const tick = () => {
      let value;
      try {
        value = predicate();
      } catch (error) {
        reject(error);
        return;
      }
      if (value) {
        resolve(value);
        return;
      }
      if (Date.now() > deadline) {
        reject(new Error(`Timed out waiting for ${label}`));
        return;
      }
      setTimeout(tick, 25);
    };
    tick();
  });
}

async function subscribe(client, topic, options = {}) {
  await new Promise((resolve, reject) => {
    client.subscribe(topic, { qos: 1, ...options }, (error) =>
      error ? reject(error) : resolve(),
    );
  });
}

async function publish(client, topic, payload, options = {}) {
  await new Promise((resolve, reject) => {
    client.publish(topic, payload, { qos: 1, ...options }, (error) =>
      error ? reject(error) : resolve(),
    );
  });
}

async function startRuntime(t) {
  const runtime = createRuntime({
    config: testConfig(),
    logger: silentLogger,
    storage: testStorage(),
  });
  await runtime.start();
  t.after(async () => {
    try {
      await runtime.stop("test cleanup");
    } catch {
      // A test that already stopped the runtime is fine.
    }
  });
  await waitFor(() => runtime.mqttService.status().connected, {
    label: "the API to connect to the broker",
  });
  return { runtime, baseUrl: `http://127.0.0.1:${runtime.address.port}` };
}

async function seedPlayback(runtime, baseUrl) {
  const timestamp = Date.now();
  for (const [id, hash] of [
    ["asset-one", "a"],
    ["asset-two", "c"],
  ]) {
    runtime.database
      .prepare(
        `INSERT INTO assets (
          id, source_sha256, display_name, original_filename, status,
          display_object_key, display_sha256, display_content_type,
          display_byte_size, display_width, display_height, created_at, updated_at
        ) VALUES (?, ?, ?, ?, 'ready', 'private/display/x.webp', ?, 'image/webp', 10, 1920, 1080, ?, ?)`,
      )
      .run(id, hash.repeat(64), id, `${id}.png`, hash.repeat(64), timestamp, timestamp);
  }
  const created = await fetch(`${baseUrl}/api/playlists`, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ name: "Broker" }),
  });
  const playlist = await created.json();
  for (const assetId of ["asset-one", "asset-two"]) {
    await fetch(`${baseUrl}/api/playlists/${playlist.id}/entries`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ assetId, durationOverrideMs: 60_000 }),
    });
  }
  await fetch(`${baseUrl}/api/playlists/${playlist.id}/activate`, { method: "POST" });
  return playlist;
}

const options = BROKER_URL ? {} : { skip: "MQTT_TEST_URL is not configured" };

test("a real broker delivers commands and retains authoritative state", options, async (t) => {
  const { runtime, baseUrl } = await startRuntime(t);
  await seedPlayback(runtime, baseUrl);

  const observer = await connectObserver();
  t.after(() => observer.endAsync?.() ?? observer.end(true));
  const availability = collect(observer, "announcements/availability");
  const modeState = collect(observer, "announcements/mode/state");
  await subscribe(observer, "announcements/availability");
  await subscribe(observer, "announcements/mode/state");

  // A late subscriber receives retained state immediately, which is what a
  // Home Assistant restart depends on.
  await waitFor(() => availability.includes("online"), {
    label: "retained availability",
  });
  await waitFor(() => modeState.length > 0, { label: "retained mode state" });
  assert.equal(modeState.at(-1), "on");

  const commander = await connectObserver();
  t.after(() => commander.endAsync?.() ?? commander.end(true));
  await publish(commander, "announcements/mode", "off");
  await waitFor(
    () =>
      runtime.database
        .prepare("SELECT mode FROM playback_state WHERE output_id = 'global'")
        .get().mode === "off",
    { label: "the mode command to commit" },
  );
  await waitFor(() => modeState.at(-1) === "off", { label: "retained mode state to follow" });

  const before = runtime.playbackService.getSnapshot();
  await publish(commander, "announcements/next", "", { qos: 0 });
  await waitFor(
    () => runtime.playbackService.getSnapshot().currentEntryId !== before.currentEntryId,
    { label: "the next command to advance playback" },
  );

  await publish(commander, "announcements/pause", "");
  await waitFor(() => runtime.playbackService.getSnapshot().transportState === "paused", {
    label: "pause",
  });
  await publish(commander, "announcements/resume", "");
  await waitFor(() => runtime.playbackService.getSnapshot().transportState === "playing", {
    label: "resume",
  });

  // An invalid payload is ignored over the real broker exactly as it is
  // locally: the last committed mode survives untouched.
  await publish(commander, "announcements/mode", " ON ");
  await waitFor(() => runtime.playbackService.getSnapshot().mode === "on", {
    label: "a whitespace-padded mode command to commit",
  });
  await publish(commander, "announcements/mode", "dim");
  await new Promise((resolve) => setTimeout(resolve, 250));
  assert.equal(runtime.playbackService.getSnapshot().mode, "on");
  assert.equal(modeState.at(-1), "on", "no fake state was published for a rejected command");
});

test("a clean API shutdown retracts its retained availability", options, async (t) => {
  const { runtime } = await startRuntime(t);

  const observer = await connectObserver();
  t.after(() => observer.endAsync?.() ?? observer.end(true));
  const availability = collect(observer, "announcements/availability");
  await subscribe(observer, "announcements/availability");
  await waitFor(() => availability.includes("online"), { label: "online availability" });

  await runtime.stop("clean shutdown test");

  await waitFor(() => availability.at(-1) === "offline", {
    label: "availability to be retracted on a planned shutdown",
  });
});

test("an ungraceful disappearance makes the broker publish the last will", options, async (t) => {
  const observer = await connectObserver();
  t.after(() => observer.endAsync?.() ?? observer.end(true));
  const availability = collect(observer, "test/availability");
  await subscribe(observer, "test/availability");

  // A crashed or power-cut API never sends DISCONNECT. Only a will can report
  // that, so the will configuration is exercised against a real broker here;
  // the unit suite proves the service registers exactly these options.
  const willed = mqtt.connect(BROKER_URL, {
    reconnectPeriod: 0,
    will: { topic: "test/availability", payload: "offline", qos: 1, retain: true },
  });
  await new Promise((resolve, reject) => {
    willed.once("connect", resolve);
    willed.once("error", reject);
  });
  await publish(willed, "test/availability", "online", { retain: true });
  await waitFor(() => availability.includes("online"), { label: "online availability" });

  willed.stream.destroy();

  await waitFor(() => availability.at(-1) === "offline", {
    label: "the broker to publish the last will",
    timeoutMs: 10_000,
  });
});