From 5711e11a0710a87e8562f078c12188c190a8da83 Mon Sep 17 00:00:00 2001 From: jtricerolph Date: Tue, 28 Jul 2026 13:32:22 +0000 Subject: [PATCH] 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 --- backend/package.json | 1 + backend/src/index.js | 4 ++ backend/src/lib/history.js | 105 +++++++++++++++++++++++++++++++ backend/src/lib/mqtt.js | 117 +++++++++++++++++++++++++++++++++++ backend/src/lib/scheduler.js | 113 +++++++++++++++++++++++++++++++++ 5 files changed, 340 insertions(+) create mode 100644 backend/src/lib/history.js create mode 100644 backend/src/lib/mqtt.js create mode 100644 backend/src/lib/scheduler.js diff --git a/backend/package.json b/backend/package.json index bec74ce..0f740f9 100644 --- a/backend/package.json +++ b/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" } } diff --git a/backend/src/index.js b/backend/src/index.js index cc48a2a..09b820a 100644 --- a/backend/src/index.js +++ b/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) diff --git a/backend/src/lib/history.js b/backend/src/lib/history.js new file mode 100644 index 0000000..e61915e --- /dev/null +++ b/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) +} diff --git a/backend/src/lib/mqtt.js b/backend/src/lib/mqtt.js new file mode 100644 index 0000000..a5d52f2 --- /dev/null +++ b/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 +} diff --git a/backend/src/lib/scheduler.js b/backend/src/lib/scheduler.js new file mode 100644 index 0000000..61e92e3 --- /dev/null +++ b/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) +}