Commit missing MQTT/history/scheduler files referenced by meters.js

routes/meters.js already imported lib/mqtt.js and lib/history.js in an
earlier uncommitted change; a prior commit staged the whole meters.js
file (picking up those imports) without staging the modules themselves,
crashing production with ERR_MODULE_NOT_FOUND. Completing the commit:
package.json (mqtt dependency), index.js (wires connectMqtt/startScheduler,
both already error-guarded if the broker isn't reachable), and the three
lib files themselves.

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

backend/package.json

@@ -13,6 +13,7 @@
     "@fastify/static": "^7.0.4",
     "fastify": "^4.28.1",
     "jose": "^5.9.6",
+    "mqtt": "^5.10.3",
     "pg": "^8.13.1"
   }
 }

backend/src/index.js

@@ -14,6 +14,8 @@ import { reportRoutes } from './routes/reports.js'
 import { estimateRoutes } from './routes/estimates.js'
 import { internalRoutes } from './routes/internal.js'
 import { settingsRoutes } from './routes/settings.js'
+import { connect as connectMqtt } from './lib/mqtt.js'
+import { startScheduler } from './lib/scheduler.js'
 
 const __dirname = dirname(fileURLToPath(import.meta.url))
 const UPLOADS_DIR = join(__dirname, '..', 'uploads')
