feat: add support for later forecasts and implement force update functionality for rules

realtime/src/index.js

@@ -44,6 +44,29 @@ app.use('/connect', require('./routes/connect'));
 app.use('/sensors', require('./routes/sensors'));
 app.use('/sessions', require('./routes/sessions'));
 
+/**
+ * POST /push-rules — Riceve rules attive dall'API e le pusha a tutti i sensori connessi.
+ * Autenticato con x-api-key (service-to-service).
+ */
+app.post('/push-rules', (req, res) => {
+    const apiKey = req.headers['x-api-key'];
+    if (!apiKey || apiKey !== process.env.INTERNAL_API_KEY) {
+        return res.status(401).json({ error: 'unauthorized' });
+    }
+
+    const payload = req.body;
+    if (!payload || Object.keys(payload).length === 0) {
+        return res.status(400).json({ error: 'empty payload' });
+    }
+
+    // Wrappa con _t per identificare il tipo di messaggio nel plugin
+    const message = { _t: 'rules_update', ...payload };
+    const sensors = wsHandler.pushToAllSensors(message);
+
+    console.log(`[PUSH-RULES] Inviato a ${sensors} sensori:`, Object.keys(payload));
+    res.json({ status: 'ok', sensors });
+});
+
 const server = app.listen(3000, '0.0.0.0', () => {
     console.log(`Realtime started`);
 });

realtime/src/store/influx.js

@@ -13,6 +13,7 @@ const writeApi = client.getWriteApi(org, bucket, 'ms', {
     batchSize: 50,
 });
 
+// Mapping legacy per sensor_data (logs telemetry)
 const fieldMap = {
     t: 'temperature',
     h: 'humidity',
@@ -24,6 +25,10 @@ const fieldMap = {
     lon: 'longitude',
 };
 
+/**
+ * Scrive dati telemetria sensore (logs) con mapping campi abbreviati.
+ * Measurement: sensor_data
+ */
 function writeSensorData(fields, sensor, session, timestamp) {
     const point = new Point('sensor_data')
         .tag('sensor', sensor)
@@ -39,6 +44,46 @@ function writeSensorData(fields, sensor, session, timestamp) {
     writeApi.writePoint(point);
 }
 
+/**
+ * Scrive dati generici (weather, forecast, ecc.) senza mapping.
+ * I campi vengono scritti con il nome originale (ref da Open-Meteo).
+ * @param {string} measurement - nome della measurement InfluxDB (es. 'weather_current', 'weather_forecast')
+ * @param {Object} fields - campi { ref: value }
+ * @param {string} sensor - nome del sensore
+ * @param {string} session - id sessione
+ * @param {number} timestamp - timestamp unix ms
+ */
+function writeGenericData(measurement, fields, sensor, session, timestamp) {
+    const point = new Point(measurement)
+        .tag('sensor', sensor)
+        .tag('session', session)
+        .timestamp(timestamp);
+
+    for (const [key, value] of Object.entries(fields)) {
+        if (value === null || value === undefined) continue;
+        if (typeof value === 'number') {
+            point.floatField(key, value);
+        } else if (typeof value === 'string') {
+            point.stringField(key, value);
+        }
+    }
+
+    writeApi.writePoint(point);
+}
+
+/**
+ * Scrive un batch di punti forecast (previsioni orarie).
+ * Ogni punto ha il proprio timestamp.
+ * @param {Array} points - array di [timestamp_ms, { ref: value, ... }]
+ * @param {string} sensor - nome del sensore
+ * @param {string} session - id sessione
+ */
+function writeForecastBatch(points, sensor, session) {
+    for (const [ts, fields] of points) {
+        writeGenericData('weather_forecast', fields, sensor, session, ts);
+    }
+}
+
 async function queryHistory(sensor, session, since) {
     const queryApi = client.getQueryApi(org);
     const query = `
@@ -62,4 +107,4 @@ async function queryHistory(sensor, session, since) {
     });
 }
 
-module.exports = { writeSensorData, queryHistory };
+module.exports = { writeSensorData, writeGenericData, writeForecastBatch, queryHistory };

realtime/src/ws/handler.js

@@ -1,7 +1,7 @@
 const { WebSocketServer } = require('ws');
 const { decode } = require('@msgpack/msgpack');
 const { consumeConnectionToken, appendAsConnection, query, hset, del } = require('../store/redis');
-const { writeSensorData, queryHistory } = require('../store/influx');
+const { writeSensorData, writeGenericData, writeForecastBatch, queryHistory } = require('../store/influx');
 
 // In-memory registries
 const sensorWatchers = new Map();   // sensorName → Set<WebSocket> (watchers)
@@ -129,8 +129,19 @@ function handleSensorConnection(ws) {
 
             const { ts, _m, ...fields } = packet;
 
-            // Usa sessionLabel (puo' cambiare a runtime dalla console)
-            writeSensorData(fields, sensorName, ws.sessionLabel, ts);
+            // Route per tipo di measurement
+            if (_m === 'weather') {
+                // Dati meteo current — salva con measurement generico
+                writeGenericData('weather_current', fields, sensorName, ws.sessionLabel, ts);
+            } else if (_m === 'forecast_batch') {
+                // Batch previsioni orarie — fields è un array [[ts, {fields}], ...]
+                if (Array.isArray(fields.points)) {
+                    writeForecastBatch(fields.points, sensorName, ws.sessionLabel);
+                }
+            } else {
+                // Dati telemetria sensore (logs) — mapping abbreviato
+                writeSensorData(fields, sensorName, ws.sessionLabel, ts);
+            }
 
             // Broadcast to watchers
             const watchers = sensorWatchers.get(sensorName);
@@ -239,4 +250,27 @@ function handleWatcherConnection(ws) {
     });
 }
 
-module.exports = { setup, connectedSensors };
+/**
+ * Invia un messaggio a tutti i sensori connessi.
+ * Usato dal push-rules endpoint per forzare l'aggiornamento delle rules.
+ * @param {Object} payload - Il payload da inviare (verrà wrappato con _t)
+ * @returns {number} Numero di sensori a cui il messaggio è stato inviato
+ */
+function pushToAllSensors(payload) {
+    const { encode } = require('@msgpack/msgpack');
+    let count = 0;
+    for (const [sensorName, ws] of connectedSensors.entries()) {
+        if (ws.readyState === ws.OPEN) {
+            try {
+                ws.send(encode(payload));
+                console.log(`[PUSH] Rules update inviato a ${sensorName}`);
+                count++;
+            } catch (err) {
+                console.error(`[PUSH] Errore invio a ${sensorName}:`, err.message);
+            }
+        }
+    }
+    return count;
+}
+
+module.exports = { setup, connectedSensors, pushToAllSensors };