Scaffold lights app: Tasmota MQTT control, schedules with sunrise/sunset triggers

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

backend/src/auth.js

@@ -0,0 +1,57 @@
+import { jwtVerify } from 'jose'
+import { isOnsite } from './ip-check.js'
+
+const APP_SLUG = process.env.APP_SLUG || 'lights'
+const secret = new TextEncoder().encode(process.env.CENTRAL_AUTH_SECRET || '')
+
+export async function requireAuth(request, reply) {
+  const token = request.cookies?.hnf_session
+  if (!token) return reply.status(401).send({ error: 'Not authenticated' })
+
+  let payload
+  try {
+    const { payload: p } = await jwtVerify(token, secret)
+    payload = p
+  } catch {
+    return reply.status(401).send({ error: 'Invalid session' })
+  }
+
+  if (!payload.apps?.includes(APP_SLUG)) {
+    return reply.status(403).send({ error: 'No permission for this app' })
+  }
+
+  if (!payload.offsite_allowed) {
+    const clientIP = request.headers['x-real-ip'] || request.ip
+    if (!(await isOnsite(clientIP))) {
+      return reply.status(403).send({ error: 'Access restricted to site network' })
+    }
+  }
+
+  const prefix = `${APP_SLUG}:`
+  let caps
+  if (Array.isArray(payload.caps)) {
+    caps = payload.caps.filter(c => c.startsWith(prefix)).map(c => c.slice(prefix.length))
+  } else {
+    // Legacy token — grant all non-settings caps until re-login
+    caps = ['view', 'control', 'schedules', 'manage_devices']
+  }
+
+  request.user = {
+    email: payload.sub,
+    name: payload.name,
+    is_admin: payload.is_admin ?? false,
+    caps,
+  }
+}
+
+export function hasCap(request, cap) {
+  return request.user?.is_admin === true || request.user?.caps?.includes(cap) === true
+}
+
+export function requireCap(cap) {
+  return async (request, reply) => {
+    if (!hasCap(request, cap)) {
+      return reply.status(403).send({ error: `Missing capability: ${cap}` })
+    }
+  }
+}

backend/src/db.js

