Merge pull request #439 from SyncrowIOT/add-check-log-to-trace-the-map-issue

feat: enhance device status handling with caching and batch processin…
2025-07-10 07:07:21 +00:00 · 2025-06-25 18:59:37 -06:00
parent 68d2d3b53d f80d097ff8
commit 8a4633b158
5 changed files with 151 additions and 134 deletions
--- a/libs/common/src/firebase/devices-status/services/devices-status.service.ts
+++ b/libs/common/src/firebase/devices-status/services/devices-status.service.ts
@ -68,22 +68,21 @@ export class DeviceStatusFirebaseService {
    }
  }
  async addBatchDeviceStatusToOurDb(
-    batch: { deviceTuyaUuid: string; status: any; log: any }[],
-    deviceCache: Map<string, any>,
+    batch: {
+      deviceTuyaUuid: string;
+      status: any;
+      log: any;
+      device: any;
+    }[],
  ): Promise<void> {
    const allLogs = [];

-    console.log(
-      `🧠 Preparing logs from batch of ${batch.length} items using cached devices only...`,
-    );
+    console.log(`🔁 Preparing logs from batch of ${batch.length} items...`);

    for (const item of batch) {
-      const device = deviceCache.get(item.deviceTuyaUuid);
-
+      const device = item.device;
      if (!device?.uuid) {
-        console.log(
-          `⛔ Ignored unknown device in batch: ${item.deviceTuyaUuid}`,
-        );
+        console.log(`⛔ Skipped unknown device: ${item.deviceTuyaUuid}`);
        continue;
      }

@ -103,61 +102,59 @@ export class DeviceStatusFirebaseService {
    }

    console.log(`📝 Total logs to insert: ${allLogs.length}`);
-    const chunkSize = 300;
-    let insertedCount = 0;

-    for (let i = 0; i < allLogs.length; i += chunkSize) {
-      const chunk = allLogs.slice(i, i + chunkSize);
-      try {
-        const result = await this.deviceStatusLogRepository
-          .createQueryBuilder()
-          .insert()
-          .into('device-status-log') // or use DeviceStatusLogEntity
-          .values(chunk)
-          .orIgnore() // skip duplicates
-          .execute();
+    const insertLogsPromise = (async () => {
+      const chunkSize = 300;
+      let insertedCount = 0;

-        insertedCount += result.identifiers.length;
-        console.log(
-          `✅ Inserted ${result.identifiers.length} / ${chunk.length} logs (chunk)`,
-        );
-      } catch (error) {
-        console.error('❌ Insert error (skipped chunk):', error.message);
+      for (let i = 0; i < allLogs.length; i += chunkSize) {
+        const chunk = allLogs.slice(i, i + chunkSize);
+        try {
+          const result = await this.deviceStatusLogRepository
+            .createQueryBuilder()
+            .insert()
+            .into('device-status-log') // or use DeviceStatusLogEntity
+            .values(chunk)
+            .orIgnore() // skip duplicates
+            .execute();
+
+          insertedCount += result.identifiers.length;
+          console.log(
+            `✅ Inserted ${result.identifiers.length} / ${chunk.length} logs (chunk)`,
+          );
+        } catch (error) {
+          console.error('❌ Insert error (skipped chunk):', error.message);
+        }
      }
-    }

-    console.log(`✅ Total logs inserted: ${insertedCount} / ${allLogs.length}`);
+      console.log(
+        `✅ Total logs inserted: ${insertedCount} / ${allLogs.length}`,
+      );
+    })();
+
+    await insertLogsPromise;
  }

  async addDeviceStatusToFirebase(
-    addDeviceStatusDto: AddDeviceStatusDto,
-    deviceCache?: Map<string, any>,
+    addDeviceStatusDto: AddDeviceStatusDto & { device?: any },
  ): Promise<AddDeviceStatusDto | null> {
    try {
-      let device;
-
-      if (deviceCache) {
-        device = deviceCache.get(addDeviceStatusDto.deviceTuyaUuid);
-      } else {
+      let device = addDeviceStatusDto.device;
+      if (!device) {
        device = await this.getDeviceByDeviceTuyaUuid(
          addDeviceStatusDto.deviceTuyaUuid,
        );
      }
-
-      if (!device?.uuid) {
-        console.log(
-          `⛔ Skipping Firebase update for unknown device: ${addDeviceStatusDto.deviceTuyaUuid}`,
-        );
-        return null;
+      if (device?.uuid) {
+        return await this.createDeviceStatusFirebase({
+          deviceUuid: device.uuid,
+          ...addDeviceStatusDto,
+          productType: device.productDevice?.prodType,
+        });
      }
-
-      addDeviceStatusDto.deviceUuid = device.uuid;
-      addDeviceStatusDto.productUuid = device.productDevice?.uuid;
-      addDeviceStatusDto.productType = device.productDevice?.prodType;
-
-      return await this.createDeviceStatusFirebase(addDeviceStatusDto);
+      // Return null if device not found or no UUID
+      return null;
    } catch (error) {
-      console.error('❌ Error in addDeviceStatusToFirebase:', error);
      return null;
    }
  }
@ -171,6 +168,15 @@ export class DeviceStatusFirebaseService {
      relations: ['productDevice'],
    });
  }
+  async getAllDevices() {
+    return await this.deviceRepository.find({
+      where: {
+        isActive: true,
+      },
+      relations: ['productDevice'],
+    });
+  }
+
  async getDevicesInstructionStatus(deviceUuid: string) {
    try {
      const deviceDetails = await this.getDeviceByDeviceUuid(deviceUuid);
--- a/libs/common/src/helper/services/sos.handler.service.ts
+++ b/libs/common/src/helper/services/sos.handler.service.ts
@ -16,53 +16,46 @@ export class SosHandlerService {
    );
  }

-  async handleSosEventFirebase(
-    devId: string,
-    logData: any,
-    deviceCache: Map<string, any>,
-  ): Promise<void> {
+  async handleSosEventFirebase(device: any, logData: any): Promise<void> {
+    const sosTrueStatus = [{ code: 'sos', value: true }];
+    const sosFalseStatus = [{ code: 'sos', value: false }];
+
    try {
-      await this.deviceStatusFirebaseService.addDeviceStatusToFirebase(
+      // ✅ Send true status
+      await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
+        deviceTuyaUuid: device.deviceTuyaUuid,
+        status: sosTrueStatus,
+        log: logData,
+        device,
+      });
+
+      await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb([
        {
-          deviceTuyaUuid: devId,
-          status: [{ code: 'sos', value: true }],
+          deviceTuyaUuid: device.deviceTuyaUuid,
+          status: sosTrueStatus,
          log: logData,
+          device,
        },
-        deviceCache,
-      );
-
-      await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb(
-        [
-          {
-            deviceTuyaUuid: devId,
-            status: [{ code: 'sos', value: true }],
-            log: logData,
-          },
-        ],
-        deviceCache,
-      );
+      ]);

+      // ✅ Schedule false status
      setTimeout(async () => {
        try {
-          await this.deviceStatusFirebaseService.addDeviceStatusToFirebase(
-            {
-              deviceTuyaUuid: devId,
-              status: [{ code: 'sos', value: false }],
-              log: logData,
-            },
-            deviceCache,
-          );
+          await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
+            deviceTuyaUuid: device.deviceTuyaUuid,
+            status: sosFalseStatus,
+            log: logData,
+            device,
+          });

-          await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb(
-            [
-              {
-                deviceTuyaUuid: devId,
-                status: [{ code: 'sos', value: false }],
-                log: logData,
-              },
-            ],
-            deviceCache,
-          );
+          await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb([
+            {
+              deviceTuyaUuid: device.deviceTuyaUuid,
+              status: sosFalseStatus,
+              log: logData,
+              device,
+            },
+          ]);
        } catch (err) {
          this.logger.error('Failed to send SOS false value', err);
        }
--- a/libs/common/src/helper/services/tuya.web.socket.service.ts
+++ b/libs/common/src/helper/services/tuya.web.socket.service.ts
@ -1,29 +1,29 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, OnModuleInit } from '@nestjs/common';
 import TuyaWebsocket from '../../config/tuya-web-socket-config';
 import { ConfigService } from '@nestjs/config';
 import { DeviceStatusFirebaseService } from '@app/common/firebase/devices-status/services/devices-status.service';
 import { SosHandlerService } from './sos.handler.service';
-import { DeviceRepository } from '@app/common/modules/device/repositories';
+import * as NodeCache from 'node-cache';

@Injectable()
-export class TuyaWebSocketService {
+export class TuyaWebSocketService implements OnModuleInit {
  private client: any;
  private readonly isDevEnv: boolean;
+  private readonly deviceCache = new NodeCache({ stdTTL: 7200 }); // TTL = 2 hour

  private messageQueue: {
    devId: string;
    status: any;
    logData: any;
+    device: any;
  }[] = [];

  private isProcessing = false;
-  private deviceCache: Map<string, any> = new Map();

  constructor(
    private readonly configService: ConfigService,
    private readonly deviceStatusFirebaseService: DeviceStatusFirebaseService,
    private readonly sosHandlerService: SosHandlerService,
-    private readonly deviceRepository: DeviceRepository,
  ) {
    this.isDevEnv =
      this.configService.get<string>('NODE_ENV') === 'development';
@ -36,38 +36,37 @@ export class TuyaWebSocketService {
      maxRetryTimes: 100,
    });

-    this.loadAllActiveDevices();
-
-    // Reload device cache every 1 hour
-    setInterval(() => this.loadAllActiveDevices(), 60 * 60 * 1000);
-
    if (this.configService.get<string>('tuya-config.TRUN_ON_TUYA_SOCKET')) {
      this.setupEventHandlers();
      this.client.start();
    }

-    // Trigger the queue processor every 15 seconds
+    // Run the queue processor every 15 seconds
    setInterval(() => this.processQueue(), 15000);
+
+    // Refresh the cache every 1 hour
+    setInterval(() => this.initializeDeviceCache(), 30 * 60 * 1000); // 30 minutes
  }

-  private async loadAllActiveDevices(): Promise<void> {
-    const devices = await this.deviceRepository.find({
-      where: { isActive: true },
-      relations: ['productDevice'],
-    });
+  async onModuleInit() {
+    await this.initializeDeviceCache();
+  }

-    this.deviceCache.clear();
-    devices.forEach((device) => {
-      this.deviceCache.set(device.deviceTuyaUuid, device);
-    });
-
-    console.log(
-      `🔄 Device cache reloaded: ${this.deviceCache.size} active devices at ${new Date().toISOString()}`,
-    );
+  private async initializeDeviceCache() {
+    try {
+      const allDevices = await this.deviceStatusFirebaseService.getAllDevices();
+      allDevices.forEach((device) => {
+        if (device.deviceTuyaUuid) {
+          this.deviceCache.set(device.deviceTuyaUuid, device);
+        }
+      });
+      console.log(`✅ Refreshed cache with ${allDevices.length} devices.`);
+    } catch (error) {
+      console.error('❌ Failed to initialize device cache:', error);
+    }
  }

  private setupEventHandlers() {
-    // Event handlers
    this.client.open(() => {
      console.log('open');
    });
@ -75,42 +74,38 @@ export class TuyaWebSocketService {
    this.client.message(async (ws: WebSocket, message: any) => {
      try {
        const { devId, status, logData } = this.extractMessageData(message);
-        if (!Array.isArray(logData?.properties)) return;
+        if (!Array.isArray(logData?.properties)) {
+          this.client.ackMessage(message.messageId);
+          return;
+        }

        const device = this.deviceCache.get(devId);
        if (!device) {
-          // console.log(`⛔ Ignored unknown device: ${devId}`);
+          // console.log(⛔ Unknown device: ${devId}, message ignored.);
+          this.client.ackMessage(message.messageId);
          return;
        }

        if (this.sosHandlerService.isSosTriggered(status)) {
-          await this.sosHandlerService.handleSosEventFirebase(
-            devId,
-            logData,
-            this.deviceCache,
-          );
+          await this.sosHandlerService.handleSosEventFirebase(devId, logData);
        } else {
-          // Firebase real-time update
-          await this.deviceStatusFirebaseService.addDeviceStatusToFirebase(
-            {
-              deviceTuyaUuid: devId,
-              status,
-              log: logData,
-            },
-            this.deviceCache,
-          );
+          await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
+            deviceTuyaUuid: devId,
+            status,
+            log: logData,
+            device,
+          });
        }

        // Push to internal queue
-        this.messageQueue.push({ devId, status, logData });
+        this.messageQueue.push({ devId, status, logData, device });

        // Acknowledge the message
        this.client.ackMessage(message.messageId);
      } catch (error) {
-        console.error('Error receiving message:', error);
+        console.error('❌ Error receiving message:', error);
      }
    });
-
    this.client.reconnect(() => {
      console.log('reconnect');
    });
@ -151,8 +146,8 @@ export class TuyaWebSocketService {
          deviceTuyaUuid: item.devId,
          status: item.status,
          log: item.logData,
+          device: item.device,
        })),
-        this.deviceCache,
      );
    } catch (error) {
      console.error('❌ Error processing batch:', error);
--- a/package-lock.json
+++ b/package-lock.json
@ -39,6 +39,7 @@
        "ioredis": "^5.3.2",
        "morgan": "^1.10.0",
        "nest-winston": "^1.10.2",
+        "node-cache": "^5.1.2",
        "nodemailer": "^6.9.10",
        "onesignal-node": "^3.4.0",
        "passport-jwt": "^4.0.1",
@ -10184,6 +10185,27 @@
        "node": "^18 || ^20 || >= 21"
      }
    },
