Merge branch 'master' of http://35.207.205.18:3000/Arminta/watermanagement-backend

src/controllers/tanksController.js

@@ -2343,7 +2343,8 @@ cron.schedule("* * * * *", async () => {
 //   }
 // };
 
-exports. publishMotorStopStatus = async (motor_id, motor_stop_status) => {
+exports.publishMotorStopStatus = async (motor_id, motor_stop_status) => {
+  const deviceTopic = `water/operation/${motor_id}`; // Target specific IoT
   const payload = {
     topic: 'operation',
     object: {
@@ -2351,9 +2352,12 @@ exports. publishMotorStopStatus = async (motor_id, motor_stop_status) => {
       control: motor_stop_status
     }
   };
-  console.log("enetred publish")
+
-  console.log(payload)
+  console.log(`📡 Publishing to ${deviceTopic}`);
-  client.publish('water/operation', JSON.stringify(payload));
+  console.log(payload);
+
+  // Publish to the specific device's control topic
+  client.publish(deviceTopic, JSON.stringify(payload));
 };
 
 const stat_stop_intervals = {};
@@ -5881,73 +5885,132 @@ exports.getBlockData = async (req, reply) => {
 //     }
 //   }
 // });
+
+
+
+
+
 const mqtt = require('mqtt');
 
-// Connect to MQTT broker
+require('dotenv').config();
-const client = mqtt.connect('mqtt://35.207.198.4:1883');
+
+// **Persistent MQTT Connection**
+const client = mqtt.connect('mqtt://35.207.198.4:1883', {
+  clientId: `mqtt_server_${Math.random().toString(16).substr(2, 8)}`,
+  clean: false, // Ensures MQTT retains subscriptions
+  reconnectPeriod: 2000, // Reconnect every 2 seconds
+});
+
+const subscribedTopics = new Set();
+const activeDevices = new Set(); // Keep track of active devices
 
 
 
 client.on('connect', () => {
-  console.log('Connected to MQTT broker');
+  console.log('✅ Connected to MQTT broker');
-  client.subscribe('water/iot-data/+', (err) => {
+
+  // **Ensure re-subscriptions after reconnect**
+  subscribedTopics.forEach(topic => {
+    client.subscribe(topic, { qos: 1 }, (err) => {
+      if (err) {
+        console.error(`❌ Error resubscribing to ${topic}:`, err);
+      } else {
+        console.log(`🔄 Resubscribed to ${topic}`);
+      }
+    });
+  });
+
+  // **Subscribe to new device announcements**
+  client.subscribe('water/iot-data/announce', { qos: 1 }, (err) => {
     if (err) {
-      console.error('Error subscribing to topic:', err);
+      console.error('❌ Error subscribing to announcement topic:', err);
     } else {
-      console.log('Subscribed to all IoT devices under water/iot-data/#');
+      console.log('📡 Subscribed to water/iot-data/announce');
     }
   });
 });
 
 client.on('message', async (topic, message) => {
-  console.log(`Message received on topic ${topic}:`, message.toString());
+  console.log(`📩 Message received on topic ${topic}:`, message.toString());
 
   try {
-    const topicParts = topic.split('/'); // ['water', 'iot-data', '140924']
+    const data = JSON.parse(message.toString());
-    const hw_Id = topicParts[2]; // Extract hardwareId from topic
 
-    if (!hw_Id) {
+    // **Handle device announcements**
-      console.error('Invalid topic format: Missing hw_Id');
+    if (topic === 'water/iot-data/announce') {
+      if (!data.objects || !data.objects.hw_Id) {
+        console.error("❌ Invalid announcement format. Missing hw_Id.");
         return;
       }
 
-    const data = JSON.parse(message.toString());
+      const hw_Id = data.objects.hw_Id;
-    const { Motor_status, tanks } = data.objects;
+      const deviceTopic = `water/iot-data/${hw_Id}`;
+
+      if (!subscribedTopics.has(deviceTopic)) {
+        client.subscribe(deviceTopic, { qos: 1 }, (err) => {
+          if (err) {
+            console.error(`❌ Error subscribing to ${deviceTopic}:`, err);
+          } else {
+            console.log(`✅ Subscribed to ${deviceTopic}`);
+            subscribedTopics.add(deviceTopic);
+            activeDevices.add(hw_Id);
+            console.log('📡 Active Devices:', Array.from(activeDevices));
+          }
+        });
+      }
+      return;
+    }
 
-    console.log(`Processing data for hardwareId: ${hw_Id}`);
+    // **Process data messages asynchronously to avoid blocking**
+    if (topic.startsWith('water/iot-data/')) {
+      setImmediate(() => {
+        const hw_Id = topic.split('/')[2];
+        processIotData(hw_Id, data);
+      });
+    }
+  } catch (err) {
+    console.error('❌ Error processing message:', err.message);
+  }
+});
 
-    // Get the current date and time in the required format
+client.on('error', (err) => console.error('❌ MQTT Error:', err));
-    const currentTime = moment().tz('Asia/Kolkata');
+client.on('close', () => console.log('⚠️ MQTT Connection Closed.'));
-    const date = currentTime.format('YYYY-MM-DD'); // Adjust format if needed
+client.on('offline', () => console.log('⚠️ MQTT Broker Offline.'));
-    const time = currentTime.format('HH:mm:ss');
+
+async function processIotData(hw_Id, data) {
+  try {
+    const { Motor_status, tanks } = data.objects;
+    console.log(`📡 Processing data for hw_Id: ${hw_Id}`);
+
+    const currentDate = new Date();
+    const date = currentDate.toISOString();
+    const time = currentDate.toLocaleTimeString('en-IN', { hour12: false, timeZone: 'Asia/Kolkata' });
 
-    // Create tank documents
     const tankDocuments = tanks.map(tank => ({
       tankhardwareId: tank.Id,
       tankHeight: tank.level,
       date,
-      time,
+      time
     }));
 
-    // Save IoT data
     const iotTankData = new IotData({
       hardwareId: hw_Id,
       Motor_status,
       tanks: tankDocuments,
       date,
-      time,
+      time
     });
     await iotTankData.save();
 
-    // Delete excess records (keep only the latest three records)
     const recordsToKeep = 3;
     const recordsToDelete = await IotData.find({ hardwareId: hw_Id })
       .sort({ date: -1, time: -1 })
       .skip(recordsToKeep);
 
-    await Promise.all(recordsToDelete.map(record => record.remove()));
+    for (const record of recordsToDelete) {
+      await record.remove();
+    }
 
-    // Process tanks and update water level
     for (const tank of tanks) {
       const { Id: tankhardwareId, level: tankHeight } = tank;
       const existingTank = await Tank.findOne({ hardwareId: hw_Id, tankhardwareId });
@@ -5956,68 +6019,53 @@ client.on('message', async (topic, message) => {
       const customerId = existingTank.customerId;
       const tank_name = existingTank.tankName;
 
-      // Convert height and calculate water level
+      const tankHeightInCm = (parseInt(existingTank.height.replace(/,/g, ''), 10)) * 30.48;
-      const tankHeightInCm = parseInt(existingTank.height.replace(/,/g, ''), 10) * 30.48;
+      const tank_height = parseInt(tankHeightInCm.toFixed(0), 10);
-      const tank_height = Math.round(tankHeightInCm);
       const waterLevelHeight = tank_height - tankHeight;
       const waterCapacityPerCm = parseInt(existingTank.waterCapacityPerCm.replace(/,/g, ''), 10);
-      const waterLevel = Math.max(0, waterLevelHeight * waterCapacityPerCm);
 
-      console.log(`${tankHeight} - Processed in MQTT subscriber`);
+      const waterLevel = parseInt(waterLevelHeight * waterCapacityPerCm, 10);
 
-      if (tankHeight > 0) {
+      if (tankHeight > 0 && waterLevel >= 0) {
         existingTank.waterlevel = waterLevel;
         await existingTank.save();
-
-        for (const outputConnection of existingTank.connections.outputConnections) {
-          const linkedTank = await Tank.findOne({ 
-            customerId, 
-            tankName: outputConnection.outputConnections, 
-            tankLocation: outputConnection.output_type 
-          });
-
-          if (linkedTank) {
-            for (const inputConnection of linkedTank.connections.inputConnections) {
-              if (inputConnection.inputConnections === tank_name) {
-                inputConnection.water_level = waterLevel;
-                await linkedTank.save();
-              }
-            }
-          }
-        }
       }
     }
 
-    // Update motor status
+    const status = Motor_status;
     const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hw_Id });
+
     if (!motorTank) {
-      console.log(`Motor not found for motor_id: ${hw_Id}`);
+      console.log('Motor not found for the specified motor_id');
       return;
     }
 
     const inputConnection = motorTank.connections.inputConnections.find(conn => conn.motor_id === hw_Id);
     if (inputConnection) {
-      inputConnection.motor_status = Motor_status;
+      inputConnection.motor_status = status;
-
+      if (inputConnection.motor_stop_status === "1" && status === 2 && inputConnection.motor_on_type !== "forced_manual") {
-      if (inputConnection.motor_stop_status === "1" && Motor_status === 2 && inputConnection.motor_on_type !== "forced_manual") {
+        const currentTime = moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm');
         inputConnection.motor_stop_status = "2";
         inputConnection.motor_on_type = "forced_manual";
-        inputConnection.startTime = currentTime.format('DD-MMM-YYYY - HH:mm');
+        inputConnection.startTime = currentTime;
       }
 
-      if (inputConnection.motor_stop_status === "2" && Motor_status === 1) {
+      if (inputConnection.motor_stop_status === "2" && status === 1) {
+        const currentTime = moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm');
         inputConnection.motor_stop_status = "1";
-        inputConnection.stopTime = currentTime.format('DD-MMM-YYYY - HH:mm');
+        inputConnection.stopTime = currentTime; 
       }
 
       await motorTank.save();
     }
 
-    console.log(`Processed successfully for hardwareId: ${hw_Id}`);
+    console.log('✅ Data processed successfully for hardwareId:', hw_Id);
   } catch (err) {
-    console.error('Error processing message:', err.message);
+    console.error('❌ Error processing IoT data:', err.message);
   }
-});
+}
+
+