@@ -0,0 +1,83 @@
+import pg from 'pg'
+
+const { Pool } = pg
+export const pool = new Pool({ connectionString: process.env.DATABASE_URL })
+
+export async function initDb() {
+  await pool.query(`
+    -- One row per controllable relay. A multi-relay Tasmota device is several
+    -- rows sharing a mqtt_topic with different channels. mqtt_topic is the
+    -- device's Tasmota "Topic" (the %topic% in cmnd/%topic%/POWER).
+    CREATE TABLE IF NOT EXISTS light_devices (
+      id             SERIAL PRIMARY KEY,
+      name           TEXT NOT NULL,
+      location       TEXT,
+      mqtt_topic     TEXT NOT NULL,
+      channel        INT NOT NULL DEFAULT 1,
+      active         BOOLEAN NOT NULL DEFAULT TRUE,
+      state          TEXT,                 -- 'on' | 'off' | NULL (unknown)
+      online         BOOLEAN,              -- from the Tasmota LWT; NULL until first seen
+      state_at       TIMESTAMPTZ,
+      last_manual_at TIMESTAMPTZ,          -- last manual/physical change; a schedule trigger older than this does not override it
+      created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      UNIQUE (mqtt_topic, channel)
+    );
+
+    CREATE TABLE IF NOT EXISTS light_groups (
+      id         SERIAL PRIMARY KEY,
+      name       TEXT NOT NULL UNIQUE,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    );
+    CREATE TABLE IF NOT EXISTS light_group_members (
+      group_id  INT NOT NULL REFERENCES light_groups(id) ON DELETE CASCADE,
+      device_id INT NOT NULL REFERENCES light_devices(id) ON DELETE CASCADE,
+      PRIMARY KEY (group_id, device_id)
+    );
+
+    -- target_type/target_id point at a device or group (no FK — the delete
+    -- routes remove a target's schedules explicitly). days uses 0=Sun..6=Sat in
+    -- site-local time. fixed schedules use time_of_day; sunrise/sunset use
+    -- offset_minutes (negative = before). last_fired_on is the site-local date
+    -- (YYYY-MM-DD) it last ran, so a restart never double-fires.
+    CREATE TABLE IF NOT EXISTS light_schedules (
+      id             SERIAL PRIMARY KEY,
+      name           TEXT NOT NULL,
+      enabled        BOOLEAN NOT NULL DEFAULT TRUE,
+      target_type    TEXT NOT NULL,        -- device | group
+      target_id      INT NOT NULL,
+      action         TEXT NOT NULL,        -- on | off
+      trigger_type   TEXT NOT NULL,        -- fixed | sunrise | sunset
+      time_of_day    TEXT,                 -- 'HH:MM' when fixed
+      offset_minutes INT NOT NULL DEFAULT 0,
+      days           INT[] NOT NULL DEFAULT '{0,1,2,3,4,5,6}',
+      last_fired_on  TEXT,
+      created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    );
+
+    CREATE TABLE IF NOT EXISTS light_events (
+      id          SERIAL PRIMARY KEY,
+      at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      device_id   INT,
+      device_name TEXT NOT NULL,
+      action      TEXT NOT NULL,           -- on | off
+      source      TEXT NOT NULL,           -- manual | schedule | reconcile | physical
+      detail      TEXT                     -- user email or schedule name
+    );
+    CREATE INDEX IF NOT EXISTS light_events_at_idx ON light_events (at DESC);
+
+    CREATE TABLE IF NOT EXISTS app_settings (
+      key        TEXT PRIMARY KEY,
+      value      TEXT,
+      updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    );
+  `)
+
+  for (const [key, value] of [['latitude', ''], ['longitude', ''], ['timezone', 'Europe/London']]) {
+    await pool.query(`INSERT INTO app_settings (key, value) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, [key, value])
+  }
+}
+
+export async function getSettings() {
+  const { rows } = await pool.query('SELECT key, value FROM app_settings')
+  return Object.fromEntries(rows.map(r => [r.key, r.value]))
+}

backend/src/index.js

@@ -0,0 +1,38 @@
+import Fastify from 'fastify'
+import cookie from '@fastify/cookie'
+import cors from '@fastify/cors'
+import { initDb } from './db.js'
+import { connect as connectMqtt } from './lib/mqtt.js'
+import { startEngine } from './lib/engine.js'
+import { deviceRoutes } from './routes/devices.js'
+import { scheduleRoutes } from './routes/schedules.js'
+import { settingsRoutes } from './routes/settings.js'
+
+const app = Fastify({ logger: true, trustProxy: true })
+const startedAt = Date.now()
+
+await app.register(cookie)
+await app.register(cors, {
+  origin: process.env.CORS_ORIGIN || false,
+  credentials: true,
+})
+
+app.get('/health', async () => ({ status: 'healthy', version: process.env.BUILD_VERSION || String(startedAt) }))
+
+await app.register(deviceRoutes)
+await app.register(scheduleRoutes)
+await app.register(settingsRoutes)
+
+try {
+  await initDb()
+
+  // The broker may be unreachable at boot — connect() handles that with its own
+  // backoff and never blocks startup.
+  connectMqtt().catch(err => app.log.warn(`MQTT connect failed at startup: ${err.message}`))
+
+  startEngine()
+  await app.listen({ port: 3001, host: '0.0.0.0' })
+} catch (err) {
+  app.log.error(err)
+  process.exit(1)
+}

backend/src/ip-check.js

@@ -0,0 +1,80 @@
+import dns from 'dns/promises'
+
+const raw = (process.env.OFFICE_IP_CHECK || 'disabled').trim()
+const matchers = raw.split(',').map(s => s.trim()).filter(Boolean)
+
+const TTL = 5 * 60 * 1000
+const cache = new Map()
+
+const PUBLIC_IP_URLS = [
+  'https://api.ipify.org',
+  'https://ifconfig.co/ip',
+  'https://icanhazip.com',
+]
+
+function normalizeIP(ip) {
+  return ip?.startsWith('::ffff:') ? ip.slice(7) : ip
+}
+
+function isIPv4(s) {
+  return /^\d{1,3}(\.\d{1,3}){3}$/.test(s)
+}
+
+function ipInCidr(ip, cidr) {
+  const [range, bits] = cidr.split('/')
+  if (!isIPv4(ip) || !isIPv4(range)) return false
+  const mask = ~(2 ** (32 - parseInt(bits)) - 1) >>> 0
+  const toInt = s => s.split('.').reduce((a, o) => (a << 8) + parseInt(o), 0) >>> 0
+  return (toInt(ip) & mask) === (toInt(range) & mask)
+}
+
+async function fetchPublicIP() {
+  for (const url of PUBLIC_IP_URLS) {
+    try {
+      const ctrl = new AbortController()
+      const timer = setTimeout(() => ctrl.abort(), 4000)
+      const res = await fetch(url, { signal: ctrl.signal })
+      clearTimeout(timer)
+      if (!res.ok) continue
+      const ip = (await res.text()).trim()
+      if (isIPv4(ip)) return ip
+    } catch {
+      // try next
+    }
+  }
+  return null
+}
+
+async function resolveDynamic(key, resolver) {
+  const hit = cache.get(key)
+  if (hit && Date.now() < hit.expiry) return hit.ip
+  const ip = await resolver()
+  if (ip) {
+    cache.set(key, { ip, expiry: Date.now() + TTL })
+    return ip
+  }
+  return hit ? hit.ip : null
+}
+
+export async function isOnsite(requestIP) {
+  if (matchers.length === 0 || matchers.includes('disabled')) return true
+  const ip = normalizeIP(requestIP)
+  if (!ip) return false
+
+  for (const m of matchers) {
+    if (m === 'auto') {
+      const pub = await resolveDynamic('auto', fetchPublicIP)
+      if (pub && ip === pub) return true
+    } else if (m.includes('/')) {
+      if (ipInCidr(ip, m)) return true
+    } else if (/[a-zA-Z]/.test(m)) {
+      const resolved = await resolveDynamic(m, async () => {
+        try { return (await dns.resolve4(m))[0] } catch { return null }
+      })
+      if (resolved && ip === resolved) return true
+    } else {
+      if (ip === m) return true
+    }
+  }
+  return false
+}

backend/src/lib/control.js

@@ -0,0 +1,33 @@
+import { pool } from '../db.js'
+import { getDevice, publishPower } from './mqtt.js'
+import { recordEvent } from './events.js'
+
+const STAGGER_MS = 150
+
+// The single path every power change goes through: publish, log, and — for
+// manual changes — stamp last_manual_at so a later reconcile won't undo it.
+export async function setPower(deviceId, state, source, detail = null) {
+  const device = getDevice(deviceId)
+  if (!device) throw new Error('Device not found or inactive')
+  const changing = device.state !== state
+  publishPower(device, state)
+  if (source === 'manual') {
+    await pool.query('UPDATE light_devices SET last_manual_at = NOW() WHERE id = $1', [deviceId])
+  }
+  if (changing) await recordEvent(device, state, source, detail)
+}
+
+export async function setPowerMany(deviceIds, state, source, detail = null) {
+  const failed = []
+  for (const id of deviceIds) {
+    try { await setPower(id, state, source, detail) } catch (err) { failed.push({ id, error: err.message }) }
+    await new Promise(r => setTimeout(r, STAGGER_MS))
+  }
+  return failed
+}
+
+export async function resolveTargetDevices(targetType, targetId) {
+  if (targetType === 'device') return [targetId]
+  const { rows } = await pool.query('SELECT device_id FROM light_group_members WHERE group_id = $1', [targetId])
+  return rows.map(r => r.device_id)
+}

backend/src/lib/engine.js

@@ -0,0 +1,123 @@
+// Schedule engine. Two jobs:
+//  1. tick(): fire any schedule whose trigger time passed within the grace
+//     window and hasn't run today.
+//  2. reconcile(): work out what each device SHOULD be right now from the
+//     schedules and correct it — used on startup, when a device comes back
+//     online, and on demand. A manual or physical change newer than the latest
+//     trigger is respected ("override until the next schedule event").
+import { pool, getSettings } from '../db.js'
+import { getDevice, isConnected, setOnlineHandler } from './mqtt.js'
+import { setPower, setPowerMany, resolveTargetDevices } from './control.js'
+import { pruneEvents } from './events.js'
+import { localParts, ymd, triggerTime, dayOffset } from './time.js'
+
+const TICK_MS = 20 * 1000
+const FIRE_GRACE_MS = 15 * 60 * 1000   // older misses are left to reconcile
+const LOOKBACK_DAYS = 7
+
+async function fire(s, trigger) {
+  const claimed = await pool.query(
+    `UPDATE light_schedules SET last_fired_on = $1 WHERE id = $2 AND last_fired_on IS DISTINCT FROM $1 RETURNING id`,
+    [s._today, s.id]
+  )
+  if (!claimed.rowCount) return
+  const ids = await resolveTargetDevices(s.target_type, s.target_id)
+  console.log(`[engine] firing "${s.name}" (${s.action}) on ${ids.length} device(s), trigger ${trigger.toISOString()}`)
+  const failed = await setPowerMany(ids, s.action, 'schedule', s.name)
+  if (failed.length) console.warn(`[engine] "${s.name}": ${failed.length} command(s) failed:`, failed.map(f => f.error).join('; '))
+}
+
+export async function tick() {
+  if (!isConnected()) return
+  const settings = await getSettings()
+  const tz = settings.timezone || 'Europe/London'
+  const now = new Date()
+  const today = localParts(now, tz)
+  const { rows } = await pool.query('SELECT * FROM light_schedules WHERE enabled = TRUE')
+  for (const s of rows) {
+    if (!s.days.includes(today.weekday)) continue
+    const t = triggerTime(s, today, settings)
+    if (!t) continue
+    const age = now - t
+    if (age >= 0 && age < FIRE_GRACE_MS) {
+      s._today = ymd(today)
+      await fire(s, t).catch(err => console.error('[engine] fire failed:', err.message))
+    }
+  }
+}
+
+// Latest trigger at or before `now` among the schedules that apply to a device.
+function latestTrigger(schedules, settings, now) {
+  const tz = settings.timezone || 'Europe/London'
+  const today = localParts(now, tz)
+  let best = null
+  for (let back = 0; back <= LOOKBACK_DAYS; back++) {
+    const day = dayOffset(today, -back, tz)
+    for (const s of schedules) {
+      if (!s.days.includes(day.weekday)) continue
+      const t = triggerTime(s, day, settings)
+      if (t && t <= now && (!best || t > best.at)) best = { at: t, action: s.action, name: s.name }
+    }
+  }
+  return best
+}
+
+// Schedules that apply to a device, directly or through any group it is in.
+async function schedulesFor(deviceId) {
+  const { rows } = await pool.query(
+    `SELECT s.* FROM light_schedules s
+      WHERE s.enabled = TRUE AND (
+        (s.target_type = 'device' AND s.target_id = $1) OR
+        (s.target_type = 'group' AND s.target_id IN (SELECT group_id FROM light_group_members WHERE device_id = $1))
+      )`,
+    [deviceId]
+  )
+  return rows
+}
+
+export async function reconcile(deviceIds = null) {
+  if (!isConnected()) return { applied: 0 }
+  const settings = await getSettings()
+  const now = new Date()
+  let ids = deviceIds
+  if (!ids) {
+    const { rows } = await pool.query('SELECT id FROM light_devices WHERE active = TRUE')
+    ids = rows.map(r => r.id)
+  }
+  let applied = 0
+  for (const id of ids) {
+    const device = getDevice(id)
+    if (!device || device.state === null || device.online === false) continue
+    const want = latestTrigger(await schedulesFor(id), settings, now)
+    if (!want || want.action === device.state) continue
+    const { rows: [row] } = await pool.query('SELECT last_manual_at FROM light_devices WHERE id = $1', [id])
+    if (row?.last_manual_at && row.last_manual_at > want.at) continue
+    console.log(`[engine] reconcile: ${device.name} is ${device.state}, schedule "${want.name}" says ${want.action}`)
+    await setPower(id, want.action, 'reconcile', want.name).catch(err => console.error('[engine] reconcile failed:', err.message))
+    applied++
+  }
+  return { applied }
+}
+
+export function startEngine() {
+  // A device announcing itself online may have missed triggers while it was
+  // down; give its state message a moment to land before comparing.
+  setOnlineHandler(ids => setTimeout(() => reconcile(ids).catch(() => {}), 5000))
+  setInterval(() => tick().catch(err => console.error('[engine] tick failed:', err.message)), TICK_MS)
+  // Startup catch-up once the broker has delivered the initial device states.
+  setTimeout(() => reconcile().catch(() => {}), 20000)
+  setInterval(() => pruneEvents().catch(() => {}), 24 * 3600 * 1000)
+}
+
+// Next time a schedule will fire (for display), searching the coming week.
+export function nextRun(s, settings, now = new Date()) {
+  const tz = settings.timezone || 'Europe/London'
+  const today = localParts(now, tz)
+  for (let ahead = 0; ahead <= 7; ahead++) {
+    const day = dayOffset(today, ahead, tz)
+    if (!s.days.includes(day.weekday)) continue
+    const t = triggerTime(s, day, settings)
+    if (t && t > now) return t
+  }
+  return null
+}

backend/src/lib/events.js

@@ -0,0 +1,14 @@
+import { pool } from '../db.js'
+
+const RETENTION_DAYS = 90
+
+export async function recordEvent(device, action, source, detail = null) {
+  await pool.query(
+    `INSERT INTO light_events (device_id, device_name, action, source, detail) VALUES ($1, $2, $3, $4, $5)`,
+    [device.id, device.name, action, source, detail]
+  ).catch(err => console.error('[events] insert failed:', err.message))
+}
+
+export async function pruneEvents() {
+  await pool.query(`DELETE FROM light_events WHERE at < NOW() - ($1 || ' days')::interval`, [String(RETENTION_DAYS)])
+}

backend/src/lib/mqtt.js

@@ -0,0 +1,214 @@
+// Tasmota over MQTT. Credentials come from the settings service's service
+// client "lights-backend" (settings → MQTT Clients → "service consumer"), same
+// runtime-fetch pattern as hvac. Reconnect/backoff is owned here, never blocks
+// startup, and every public function degrades when the broker is unreachable.
+//
+// Tasmota topic layout (defaults): commands go to cmnd/<topic>/POWER[n]; the
+// device answers on stat/<topic>/RESULT and reports periodically on
+// tele/<topic>/STATE; tele/<topic>/LWT is retained "Online"/"Offline".
+import mqtt from 'mqtt'
+import { pool } from '../db.js'
+import { recordEvent } from './events.js'
+
+const CLIENT_NAME = 'lights-backend'
+const BROKER_RECONNECT_BACKOFF_MS = [30, 60, 120, 300, 600, 1800].map(s => s * 1000)
+const PENDING_MS = 15000
+
+let client = null
+let connecting = false
+let reconnectAttempt = 0
+let devices = new Map()          // id -> device row (live state cached here)
+let topicMap = new Map()         // mqtt_topic -> [device]
+const discovered = new Map()     // unregistered topic -> { topic, online, channels, last_seen }
+const pending = new Map()        // device id -> expiry ms; our own command, so its echo isn't a "physical" change
+let onlineHandler = null
+
+export function setOnlineHandler(fn) { onlineHandler = fn }
+
+async function getCredentials() {
+  const url = `${process.env.SETTINGS_URL}/settings/api/internal/mqtt-client/${CLIENT_NAME}`
+  const res = await fetch(url, {
+    headers: { Authorization: `Bearer ${process.env.SETTINGS_SECRET}` },
+    signal: AbortSignal.timeout(5000),
+  })
+  if (!res.ok) throw new Error(`Settings service returned ${res.status} fetching MQTT credentials for "${CLIENT_NAME}"`)
+  const s = await res.json()
+  if (!s.host || !s.username || !s.password) throw new Error('MQTT broker credentials not configured')
+  return { host: s.host, port: s.port || 1883, username: s.username, password: s.password }
+}
+
+export async function refreshDevices() {
+  const { rows } = await pool.query('SELECT * FROM light_devices WHERE active = TRUE')
+  const next = new Map()
+  for (const r of rows) {
+    // Keep the in-memory state if we already have it — the DB copy is only ever
+    // behind it, never ahead.
+    const prev = devices.get(r.id)
+    next.set(r.id, prev ? { ...r, state: prev.state, online: prev.online } : r)
+  }
+  devices = next
+  topicMap = new Map()
+  for (const d of devices.values()) {
+    if (!topicMap.has(d.mqtt_topic)) topicMap.set(d.mqtt_topic, [])
+    topicMap.get(d.mqtt_topic).push(d)
+    discovered.delete(d.mqtt_topic)
+  }
+  if (client?.connected) queryAll()
+}
+
+export function getDevice(id) { return devices.get(id) }
+
+function queryAll() {
+  for (const topic of topicMap.keys()) client.publish(`cmnd/${topic}/State`, '')
+}
+
+function channelValue(obj, channel) {
+  const v = channel === 1 ? (obj.POWER1 ?? obj.POWER) : obj[`POWER${channel}`]
+  return typeof v === 'string' ? v.toUpperCase() : null
+}
+
+function channelsIn(obj) {
+  const out = []
+  if (obj.POWER !== undefined || obj.POWER1 !== undefined) out.push(1)
+  for (let n = 2; n <= 8; n++) if (obj[`POWER${n}`] !== undefined) out.push(n)
+  return out
+}
+
+async function applyState(device, value, now) {
+  const next = value === 'ON' ? 'on' : value === 'OFF' ? 'off' : null
+  if (!next) return
+  const prev = device.state
+  device.state = next
+  if (prev === next) {
+    pool.query('UPDATE light_devices SET state_at = $1 WHERE id = $2', [now, device.id]).catch(() => {})
+    return
+  }
+  const expected = pending.get(device.id)
+  const ours = expected && expected > Date.now()
+  pending.delete(device.id)
+  // A change we didn't ask for (wall switch, Tasmota's own timer, the web UI)
+  // counts as a manual override, so the next reconcile doesn't fight it.
+  const physical = prev !== null && !ours
+  await pool.query(
+    `UPDATE light_devices SET state = $1, state_at = $2${physical ? ', last_manual_at = NOW()' : ''} WHERE id = $3`,
+    [next, now, device.id]
+  ).catch(err => console.error('[mqtt] state update failed:', err.message))
+  if (physical) await recordEvent(device, next, 'physical')
+}
+
+async function handleMessage(topic, message) {
+  const [prefix, devTopic, kind] = topic.split('/')
+  const text = message.toString().trim()
+  const list = topicMap.get(devTopic)
+
+  if (kind === 'LWT') {
+    const online = text.toLowerCase() === 'online'
+    if (!list) {
+      const d = discovered.get(devTopic) || { topic: devTopic, channels: [1] }
+      discovered.set(devTopic, { ...d, online, last_seen: new Date() })
+      return
+    }
+    const cameOnline = online && list.some(d => d.online !== true)
+    for (const d of list) d.online = online
+    await pool.query('UPDATE light_devices SET online = $1 WHERE mqtt_topic = $2', [online, devTopic]).catch(() => {})
+    if (cameOnline) {
+      client?.publish(`cmnd/${devTopic}/State`, '')
+      onlineHandler?.(list.map(d => d.id))
+    }
+    return
+  }
+
+  if (prefix === 'tele' ? kind !== 'STATE' : kind !== 'RESULT') return
+  let obj
+  try { obj = JSON.parse(text) } catch { return }
+  if (!obj || typeof obj !== 'object') return
+  const channels = channelsIn(obj)
+  if (!channels.length) return
+
+  if (!list) {
+    if (discovered.size < 200 || discovered.has(devTopic)) {
+      const d = discovered.get(devTopic) || { topic: devTopic, online: null }
+      discovered.set(devTopic, { ...d, channels, last_seen: new Date() })
+    }
+    return
+  }
+  const now = new Date()
+  for (const d of list) {
+    const v = channelValue(obj, d.channel)
+    if (v) await applyState(d, v, now)
+  }
+  // A state message from a device we thought offline proves it's alive.
+  if (list.some(d => d.online === false)) {
+    for (const d of list) d.online = true
+    pool.query('UPDATE light_devices SET online = TRUE WHERE mqtt_topic = $1', [devTopic]).catch(() => {})
+  }
+}
+
+function scheduleReconnect() {
+  if (connecting) return
+  const delay = BROKER_RECONNECT_BACKOFF_MS[Math.min(reconnectAttempt, BROKER_RECONNECT_BACKOFF_MS.length - 1)]
+  reconnectAttempt++
+  console.warn(`[mqtt] broker unreachable — retrying in ${delay / 1000}s`)
+  setTimeout(() => { connect().catch(() => {}) }, delay)
+}
+
+export async function connect() {
+  if (connecting || client?.connected) return
+  connecting = true
+  try {
+    const creds = await getCredentials()
+    await refreshDevices()
+
+    await new Promise((resolve, reject) => {
+      const c = mqtt.connect(`mqtt://${creds.host}:${creds.port}`, {
+        username: creds.username,
+        password: creds.password,
+        clientId: `${CLIENT_NAME}-${Math.random().toString(16).slice(2, 8)}`,
+        connectTimeout: 10000,
+        reconnectPeriod: 0, // we own reconnect/backoff ourselves
+      })
+
+      c.on('connect', () => {
+        client = c
+        reconnectAttempt = 0
+        connecting = false
+        console.log('[mqtt] connected to broker')
+        c.subscribe(['tele/+/LWT', 'tele/+/STATE', 'stat/+/RESULT'])
+        queryAll()
+        resolve()
+      })
+      c.on('message', (topic, message) => {
+        handleMessage(topic, message).catch(err => console.error('[mqtt] message handling failed:', err.message))
+      })
+      c.on('error', err => console.error('[mqtt] connection error:', err.message))
+      c.on('close', () => {
+        if (client === c) client = null
+        connecting = false
+        scheduleReconnect()
+      })
+
+      setTimeout(() => {
+        if (!client) { c.end(true); connecting = false; reject(new Error('MQTT connect timed out')) }
+      }, 12000)
+    })
+  } catch (err) {
+    connecting = false
+    console.error('[mqtt] could not connect (broker/credentials unavailable):', err.message)
+    scheduleReconnect()
+  }
+}
+
+export function isConnected() { return !!client?.connected }
+
+export function listDiscovered() {
+  return [...discovered.values()].sort((a, b) => a.topic.localeCompare(b.topic))
+}
+
+// state: 'on' | 'off'. Throws if the broker isn't connected so callers can tell
+// the user instead of silently recording a command that never went out.
+export function publishPower(device, state) {
+  if (!client?.connected) throw new Error('MQTT broker not connected')
+  pending.set(device.id, Date.now() + PENDING_MS)
+  const cmd = device.channel > 1 ? `POWER${device.channel}` : 'POWER'
+  client.publish(`cmnd/${device.mqtt_topic}/${cmd}`, state === 'on' ? 'ON' : 'OFF')
+}

