Add per-meter MQTT value scale + expose device topic/scale in Meters UI

The water-softener totaliser publishes litres but the water category is
m3 — without a conversion factor its readings would land in the DB
1000x too high. Also, mqtt_topic was only settable via direct SQL since
the Meters form never exposed it.

backend/src/db.js

@@ -47,6 +47,12 @@ export async function initDb() {
     ALTER TABLE meters ADD COLUMN IF NOT EXISTS mqtt_topic TEXT;
     CREATE UNIQUE INDEX IF NOT EXISTS meters_mqtt_topic_uidx ON meters (mqtt_topic) WHERE mqtt_topic IS NOT NULL;
 
+    -- multiplied onto a raw device-published value before it's stored, so a
+    -- device publishing in a different unit than the category's billed unit
+    -- (e.g. the water-softener totaliser publishes litres, category is m3)
+    -- lands correctly with no per-category special-casing. 1 = no conversion.
+    ALTER TABLE meters ADD COLUMN IF NOT EXISTS mqtt_value_scale NUMERIC(12,6) NOT NULL DEFAULT 1;
+
     -- gas meters read raw volume off the dial (m3 on newer metric meters, ft3 on
     -- older imperial ones like ours), never kWh directly — these two columns let
     -- cost-calc.js convert a volume reading to billed kWh using the same

backend/src/lib/mqtt.js

@@ -14,7 +14,7 @@ const BROKER_RECONNECT_BACKOFF_MS = [30, 60, 120, 300, 600, 1800].map(s => s * 1
 let client = null
 let connecting = false
 let reconnectAttempt = 0
-let topicMap = new Map() // mqtt_topic -> meter_id
+let topicMap = new Map() // mqtt_topic -> { meterId, scale }
 let subscribedTopics = new Set()
 
 async function getCredentials() {
@@ -31,9 +31,9 @@ async function getCredentials() {
 
 export async function refreshTopicMap() {
   const { rows } = await pool.query(
-    `SELECT id, mqtt_topic FROM meters WHERE mqtt_topic IS NOT NULL AND active = TRUE`
+    `SELECT id, mqtt_topic, mqtt_value_scale FROM meters WHERE mqtt_topic IS NOT NULL AND active = TRUE`
   )
-  topicMap = new Map(rows.map(r => [r.mqtt_topic, r.id]))
+  topicMap = new Map(rows.map(r => [r.mqtt_topic, { meterId: r.id, scale: Number(r.mqtt_value_scale) }]))
   if (client?.connected) subscribeToKnownTopics()
 }
 
@@ -47,11 +47,12 @@ function subscribeToKnownTopics() {
 }
 
 function handleMessage(topic, message) {
-  const meterId = topicMap.get(topic)
+  const entry = topicMap.get(topic)
-  if (!meterId) return
+  if (!entry) return
-  const value = Number(message.toString().trim())
+  const raw = Number(message.toString().trim())
-  if (!Number.isFinite(value)) return
+  if (!Number.isFinite(raw)) return
-  pool.query(`INSERT INTO meter_telemetry_raw (meter_id, value) VALUES ($1, $2)`, [meterId, value])
+  const value = raw * entry.scale
+  pool.query(`INSERT INTO meter_telemetry_raw (meter_id, value) VALUES ($1, $2)`, [entry.meterId, value])
     .catch(err => console.error('[mqtt] telemetry insert failed:', err.message))
 }
 

backend/src/routes/meters.js

@@ -93,11 +93,12 @@ export async function meterRoutes(app, opts) {
     if (!b.name || !b.category_id) return reply.status(400).send({ error: 'name and category_id required' })
     const { rows } = await pool.query(
       `INSERT INTO meters (category_id, parent_meter_id, name, location, serial_number, install_date, notes, mqtt_topic,
-                           gas_volume_unit, gas_calorific_value)
+                           mqtt_value_scale, gas_volume_unit, gas_calorific_value)
-       VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) RETURNING *`,
+       VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) RETURNING *`,
       [
         b.category_id, b.parent_meter_id || null, b.name, b.location || null, b.serial_number || null,
         b.install_date || null, b.notes || null, b.mqtt_topic || null,
+        b.mqtt_value_scale || 1,
         b.gas_volume_unit || null, b.gas_calorific_value || null,
       ]
     )
@@ -116,8 +117,8 @@ export async function meterRoutes(app, opts) {
     const { rows } = await pool.query(
       `UPDATE meters SET category_id = $1, parent_meter_id = $2, name = $3, location = $4,
                          serial_number = $5, install_date = $6, notes = $7, active = $8, mqtt_topic = $9,
-                         gas_volume_unit = $10, gas_calorific_value = $11
+                         mqtt_value_scale = $10, gas_volume_unit = $11, gas_calorific_value = $12
-       WHERE id = $12 RETURNING *`,
+       WHERE id = $13 RETURNING *`,
       [
         b.category_id ?? m.category_id,
         b.parent_meter_id !== undefined ? b.parent_meter_id : m.parent_meter_id,
@@ -128,6 +129,7 @@ export async function meterRoutes(app, opts) {
         b.notes !== undefined ? b.notes : m.notes,
         b.active ?? m.active,
         b.mqtt_topic !== undefined ? b.mqtt_topic : m.mqtt_topic,
+        b.mqtt_value_scale !== undefined ? b.mqtt_value_scale : m.mqtt_value_scale,
         b.gas_volume_unit !== undefined ? b.gas_volume_unit : m.gas_volume_unit,
         b.gas_calorific_value !== undefined ? b.gas_calorific_value : m.gas_calorific_value,
         req.params.id,

frontend/src/pages/Meters.tsx

@@ -10,6 +10,7 @@ const emptyForm = {
   id: 0, category_id: 0, parent_meter_id: '' as number | '', name: '', location: '',
   serial_number: '', install_date: '', notes: '', active: true,
   gas_volume_unit: '' as 'm3' | 'ft3' | '', gas_calorific_value: '' as number | '',
+  mqtt_topic: '', mqtt_value_scale: 1 as number | '',
 }
 
 export default function Meters() {
@@ -54,6 +55,7 @@ export default function Meters() {
       install_date: m.install_date ? m.install_date.slice(0, 10) : '', notes: m.notes || '',
       active: m.active,
       gas_volume_unit: m.gas_volume_unit || '', gas_calorific_value: m.gas_calorific_value != null ? Number(m.gas_calorific_value) : '',
+      mqtt_topic: m.mqtt_topic || '', mqtt_value_scale: m.mqtt_value_scale != null ? Number(m.mqtt_value_scale) : 1,
     })
     setImageFile(null)
   }
@@ -76,6 +78,8 @@ export default function Meters() {
         active: form.active,
         gas_volume_unit: isGas && form.gas_volume_unit ? form.gas_volume_unit : null,
         gas_calorific_value: isGas && form.gas_calorific_value !== '' ? form.gas_calorific_value : null,
+        mqtt_topic: form.mqtt_topic.trim() || null,
+        mqtt_value_scale: form.mqtt_value_scale === '' ? 1 : form.mqtt_value_scale,
       }
       const saved = form.id ? await api.updateMeter(form.id, body) : await api.createMeter(body)
       if (imageFile) await api.uploadMeterImage(saved.id, imageFile)
@@ -220,6 +224,21 @@ export default function Meters() {
               </div>
             )}
 
+            <div className="field-row">
+              <div className="field">
+                <label>MQTT topic (device feed)</label>
+                <input type="text" value={form.mqtt_topic} onChange={e => setForm({ ...form, mqtt_topic: e.target.value })}
+                  placeholder="e.g. utilities/water-softener/total_usage_l" />
+                <div className="field-hint">Leave blank for a manually-read meter. If set, readings are ingested automatically from the shared MQTT broker.</div>
+              </div>
+              <div className="field">
+                <label>Device value scale</label>
+                <input type="number" step="0.000001" value={form.mqtt_value_scale}
+                  onChange={e => setForm({ ...form, mqtt_value_scale: e.target.value === '' ? '' : parseFloat(e.target.value) })} />
+                <div className="field-hint">Multiplied onto the raw published value, e.g. 0.001 if the device publishes litres but this meter's unit is m³.</div>
+              </div>
+            </div>
+
             <div className="field-row">
               <div className="field">
                 <label>Parent meter (sub-metering)</label>

frontend/src/types.ts

@@ -40,6 +40,8 @@ export interface Meter {
   reading_unit_label: string
   gas_volume_unit: 'm3' | 'ft3' | null
   gas_calorific_value: number | null
+  mqtt_topic: string | null
+  mqtt_value_scale: number
   parent_meter_id: number | null
   parent_name: string | null
   name: string