From 2c9d218211264ece72a4c226f57d885024082359 Mon Sep 17 00:00:00 2001
From: Christian Mueller <christian@cynfo.com>
Date: Mon, 6 Apr 2026 12:56:22 +0200
Subject: [PATCH] feat(mqtt): add topic discovery with 30s wildcard subscribe
 (Task 10)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../src/modules/mqtt/topic-discovery.ts       | 169 ++++++++++++++++++
 1 file changed, 169 insertions(+)
 create mode 100644 display/middleware/src/modules/mqtt/topic-discovery.ts

diff --git a/display/middleware/src/modules/mqtt/topic-discovery.ts b/display/middleware/src/modules/mqtt/topic-discovery.ts
new file mode 100644
index 0000000..7603374
--- /dev/null
+++ b/display/middleware/src/modules/mqtt/topic-discovery.ts
@@ -0,0 +1,169 @@
+import * as mqtt from 'mqtt';
+import { getDb } from '../../db/index';
+import { getConnection } from './broker-manager';
+
+export interface MqttTopic {
+  id: number;
+  broker_id: number;
+  topic: string;
+  json_path: string | null;
+  selected: number;
+  ignored: number;
+  last_payload: string | null;
+  discovered_at: string;
+}
+
+export interface UpdateTopicData {
+  json_path?: string | null;
+  selected?: number;
+  ignored?: number;
+}
+
+export function getTopicsForBroker(brokerId: number): MqttTopic[] {
+  const db = getDb();
+  return db
+    .prepare('SELECT * FROM mqtt_topics WHERE broker_id = ? ORDER BY id ASC')
+    .all(brokerId) as MqttTopic[];
+}
+
+export function updateTopic(id: number, data: UpdateTopicData): MqttTopic | undefined {
+  const db = getDb();
+  const fields: string[] = [];
+  const values: unknown[] = [];
+
+  if (data.json_path !== undefined) { fields.push('json_path = ?'); values.push(data.json_path); }
+  if (data.selected !== undefined) { fields.push('selected = ?'); values.push(data.selected); }
+  if (data.ignored !== undefined) { fields.push('ignored = ?'); values.push(data.ignored); }
+
+  if (fields.length === 0) {
+    return db.prepare('SELECT * FROM mqtt_topics WHERE id = ?').get(id) as MqttTopic | undefined;
+  }
+
+  values.push(id);
+  db.run(`UPDATE mqtt_topics SET ${fields.join(', ')} WHERE id = ?`, values);
+  return db.prepare('SELECT * FROM mqtt_topics WHERE id = ?').get(id) as MqttTopic | undefined;
+}
+
+function ensureTopicExists(brokerId: number, topic: string): void {
+  const db = getDb();
+  const existing = db
+    .prepare('SELECT id FROM mqtt_topics WHERE broker_id = ? AND topic = ?')
+    .get(brokerId, topic);
+
+  if (!existing) {
+    db.run(
+      `INSERT INTO mqtt_topics (broker_id, topic, selected, ignored)
+       VALUES (?, ?, 0, 0)`,
+      [brokerId, topic]
+    );
+  }
+}
+
+export function startDiscovery(brokerId: number, prefix?: string): Promise<MqttTopic[]> {
+  return new Promise((resolve) => {
+    const db = getDb();
+    const broker = db
+      .prepare('SELECT * FROM mqtt_brokers WHERE id = ?')
+      .get(brokerId) as {
+        id: number;
+        host: string;
+        port: number;
+        username: string | null;
+        password: string | null;
+        use_tls: number;
+      } | undefined;
+
+    if (!broker) {
+      resolve([]);
+      return;
+    }
+
+    // Try to use existing connection first, otherwise create a temporary one
+    const existingClient = getConnection(brokerId);
+    let tempClient: mqtt.MqttClient | null = null;
+    let client: mqtt.MqttClient;
+
+    const subscribePattern = prefix ? `${prefix}#` : '#';
+    const discoveredTopics: Set<string> = new Set();
+
+    const finalize = (): void => {
+      if (tempClient) {
+        tempClient.end(true);
+      }
+      resolve(getTopicsForBroker(brokerId));
+    };
+
+    const onMessage = (topic: string): void => {
+      if (!discoveredTopics.has(topic)) {
+        discoveredTopics.add(topic);
+        ensureTopicExists(brokerId, topic);
+      }
+    };
+
+    if (existingClient && existingClient.connected) {
+      client = existingClient;
+      client.subscribe(subscribePattern, (err) => {
+        if (err) {
+          resolve(getTopicsForBroker(brokerId));
+          return;
+        }
+        client.on('message', onMessage);
+        setTimeout(() => {
+          client.unsubscribe(subscribePattern);
+          client.removeListener('message', onMessage);
+          finalize();
+        }, 30000);
+      });
+    } else {
+      const protocol = broker.use_tls ? 'mqtts' : 'mqtt';
+      const url = `${protocol}://${broker.host}:${broker.port}`;
+
+      const options: mqtt.IClientOptions = {
+        clientId: `display-discovery-${brokerId}-${Date.now()}`,
+        connectTimeout: 10000,
+        reconnectPeriod: 0,
+      };
+      if (broker.username) options.username = broker.username;
+      if (broker.password) options.password = broker.password;
+
+      tempClient = mqtt.connect(url, options);
+      client = tempClient;
+
+      tempClient.on('connect', () => {
+        client.subscribe(subscribePattern, (err) => {
+          if (err) {
+            finalize();
+            return;
+          }
+          client.on('message', onMessage);
+          setTimeout(() => {
+            finalize();
+          }, 30000);
+        });
+      });
+
+      tempClient.on('error', () => {
+        finalize();
+      });
+    }
+  });
+}
+
+export function addManualTopic(brokerId: number, topic: string): MqttTopic {
+  const db = getDb();
+
+  // Insert or ignore duplicate
+  const existing = db
+    .prepare('SELECT * FROM mqtt_topics WHERE broker_id = ? AND topic = ?')
+    .get(brokerId, topic) as MqttTopic | undefined;
+
+  if (existing) return existing;
+
+  const result = db.run(
+    `INSERT INTO mqtt_topics (broker_id, topic, selected, ignored)
+     VALUES (?, ?, 0, 0)`,
+    [brokerId, topic]
+  );
+  const id = result.lastInsertRowid as number;
+  return db.prepare('SELECT * FROM mqtt_topics WHERE id = ?').get(id) as MqttTopic;
+}