backend/src/lib/time.js

@@ -0,0 +1,79 @@
+import SunCalc from 'suncalc'
+
+const PART_FMT = new Map()
+function fmt(tz) {
+  if (!PART_FMT.has(tz)) {
+    PART_FMT.set(tz, new Intl.DateTimeFormat('en-GB', {
+      timeZone: tz, year: 'numeric', month: '2-digit', day: '2-digit', weekday: 'short',
+      hour: '2-digit', minute: '2-digit', hourCycle: 'h23',
+    }))
+  }
+  return PART_FMT.get(tz)
+}
+
+const WEEKDAYS = { Sun: 0, Mon: 1, Tue: 2, Wed: 3, Thu: 4, Fri: 5, Sat: 6 }
+
+// Wall-clock parts of an instant in the site timezone.
+export function localParts(date, tz) {
+  const p = Object.fromEntries(fmt(tz).formatToParts(date).map(x => [x.type, x.value]))
+  return {
+    y: Number(p.year), m: Number(p.month), d: Number(p.day),
+    hh: Number(p.hour), mm: Number(p.minute), weekday: WEEKDAYS[p.weekday],
+  }
+}
+
+export function ymd({ y, m, d }) {
+  return `${y}-${String(m).padStart(2, '0')}-${String(d).padStart(2, '0')}`
+}
+
+function offsetMs(ts, tz) {
+  const t = ts - (ts % 60000)
+  const p = localParts(new Date(t), tz)
+  return Date.UTC(p.y, p.m - 1, p.d, p.hh, p.mm) - t
+}
+
+// Site-local wall-clock time -> UTC instant (handles DST by re-checking the offset once).
+export function zonedToUtc(y, m, d, hh, mm, tz) {
+  const guess = Date.UTC(y, m - 1, d, hh, mm)
+  let ts = guess - offsetMs(guess, tz)
+  ts = guess - offsetMs(ts, tz)
+  return new Date(ts)
+}
+
+export function validTimezone(tz) {
+  try { fmt(tz); return true } catch { return false }
+}
+
+// The instant a schedule fires on the given site-local calendar date, or null
+// (sun trigger with no location configured, or no sunrise/sunset that day).
+export function triggerTime(s, day, settings) {
+  const tz = settings.timezone || 'Europe/London'
+  if (s.trigger_type === 'fixed') {
+    const [hh, mm] = (s.time_of_day || '').split(':').map(Number)
+    if (!Number.isInteger(hh) || !Number.isInteger(mm)) return null
+    return zonedToUtc(day.y, day.m, day.d, hh, mm, tz)
+  }
+  const lat = parseFloat(settings.latitude)
+  const lon = parseFloat(settings.longitude)
+  if (!Number.isFinite(lat) || !Number.isFinite(lon)) return null
+  const t = sunTimes(day, settings)?.[s.trigger_type]
+  if (!t) return null
+  return new Date(t.getTime() + (s.offset_minutes || 0) * 60000)
+}
+
+export function sunTimes(day, settings) {
+  const lat = parseFloat(settings.latitude)
+  const lon = parseFloat(settings.longitude)
+  if (!Number.isFinite(lat) || !Number.isFinite(lon)) return null
+  const noon = zonedToUtc(day.y, day.m, day.d, 12, 0, settings.timezone || 'Europe/London')
+  const t = SunCalc.getTimes(noon, lat, lon)
+  const ok = d => (d instanceof Date && !isNaN(d) ? d : null)
+  return { sunrise: ok(t.sunrise), sunset: ok(t.sunset) }
+}
+
+// Site-local calendar day `offset` days from `from` (offset may be negative).
+export function dayOffset(from, offset, tz) {
+  const noon = zonedToUtc(from.y, from.m, from.d, 12, 0, tz)
+  const p = localParts(new Date(noon.getTime() + offset * 86400000), tz)
+  return { y: p.y, m: p.m, d: p.d, weekday: p.weekday }
+}

