Announcements/server/test/mqtt.test.js

const test = require("node:test");
const assert = require("node:assert/strict");
const { COMMANDS, createMqttService } = require("../src/mqtt");

const PREFIX = "announcements";

function createFakeClient() {
  const listeners = new Map();
  const client = {
    published: [],
    subscriptions: [],
    unsubscribed: [],
    ended: false,
    options: null,
    on(event, handler) {
      listeners.set(event, [...(listeners.get(event) || []), handler]);
      return client;
    },
    emit(event, ...args) {
      for (const handler of listeners.get(event) || []) handler(...args);
    },
    publish(topic, payload, options, callback) {
      client.published.push({ topic, payload: String(payload), options });
      callback?.(null);
    },
    subscribe(topic, options, callback) {
      client.subscriptions.push({ topic, qos: options.qos });
      callback?.(null);
    },
    unsubscribe(topic) {
      client.unsubscribed.push(topic);
    },
    end(force, options, callback) {
      client.ended = true;
      callback?.();
    },
  };
  return client;
}

function createPlaybackStub(snapshot = {}) {
  const calls = [];
  const base = {
    mode: "on",
    transportState: "playing",
    transport: { rotationPaused: false },
    activePlaylistId: "playlist-a",
    currentEntryId: "entry-a",
    normalTarget: { assetId: "asset-a" },
    staticTarget: null,
    playbackRevision: 7,
    ...snapshot,
  };
  return {
    calls,
    getSnapshot: () => base,
    setMode(mode) {
      calls.push(["setMode", mode]);
    },
    next() {
      calls.push(["next"]);
    },
    previous() {
      calls.push(["previous"]);
    },
    pause() {
      calls.push(["pause"]);
    },
    resume() {
      calls.push(["resume"]);
    },
    reload() {
      calls.push(["reload"]);
    },
    enterStatic(assetId) {
      calls.push(["enterStatic", assetId]);
      if (assetId === "11111111-1111-4111-8111-111111111111") {
        const error = new Error("Library asset not found");
        error.code = "ASSET_NOT_FOUND";
        throw error;
      }
      if (assetId === "22222222-2222-4222-8222-222222222222") {
        const error = new Error("Static playback requires a ready library asset");
        error.code = "ASSET_NOT_READY";
        throw error;
      }
      if (assetId === "33333333-3333-4333-8333-333333333333") {
        const error = new Error("Expired library assets cannot be held statically");
        error.code = "ASSET_EXPIRED";
        throw error;
      }
    },
  };
}

function createHarness({ snapshot, url = "mqtt://broker:1883" } = {}) {
  const client = createFakeClient();
  const playbackService = createPlaybackStub(snapshot);
  const service = createMqttService({
    config: {
      mqtt: {
        url,
        username: "user",
        password: "must-not-leak",
        topicPrefix: PREFIX,
        reconnectPeriodMs: 5_000,
        connectTimeoutMs: 10_000,
      },
    },
    playbackService,
    logger: { info() {}, warn() {} },
    connect: (connectUrl, options) => {
      client.options = options;
      client.url = connectUrl;
      return client;
    },
  });
  return { client, playbackService, service };
}

function connectHarness(options) {
  const harness = createHarness(options);
  harness.service.start();
  harness.client.emit("connect");
  return harness;
}

test("the default configuration resolves the exact required command topics", () => {
  const harness = createHarness();
  assert.deepEqual(harness.service.commandTopics.sort(), [
    "announcements/mode",
    "announcements/next",
    "announcements/pause",
    "announcements/previous",
    "announcements/reload",
    "announcements/resume",
    "announcements/static",
  ]);
  assert.equal(harness.service.availabilityTopic, "announcements/availability");
  assert.equal(harness.service.modeStateTopic, "announcements/mode/state");
  assert.equal(harness.service.playbackStateTopic, "announcements/playback/state");
});

test("connecting configures a retained last will and publishes availability", () => {
  const harness = connectHarness();

  assert.deepEqual(harness.client.options.will, {
    topic: "announcements/availability",
    payload: "offline",
    qos: 1,
    retain: true,
  });
  assert.equal(harness.client.options.reconnectPeriod, 5_000);
  const availability = harness.client.published.find(
    (message) => message.topic === "announcements/availability",
  );
  assert.equal(availability.payload, "online");
  assert.equal(availability.options.retain, true);
  assert.equal(availability.options.qos, 1);
});

