feat: add ioBroker REST API integration

- New iobroker_sources + iobroker_states tables
- Source manager with discover, test, CRUD
- State poller (10s interval, getBulk endpoint)
- REST API routes under /api/iobroker
- Frontend page with search + pagination
- Sidebar link, app route

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

display/middleware/src/api/index.ts

@@ -6,6 +6,7 @@ import displayApiRouter from './routes/display-api';
 import mqttBrokersRouter from './routes/mqtt-brokers';
 import mqttTopicsRouter from './routes/mqtt-topics';
 import pgSourcesRouter from './routes/pg-sources';
+import iobrokerRouter from './routes/iobroker';
 import datapointsRouter from './routes/datapoints';
 import displaysRouter from './routes/displays';
 import layoutsRouter from './routes/layouts';
@@ -30,6 +31,7 @@ export function createApp(): express.Application {
   app.use('/api/mqtt/brokers/:brokerId/topics', authMiddleware, mqttTopicsRouter);
   app.use('/api/mqtt/brokers', authMiddleware, mqttBrokersRouter);
   app.use('/api/pg/sources', authMiddleware, pgSourcesRouter);
+  app.use('/api/iobroker', authMiddleware, iobrokerRouter);
   app.use('/api/datapoints', authMiddleware, datapointsRouter);
   app.use('/api/displays', authMiddleware, displaysRouter);
   app.use('/api/layouts', authMiddleware, layoutsRouter);

display/middleware/src/api/routes/iobroker.ts

@@ -0,0 +1,132 @@
+import { Router } from 'express';
+import {
+  getAllSources,
+  getSourceById,
+  createSource,
+  updateSource,
+  deleteSource,
+  testSource,
+  discoverStates,
+  getStatesForSource,
+  updateState,
+} from '../../modules/iobroker/source-manager';
+import {
+  createDatapointFromIoBroker,
+  getDatapointByIoBrokerStateId,
+} from '../../modules/datapoints/datapoint-manager';
+import { startPolling } from '../../modules/iobroker/state-poller';
+import { getDb } from '../../db/index';
+
+const router = Router();
+
+// GET /api/iobroker/sources
+router.get('/sources', (_req, res): void => {
+  try {
+    res.json(getAllSources());
+  } catch (err) {
+    res.status(500).json({ error: (err as Error).message });
+  }
+});
+
+// POST /api/iobroker/sources
+router.post('/sources', (req, res): void => {
+  try {
+    const { name, host } = req.body;
+    if (!name || !host) {
+      res.status(400).json({ error: 'name and host are required' });
+      return;
+    }
+    res.status(201).json(createSource(req.body));
+  } catch (err) {
+    res.status(500).json({ error: (err as Error).message });
+  }
+});
+
+// PUT /api/iobroker/sources/:id
+router.put('/sources/:id', (req, res): void => {
+  try {
+    const source = updateSource(parseInt(req.params.id), req.body);
+    if (!source) { res.status(404).json({ error: 'Source not found' }); return; }
+    res.json(source);
+  } catch (err) {
+    res.status(500).json({ error: (err as Error).message });
+  }
+});
+
+// DELETE /api/iobroker/sources/:id
+router.delete('/sources/:id', (req, res): void => {
+  try {
+    const deleted = deleteSource(parseInt(req.params.id));
+    if (!deleted) { res.status(404).json({ error: 'Source not found' }); return; }
+    res.status(204).send();
+  } catch (err) {
+    res.status(500).json({ error: (err as Error).message });
+  }
+});
+
+// POST /api/iobroker/sources/:id/test
+router.post('/sources/:id/test', (req, res): void => {
+  const id = parseInt(req.params.id);
+  const source = getSourceById(id);
+  if (!source) { res.status(404).json({ error: 'Source not found' }); return; }
+
+  testSource(source.host, source.port)
+    .then((ok) => res.json({ success: ok }))
+    .catch((err) => res.status(500).json({ error: (err as Error).message }));
+});
+
+// POST /api/iobroker/sources/:id/discover
+router.post('/sources/:id/discover', (req, res): void => {
+  const id = parseInt(req.params.id);
+
+  res.json({ message: 'Discovery started', sourceId: id });
+
+  discoverStates(id).then((count) => {
+    console.log(`ioBroker discovery for source ${id}: found ${count} states`);
+  }).catch((err) => {
+    console.error(`ioBroker discovery error for source ${id}:`, (err as Error).message);
+  });
+});
+
+// GET /api/iobroker/sources/:id/states?search=&selected=1&limit=50&offset=0
+router.get('/sources/:id/states', (req, res): void => {
+  try {
+    const sourceId = parseInt(req.params.id);
+    const search = req.query.search as string | undefined;
+    const selectedOnly = req.query.selected === '1';
+    const limit = parseInt(req.query.limit as string) || 50;
+    const offset = parseInt(req.query.offset as string) || 0;
+    res.json(getStatesForSource(sourceId, { search, selectedOnly, limit, offset }));
+  } catch (err) {
+    res.status(500).json({ error: (err as Error).message });
+  }
+});
+
+// PUT /api/iobroker/states/:id
+router.put('/states/:id', (req, res): void => {
+  try {
+    const stateRowId = parseInt(req.params.id);
+    const stateRow = getDb()
+      .prepare('SELECT * FROM iobroker_states WHERE id = ?')
+      .get(stateRowId) as { id: number; source_id: number; state_id: string; selected: number } | undefined;
+
+    if (!stateRow) { res.status(404).json({ error: 'State not found' }); return; }
+
+    const updated = updateState(stateRowId, req.body);
+
+    // If selecting a state, auto-create a datapoint and ensure poller is running
+    if (req.body.selected === 1 || req.body.selected === true) {
+      const existing = getDatapointByIoBrokerStateId(stateRow.state_id);
+      if (!existing) {
+        createDatapointFromIoBroker(stateRow.source_id, stateRow.state_id, stateRow.state_id);
+      }
+      startPolling(stateRow.source_id);
+    }
+
+    res.json(updated);
+  } catch (err) {
+    res.status(500).json({ error: (err as Error).message });
+  }
+});
+
+export default router;

display/middleware/src/db/index.ts

@@ -1,5 +1,6 @@
 import Database from 'better-sqlite3';
 import { migrate001 } from './migrations/001-initial';
+import { migrate002 } from './migrations/002-iobroker';
 
 let dbInstance: Database.Database | null = null;
 
@@ -11,6 +12,7 @@ export function initDb(path: string): Database.Database {
 
   // Run migrations
   migrate001(db);
+  migrate002(db);
 
   dbInstance = db;
   return db;

display/middleware/src/db/migrations/002-iobroker.ts

@@ -0,0 +1,57 @@
+import Database from 'better-sqlite3';
+
+export function migrate002(db: Database.Database): void {
+  // Create ioBroker sources table
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS iobroker_sources (
+      id INTEGER PRIMARY KEY AUTOINCREMENT,
+      name TEXT NOT NULL,
+      host TEXT NOT NULL,
+      port INTEGER DEFAULT 8087,
+      enabled INTEGER DEFAULT 1,
+      status TEXT DEFAULT 'disconnected',
+      created_at TEXT DEFAULT (datetime('now'))
+    );
+
+    CREATE TABLE IF NOT EXISTS iobroker_states (
+      id INTEGER PRIMARY KEY AUTOINCREMENT,
+      source_id INTEGER NOT NULL REFERENCES iobroker_sources(id) ON DELETE CASCADE,
+      state_id TEXT NOT NULL,
+      selected INTEGER DEFAULT 0,
+      last_value TEXT,
+      last_ts INTEGER,
+      discovered_at TEXT DEFAULT (datetime('now'))
+    );
+  `);
+
+  // Update datapoints CHECK constraint to allow 'iobroker'
+  // SQLite does not support ALTER CHECK, so recreate the table if needed
+  const row = db
+    .prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='datapoints'")
+    .get() as { sql: string } | undefined;
+
+  if (row && !row.sql.includes('iobroker')) {
+    db.exec(`
+      CREATE TABLE datapoints_new (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        name TEXT NOT NULL,
+        label TEXT NOT NULL,
+        unit TEXT,
+        value TEXT,
+        status TEXT DEFAULT 'unknown',
+        source_type TEXT NOT NULL CHECK(source_type IN ('mqtt','postgresql','iobroker')),
+        source_id INTEGER NOT NULL,
+        topic_id INTEGER,
+        query_id INTEGER,
+        transform TEXT,
+        enabled INTEGER DEFAULT 1,
+        updated_at TEXT DEFAULT (datetime('now')),
+        FOREIGN KEY (topic_id) REFERENCES mqtt_topics(id) ON DELETE SET NULL,
+        FOREIGN KEY (query_id) REFERENCES pg_queries(id) ON DELETE SET NULL
+      );
+      INSERT INTO datapoints_new SELECT * FROM datapoints;
+      DROP TABLE datapoints;
+      ALTER TABLE datapoints_new RENAME TO datapoints;
+    `);
+  }
+}

display/middleware/src/index.ts

@@ -4,6 +4,7 @@ import { createApp } from './api/index';
 import { connectAllEnabledBrokers } from './modules/mqtt/broker-manager';
 import { connectAllEnabledSources } from './modules/postgres/source-manager';
 import { startAllQueryTimers } from './modules/postgres/query-scheduler';
+import { startAllPollers } from './modules/iobroker/state-poller';
 
 const PORT = process.env.PORT ? parseInt(process.env.PORT) : 4000;
 const DB_PATH = process.env.DB_PATH ?? './display.db';
@@ -32,6 +33,9 @@ async function main(): Promise<void> {
   await connectAllEnabledSources();
   startAllQueryTimers();
 
+  // Start ioBroker state pollers for all enabled sources
+  startAllPollers();
+
   // Create and start Express app
   const app = createApp();
 

display/middleware/src/modules/datapoints/datapoint-manager.ts

@@ -137,3 +137,24 @@ export function getDatapointByQueryId(queryId: number): Datapoint | undefined {
     .prepare('SELECT * FROM datapoints WHERE query_id = ?')
     .get(queryId) as Datapoint | undefined;
 }
+
+export function createDatapointFromIoBroker(
+  sourceId: number,
+  stateId: string,
+  stateName: string
+): Datapoint {
+  const db = getDb();
+  const result = db.prepare(
+    `INSERT INTO datapoints (name, label, source_type, source_id, enabled)
+     VALUES (?, ?, 'iobroker', ?, 1)`
+  ).run(stateId, stateName, sourceId);
+  const id = result.lastInsertRowid as number;
+  return db.prepare('SELECT * FROM datapoints WHERE id = ?').get(id) as Datapoint;
+}
+
+export function getDatapointByIoBrokerStateId(stateId: string): Datapoint | undefined {
+  const db = getDb();
+  return db
+    .prepare("SELECT * FROM datapoints WHERE source_type = 'iobroker' AND name = ?")
+    .get(stateId) as Datapoint | undefined;
+}

display/middleware/src/modules/iobroker/source-manager.ts

@@ -0,0 +1,187 @@
+import { getDb } from '../../db/index';
+
+export interface IoBrokerSource {
+  id: number;
+  name: string;
+  host: string;
+  port: number;
+  enabled: number;
+  status: string;
+  created_at: string;
+}
+
+export interface CreateSourceData {
+  name: string;
+  host: string;
+  port?: number;
+  enabled?: number;
+}
+
+export type UpdateSourceData = Partial<CreateSourceData>;
+
+export interface IoBrokerState {
+  id: number;
+  source_id: number;
+  state_id: string;
+  selected: number;
+  last_value: string | null;
+  last_ts: number | null;
+  discovered_at: string;
+}
+
+export function getAllSources(): IoBrokerSource[] {
+  const db = getDb();
+  return db.prepare('SELECT * FROM iobroker_sources ORDER BY id ASC').all() as IoBrokerSource[];
+}
+
+export function getSourceById(id: number): IoBrokerSource | undefined {
+  const db = getDb();
+  return db.prepare('SELECT * FROM iobroker_sources WHERE id = ?').get(id) as IoBrokerSource | undefined;
+}
+
+export function createSource(data: CreateSourceData): IoBrokerSource {
+  const db = getDb();
+  const result = db.prepare(
+    `INSERT INTO iobroker_sources (name, host, port, enabled)
+     VALUES (?, ?, ?, ?)`
+  ).run(
+    data.name,
+    data.host,
+    data.port ?? 8087,
+    data.enabled ?? 1,
+  );
+  const id = result.lastInsertRowid as number;
+  return db.prepare('SELECT * FROM iobroker_sources WHERE id = ?').get(id) as IoBrokerSource;
+}
+
+export function updateSource(id: number, data: UpdateSourceData): IoBrokerSource | 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.enabled !== undefined) { fields.push('enabled = ?'); values.push(data.enabled); }
+
+  if (fields.length === 0) return getSourceById(id);
+
+  values.push(id);
+  db.prepare(`UPDATE iobroker_sources SET ${fields.join(', ')} WHERE id = ?`).run(...values);
+  return getSourceById(id);
+}
+
+export function deleteSource(id: number): boolean {
+  const db = getDb();
+  const result = db.prepare('DELETE FROM iobroker_sources WHERE id = ?').run(id);
+  return (result.changes as number) > 0;
+}
+
+function setStatus(id: number, status: string): void {
+  const db = getDb();
+  db.prepare('UPDATE iobroker_sources SET status = ? WHERE id = ?').run(status, id);
+}
+
+export async function testSource(host: string, port: number): Promise<boolean> {
+  try {
+    const url = `http://${host}:${port}/get/system.adapter.admin.0.alive`;
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(), 5000);
+    const res = await fetch(url, { signal: controller.signal });
+    clearTimeout(timeout);
+    return res.ok;
+  } catch {
+    return false;
+  }
+}
+
+export function getStatesForSource(
+  sourceId: number,
+  opts?: { search?: string; selectedOnly?: boolean; limit?: number; offset?: number }
+): { states: IoBrokerState[]; total: number } {
+  const db = getDb();
+  const conditions = ['source_id = ?'];
+  const params: unknown[] = [sourceId];
+
+  if (opts?.search) {
+    conditions.push('state_id LIKE ?');
+    params.push(`%${opts.search}%`);
+  }
+  if (opts?.selectedOnly) {
+    conditions.push('selected = 1');
+  }
+
+  const where = conditions.join(' AND ');
+  const total = (db.prepare(`SELECT count(*) as c FROM iobroker_states WHERE ${where}`).get(...params) as { c: number }).c;
+
+  const limit = opts?.limit ?? 50;
+  const offset = opts?.offset ?? 0;
+  params.push(limit, offset);
+
+  const states = db
+    .prepare(`SELECT * FROM iobroker_states WHERE ${where} ORDER BY selected DESC, state_id ASC LIMIT ? OFFSET ?`)
+    .all(...params) as IoBrokerState[];
+
+  return { states, total };
+}
+
+export function updateState(id: number, data: { selected?: number; last_value?: string; last_ts?: number }): IoBrokerState | undefined {
+  const db = getDb();
+  const fields: string[] = [];
+  const values: unknown[] = [];
+
+  if (data.selected !== undefined) { fields.push('selected = ?'); values.push(data.selected); }
+  if (data.last_value !== undefined) { fields.push('last_value = ?'); values.push(data.last_value); }
+  if (data.last_ts !== undefined) { fields.push('last_ts = ?'); values.push(data.last_ts); }
+
+  if (fields.length === 0) {
+    return db.prepare('SELECT * FROM iobroker_states WHERE id = ?').get(id) as IoBrokerState | undefined;
+  }
+
+  values.push(id);
+  db.prepare(`UPDATE iobroker_states SET ${fields.join(', ')} WHERE id = ?`).run(...values);
+  return db.prepare('SELECT * FROM iobroker_states WHERE id = ?').get(id) as IoBrokerState | undefined;
+}
+
+export async function discoverStates(sourceId: number): Promise<number> {
+  const source = getSourceById(sourceId);
+  if (!source) throw new Error(`ioBroker source ${sourceId} not found`);
+
+  const db = getDb();
+
+  try {
+    setStatus(sourceId, 'discovering');
+
+    const url = `http://${source.host}:${source.port}/objects?type=state&pattern=*`;
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(), 15000);
+    const res = await fetch(url, { signal: controller.signal });
+    clearTimeout(timeout);
+
+    if (!res.ok) {
+      setStatus(sourceId, 'error');
+      throw new Error(`ioBroker API returned ${res.status}`);
+    }
+
+    const data = await res.json() as Record<string, unknown>;
+    const stateIds = Object.keys(data).slice(0, 5000);
+
+    const insertStmt = db.prepare(
+      `INSERT OR IGNORE INTO iobroker_states (source_id, state_id) VALUES (?, ?)`
+    );
+
+    const insertMany = db.transaction((ids: string[]) => {
+      for (const sid of ids) {
+        insertStmt.run(sourceId, sid);
+      }
+    });
+
+    insertMany(stateIds);
+    setStatus(sourceId, 'connected');
+
+    return stateIds.length;
+  } catch (err) {
+    setStatus(sourceId, 'error');
+    throw err;
+  }
+}