backend/src/routes/devices.js

@@ -0,0 +1,152 @@
+import { requireAuth, requireCap } from '../auth.js'
+import { pool, getSettings } from '../db.js'
+import { refreshDevices, listDiscovered, isConnected } from '../lib/mqtt.js'
+import { setPower, setPowerMany, resolveTargetDevices } from '../lib/control.js'
+import { reconcile } from '../lib/engine.js'
+import { localParts, sunTimes } from '../lib/time.js'
+
+const TOPIC_RE = /^[A-Za-z0-9_.-]+$/
+
+async function loadState() {
+  const { rows: devices } = await pool.query(
+    `SELECT id, name, location, mqtt_topic, channel, active, state, online, state_at FROM light_devices ORDER BY location NULLS LAST, name`
+  )
+  const { rows: groups } = await pool.query(
+    `SELECT g.id, g.name, COALESCE(array_agg(m.device_id) FILTER (WHERE m.device_id IS NOT NULL), '{}') AS device_ids
+       FROM light_groups g LEFT JOIN light_group_members m ON m.group_id = g.id GROUP BY g.id ORDER BY g.name`
+  )
+  return { devices, groups }
+}
+
+export async function deviceRoutes(app) {
+  app.addHook('preHandler', requireAuth)
+
+  app.get('/api/state', { preHandler: requireCap('view') }, async () => {
+    const settings = await getSettings()
+    const tz = settings.timezone || 'Europe/London'
+    const sun = sunTimes(localParts(new Date(), tz), settings)
+    return { ...(await loadState()), mqtt_connected: isConnected(), sun, timezone: tz }
+  })
+
+  // ── manual control ──
+  app.post('/api/devices/:id/power', { preHandler: requireCap('control') }, async (req, reply) => {
+    const state = req.body?.state
+    if (state !== 'on' && state !== 'off') return reply.status(400).send({ error: 'state must be on or off' })
+    try {
+      await setPower(Number(req.params.id), state, 'manual', req.user.email)
+    } catch (e) {
+      return reply.status(e.message.includes('not connected') ? 503 : 404).send({ error: e.message })
+    }
+    return { ok: true }
+  })
+
+  app.post('/api/groups/:id/power', { preHandler: requireCap('control') }, async (req, reply) => {
+    const state = req.body?.state
+    if (state !== 'on' && state !== 'off') return reply.status(400).send({ error: 'state must be on or off' })
+    if (!isConnected()) return reply.status(503).send({ error: 'MQTT broker not connected' })
+    const ids = await resolveTargetDevices('group', Number(req.params.id))
+    const failed = await setPowerMany(ids, state, 'manual', req.user.email)
+    return { ok: failed.length === 0, failed }
+  })
+
+  app.post('/api/reconcile', { preHandler: requireCap('control') }, async (req, reply) => {
+    if (!isConnected()) return reply.status(503).send({ error: 'MQTT broker not connected' })
+    return reconcile()
+  })
+
+  // ── devices ──
+  app.get('/api/devices/discovered', { preHandler: requireCap('manage_devices') }, async () => listDiscovered())
+
+  app.post('/api/devices', { preHandler: requireCap('manage_devices') }, async (req, reply) => {
+    const { name, location, mqtt_topic, channel } = req.body || {}
+    const ch = Number(channel) || 1
+    if (!name?.trim()) return reply.status(400).send({ error: 'Name is required' })
+    if (!TOPIC_RE.test(mqtt_topic || '')) return reply.status(400).send({ error: 'Tasmota topic is required (letters, numbers, - _ .)' })
+    if (!Number.isInteger(ch) || ch < 1 || ch > 8) return reply.status(400).send({ error: 'Channel must be 1-8' })
+    try {
+      const { rows: [row] } = await pool.query(
+        `INSERT INTO light_devices (name, location, mqtt_topic, channel) VALUES ($1, $2, $3, $4) RETURNING *`,
+        [name.trim(), location?.trim() || null, mqtt_topic, ch]
+      )
+      await refreshDevices()
+      return row
+    } catch (e) {
+      if (e.code === '23505') return reply.status(409).send({ error: 'That topic + channel is already registered' })
+      throw e
+    }
+  })
+
+  app.patch('/api/devices/:id', { preHandler: requireCap('manage_devices') }, async (req, reply) => {
+    const { name, location, mqtt_topic, channel, active } = req.body || {}
+    if (mqtt_topic !== undefined && !TOPIC_RE.test(mqtt_topic)) return reply.status(400).send({ error: 'Invalid Tasmota topic' })
+    try {
+      const { rows: [row] } = await pool.query(
+        `UPDATE light_devices SET
+           name       = COALESCE($1, name),
+           location   = CASE WHEN $2::boolean THEN $3 ELSE location END,
+           mqtt_topic = COALESCE($4, mqtt_topic),
+           channel    = COALESCE($5, channel),
+           active     = COALESCE($6, active)
+         WHERE id = $7 RETURNING *`,
+        [name?.trim() || null, location !== undefined, location?.trim() || null, mqtt_topic ?? null, channel ?? null, active ?? null, req.params.id]
+      )
+      if (!row) return reply.status(404).send({ error: 'Not found' })
+      await refreshDevices()
+      return row
+    } catch (e) {
+      if (e.code === '23505') return reply.status(409).send({ error: 'That topic + channel is already registered' })
+      throw e
+    }
+  })
+
+  app.delete('/api/devices/:id', { preHandler: requireCap('manage_devices') }, async (req) => {
+    await pool.query(`DELETE FROM light_schedules WHERE target_type = 'device' AND target_id = $1`, [req.params.id])
+    await pool.query('DELETE FROM light_devices WHERE id = $1', [req.params.id])
+    await refreshDevices()
+    return { ok: true }
+  })
+
+  // ── groups ──
+  async function setMembers(groupId, deviceIds) {
+    await pool.query('DELETE FROM light_group_members WHERE group_id = $1', [groupId])
+    for (const id of deviceIds) {
+      await pool.query('INSERT INTO light_group_members (group_id, device_id) VALUES ($1, $2) ON CONFLICT DO NOTHING', [groupId, id])
+    }
+  }
+
+  app.post('/api/groups', { preHandler: requireCap('manage_devices') }, async (req, reply) => {
+    const { name, device_ids } = req.body || {}
+    if (!name?.trim()) return reply.status(400).send({ error: 'Name is required' })
+    try {
+      const { rows: [row] } = await pool.query('INSERT INTO light_groups (name) VALUES ($1) RETURNING *', [name.trim()])
+      await setMembers(row.id, Array.isArray(device_ids) ? device_ids.map(Number) : [])
+      return row
+    } catch (e) {
+      if (e.code === '23505') return reply.status(409).send({ error: 'A group with that name already exists' })
+      throw e
+    }
+  })
+
+  app.patch('/api/groups/:id', { preHandler: requireCap('manage_devices') }, async (req, reply) => {
+    const { name, device_ids } = req.body || {}
+    const { rows: [row] } = await pool.query(
+      'UPDATE light_groups SET name = COALESCE($1, name) WHERE id = $2 RETURNING *', [name?.trim() || null, req.params.id]
+    ).catch(e => { if (e.code === '23505') return { rows: [] }; throw e })
+    if (!row) return reply.status(404).send({ error: 'Not found (or name already used)' })
+    if (Array.isArray(device_ids)) await setMembers(row.id, device_ids.map(Number))
+    return row
+  })
+
+  app.delete('/api/groups/:id', { preHandler: requireCap('manage_devices') }, async (req) => {
+    await pool.query(`DELETE FROM light_schedules WHERE target_type = 'group' AND target_id = $1`, [req.params.id])
+    await pool.query('DELETE FROM light_groups WHERE id = $1', [req.params.id])
+    return { ok: true }
+  })
+
+  // ── activity log ──
+  app.get('/api/events', { preHandler: requireCap('view') }, async (req) => {
+    const limit = Math.min(Number(req.query.limit) || 100, 500)
+    const { rows } = await pool.query('SELECT * FROM light_events ORDER BY at DESC LIMIT $1', [limit])
+    return rows
+  })
+}