test("every required command topic is subscribed with its delivery guarantee", () => {
  const harness = connectHarness();
  const byTopic = Object.fromEntries(
    harness.client.subscriptions.map((entry) => [entry.topic, entry.qos]),
  );

  assert.equal(byTopic["announcements/mode"], 1);
  assert.equal(byTopic["announcements/pause"], 1);
  assert.equal(byTopic["announcements/resume"], 1);
  assert.equal(byTopic["announcements/reload"], 1);
  assert.equal(byTopic["announcements/static"], 1);
  // Redelivering a non-idempotent edge command would silently skip a slide.
  assert.equal(byTopic["announcements/next"], 0);
  assert.equal(byTopic["announcements/previous"], 0);
  assert.equal(harness.client.subscriptions.length, COMMANDS.length);
});

test("mode commands delegate to the playback service and tolerate whitespace and case", () => {
  const harness = connectHarness();

  harness.client.emit("message", "announcements/mode", Buffer.from("off"));
  harness.client.emit("message", "announcements/mode", Buffer.from(" ON \n"));
  harness.client.emit("message", "announcements/mode", Buffer.from("OFF"));

  assert.deepEqual(harness.playbackService.calls, [
    ["setMode", "off"],
    ["setMode", "on"],
    ["setMode", "off"],
  ]);
});

test("an invalid mode payload changes nothing and publishes no fake state", () => {
  const harness = connectHarness();
  const publishedBefore = harness.client.published.length;

  for (const payload of ["dim", "", "   ", "1", "{\"mode\":\"on\"}"]) {
    harness.client.emit("message", "announcements/mode", Buffer.from(payload));
  }

  assert.deepEqual(harness.playbackService.calls, []);
  assert.equal(harness.client.published.length, publishedBefore);
  assert.equal(harness.service.status().commandCount, 0);
});

test("edge commands invoke the exact playback operation regardless of payload", () => {
  const harness = connectHarness();

  harness.client.emit("message", "announcements/next", Buffer.from(""));
  harness.client.emit("message", "announcements/previous", Buffer.from("  "));
  harness.client.emit("message", "announcements/pause", Buffer.from("press"));
  harness.client.emit("message", "announcements/resume", Buffer.from("\n"));
  harness.client.emit("message", "announcements/reload", null);

  assert.deepEqual(harness.playbackService.calls, [
    ["next"],
    ["previous"],
    ["pause"],
    ["resume"],
    ["reload"],
  ]);
});

test("static accepts a LibraryAsset UUID and rejects anything that is not one", () => {
  const harness = connectHarness();
  const valid = "44444444-4444-4444-8444-444444444444";

  harness.client.emit("message", "announcements/static", Buffer.from(` ${valid} `));
  for (const payload of [
    "not-a-uuid",
    "https://example.com/image.png",
    "private/display/asset.webp",
    "bulletin.png",
    "../../etc/passwd",
    "",
  ]) {
    harness.client.emit("message", "announcements/static", Buffer.from(payload));
  }

  assert.deepEqual(
    harness.playbackService.calls,
    [["enterStatic", valid]],
    "only a stable asset UUID may reach the playback service",
  );
});

test("a static command the playback service rejects leaves playback untouched", () => {
  const harness = connectHarness();
  const publishedBefore = harness.client.published.length;

  for (const assetId of [
    "11111111-1111-4111-8111-111111111111",
    "22222222-2222-4222-8222-222222222222",
    "33333333-3333-4333-8333-333333333333",
  ]) {
    harness.client.emit("message", "announcements/static", Buffer.from(assetId));
  }

  // The rejection is logged; nothing is published and the output never goes
  // black merely because an operator sent a bad asset ID.
  assert.equal(harness.client.published.length, publishedBefore);
  assert.equal(harness.service.status().connected, true);
});

