hotel-manage-stack/plant
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Scaffold plant app - MQTT monitoring for boiler-room equipment

Browse source 
Read-only monitoring/alerting for boilers, water softener, calorifiers
and pumps via a generic MQTT-topic-prefix asset model, so new
equipment can be onboarded without new ingestion code. Threshold and
stale-data alert rules with email + in-app notification.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
jtricerolph 2026-07-28 21:16:02 +00:00
commit 503b397dff
38 changed files with 5044 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

57
backend/src/auth.js Normal file
View file

@ -0,0 +1,57 @@
import { jwtVerify } from 'jose'
import { isOnsite } from './ip-check.js'
const APP_SLUG = process.env.APP_SLUG || 'plant'
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', 'guest_details', 'rate_details', 'view_all_notes', 'complete_tasks', 'update_status']
}
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}` })
}
}
}

145
backend/src/db.js Normal file
View file

@ -0,0 +1,145 @@
import pg from 'pg'
const { Pool } = pg
export const pool = new Pool({ connectionString: process.env.DATABASE_URL })
export async function initDb() {
await pool.query(`
-- Plant-room equipment register — boilers, water softeners, calorifiers,
-- pump/pressurisation sets. Read-only monitoring only: no target/setpoint
-- columns here and no driver abstraction, unlike hvac's zone_devices.
CREATE TABLE IF NOT EXISTS plant_assets (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
asset_type TEXT NOT NULL, -- boiler | water_softener | calorifier | pump
location TEXT,
make_model TEXT,
serial_no TEXT,
install_date DATE,
notes TEXT,
active BOOLEAN NOT NULL DEFAULT TRUE,
mqtt_topic_prefix TEXT, -- e.g. 'plant/water-softener' — nullable until wired to a device
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS plant_assets_type_idx ON plant_assets (asset_type);
CREATE UNIQUE INDEX IF NOT EXISTS plant_assets_topic_prefix_idx ON plant_assets (mqtt_topic_prefix) WHERE mqtt_topic_prefix IS NOT NULL;
-- Raw telemetry history, one row per MQTT message. field_key is whatever
-- topic segment(s) follow the asset's mqtt_topic_prefix (see lib/mqtt.js) —
-- generic across any asset type, never a fixed column set. Both a numeric
-- and text value are kept: numeric fields (salt_level_pct) populate
-- value_numeric, boolean/enum fields (regeneration_active, salt_level_status)
-- populate value_text (and value_numeric too when the text is exactly
-- "true"/"false", so booleans can still be charted).
CREATE TABLE IF NOT EXISTS plant_telemetry_raw (
id SERIAL PRIMARY KEY,
asset_id INT NOT NULL REFERENCES plant_assets(id) ON DELETE CASCADE,
field_key TEXT NOT NULL,
value_numeric NUMERIC,
value_text TEXT,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS plant_telemetry_raw_asset_field_idx ON plant_telemetry_raw (asset_id, field_key, recorded_at DESC);
-- Latest known value per asset+field — what the Dashboard and stale-minutes
-- alert sweep read from, upserted on every telemetry insert.
CREATE TABLE IF NOT EXISTS plant_asset_latest (
asset_id INT NOT NULL REFERENCES plant_assets(id) ON DELETE CASCADE,
field_key TEXT NOT NULL,
value_numeric NUMERIC,
value_text TEXT,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (asset_id, field_key)
);
-- Alert rules — either scoped to one asset (asset_id set) or to every asset
-- of a type (asset_type set, asset_id NULL). 'stale_minutes' rules are swept
-- periodically by lib/scheduler.js; lt/gt/eq rules are evaluated inline as
-- telemetry arrives (lib/mqtt.js -> lib/alert-engine.js).
CREATE TABLE IF NOT EXISTS plant_alert_rules (
id SERIAL PRIMARY KEY,
asset_id INT REFERENCES plant_assets(id) ON DELETE CASCADE,
asset_type TEXT,
field_key TEXT NOT NULL,
condition TEXT NOT NULL, -- lt | gt | eq | stale_minutes
threshold NUMERIC NOT NULL,
severity TEXT NOT NULL DEFAULT 'warning', -- warning | critical
active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CHECK (asset_id IS NOT NULL OR asset_type IS NOT NULL)
);
CREATE INDEX IF NOT EXISTS plant_alert_rules_lookup_idx ON plant_alert_rules (field_key, condition) WHERE active = TRUE;
-- Triggered alerts. Auto-resolved by the same check that raised them once
-- the underlying condition clears (see lib/alert-engine.js) — "acknowledge"
-- is a separate, purely human action (someone has seen it) from "resolved"
-- (the condition itself is no longer breaching).
CREATE TABLE IF NOT EXISTS plant_alerts (
id SERIAL PRIMARY KEY,
rule_id INT NOT NULL REFERENCES plant_alert_rules(id) ON DELETE CASCADE,
asset_id INT NOT NULL REFERENCES plant_assets(id) ON DELETE CASCADE,
field_key TEXT NOT NULL,
value_at_trigger TEXT,
status TEXT NOT NULL DEFAULT 'open', -- open | acknowledged | resolved
triggered_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
acknowledged_at TIMESTAMPTZ,
acknowledged_by TEXT,
resolved_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS plant_alerts_status_idx ON plant_alerts (status, triggered_at DESC);
CREATE INDEX IF NOT EXISTS plant_alerts_open_lookup_idx ON plant_alerts (rule_id, asset_id, field_key) WHERE status = 'open';
-- Asset photos — same shape as hvac's device_photos (multipart + sharp,
-- shared uploads_data volume). photo_type 'asset' (general shot) or
-- 'serial_plate' (model/serial plate close-up).
CREATE TABLE IF NOT EXISTS asset_photos (
id SERIAL PRIMARY KEY,
asset_id INT NOT NULL REFERENCES plant_assets(id) ON DELETE CASCADE,
file_name TEXT NOT NULL,
file_path TEXT NOT NULL,
mime_type TEXT,
file_size INT,
photo_type TEXT NOT NULL DEFAULT 'asset', -- asset | serial_plate
uploaded_by TEXT,
uploaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS asset_photos_asset_idx ON asset_photos (asset_id);
-- Generic app settings (cross-app-integration pattern, see conventions doc) —
-- plant has no other app to integrate with yet, just an alert-notification
-- recipient, but the table/route/page shape is kept standard so a future
-- integration is a one-row addition, not a new pattern.
CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
`)
await seedDefaultSettings()
}
async function seedDefaultSettings() {
const defaults = {
alert_notify_email: '',
}
for (const [key, value] of Object.entries(defaults)) {
await pool.query(
`INSERT INTO app_settings (key, value) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`,
[key, value]
)
}
}
export async function getSetting(key) {
const { rows } = await pool.query('SELECT value FROM app_settings WHERE key = $1', [key])
return rows.length ? rows[0].value : null
}
export async function setSetting(key, value) {
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, value]
)
}

55
backend/src/index.js Normal file
View file

@ -0,0 +1,55 @@
import Fastify from 'fastify'
import cookie from '@fastify/cookie'
import cors from '@fastify/cors'
import multipart from '@fastify/multipart'
import staticFiles from '@fastify/static'
import { fileURLToPath } from 'url'
import { dirname, join } from 'path'
import { initDb } from './db.js'
import { connect as connectMqtt } from './lib/mqtt.js'
import { startScheduler } from './lib/scheduler.js'
import { assetRoutes } from './routes/assets.js'
import { statusRoutes } from './routes/status.js'
import { alertRoutes } from './routes/alerts.js'
import { alertRuleRoutes } from './routes/alert-rules.js'
import { settingsRoutes } from './routes/settings.js'
const __dirname = dirname(fileURLToPath(import.meta.url))
const UPLOADS_DIR = join(__dirname, '..', 'uploads')
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,
})
await app.register(multipart, { limits: { fileSize: 10 * 1024 * 1024 } })
await app.register(staticFiles, {
root: UPLOADS_DIR,
prefix: '/api/uploads/',
decorateReply: false,
})
app.get('/health', async () => ({ status: 'healthy', version: process.env.BUILD_VERSION || String(startedAt) }))
await app.register(assetRoutes, { uploadsDir: UPLOADS_DIR })
await app.register(statusRoutes)
await app.register(alertRoutes)
await app.register(alertRuleRoutes)
await app.register(settingsRoutes)
try {
await initDb()
// MQTT broker (shared infra, LXC 104) may not exist or be reachable yet —
// connect() handles that gracefully with its own backoff and never blocks startup.
connectMqtt().catch(err => app.log.warn(`MQTT connect failed at startup: ${err.message}`))
startScheduler()
await app.listen({ port: 3001, host: '0.0.0.0' })
} catch (err) {
app.log.error(err)
process.exit(1)
}

80
backend/src/ip-check.js Normal file
View file

@ -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
}

108
backend/src/lib/alert-engine.js Normal file
View file

@ -0,0 +1,108 @@
// Shared alert evaluation for both ingestion paths: lib/mqtt.js calls
// checkThresholdRules() inline on every message (lt/gt/eq), and lib/scheduler.js
// calls checkStaleRules() on its periodic sweep (stale_minutes only). Both
// funnel through the same trigger/auto-resolve helpers so an alert is never
// raised or cleared differently depending on which path noticed it.
import { pool } from '../db.js'
import { notifyPlantAlert } from './mailer.js'
function breaches(condition, threshold, numericValue) {
if (numericValue === null || numericValue === undefined) return false
const v = Number(numericValue)
if (!Number.isFinite(v)) return false
const t = Number(threshold)
if (condition === 'lt') return v < t
if (condition === 'gt') return v > t
if (condition === 'eq') return v === t
return false
}
async function findRules(fieldKey, conditions, assetId, assetType) {
const { rows } = await pool.query(
`SELECT * FROM plant_alert_rules
WHERE active = TRUE AND field_key = $1 AND condition = ANY($2)
AND (asset_id = $3 OR (asset_id IS NULL AND asset_type = $4))`,
[fieldKey, conditions, assetId, assetType]
)
return rows
}
async function triggerAlert(rule, assetId, fieldKey, valueAtTrigger) {
const { rows: openRows } = await pool.query(
`SELECT id FROM plant_alerts WHERE rule_id = $1 AND asset_id = $2 AND field_key = $3 AND status = 'open'`,
[rule.id, assetId, fieldKey]
)
if (openRows.length) return // already open — don't re-notify on every message/sweep tick
const { rows: inserted } = await pool.query(
`INSERT INTO plant_alerts (rule_id, asset_id, field_key, value_at_trigger)
VALUES ($1, $2, $3, $4) RETURNING *`,
[rule.id, assetId, fieldKey, String(valueAtTrigger)]
)
const { rows: assetRows } = await pool.query('SELECT * FROM plant_assets WHERE id = $1', [assetId])
if (assetRows.length) {
notifyPlantAlert(inserted[0], assetRows[0], rule).catch(err =>
console.error('[alert-engine] notify failed:', err.message)
)
}
}
async function autoResolveIfOpen(ruleId, assetId, fieldKey) {
await pool.query(
`UPDATE plant_alerts SET status = 'resolved', resolved_at = NOW()
WHERE rule_id = $1 AND asset_id = $2 AND field_key = $3 AND status = 'open'`,
[ruleId, assetId, fieldKey]
)
}
// Called by lib/mqtt.js on every message with a parsed numeric value (may be
// null for pure-text fields, in which case lt/gt/eq rules simply never match).
export async function checkThresholdRules(assetId, fieldKey, numericValue) {
const { rows: assetRows } = await pool.query('SELECT asset_type FROM plant_assets WHERE id = $1', [assetId])
if (!assetRows.length) return
const assetType = assetRows[0].asset_type
const rules = await findRules(fieldKey, ['lt', 'gt', 'eq'], assetId, assetType)
for (const rule of rules) {
if (breaches(rule.condition, rule.threshold, numericValue)) {
await triggerAlert(rule, assetId, fieldKey, numericValue)
} else {
await autoResolveIfOpen(rule.id, assetId, fieldKey)
}
}
}
// Called by lib/scheduler.js's periodic sweep — every active stale_minutes
// rule, against every asset it applies to (its own asset_id, or every active
// asset of its asset_type).
export async function checkStaleRules() {
const { rows: rules } = await pool.query(
`SELECT * FROM plant_alert_rules WHERE active = TRUE AND condition = 'stale_minutes'`
)
if (!rules.length) return
for (const rule of rules) {
const { rows: assets } = await pool.query(
`SELECT id FROM plant_assets
WHERE active = TRUE
AND ($1::int IS NULL OR id = $1)
AND ($2::text IS NULL OR asset_type = $2)`,
[rule.asset_id, rule.asset_type]
)
for (const asset of assets) {
const { rows: latest } = await pool.query(
`SELECT updated_at FROM plant_asset_latest WHERE asset_id = $1 AND field_key = $2`,
[asset.id, rule.field_key]
)
const lastUpdate = latest[0]?.updated_at
const ageMinutes = lastUpdate ? (Date.now() - new Date(lastUpdate).getTime()) / 60000 : Infinity
if (ageMinutes >= Number(rule.threshold)) {
await triggerAlert(rule, asset.id, rule.field_key, `stale for ${Math.round(ageMinutes)}m`)
} else {
await autoResolveIfOpen(rule.id, asset.id, rule.field_key)
}
}
}
}

77
backend/src/lib/mailer.js Normal file
View file

@ -0,0 +1,77 @@
import nodemailer from 'nodemailer'
import { getSetting } from '../db.js'
const SETTINGS_URL = process.env.SETTINGS_URL || ''
const SETTINGS_SECRET = process.env.SETTINGS_SECRET || ''
let _smtpCache = null // { config, expires_at }
let _transporter = null
async function getSmtpConfig() {
if (_smtpCache && Date.now() < _smtpCache.expires_at) return _smtpCache.config
const res = await fetch(`${SETTINGS_URL}/settings/api/internal/integration/smtp`, {
headers: { Authorization: `Bearer ${SETTINGS_SECRET}` },
signal: AbortSignal.timeout(5000),
})
if (!res.ok) throw new Error(`Failed to fetch SMTP config from settings: ${res.status}`)
const config = await res.json()
if (!config.host) throw new Error('SMTP not configured in settings')
_smtpCache = { config, expires_at: Date.now() + 5 * 60_000 }
_transporter = null
return config
}
async function getTransporter() {
if (_transporter) return _transporter
const config = await getSmtpConfig()
const port = parseInt(config.port || '587')
_transporter = nodemailer.createTransport({
host: config.host,
port,
secure: port === 465,
auth: config.user ? { user: config.user, pass: config.pass } : undefined,
})
return _transporter
}
const HOTEL_NAME = process.env.VITE_HOTEL_NAME || 'Hotel'
// Fire-and-forget: email failure must never block telemetry ingestion or the alert write.
async function send(to, subject, text) {
if (!to) return
try {
const config = await getSmtpConfig()
const transport = await getTransporter()
await transport.sendMail({
from: config.from || `"${HOTEL_NAME} Plant Room" <noreply@localhost>`,
to,
subject,
text,
})
} catch (err) {
console.error(`Plant mail to ${to} failed: ${err.message}`)
}
}
const CONDITION_LABEL = { lt: 'below', gt: 'above', eq: 'equal to', stale_minutes: 'stale beyond (minutes)' }
export async function notifyPlantAlert(alert, asset, rule) {
const to = await getSetting('alert_notify_email')
if (!to) return
const lines = [
`Asset: ${asset.name} (${asset.asset_type})`,
`Location: ${asset.location || 'n/a'}`,
`Field: ${rule.field_key}`,
`Condition: ${rule.field_key} ${CONDITION_LABEL[rule.condition] || rule.condition} ${rule.threshold}`,
`Value: ${alert.value_at_trigger}`,
`Severity: ${rule.severity}`,
`Triggered: ${new Date(alert.triggered_at).toLocaleString('en-GB')}`,
]
return send(
to,
`[Plant] ${rule.severity.toUpperCase()}: ${asset.name} — ${rule.field_key}`,
lines.join('\n')
)
}

157
backend/src/lib/mqtt.js Normal file
View file

@ -0,0 +1,157 @@
// MQTT client for plant-room asset telemetry. Connection/backoff skeleton
// follows utilities/backend/src/lib/mqtt.js's shape (own reconnect/backoff,
// credentials fetched at runtime from settings). Unlike utilities' single
// exact-topic -> meter_id map, this subscribes to a whole prefix per asset
// (`<mqtt_topic_prefix>/#`) since one plant asset publishes many fields under
// its own topic tree (e.g. the water softener's salt_level_pct,
// tank_a_capacity_remaining, regeneration_active, ...). The topic segment(s)
// after the prefix become the field_key — generic for any asset type, never
// hardcoded per device.
import mqtt from 'mqtt'
import { pool } from '../db.js'
import { checkThresholdRules } from './alert-engine.js'
const CLIENT_NAME = 'plant-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 prefixMap = [] // [{ prefix, assetId }], longest prefix first so overlapping prefixes resolve to the most specific asset
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 refreshAssetMap() {
const { rows } = await pool.query(
`SELECT id, mqtt_topic_prefix FROM plant_assets WHERE mqtt_topic_prefix IS NOT NULL AND active = TRUE`
)
prefixMap = rows
.map(r => ({ prefix: r.mqtt_topic_prefix.replace(/\/+$/, ''), assetId: r.id }))
.sort((a, b) => b.prefix.length - a.prefix.length)
if (client?.connected) subscribeToKnownTopics()
}
function subscribeToKnownTopics() {
const nextTopics = new Set(prefixMap.map(p => `${p.prefix}/#`))
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 resolveAsset(topic) {
for (const p of prefixMap) {
if (topic === p.prefix || topic.startsWith(p.prefix + '/')) return p
}
return null
}
// Booleans (e.g. regeneration_active: "true"/"false") get both a text value
// and a 0/1 numeric value so they can still be charted or used in lt/gt rules;
// enums (e.g. salt_level_status: "ok"/"low") only ever populate value_text.
function parseValue(raw) {
const text = raw.trim()
const lower = text.toLowerCase()
if (lower === 'true' || lower === 'false') return { numeric: lower === 'true' ? 1 : 0, text }
const num = Number(text)
return { numeric: Number.isFinite(num) && text !== '' ? num : null, text }
}
async function handleMessage(topic, message) {
const match = resolveAsset(topic)
if (!match) return
const remainder = topic.slice(match.prefix.length).replace(/^\/+/, '')
const fieldKey = (remainder ? remainder.replace(/\//g, '_') : topic)
const { numeric, text } = parseValue(message.toString())
try {
await pool.query(
`INSERT INTO plant_telemetry_raw (asset_id, field_key, value_numeric, value_text) VALUES ($1, $2, $3, $4)`,
[match.assetId, fieldKey, numeric, text]
)
await pool.query(
`INSERT INTO plant_asset_latest (asset_id, field_key, value_numeric, value_text, updated_at)
VALUES ($1, $2, $3, $4, NOW())
ON CONFLICT (asset_id, field_key) DO UPDATE
SET value_numeric = EXCLUDED.value_numeric, value_text = EXCLUDED.value_text, updated_at = NOW()`,
[match.assetId, fieldKey, numeric, text]
)
await checkThresholdRules(match.assetId, fieldKey, numeric)
} catch (err) {
console.error('[mqtt] telemetry insert/alert-check 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 refreshAssetMap()
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
}

26
backend/src/lib/scheduler.js Normal file
View file

@ -0,0 +1,26 @@
// Periodic sweep for 'stale_minutes' alert rules only — breach-on-message
// (lt/gt/eq) is handled inline as telemetry arrives, in lib/mqtt.js. This
// stack has no cron library anywhere (see utilities/backend/src/lib/scheduler.js);
// a 1-minute setInterval wall-clock poll is plenty granular for a rule whose
// own threshold is expressed in minutes.
import { checkStaleRules } from './alert-engine.js'
const SWEEP_MS = 60 * 1000
let running = false
export function startScheduler() {
const tick = async () => {
if (running) return
running = true
try {
await checkStaleRules()
} catch (err) {
console.error('[scheduler] stale-rule sweep failed:', err.message)
} finally {
running = false
}
}
setTimeout(tick, 10_000)
setInterval(tick, SWEEP_MS)
}

93
backend/src/routes/alert-rules.js Normal file
View file

@ -0,0 +1,93 @@
import { requireAuth, requireCap } from '../auth.js'
import { pool } from '../db.js'
const CONDITIONS = ['lt', 'gt', 'eq', 'stale_minutes']
const SEVERITIES = ['warning', 'critical']
const ASSET_TYPES = ['boiler', 'water_softener', 'calorifier', 'pump']
export async function alertRuleRoutes(app) {
app.addHook('preHandler', requireAuth)
// All of alert-rules CRUD (including read) is gated on manage_assets — this
// is an admin/maintenance-lead configuration surface, not something every
// 'view' user needs to see; open alerts themselves are visible via /api/alerts.
app.addHook('preHandler', requireCap('manage_assets'))
app.get('/api/alert-rules', async () => {
const { rows } = await pool.query(`
SELECT r.*, a.name AS asset_name
FROM plant_alert_rules r
LEFT JOIN plant_assets a ON a.id = r.asset_id
ORDER BY r.created_at DESC
`)
return rows
})
app.post('/api/alert-rules', async (req, reply) => {
const b = req.body || {}
if (!b.field_key || !String(b.field_key).trim()) return reply.status(400).send({ error: 'field_key is required' })
if (!CONDITIONS.includes(b.condition)) return reply.status(400).send({ error: `condition must be one of: ${CONDITIONS.join(', ')}` })
if (b.threshold === undefined || b.threshold === null || isNaN(Number(b.threshold))) {
return reply.status(400).send({ error: 'threshold must be a number' })
}
if (b.severity && !SEVERITIES.includes(b.severity)) return reply.status(400).send({ error: `severity must be one of: ${SEVERITIES.join(', ')}` })
if (!b.asset_id && !b.asset_type) return reply.status(400).send({ error: 'Either asset_id or asset_type must be set' })
if (b.asset_type && !ASSET_TYPES.includes(b.asset_type)) return reply.status(400).send({ error: `asset_type must be one of: ${ASSET_TYPES.join(', ')}` })
const { rows } = await pool.query(
`INSERT INTO plant_alert_rules (asset_id, asset_type, field_key, condition, threshold, severity, active)
VALUES ($1,$2,$3,$4,$5,$6,$7)
RETURNING *`,
[
b.asset_id || null, b.asset_id ? null : b.asset_type, b.field_key.trim(), b.condition,
Number(b.threshold), b.severity || 'warning', b.active !== undefined ? b.active : true,
]
)
return reply.status(201).send(rows[0])
})
app.patch('/api/alert-rules/:id', async (req, reply) => {
const { rows: existing } = await pool.query('SELECT * FROM plant_alert_rules WHERE id = $1', [req.params.id])
if (!existing.length) return reply.status(404).send({ error: 'Rule not found' })
const cur = existing[0]
const b = req.body || {}
if (b.condition !== undefined && !CONDITIONS.includes(b.condition)) {
return reply.status(400).send({ error: `condition must be one of: ${CONDITIONS.join(', ')}` })
}
if (b.severity !== undefined && !SEVERITIES.includes(b.severity)) {
return reply.status(400).send({ error: `severity must be one of: ${SEVERITIES.join(', ')}` })
}
if (b.asset_type !== undefined && b.asset_type !== null && !ASSET_TYPES.includes(b.asset_type)) {
return reply.status(400).send({ error: `asset_type must be one of: ${ASSET_TYPES.join(', ')}` })
}
const next = {
asset_id: b.asset_id !== undefined ? b.asset_id : cur.asset_id,
asset_type: b.asset_type !== undefined ? b.asset_type : cur.asset_type,
field_key: b.field_key !== undefined ? b.field_key : cur.field_key,
condition: b.condition !== undefined ? b.condition : cur.condition,
threshold: b.threshold !== undefined ? Number(b.threshold) : cur.threshold,
severity: b.severity !== undefined ? b.severity : cur.severity,
active: b.active !== undefined ? b.active : cur.active,
}
if (next.asset_id) next.asset_type = null
if (!next.asset_id && !next.asset_type) {
return reply.status(400).send({ error: 'Either asset_id or asset_type must be set' })
}
const { rows } = await pool.query(
`UPDATE plant_alert_rules SET
asset_id = $1, asset_type = $2, field_key = $3, condition = $4, threshold = $5, severity = $6, active = $7
WHERE id = $8
RETURNING *`,
[next.asset_id, next.asset_type, next.field_key, next.condition, next.threshold, next.severity, next.active, req.params.id]
)
return rows[0]
})
app.delete('/api/alert-rules/:id', async (req, reply) => {
const { rows } = await pool.query('DELETE FROM plant_alert_rules WHERE id = $1 RETURNING id', [req.params.id])
if (!rows.length) return reply.status(404).send({ error: 'Rule not found' })
return { ok: true }
})
}

64
backend/src/routes/alerts.js Normal file
View file

@ -0,0 +1,64 @@
import { requireAuth, requireCap } from '../auth.js'
import { pool } from '../db.js'
const STATUSES = ['open', 'acknowledged', 'resolved']
const SEVERITIES = ['warning', 'critical']
export async function alertRoutes(app) {
app.addHook('preHandler', requireAuth)
// GET /api/alerts?status=open&severity=critical — joined with asset + rule detail
app.get('/api/alerts', { preHandler: requireCap('view') }, async (req) => {
const { status, severity } = req.query || {}
const conditions = []
const params = []
if (status && STATUSES.includes(status)) {
params.push(status)
conditions.push(`al.status = $${params.length}`)
}
if (severity && SEVERITIES.includes(severity)) {
params.push(severity)
conditions.push(`r.severity = $${params.length}`)
}
const where = conditions.length ? `WHERE ${conditions.join(' AND ')}` : ''
const { rows } = await pool.query(
`SELECT al.*, a.name AS asset_name, a.asset_type, r.condition, r.threshold, r.severity
FROM plant_alerts al
JOIN plant_assets a ON a.id = al.asset_id
JOIN plant_alert_rules r ON r.id = al.rule_id
${where}
ORDER BY al.triggered_at DESC
LIMIT 500`,
params
)
return rows
})
// PATCH /api/alerts/:id — { action: 'acknowledge' | 'resolve' }
app.patch('/api/alerts/:id', { preHandler: requireCap('acknowledge_alerts') }, async (req, reply) => {
const { rows: existing } = await pool.query('SELECT * FROM plant_alerts WHERE id = $1', [req.params.id])
if (!existing.length) return reply.status(404).send({ error: 'Alert not found' })
const { action } = req.body || {}
if (!['acknowledge', 'resolve'].includes(action)) {
return reply.status(400).send({ error: "action must be 'acknowledge' or 'resolve'" })
}
let rows
if (action === 'acknowledge') {
;({ rows } = await pool.query(
`UPDATE plant_alerts SET status = 'acknowledged', acknowledged_at = NOW(), acknowledged_by = $1
WHERE id = $2 RETURNING *`,
[req.user.email, req.params.id]
))
} else {
;({ rows } = await pool.query(
`UPDATE plant_alerts SET status = 'resolved', resolved_at = NOW() WHERE id = $1 RETURNING *`,
[req.params.id]
))
}
return rows[0]
})
}

152
backend/src/routes/assets.js Normal file
View file

@ -0,0 +1,152 @@
import { mkdir, unlink, writeFile } from 'fs/promises'
import { randomUUID } from 'crypto'
import { join } from 'path'
import sharp from 'sharp'
import { requireAuth, requireCap } from '../auth.js'
import { pool } from '../db.js'
import { refreshAssetMap } from '../lib/mqtt.js'
const ALLOWED_IMAGES = ['image/jpeg', 'image/jpg', 'image/png', 'image/webp']
const PHOTO_TYPES = ['asset', 'serial_plate']
const ASSET_TYPES = ['boiler', 'water_softener', 'calorifier', 'pump']
const WRITABLE_FIELDS = [
'name', 'asset_type', 'location', 'make_model', 'serial_no',
'install_date', 'notes', 'active', 'mqtt_topic_prefix',
]
export async function assetRoutes(app, opts) {
const UPLOADS_DIR = opts.uploadsDir
app.addHook('preHandler', requireAuth)
// GET /api/assets — every asset, with photo count
app.get('/api/assets', { preHandler: requireCap('view') }, async () => {
const { rows } = await pool.query(`
SELECT a.*, (SELECT COUNT(*)::int FROM asset_photos p WHERE p.asset_id = a.id) AS photo_count
FROM plant_assets a
ORDER BY a.asset_type, a.name
`)
return rows
})
app.get('/api/assets/:id', { preHandler: requireCap('view') }, async (req, reply) => {
const { rows } = await pool.query('SELECT * FROM plant_assets WHERE id = $1', [req.params.id])
if (!rows.length) return reply.status(404).send({ error: 'Asset not found' })
return rows[0]
})
// POST /api/assets — create
app.post('/api/assets', { preHandler: requireCap('manage_assets') }, async (req, reply) => {
const b = req.body || {}
if (!b.name || !String(b.name).trim()) return reply.status(400).send({ error: 'name is required' })
if (!ASSET_TYPES.includes(b.asset_type)) {
return reply.status(400).send({ error: `asset_type must be one of: ${ASSET_TYPES.join(', ')}` })
}
const { rows } = await pool.query(
`INSERT INTO plant_assets
(name, asset_type, location, make_model, serial_no, install_date, notes, active, mqtt_topic_prefix)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
RETURNING *`,
[
b.name.trim(), b.asset_type, b.location || null, b.make_model || null, b.serial_no || null,
b.install_date || null, b.notes || null, b.active !== undefined ? b.active : true,
b.mqtt_topic_prefix || null,
]
)
await refreshAssetMap().catch(err => app.log.warn(`mqtt asset map refresh failed: ${err.message}`))
return reply.status(201).send(rows[0])
})
// PATCH /api/assets/:id — partial update
app.patch('/api/assets/:id', { preHandler: requireCap('manage_assets') }, async (req, reply) => {
const { rows: existing } = await pool.query('SELECT * FROM plant_assets WHERE id = $1', [req.params.id])
if (!existing.length) return reply.status(404).send({ error: 'Asset not found' })
const current = existing[0]
const b = req.body || {}
if (b.asset_type !== undefined && !ASSET_TYPES.includes(b.asset_type)) {
return reply.status(400).send({ error: `asset_type must be one of: ${ASSET_TYPES.join(', ')}` })
}
const next = { ...current }
for (const field of WRITABLE_FIELDS) {
if (b[field] !== undefined) next[field] = b[field]
}
const { rows } = await pool.query(
`UPDATE plant_assets SET
name = $1, asset_type = $2, location = $3, make_model = $4, serial_no = $5,
install_date = $6, notes = $7, active = $8, mqtt_topic_prefix = $9
WHERE id = $10
RETURNING *`,
[
next.name, next.asset_type, next.location, next.make_model, next.serial_no,
next.install_date, next.notes, next.active, next.mqtt_topic_prefix, req.params.id,
]
)
await refreshAssetMap().catch(err => app.log.warn(`mqtt asset map refresh failed: ${err.message}`))
return rows[0]
})
// POST /api/assets/:id/photos — multipart: file + photo_type (asset | serial_plate)
app.post('/api/assets/:id/photos', { preHandler: requireCap('manage_assets') }, async (req, reply) => {
const assetId = parseInt(req.params.id)
const { rows } = await pool.query('SELECT id FROM plant_assets WHERE id = $1', [assetId])
if (!rows.length) return reply.status(404).send({ error: 'Asset not found' })
let fileData = null, photoType = 'asset'
for await (const part of req.parts()) {
if (part.type === 'file') {
if (!ALLOWED_IMAGES.includes(part.mimetype)) {
return reply.status(400).send({ error: 'Only JPEG, PNG and WebP images are allowed' })
}
const chunks = []
for await (const chunk of part.file) chunks.push(chunk)
const raw = Buffer.concat(chunks)
// Auto-rotate (phone EXIF), resize to 1800px max, re-encode as JPEG — same
// pattern as hvac's device_photos / maintenance's task_photos.
const processed = await sharp(raw)
.rotate()
.resize(1800, 1800, { fit: 'inside', withoutEnlargement: true })
.jpeg({ quality: 82, progressive: true })
.toBuffer()
const filename = randomUUID() + '.jpg'
const dir = join(UPLOADS_DIR, 'assets', String(assetId))
await mkdir(dir, { recursive: true })
await writeFile(join(dir, filename), processed)
fileData = { originalName: part.filename, savedAs: filename, size: processed.length }
} else {
const val = await part.value
if (part.fieldname === 'photo_type' && PHOTO_TYPES.includes(String(val))) photoType = String(val)
}
}
if (!fileData?.savedAs) return reply.status(400).send({ error: 'No file uploaded' })
const filePath = `/assets/${assetId}/${fileData.savedAs}`
const { rows: ins } = await pool.query(
`INSERT INTO asset_photos (asset_id, file_name, file_path, mime_type, file_size, photo_type, uploaded_by)
VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING *`,
[assetId, fileData.originalName, filePath, 'image/jpeg', fileData.size, photoType, req.user.email]
)
return ins[0]
})
app.get('/api/assets/:id/photos', { preHandler: requireCap('view') }, async (req) => {
const { rows } = await pool.query(
'SELECT * FROM asset_photos WHERE asset_id = $1 ORDER BY uploaded_at DESC',
[req.params.id]
)
return rows
})
app.delete('/api/photos/:id', { preHandler: requireCap('manage_assets') }, async (req, reply) => {
const { rows } = await pool.query('SELECT * FROM asset_photos WHERE id = $1', [req.params.id])
if (!rows.length) return reply.status(404).send({ error: 'Not found' })
await unlink(join(UPLOADS_DIR, rows[0].file_path)).catch(() => {})
await pool.query('DELETE FROM asset_photos WHERE id = $1', [req.params.id])
return { ok: true }
})
}

42
backend/src/routes/settings.js Normal file
View file

@ -0,0 +1,42 @@
import { requireAuth, requireCap } from '../auth.js'
import { pool } from '../db.js'
import { isConnected as mqttConnected } from '../lib/mqtt.js'
// Standard app_settings pattern (see conventions doc's "Cross-app integration
// credentials" section) — plant only has one setting today (who gets alert
// emails), but the shape is kept generic so a future integration is a one-row
// addition to ALLOWED_KEYS + the seed default in db.js, not a new pattern.
const ALLOWED_KEYS = new Set(['alert_notify_email'])
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
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, value ?? '']
)
}
return { ok: true }
})
app.get('/api/settings/mqtt-status', { preHandler: requireCap('settings') }, async () => {
return {
connected: mqttConnected(),
note: mqttConnected()
? 'Connected to the shared MQTT broker.'
: 'Not connected — the shared MQTT broker (LXC 104) may be unreachable, or plant-backend has no credentials in settings.',
}
})
}

54
backend/src/routes/status.js Normal file
View file

@ -0,0 +1,54 @@
import { requireAuth, requireCap } from '../auth.js'
import { pool } from '../db.js'
import { isConnected as mqttConnected } from '../lib/mqtt.js'
const SEVERITY_RANK = { critical: 2, warning: 1 }
export async function statusRoutes(app) {
app.addHook('preHandler', requireAuth)
// GET /api/status — every active asset, its latest telemetry fields, and a
// worst-open-alert severity for the Dashboard's per-card colour + top banner.
app.get('/api/status', { preHandler: requireCap('view') }, async () => {
const { rows: assets } = await pool.query(`
SELECT * FROM plant_assets WHERE active = TRUE ORDER BY asset_type, name
`)
const { rows: latest } = await pool.query(`
SELECT asset_id, field_key, value_numeric, value_text, updated_at
FROM plant_asset_latest
`)
const latestByAsset = new Map()
for (const l of latest) {
if (!latestByAsset.has(l.asset_id)) latestByAsset.set(l.asset_id, [])
latestByAsset.get(l.asset_id).push(l)
}
const { rows: openAlerts } = await pool.query(`
SELECT al.asset_id, r.severity
FROM plant_alerts al
JOIN plant_alert_rules r ON r.id = al.rule_id
WHERE al.status = 'open'
`)
const alertSummaryByAsset = new Map()
for (const a of openAlerts) {
const cur = alertSummaryByAsset.get(a.asset_id) || { count: 0, max_severity: null }
cur.count++
if (!cur.max_severity || SEVERITY_RANK[a.severity] > SEVERITY_RANK[cur.max_severity]) {
cur.max_severity = a.severity
}
alertSummaryByAsset.set(a.asset_id, cur)
}
return {
open_alerts: openAlerts.length,
assets: assets.map(a => ({
...a,
latest: latestByAsset.get(a.id) || [],
open_alert_count: alertSummaryByAsset.get(a.id)?.count || 0,
max_severity: alertSummaryByAsset.get(a.id)?.max_severity || null,
})),
mqtt_connected: mqttConnected(),
}
})
}