backend/src/routes/schedules.js

@@ -0,0 +1,75 @@
+import { requireAuth, requireCap } from '../auth.js'
+import { pool, getSettings } from '../db.js'
+import { nextRun } from '../lib/engine.js'
+
+function validate(b, partial = false) {
+  const out = {}
+  const need = (cond, msg) => { if (!cond) throw Object.assign(new Error(msg), { statusCode: 400 }) }
+  if (!partial || b.name !== undefined) { need(b.name?.trim(), 'Name is required'); out.name = b.name.trim() }
+  if (!partial || b.target_type !== undefined) { need(['device', 'group'].includes(b.target_type), 'target_type must be device or group'); out.target_type = b.target_type }
+  if (!partial || b.target_id !== undefined) { need(Number.isInteger(Number(b.target_id)), 'target_id is required'); out.target_id = Number(b.target_id) }
+  if (!partial || b.action !== undefined) { need(['on', 'off'].includes(b.action), 'action must be on or off'); out.action = b.action }
+  if (!partial || b.trigger_type !== undefined) { need(['fixed', 'sunrise', 'sunset'].includes(b.trigger_type), 'trigger_type must be fixed, sunrise or sunset'); out.trigger_type = b.trigger_type }
+  if (b.time_of_day !== undefined) { need(/^([01]\d|2[0-3]):[0-5]\d$/.test(b.time_of_day || ''), 'time_of_day must be HH:MM'); out.time_of_day = b.time_of_day }
+  if (b.offset_minutes !== undefined) {
+    const o = Number(b.offset_minutes)
+    need(Number.isInteger(o) && o >= -720 && o <= 720, 'offset_minutes must be a whole number between -720 and 720')
+    out.offset_minutes = o
+  }
+  if (b.days !== undefined) {
+    need(Array.isArray(b.days) && b.days.length > 0 && b.days.every(d => Number.isInteger(d) && d >= 0 && d <= 6), 'days must be a non-empty list of 0-6')
+    out.days = [...new Set(b.days)].sort()
+  }
+  if (b.enabled !== undefined) out.enabled = !!b.enabled
+  return out
+}
+
+export async function scheduleRoutes(app) {
+  app.addHook('preHandler', requireAuth)
+
+  app.setErrorHandler((err, req, reply) => {
+    if (err.statusCode === 400) return reply.status(400).send({ error: err.message })
+    req.log.error(err)
+    reply.status(err.statusCode || 500).send({ error: err.message })
+  })
+
+  app.get('/api/schedules', { preHandler: requireCap('view') }, async () => {
+    const settings = await getSettings()
+    const { rows } = await pool.query(`SELECT * FROM light_schedules ORDER BY trigger_type, time_of_day NULLS LAST, name`)
+    return rows.map(s => ({ ...s, next_run: s.enabled ? nextRun(s, settings) : null }))
+  })
+
+  app.post('/api/schedules', { preHandler: requireCap('schedules') }, async (req) => {
+    const v = validate(req.body || {})
+    if (v.trigger_type === 'fixed' && !v.time_of_day) throw Object.assign(new Error('time_of_day is required for fixed schedules'), { statusCode: 400 })
+    const { rows: [row] } = await pool.query(
+      `INSERT INTO light_schedules (name, enabled, target_type, target_id, action, trigger_type, time_of_day, offset_minutes, days)
+       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING *`,
+      [v.name, v.enabled ?? true, v.target_type, v.target_id, v.action, v.trigger_type,
+       v.trigger_type === 'fixed' ? v.time_of_day : null, v.trigger_type === 'fixed' ? 0 : (v.offset_minutes ?? 0),
+       v.days ?? [0, 1, 2, 3, 4, 5, 6]]
+    )
+    return row
+  })
+
+  app.patch('/api/schedules/:id', { preHandler: requireCap('schedules') }, async (req, reply) => {
+    const v = validate(req.body || {}, true)
+    const { rows: [cur] } = await pool.query('SELECT * FROM light_schedules WHERE id = $1', [req.params.id])
+    if (!cur) return reply.status(404).send({ error: 'Not found' })
+    const m = { ...cur, ...v }
+    if (m.trigger_type === 'fixed') { m.offset_minutes = 0; if (!m.time_of_day) return reply.status(400).send({ error: 'time_of_day is required for fixed schedules' }) }
+    else m.time_of_day = null
+    // Editing a trigger re-arms it, so moving a time later today still fires.
+    const { rows: [row] } = await pool.query(
+      `UPDATE light_schedules SET name=$1, enabled=$2, target_type=$3, target_id=$4, action=$5, trigger_type=$6,
+         time_of_day=$7, offset_minutes=$8, days=$9, last_fired_on=NULL WHERE id=$10 RETURNING *`,
+      [m.name, m.enabled, m.target_type, m.target_id, m.action, m.trigger_type, m.time_of_day, m.offset_minutes, m.days, req.params.id]
+    )
+    return row
+  })
+
+  app.delete('/api/schedules/:id', { preHandler: requireCap('schedules') }, async (req) => {
+    await pool.query('DELETE FROM light_schedules WHERE id = $1', [req.params.id])
+    return { ok: true }
+  })
+}