test("authoritative state is retained on state topics and never echoed to commands", () => {
  const harness = connectHarness();

  harness.service.publishState({
    mode: "off",
    transport: { rotationPaused: true },
    activePlaylistId: "playlist-b",
    currentEntryId: "entry-b",
    normalTarget: { assetId: "asset-b" },
    staticTarget: null,
    playbackRevision: 12,
  });

  const modeState = harness.client.published.filter(
    (message) => message.topic === "announcements/mode/state",
  );
  assert.equal(modeState.at(-1).payload, "off");
  assert.equal(modeState.at(-1).options.retain, true);
  assert.equal(modeState.at(-1).options.qos, 1);

  const compact = JSON.parse(
    harness.client.published
      .filter((message) => message.topic === "announcements/playback/state")
      .at(-1).payload,
  );
  assert.deepEqual(compact, {
    version: 1,
    mode: "off",
    transport: "paused",
    activePlaylistId: "playlist-b",
    currentEntryId: "entry-b",
    currentAssetId: "asset-b",
    staticAssetId: null,
    revision: 12,
  });

  const commandTopics = new Set(harness.service.commandTopics);
  assert.ok(
    harness.client.published.every((message) => !commandTopics.has(message.topic)),
    "the server must never publish onto a command topic",
  );
});

test("the compact state topic never carries the manifest or private storage details", () => {
  const harness = connectHarness();
  harness.service.publishState({
    mode: "on",
    transport: { rotationPaused: false },
    activePlaylistId: "playlist-a",
    currentEntryId: "entry-a",
    normalTarget: {
      assetId: "asset-a",
      asset: { media: { url: "/media/assets/a/display/abc.webp" } },
    },
    staticTarget: null,
    playbackRevision: 3,
    manifest: { entries: [{ id: "entry-a" }, { id: "entry-b" }] },
  });

  const payload = harness.client.published
    .filter((message) => message.topic === "announcements/playback/state")
    .at(-1).payload;
  assert.ok(payload.length < 400, "the compact topic stays small");
  assert.doesNotMatch(payload, /manifest|media|object_key|bucket|private/i);
});

test("reconnecting resubscribes and republishes availability and authoritative state", () => {
  const harness = connectHarness();
  harness.client.published.length = 0;
  harness.client.subscriptions.length = 0;

  harness.client.emit("close");
  assert.equal(harness.service.status().connected, false);
  assert.equal(harness.service.status().status, "disconnected");

  harness.client.emit("connect");

  assert.equal(harness.client.subscriptions.length, COMMANDS.length);
  assert.equal(
    harness.client.published.find(
      (message) => message.topic === "announcements/availability",
    ).payload,
    "online",
  );
  assert.equal(
    harness.client.published.find(
      (message) => message.topic === "announcements/mode/state",
    ).payload,
    "on",
  );
  assert.equal(harness.service.status().connectionCount, 2);
});

test("a broker error degrades the integration without throwing or leaking the password", () => {
  const harness = connectHarness();

  assert.doesNotThrow(() =>
    harness.client.emit("error", new Error("connect ECONNREFUSED 10.0.0.5:1883")),
  );

  const status = harness.service.status();
  assert.equal(status.connected, false);
  assert.equal(status.status, "disconnected");
  assert.match(status.lastErrorMessage, /ECONNREFUSED/);
  assert.doesNotMatch(JSON.stringify(status), /must-not-leak/);
});

test("an unconfigured broker leaves the integration disabled and inert", async () => {
  const harness = createHarness({ url: null });

  assert.equal(harness.service.start(), null);
  assert.equal(harness.service.status().status, "disabled");
  assert.equal(harness.service.status().configured, false);
  assert.doesNotThrow(() => harness.service.publishState({ mode: "on" }));
  await harness.service.stop();
  assert.equal(harness.client.published.length, 0);
});

test("messages on unknown topics are ignored without crashing", () => {
  const harness = connectHarness();
  assert.doesNotThrow(() => {
    harness.client.emit("message", "announcements/unknown", Buffer.from("x"));
    harness.client.emit("message", "other/mode", Buffer.from("on"));
  });
  assert.deepEqual(harness.playbackService.calls, []);
});

test("a clean shutdown retracts availability and unsubscribes every command", async () => {
  const harness = connectHarness();
  await harness.service.stop();

  const availability = harness.client.published
    .filter((message) => message.topic === "announcements/availability")
    .at(-1);
  assert.equal(availability.payload, "offline");
  assert.equal(availability.options.retain, true);
  assert.equal(harness.client.unsubscribed.length, COMMANDS.length);
  assert.equal(harness.client.ended, true);
});