Add methods to handle SOS events and device status updates in Firebase and our DB

Revert "SP-1778-be-fix-time-out-connections-in-the-db"

libs/common/src/database/database.module.ts

@@ -125,7 +125,7 @@ import { VisitorPasswordEntity } from '../modules/visitor-password/entities';
           logger: typeOrmLogger,
           extra: {
             charset: 'utf8mb4',
-            max: 50, // set pool max size
+            max: 100, // set pool max size
             idleTimeoutMillis: 5000, // close idle clients after 5 second
             connectionTimeoutMillis: 12_000, // return an error after 11 second if connection could not be established
             maxUses: 7500, // close (and replace) a connection after it has been used 7500 times (see below for discussion)

libs/common/src/firebase/devices-status/services/devices-status.service.ts

@@ -24,7 +24,6 @@ import { PowerClampEnergyEnum } from '@app/common/constants/power.clamp.enargy.e
 import { PresenceSensorEnum } from '@app/common/constants/presence.sensor.enum';
 import { OccupancyService } from '@app/common/helper/services/occupancy.service';
 import { AqiDataService } from '@app/common/helper/services/aqi.data.service';
-import { DataSource, QueryRunner } from 'typeorm';
 @Injectable()
 export class DeviceStatusFirebaseService {
   private tuya: TuyaContext;
@@ -36,7 +35,6 @@ export class DeviceStatusFirebaseService {
     private readonly occupancyService: OccupancyService,
     private readonly aqiDataService: AqiDataService,
     private deviceStatusLogRepository: DeviceStatusLogRepository,
-    private readonly dataSource: DataSource,
   ) {
     const accessKey = this.configService.get<string>('auth-config.ACCESS_KEY');
     const secretKey = this.configService.get<string>('auth-config.SECRET_KEY');
@@ -78,49 +76,53 @@ export class DeviceStatusFirebaseService {
       );
     }
   }
-  async addDeviceStatusToFirebase(
+  async addDeviceStatusToOurDb(
     addDeviceStatusDto: AddDeviceStatusDto,
   ): Promise<AddDeviceStatusDto | null> {
-    const queryRunner = this.dataSource.createQueryRunner();
-    await queryRunner.connect();
-    await queryRunner.startTransaction();
     try {
       const device = await this.getDeviceByDeviceTuyaUuid(
         addDeviceStatusDto.deviceTuyaUuid,
-        queryRunner,
       );
 
       if (device?.uuid) {
-        const result = await this.createDeviceStatusFirebase(
-          {
-            deviceUuid: device.uuid,
-            ...addDeviceStatusDto,
-            productType: device.productDevice.prodType,
-          },
-          queryRunner,
-        );
-        await queryRunner.commitTransaction();
-        return result;
+        return await this.createDeviceStatusInOurDb({
+          deviceUuid: device.uuid,
+          ...addDeviceStatusDto,
+          productType: device.productDevice.prodType,
+        });
       }
       // Return null if device not found or no UUID
-      await queryRunner.rollbackTransaction();
       return null;
     } catch (error) {
-      await queryRunner.rollbackTransaction();
+      // Handle the error silently, perhaps log it internally or ignore it
       return null;
-    } finally {
-      await queryRunner.release();
     }
   }
-  async getDeviceByDeviceTuyaUuid(
-    deviceTuyaUuid: string,
-    queryRunner?: QueryRunner,
-  ) {
-    const repo = queryRunner
-      ? queryRunner.manager.getRepository(this.deviceRepository.target)
-      : this.deviceRepository;
+  async addDeviceStatusToFirebase(
+    addDeviceStatusDto: AddDeviceStatusDto,
+  ): Promise<AddDeviceStatusDto | null> {
+    try {
+      const device = await this.getDeviceByDeviceTuyaUuid(
+        addDeviceStatusDto.deviceTuyaUuid,
+      );
 
-    return await repo.findOne({
+      if (device?.uuid) {
+        return await this.createDeviceStatusFirebase({
+          deviceUuid: device.uuid,
+          ...addDeviceStatusDto,
+          productType: device.productDevice.prodType,
+        });
+      }
+      // Return null if device not found or no UUID
+      return null;
+    } catch (error) {
+      // Handle the error silently, perhaps log it internally or ignore it
+      return null;
+    }
+  }
+
+  async getDeviceByDeviceTuyaUuid(deviceTuyaUuid: string) {
+    return await this.deviceRepository.findOne({
       where: {
         deviceTuyaUuid,
         isActive: true,
@@ -128,7 +130,6 @@ export class DeviceStatusFirebaseService {
       relations: ['productDevice'],
     });
   }
-
   async getDevicesInstructionStatus(deviceUuid: string) {
     try {
       const deviceDetails = await this.getDeviceByDeviceUuid(deviceUuid);
@@ -174,14 +175,9 @@ export class DeviceStatusFirebaseService {
   }
   async getDeviceByDeviceUuid(
     deviceUuid: string,
-    withProductDevice = true,
-    queryRunner?: QueryRunner,
+    withProductDevice: boolean = true,
   ) {
-    const repo = queryRunner
-      ? queryRunner.manager.getRepository(this.deviceRepository.target)
-      : this.deviceRepository;
-
-    return await repo.findOne({
+    return await this.deviceRepository.findOne({
       where: {
         uuid: deviceUuid,
         isActive: true,
@@ -189,20 +185,21 @@ export class DeviceStatusFirebaseService {
       ...(withProductDevice && { relations: ['productDevice'] }),
     });
   }
-
   async createDeviceStatusFirebase(
     addDeviceStatusDto: AddDeviceStatusDto,
-    queryRunner?: QueryRunner,
   ): Promise<any> {
     const dataRef = ref(
       this.firebaseDb,
       `device-status/${addDeviceStatusDto.deviceUuid}`,
     );
 
-    // Step 1: Update Firebase Realtime Database
+    // Use a transaction to handle concurrent updates
     await runTransaction(dataRef, (existingData) => {
-      if (!existingData) existingData = {};
+      if (!existingData) {
+        existingData = {};
+      }
 
+      // Assign default values if fields are not present
       if (!existingData.deviceTuyaUuid) {
         existingData.deviceTuyaUuid = addDeviceStatusDto.deviceTuyaUuid;
       }
@@ -216,15 +213,18 @@ export class DeviceStatusFirebaseService {
         existingData.status = [];
       }
 
-      // Merge incoming status with existing status
+      // Create a map to track existing status codes
       const statusMap = new Map(
         existingData.status.map((item) => [item.code, item.value]),
       );
 
+      // Update or add status codes
+
       for (const statusItem of addDeviceStatusDto.status) {
         statusMap.set(statusItem.code, statusItem.value);
       }
 
+      // Convert the map back to an array format
       existingData.status = Array.from(statusMap, ([code, value]) => ({
         code,
         value,
@@ -233,9 +233,16 @@ export class DeviceStatusFirebaseService {
       return existingData;
     });
 
-    // Step 2: Save device status log entries
-    const newLogs = addDeviceStatusDto.log.properties.map((property) =>
-      this.deviceStatusLogRepository.create({
+    // Return the updated data
+    const snapshot: DataSnapshot = await get(dataRef);
+    return snapshot.val();
+  }
+  async createDeviceStatusInOurDb(
+    addDeviceStatusDto: AddDeviceStatusDto,
+  ): Promise<any> {
+    // Save logs to your repository
+    const newLogs = addDeviceStatusDto.log.properties.map((property) => {
+      return this.deviceStatusLogRepository.create({
         deviceId: addDeviceStatusDto.deviceUuid,
         deviceTuyaId: addDeviceStatusDto.deviceTuyaUuid,
         productId: addDeviceStatusDto.log.productId,
@@ -244,19 +251,10 @@ export class DeviceStatusFirebaseService {
         value: property.value,
         eventId: addDeviceStatusDto.log.dataId,
         eventTime: new Date(property.time).toISOString(),
-      }),
-    );
+      });
+    });
+    await this.deviceStatusLogRepository.save(newLogs);
 
-    if (queryRunner) {
-      const repo = queryRunner.manager.getRepository(
-        this.deviceStatusLogRepository.target,
-      );
-      await repo.save(newLogs);
-    } else {
-      await this.deviceStatusLogRepository.save(newLogs);
-    }
-
-    // Step 3: Trigger additional data services
     if (addDeviceStatusDto.productType === ProductType.PC) {
       const energyCodes = new Set([
         PowerClampEnergyEnum.ENERGY_CONSUMED,
@@ -300,9 +298,5 @@ export class DeviceStatusFirebaseService {
         addDeviceStatusDto.deviceUuid,
       );
     }
-
-    // Step 4: Return updated Firebase status
-    const snapshot: DataSnapshot = await get(dataRef);
-    return snapshot.val();
   }
 }

libs/common/src/helper/services/aqi.data.service.ts

@@ -36,18 +36,9 @@ export class AqiDataService {
     procedureFileName: string,
     params: (string | number | null)[],
   ): Promise<void> {
-    const queryRunner = this.dataSource.createQueryRunner();
-    await queryRunner.connect();
-    try {
-      const query = this.loadQuery(procedureFolderName, procedureFileName);
-      await queryRunner.query(query, params);
-      console.log(`Procedure ${procedureFileName} executed successfully.`);
-    } catch (err) {
-      console.error(`Failed to execute procedure ${procedureFileName}:`, err);
-      throw err;
-    } finally {
-      await queryRunner.release();
-    }
+    const query = this.loadQuery(procedureFolderName, procedureFileName);
+    await this.dataSource.query(query, params);
+    console.log(`Procedure ${procedureFileName} executed successfully.`);
   }
 
   private loadQuery(folderName: string, fileName: string): string {

libs/common/src/helper/services/occupancy.service.ts

@@ -57,18 +57,9 @@ export class OccupancyService {
     procedureFileName: string,
     params: (string | number | null)[],
   ): Promise<void> {
-    const queryRunner = this.dataSource.createQueryRunner();
-    await queryRunner.connect();
-    try {
-      const query = this.loadQuery(procedureFolderName, procedureFileName);
-      await queryRunner.query(query, params);
-      console.log(`Procedure ${procedureFileName} executed successfully.`);
-    } catch (err) {
-      console.error(`Failed to execute procedure ${procedureFileName}:`, err);
-      throw err;
-    } finally {
-      await queryRunner.release();
-    }
+    const query = this.loadQuery(procedureFolderName, procedureFileName);
+    await this.dataSource.query(query, params);
+    console.log(`Procedure ${procedureFileName} executed successfully.`);
   }
 
   private loadQuery(folderName: string, fileName: string): string {

libs/common/src/helper/services/power.clamp.service.ts

@@ -46,21 +46,12 @@ export class PowerClampService {
     procedureFileName: string,
     params: (string | number | null)[],
   ): Promise<void> {
-    const queryRunner = this.dataSource.createQueryRunner();
-    await queryRunner.connect();
-    try {
-      const query = this.loadQuery(
-        'fact_device_energy_consumed',
-        procedureFileName,
-      );
-      await queryRunner.query(query, params);
-      console.log(`Procedure ${procedureFileName} executed successfully.`);
-    } catch (err) {
-      console.error(`Failed to execute procedure ${procedureFileName}:`, err);
-      throw err;
-    } finally {
-      await queryRunner.release();
-    }
+    const query = this.loadQuery(
+      'fact_device_energy_consumed',
+      procedureFileName,
+    );
+    await this.dataSource.query(query, params);
+    console.log(`Procedure ${procedureFileName} executed successfully.`);
   }
 
   private loadQuery(folderName: string, fileName: string): string {

libs/common/src/helper/services/sos.handler.service.ts

@@ -16,7 +16,7 @@ export class SosHandlerService {
     );
   }
 
-  async handleSosEvent(devId: string, logData: any): Promise<void> {
+  async handleSosEventFirebase(devId: string, logData: any): Promise<void> {
     try {
       await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
         deviceTuyaUuid: devId,
@@ -39,4 +39,28 @@ export class SosHandlerService {
       this.logger.error('Failed to send SOS true value', err);
     }
   }
+
+  async handleSosEventOurDb(devId: string, logData: any): Promise<void> {
+    try {
+      await this.deviceStatusFirebaseService.addDeviceStatusToOurDb({
+        deviceTuyaUuid: devId,
+        status: [{ code: 'sos', value: true }],
+        log: logData,
+      });
+
+      setTimeout(async () => {
+        try {
+          await this.deviceStatusFirebaseService.addDeviceStatusToOurDb({
+            deviceTuyaUuid: devId,
+            status: [{ code: 'sos', value: false }],
+            log: logData,
+          });
+        } catch (err) {
+          this.logger.error('Failed to send SOS false value', err);
+        }
+      }, 2000);
+    } catch (err) {
+      this.logger.error('Failed to send SOS true value', err);
+    }
+  }
 }

libs/common/src/helper/services/tuya.web.socket.service.ts

@@ -9,6 +9,14 @@ export class TuyaWebSocketService {
   private client: any;
   private readonly isDevEnv: boolean;
 
+  private messageQueue: {
+    devId: string;
+    status: any;
+    logData: any;
+  }[] = [];
+
+  private isProcessing = false;
+
   constructor(
     private readonly configService: ConfigService,
     private readonly deviceStatusFirebaseService: DeviceStatusFirebaseService,
@@ -26,12 +34,12 @@ export class TuyaWebSocketService {
     });
 
     if (this.configService.get<string>('tuya-config.TRUN_ON_TUYA_SOCKET')) {
-      // Set up event handlers
       this.setupEventHandlers();
-
-      // Start receiving messages
       this.client.start();
     }
+
+    // Trigger the queue processor every 2 seconds
+    setInterval(() => this.processQueue(), 10000);
   }
 
   private setupEventHandlers() {
@@ -43,10 +51,10 @@ export class TuyaWebSocketService {
     this.client.message(async (ws: WebSocket, message: any) => {
       try {
         const { devId, status, logData } = this.extractMessageData(message);
-
         if (this.sosHandlerService.isSosTriggered(status)) {
-          await this.sosHandlerService.handleSosEvent(devId, logData);
+          await this.sosHandlerService.handleSosEventFirebase(devId, logData);
         } else {
+          // Firebase real-time update
           await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
             deviceTuyaUuid: devId,
             status: status,
@@ -54,9 +62,13 @@ export class TuyaWebSocketService {
           });
         }
 
+        // Push to internal queue
+        this.messageQueue.push({ devId, status, logData });
+
+        // Acknowledge the message
         this.client.ackMessage(message.messageId);
       } catch (error) {
-        console.error('Error processing message:', error);
+        console.error('Error receiving message:', error);
       }
     });
 
@@ -80,6 +92,38 @@ export class TuyaWebSocketService {
       console.error('WebSocket error:', error);
     });
   }
+  private async processQueue() {
+    if (this.isProcessing || this.messageQueue.length === 0) return;
+
+    this.isProcessing = true;
+
+    const batch = [...this.messageQueue];
+    this.messageQueue = [];
+
+    try {
+      for (const item of batch) {
+        if (this.sosHandlerService.isSosTriggered(item.status)) {
+          await this.sosHandlerService.handleSosEventOurDb(
+            item.devId,
+            item.logData,
+          );
+        } else {
+          await this.deviceStatusFirebaseService.addDeviceStatusToOurDb({
+            deviceTuyaUuid: item.devId,
+            status: item.status,
+            log: item.logData,
+          });
+        }
+      }
+    } catch (error) {
+      console.error('Error processing batch:', error);
+      // Re-add the batch to the queue for retry
+      this.messageQueue.unshift(...batch);
+    } finally {
+      this.isProcessing = false;
+    }
+  }
+
   private extractMessageData(message: any): {
     devId: string;
     status: any;