backend/src/routes/settings.js

@@ -0,0 +1,42 @@
+import { requireAuth, requireCap } from '../auth.js'
+import { pool } from '../db.js'
+import { isConnected } from '../lib/mqtt.js'
+import { validTimezone } from '../lib/time.js'
+
+// Standard app_settings pattern (see conventions doc). Latitude/longitude drive
+// the sunrise/sunset triggers; timezone decides what "18:00" and "Mon-Fri" mean.
+const ALLOWED_KEYS = new Set(['latitude', 'longitude', 'timezone'])
+
+export async function settingsRoutes(app) {
+  app.addHook('preHandler', requireAuth)
+
+  app.get('/api/settings', { preHandler: requireCap('settings') }, async () => {
+    const { rows } = await pool.query('SELECT key, value, updated_at FROM app_settings ORDER BY key')
+    return { settings: rows }
+  })
+
+  app.put('/api/settings', { preHandler: requireCap('settings') }, async (req, reply) => {
+    const { settings } = req.body || {}
+    if (!Array.isArray(settings)) return reply.status(400).send({ error: 'settings must be an array' })
+    for (const { key, value } of settings) {
+      if (!ALLOWED_KEYS.has(key)) continue
+      const v = String(value ?? '').trim()
+      if (key === 'latitude' && v && !(Math.abs(parseFloat(v)) <= 90)) return reply.status(400).send({ error: 'Latitude must be between -90 and 90' })
+      if (key === 'longitude' && v && !(Math.abs(parseFloat(v)) <= 180)) return reply.status(400).send({ error: 'Longitude must be between -180 and 180' })
+      if (key === 'timezone' && !validTimezone(v)) return reply.status(400).send({ error: `Unknown timezone "${v}"` })
+      await pool.query(
+        `INSERT INTO app_settings (key, value, updated_at) VALUES ($1, $2, NOW())
+         ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()`,
+        [key, v]
+      )
+    }
+    return { ok: true }
+  })
+
+  app.get('/api/settings/mqtt-status', { preHandler: requireCap('settings') }, async () => ({
+    connected: isConnected(),
+    note: isConnected()
+      ? 'Connected to the MQTT broker.'
+      : 'Not connected — create a service client named "lights-backend" under Settings → MQTT Clients, or check the broker (LXC 104) is reachable.',
+  }))
+}