diff --git a/display/middleware/src/modules/mqtt/broker-manager.ts b/display/middleware/src/modules/mqtt/broker-manager.ts
new file mode 100644
index 0000000..fc4d368
--- /dev/null
+++ b/display/middleware/src/modules/mqtt/broker-manager.ts
@@ -0,0 +1,287 @@
+import * as mqtt from 'mqtt';
+import { getDb } from '../../db/index';
+import { extractValue } from './payload-parser';
+import {
+  getDatapointByTopicId,
+  updateDatapointValue,
+} from '../datapoints/datapoint-manager';
+
+export interface MqttBroker {
+  id: number;
+  name: string;
+  host: string;
+  port: number;
+  username: string | null;
+  password: string | null;
+  use_tls: number;
+  enabled: number;
+  status: string;
+  created_at: string;
+  updated_at: string;
+}
+
+export interface CreateBrokerData {
+  name: string;
+  host: string;
+  port?: number;
+  username?: string;
+  password?: string;
+  use_tls?: number;
+  enabled?: number;
+}
+
+export type UpdateBrokerData = Partial<CreateBrokerData>;
+
+interface MqttTopic {
+  id: number;
+  broker_id: number;
+  topic: string;
+  json_path: string | null;
+  selected: number;
+  ignored: number;
+}
+
+const connections = new Map<number, mqtt.MqttClient>();
+
+export function getAllBrokers(): MqttBroker[] {
+  const db = getDb();
+  return db.prepare('SELECT * FROM mqtt_brokers ORDER BY id ASC').all() as MqttBroker[];
+}
+
+export function getBrokerById(id: number): MqttBroker | undefined {
+  const db = getDb();
+  return db.prepare('SELECT * FROM mqtt_brokers WHERE id = ?').get(id) as MqttBroker | undefined;
+}
+
+export function createBroker(data: CreateBrokerData): MqttBroker {
+  const db = getDb();
+  const result = db.run(
+    `INSERT INTO mqtt_brokers (name, host, port, username, password, use_tls, enabled)
+     VALUES (?, ?, ?, ?, ?, ?, ?)`,
+    [
+      data.name,
+      data.host,
+      data.port ?? 1883,
+      data.username ?? null,
+      data.password ?? null,
+      data.use_tls ?? 0,
+      data.enabled ?? 1,
+    ]
+  );
+  const id = result.lastInsertRowid as number;
+  return db.prepare('SELECT * FROM mqtt_brokers WHERE id = ?').get(id) as MqttBroker;
+}
+
+export function updateBroker(id: number, data: UpdateBrokerData): MqttBroker | undefined {
+  const db = getDb();
+  const fields: string[] = [];
+  const values: unknown[] = [];
+
+  if (data.name !== undefined) { fields.push('name = ?'); values.push(data.name); }
+  if (data.host !== undefined) { fields.push('host = ?'); values.push(data.host); }
+  if (data.port !== undefined) { fields.push('port = ?'); values.push(data.port); }
+  if (data.username !== undefined) { fields.push('username = ?'); values.push(data.username); }
+  if (data.password !== undefined) { fields.push('password = ?'); values.push(data.password); }
+  if (data.use_tls !== undefined) { fields.push('use_tls = ?'); values.push(data.use_tls); }
+  if (data.enabled !== undefined) { fields.push('enabled = ?'); values.push(data.enabled); }
+
+  if (fields.length === 0) return getBrokerById(id);
+
+  fields.push("updated_at = datetime('now')");
+  values.push(id);
+  db.run(`UPDATE mqtt_brokers SET ${fields.join(', ')} WHERE id = ?`, values);
+  return getBrokerById(id);
+}
+
+export function deleteBroker(id: number): boolean {
+  const db = getDb();
+  disconnectBroker(id);
+  const result = db.run('DELETE FROM mqtt_brokers WHERE id = ?', [id]);
+  return (result.changes as number) > 0;
+}
+
+function buildConnectUrl(broker: MqttBroker): string {
+  const protocol = broker.use_tls ? 'mqtts' : 'mqtt';
+  return `${protocol}://${broker.host}:${broker.port}`;
+}
+
+function setStatus(id: number, status: string): void {
+  const db = getDb();
+  db.run(
+    `UPDATE mqtt_brokers SET status = ?, updated_at = datetime('now') WHERE id = ?`,
+    [status, id]
+  );
+}
+
+export function connectBroker(id: number): Promise<mqtt.MqttClient> {
+  return new Promise((resolve, reject) => {
+    const broker = getBrokerById(id);
+    if (!broker) {
+      reject(new Error(`Broker ${id} not found`));
+      return;
+    }
+
+    // Disconnect existing connection if any
+    if (connections.has(id)) {
+      connections.get(id)!.end(true);
+      connections.delete(id);
+    }
+
+    const url = buildConnectUrl(broker);
+    const options: mqtt.IClientOptions = {
+      clientId: `display-middleware-${id}-${Date.now()}`,
+      reconnectPeriod: 5000,
+      connectTimeout: 10000,
+    };
+
+    if (broker.username) options.username = broker.username;
+    if (broker.password) options.password = broker.password;
+
+    const client = mqtt.connect(url, options);
+
+    client.on('connect', () => {
+      setStatus(id, 'connected');
+      connections.set(id, client);
+
+      // Subscribe to all selected topics for this broker
+      const db = getDb();
+      const topics = db
+        .prepare(
+          'SELECT * FROM mqtt_topics WHERE broker_id = ? AND selected = 1 AND ignored = 0'
+        )
+        .all(id) as MqttTopic[];
+
+      for (const t of topics) {
+        client.subscribe(t.topic, (err) => {
+          if (err) {
+            console.error(`Failed to subscribe to ${t.topic}:`, err.message);
+          }
+        });
+      }
+
+      resolve(client);
+    });
+
+    client.on('message', (topic: string, payloadBuf: Buffer) => {
+      const db = getDb();
+      const payload = payloadBuf.toString();
+
+      // Find the topic record
+      const topicRecord = db
+        .prepare(
+          'SELECT * FROM mqtt_topics WHERE broker_id = ? AND topic = ?'
+        )
+        .get(id, topic) as MqttTopic | undefined;
+
+      if (!topicRecord) return;
+
+      // Update last_payload
+      db.run(
+        'UPDATE mqtt_topics SET last_payload = ? WHERE id = ?',
+        [payload, topicRecord.id]
+      );
+
+      // Extract value and update datapoint
+      const extracted = extractValue(payload, topicRecord.json_path);
+      if (extracted === null) return;
+
+      const dp = getDatapointByTopicId(topicRecord.id);
+      if (dp) {
+        updateDatapointValue(dp.id, extracted);
+      }
+    });
+
+    client.on('error', (err: Error) => {
+      setStatus(id, 'error');
+      console.error(`MQTT broker ${id} error:`, err.message);
+    });
+
+    client.on('offline', () => {
+      setStatus(id, 'disconnected');
+    });
+
+    client.on('reconnect', () => {
+      setStatus(id, 'reconnecting');
+    });
+
+    // Reject if connection fails within timeout
+    const timer = setTimeout(() => {
+      if (!connections.has(id)) {
+        client.end(true);
+        setStatus(id, 'error');
+        reject(new Error(`Connection to broker ${id} timed out`));
+      }
+    }, 15000);
+
+    client.on('connect', () => clearTimeout(timer));
+  });
+}
+
+export function disconnectBroker(id: number): void {
+  const client = connections.get(id);
+  if (client) {
+    client.end(true);
+    connections.delete(id);
+    setStatus(id, 'disconnected');
+  }
+}
+
+export function testBroker(data: {
+  host: string;
+  port?: number;
+  username?: string;
+  password?: string;
+  use_tls?: number;
+}): Promise<boolean> {
+  return new Promise((resolve) => {
+    const protocol = data.use_tls ? 'mqtts' : 'mqtt';
+    const port = data.port ?? 1883;
+    const url = `${protocol}://${data.host}:${port}`;
+
+    const options: mqtt.IClientOptions = {
+      clientId: `display-test-${Date.now()}`,
+      connectTimeout: 5000,
+      reconnectPeriod: 0,
+    };
+    if (data.username) options.username = data.username;
+    if (data.password) options.password = data.password;
+
+    const client = mqtt.connect(url, options);
+
+    const timer = setTimeout(() => {
+      client.end(true);
+      resolve(false);
+    }, 5000);
+
+    client.on('connect', () => {
+      clearTimeout(timer);
+      client.end(true);
+      resolve(true);
+    });
+
+    client.on('error', () => {
+      clearTimeout(timer);
+      client.end(true);
+      resolve(false);
+    });
+  });
+}
+
+export async function connectAllEnabledBrokers(): Promise<void> {
+  const db = getDb();
+  const brokers = db
+    .prepare('SELECT * FROM mqtt_brokers WHERE enabled = 1')
+    .all() as MqttBroker[];
+
+  for (const broker of brokers) {
+    try {
+      await connectBroker(broker.id);
+    } catch (err) {
+      console.error(`Failed to connect broker ${broker.id}:`, (err as Error).message);
+    }
+  }
+}
+
+export function getConnection(id: number): mqtt.MqttClient | undefined {
+  return connections.get(id);
+}