+    "node_modules/node-cache": {
+      "version": "5.1.2",
+      "resolved": "https://registry.npmjs.org/node-cache/-/node-cache-5.1.2.tgz",
+      "integrity": "sha512-t1QzWwnk4sjLWaQAS8CHgOJ+RAfmHpxFWmc36IWTiWHQfs0w5JDMBS1b1ZxQteo0vVVuWJvIUKHDkkeK7vIGCg==",
+      "license": "MIT",
+      "dependencies": {
+        "clone": "2.x"
+      },
+      "engines": {
+        "node": ">= 8.0.0"
+      }
+    },
+    "node_modules/node-cache/node_modules/clone": {
+      "version": "2.1.2",
+      "resolved": "https://registry.npmjs.org/clone/-/clone-2.1.2.tgz",
+      "integrity": "sha512-3Pe/CF1Nn94hyhIYpjtiLhdCoEoz0DqQ+988E9gmeEdQZlojxnOb74wctFyuwWQHzqyf9X7C7MG8juUpqBJT8w==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=0.8"
+      }
+    },
    "node_modules/node-emoji": {
      "version": "1.11.0",
      "resolved": "https://registry.npmjs.org/node-emoji/-/node-emoji-1.11.0.tgz",
--- a/package.json
+++ b/package.json
@ -51,6 +51,7 @@
    "ioredis": "^5.3.2",
    "morgan": "^1.10.0",
    "nest-winston": "^1.10.2",
+    "node-cache": "^5.1.2",
    "nodemailer": "^6.9.10",
    "onesignal-node": "^3.4.0",
    "passport-jwt": "^4.0.1",