display/middleware/src/modules/iobroker/state-poller.ts

@@ -0,0 +1,86 @@
+import { getDb } from '../../db/index';
+import { getAllSources, getSourceById, type IoBrokerState } from './source-manager';
+import {
+  getDatapointByIoBrokerStateId,
+  updateDatapointValue,
+} from '../datapoints/datapoint-manager';
+
+const pollerTimers = new Map<number, ReturnType<typeof setInterval>>();
+
+const POLL_INTERVAL_MS = 10_000;
+
+async function pollSource(sourceId: number): Promise<void> {
+  const source = getSourceById(sourceId);
+  if (!source || !source.enabled) return;
+
+  const db = getDb();
+  const selectedStates = db
+    .prepare('SELECT * FROM iobroker_states WHERE source_id = ? AND selected = 1')
+    .all(sourceId) as IoBrokerState[];
+
+  if (selectedStates.length === 0) return;
+
+  const stateIds = selectedStates.map((s) => s.state_id).join(',');
+  const url = `http://${source.host}:${source.port}/getBulk/${encodeURIComponent(stateIds)}`;
+
+  try {
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(), 8000);
+    const res = await fetch(url, { signal: controller.signal });
+    clearTimeout(timeout);
+
+    if (!res.ok) return;
+
+    const results = await res.json() as Array<{ id: string; val: unknown; ts?: number }>;
+
+    for (const item of results) {
+      if (item.val === null || item.val === undefined) continue;
+      const rawValue = String(item.val);
+
+      // Update last_value in iobroker_states
+      db.prepare(
+        'UPDATE iobroker_states SET last_value = ?, last_ts = ? WHERE source_id = ? AND state_id = ?'
+      ).run(rawValue, item.ts ?? null, sourceId, item.id);
+
+      // Update datapoint value if one exists
+      const dp = getDatapointByIoBrokerStateId(item.id);
+      if (dp) {
+        updateDatapointValue(dp.id, rawValue);
+      }
+    }
+  } catch (err) {
+    // Silently ignore polling errors — source may be temporarily unreachable
+    console.warn(`ioBroker poll error for source ${sourceId}:`, (err as Error).message);
+  }
+}
+
+export function startPolling(sourceId: number): void {
+  if (pollerTimers.has(sourceId)) return; // already running
+
+  const timer = setInterval(() => {
+    pollSource(sourceId).catch((err) =>
+      console.error(`ioBroker poller ${sourceId}:`, (err as Error).message)
+    );
+  }, POLL_INTERVAL_MS);
+
+  pollerTimers.set(sourceId, timer);
+  console.log(`ioBroker poller started for source ${sourceId}`);
+}
+
+export function stopPolling(sourceId: number): void {
+  const timer = pollerTimers.get(sourceId);
+  if (timer) {
+    clearInterval(timer);
+    pollerTimers.delete(sourceId);
+    console.log(`ioBroker poller stopped for source ${sourceId}`);
+  }
+}
+
+export function startAllPollers(): void {
+  const sources = getAllSources();
+  for (const source of sources) {
+    if (source.enabled) {
+      startPolling(source.id);
+    }
+  }
+}