@@ -47,6 +49,8 @@ await app.register(settingsRoutes)
 try {
   await initDb()
   await app.listen({ port: 3001, host: '0.0.0.0' })
+  connectMqtt().catch(err => app.log.error(err, '[mqtt] initial connect failed'))
+  startScheduler()
 } catch (err) {
   app.log.error(err)
   process.exit(1)

backend/src/lib/history.js

@@ -0,0 +1,105 @@
+// Stitches a meter's telemetry + `readings` into one continuous history series
+// for a requested date range — the data API a future week/month graph would
+// call. Recent range comes from telemetry (fine-grained, but purged after
+// telemetry_raw/hourly_retention); anything older falls back to the meter's own
+// `readings` rows (source-agnostic, never purged, but only daily granularity).
+import { pool, getConfig } from '../db.js'
+
+// Ordered { date, value, delta, resolution } series over [from, to] (inclusive).
+// resolution: 'raw' | 'hourly' | 'daily-reading'.
+export async function getMeterHistorySeries(meterId, from, to) {
+  const cfg = await getConfig()
+  const rawRetentionHours = cfg.telemetry_raw_retention_hours ?? 336
+  const hourlyRetentionDays = cfg.telemetry_hourly_retention_days ?? 60
+
+  const now = new Date()
+  const rawCutoff = new Date(now.getTime() - rawRetentionHours * 60 * 60 * 1000)
+  const hourlyCutoff = new Date(now.getTime() - hourlyRetentionDays * 24 * 60 * 60 * 1000)
+
+  const fromDate = new Date(from)
+  const toDate = new Date(to)
+
+  const series = []
+
+  // Oldest portion: the meter's own daily readings, as successive deltas —
+  // same logic as cost-calc.js's getPeriodConsumption, returning every point
+  // in the range instead of one period total.
+  if (fromDate < hourlyCutoff) {
+    const dailyEnd = toISODate(toDate < hourlyCutoff ? toDate : hourlyCutoff)
+    const { rows } = await pool.query(
+      `SELECT reading_value, reading_date FROM readings
+       WHERE meter_id = $1 AND reading_date <= $2
+       ORDER BY reading_date ASC, id ASC`,
+      [meterId, dailyEnd]
+    )
+    const fromISO = toISODate(fromDate)
+    // Include one reading immediately before the range (if any) so the first
+    // in-range point still gets a delta instead of starting null.
+    const firstInRangeIdx = rows.findIndex(r => toISODate(r.reading_date) >= fromISO)
+    const sliceStart = firstInRangeIdx === -1 ? rows.length : Math.max(0, firstInRangeIdx - 1)
+    const withPrior = rows.slice(sliceStart)
+
+    for (let i = 0; i < withPrior.length; i++) {
+      const row = withPrior[i]
+      if (toISODate(row.reading_date) < fromISO) continue
+      const prev = withPrior[i - 1]
+      series.push({
+        date: toISODate(row.reading_date),
+        value: Number(row.reading_value),
+        delta: prev ? Number(row.reading_value) - Number(prev.reading_value) : null,
+        resolution: 'daily-reading',
+      })
+    }
+  }
+
+  // Middle portion: hourly aggregates.
+  if (toDate >= hourlyCutoff && fromDate < rawCutoff) {
+    const hourlyStart = fromDate > hourlyCutoff ? fromDate : hourlyCutoff
+    const hourlyEnd = toDate < rawCutoff ? toDate : rawCutoff
+    const { rows } = await pool.query(
+      `SELECT bucket_start, value FROM meter_telemetry_hourly
+       WHERE meter_id = $1 AND bucket_start >= $2 AND bucket_start <= $3
+       ORDER BY bucket_start ASC`,
+      [meterId, hourlyStart.toISOString(), hourlyEnd.toISOString()]
+    )
+    let prevValue = null
+    for (const row of rows) {
+      const value = Number(row.value)
+      series.push({
+        date: row.bucket_start.toISOString(),
+        value,
+        delta: prevValue != null ? value - prevValue : null,
+        resolution: 'hourly',
+      })
+      prevValue = value
+    }
+  }
+
+  // Most recent portion: raw telemetry.
+  if (toDate >= rawCutoff) {
+    const rawStart = fromDate > rawCutoff ? fromDate : rawCutoff
+    const { rows } = await pool.query(
+      `SELECT recorded_at, value FROM meter_telemetry_raw
+       WHERE meter_id = $1 AND recorded_at >= $2 AND recorded_at <= $3
+       ORDER BY recorded_at ASC`,
+      [meterId, rawStart.toISOString(), toDate.toISOString()]
+    )
+    let prevValue = null
+    for (const row of rows) {
+      const value = Number(row.value)
+      series.push({
+        date: row.recorded_at.toISOString(),
+        value,
+        delta: prevValue != null ? value - prevValue : null,
+        resolution: 'raw',
+      })
+      prevValue = value
+    }
+  }
+
+  return series
+}
+
+function toISODate(d) {
+  return new Date(d).toISOString().slice(0, 10)
+}

backend/src/lib/mqtt.js

@@ -0,0 +1,117 @@
+// MQTT client for device-fed meters (e.g. an ESPHome water-softener totalizer).
+//
+// Connection/backoff skeleton follows hvac/backend/src/lib/mqtt.js's shape —
+// shared broker, own reconnect/backoff, credentials fetched at runtime from
+// settings rather than baked into docker-compose. Unlike hvac (which dispatches
+// hardcoded Shelly topics), this needs a generic topic -> meter_id router since
+// any meter with `mqtt_topic` set should be ingested with zero new code.
+import mqtt from 'mqtt'
+import { pool } from '../db.js'
+
+const CLIENT_NAME = 'utilities-backend'
+const BROKER_RECONNECT_BACKOFF_MS = [30, 60, 120, 300, 600, 1800].map(s => s * 1000) // 30s -> 30min
+
+let client = null
+let connecting = false
+let reconnectAttempt = 0
+let topicMap = new Map() // mqtt_topic -> meter_id
+let subscribedTopics = new Set()
+
+async function getCredentials() {
+  const url = `${process.env.SETTINGS_URL}/settings/api/internal/integration/mqtt`
+  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`)
+  const s = await res.json()
+  if (!s.host || !s.username || !s.password) throw new Error('MQTT broker credentials not configured in settings')
+  return { host: s.host, port: s.port || 1883, username: s.username, password: s.password }
+}
+
+export async function refreshTopicMap() {
+  const { rows } = await pool.query(
+    `SELECT id, mqtt_topic FROM meters WHERE mqtt_topic IS NOT NULL AND active = TRUE`
+  )
+  topicMap = new Map(rows.map(r => [r.mqtt_topic, r.id]))
+  if (client?.connected) subscribeToKnownTopics()
+}
+
+function subscribeToKnownTopics() {
+  const nextTopics = new Set(topicMap.keys())
+  const stale = [...subscribedTopics].filter(t => !nextTopics.has(t))
+  const fresh = [...nextTopics].filter(t => !subscribedTopics.has(t))
+  if (stale.length) client.unsubscribe(stale)
+  if (fresh.length) client.subscribe(fresh)
+  subscribedTopics = nextTopics
+}
+
+function handleMessage(topic, message) {
+  const meterId = topicMap.get(topic)
+  if (!meterId) return
+  const value = Number(message.toString().trim())
+  if (!Number.isFinite(value)) return
+  pool.query(`INSERT INTO meter_telemetry_raw (meter_id, value) VALUES ($1, $2)`, [meterId, value])
+    .catch(err => console.error('[mqtt] telemetry insert failed:', err.message))
+}
+
+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 refreshTopicMap()
+
+    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')
+        subscribedTopics = new Set()
+        subscribeToKnownTopics()
+        resolve()
+      })
+
+      c.on('message', handleMessage)
+
+      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
+}

backend/src/lib/scheduler.js

@@ -0,0 +1,113 @@
+// Telemetry downsample/purge + daily device-reading insert. Follows wages'
+// two-interval wall-clock idiom (wages/backend/src/lib/scheduler.js) — this
+// stack has no cron library anywhere, "run once daily" is always a 1-minute
+// wall-clock poll guarded by a last-run-date check.
+import { pool, getConfig } from '../db.js'
+
+const HOURLY_MS = 60 * 60 * 1000
+const DAILY_CHECK_MS = 60 * 1000
+
+let hourlyRunning = false
+let lastDailyRunDate = null
+
+// Recomputes the last 2 hourly buckets (not just the just-completed one) so
+// late-arriving raw rows near an hour boundary still get folded in cheaply.
+async function downsampleAndPurgeRaw() {
+  const cfg = await getConfig()
+  const rawRetentionHours = cfg.telemetry_raw_retention_hours ?? 336
+
+  await pool.query(`
+    INSERT INTO meter_telemetry_hourly (meter_id, bucket_start, value, sample_count)
+    SELECT meter_id, date_trunc('hour', recorded_at) AS bucket_start,
+           (array_agg(value ORDER BY recorded_at DESC))[1] AS value,
+           COUNT(*) AS sample_count
+    FROM meter_telemetry_raw
+    WHERE recorded_at >= date_trunc('hour', NOW()) - INTERVAL '2 hours'
+    GROUP BY meter_id, date_trunc('hour', recorded_at)
+    ON CONFLICT (meter_id, bucket_start) DO UPDATE
+      SET value = EXCLUDED.value, sample_count = EXCLUDED.sample_count
+  `)
+  await pool.query(
+    `DELETE FROM meter_telemetry_raw WHERE recorded_at < NOW() - ($1 || ' hours')::interval`,
+    [rawRetentionHours]
+  )
+}
+
+// One `readings` row per device-fed meter per day (source='device') — keeps
+// the same sparse, low-frequency shape cost-calc.js's bracket queries assume.
+async function insertDailyDeviceReadings() {
+  const today = new Date().toISOString().slice(0, 10)
+  const { rows: meters } = await pool.query(
+    `SELECT id FROM meters WHERE mqtt_topic IS NOT NULL AND active = TRUE`
+  )
+  for (const m of meters) {
+    const { rows: latest } = await pool.query(
+      `SELECT value FROM meter_telemetry_raw WHERE meter_id = $1 ORDER BY recorded_at DESC LIMIT 1`,
+      [m.id]
+    )
+    if (!latest.length) continue
+
+    const { rows: existing } = await pool.query(
+      `SELECT id FROM readings WHERE meter_id = $1 AND reading_date = $2 AND source = 'device'`,
+      [m.id, today]
+    )
+    if (existing.length) {
+      await pool.query(
+        `UPDATE readings SET reading_value = $1, recorded_at = NOW() WHERE id = $2`,
+        [latest[0].value, existing[0].id]
+      )
+    } else {
+      await pool.query(
+        `INSERT INTO readings (meter_id, reading_value, reading_date, source, notes)
+         VALUES ($1, $2, $3, 'device', 'auto (MQTT)')`,
+        [m.id, latest[0].value, today]
+      )
+    }
+  }
+}
+
+async function purgeOldHourlyAggregates() {
+  const cfg = await getConfig()
+  const days = cfg.telemetry_hourly_retention_days ?? 60
+  await pool.query(
+    `DELETE FROM meter_telemetry_hourly WHERE bucket_start < NOW() - ($1 || ' days')::interval`,
+    [days]
+  )
+}
+
+export function startScheduler() {
+  const hourlyTick = async () => {
+    if (hourlyRunning) return
+    hourlyRunning = true
+    try {
+      await downsampleAndPurgeRaw()
+    } catch (err) {
+      console.error('[scheduler] downsample/purge failed:', err.message)
+    } finally {
+      hourlyRunning = false
+    }
+  }
+  setTimeout(hourlyTick, 10_000)
+  setInterval(hourlyTick, HOURLY_MS)
+
+  setInterval(async () => {
+    const cfg = await getConfig()
+    const [hour, minute] = (cfg.telemetry_daily_schedule_time || '03:15').split(':').map(Number)
+    const now = new Date()
+    const todayKey = now.toISOString().slice(0, 10)
+    if (lastDailyRunDate === todayKey) return
+    if (now.getHours() !== hour || now.getMinutes() !== minute) return
+
+    lastDailyRunDate = todayKey
+    try {
+      await insertDailyDeviceReadings()
+    } catch (err) {
+      console.error('[scheduler] daily reading insert failed:', err.message)
+    }
+    try {
+      await purgeOldHourlyAggregates()
+    } catch (err) {
+      console.error('[scheduler] hourly purge failed:', err.message)
+    }
+  }, DAILY_CHECK_MS)
+}