mirror of
https://github.com/SyncrowIOT/backend.git
synced 2025-07-11 07:38:49 +00:00
Compare commits
12 Commits
fix-time-o
...
add-check-
Author | SHA1 | Date | |
---|---|---|---|
f80d097ff8 | |||
04bd156df1 | |||
731819aeaa | |||
68d2d3b53d | |||
3fcfe2d92f | |||
c0a069b460 | |||
30724d7d37 | |||
324661e1ee | |||
a83424f45b | |||
71f6ccb4db | |||
68692b7c8b | |||
4d60c1ed54 |
@ -68,33 +68,23 @@ export class DeviceStatusFirebaseService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
async addBatchDeviceStatusToOurDb(
|
async addBatchDeviceStatusToOurDb(
|
||||||
batch: { deviceTuyaUuid: string; status: any; log: any }[],
|
batch: {
|
||||||
|
deviceTuyaUuid: string;
|
||||||
|
status: any;
|
||||||
|
log: any;
|
||||||
|
device: any;
|
||||||
|
}[],
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
const allLogs = [];
|
const allLogs = [];
|
||||||
const deviceMap = new Map<string, any>();
|
|
||||||
|
|
||||||
console.log(
|
console.log(`🔁 Preparing logs from batch of ${batch.length} items...`);
|
||||||
`🧠 Starting device lookups for batch of ${batch.length} items...`,
|
|
||||||
);
|
|
||||||
|
|
||||||
// Step 1: Parallel device fetching
|
|
||||||
await Promise.all(
|
|
||||||
batch.map(async (item) => {
|
|
||||||
if (!deviceMap.has(item.deviceTuyaUuid)) {
|
|
||||||
const device = await this.getDeviceByDeviceTuyaUuid(
|
|
||||||
item.deviceTuyaUuid,
|
|
||||||
);
|
|
||||||
device?.uuid && deviceMap.set(item.deviceTuyaUuid, device);
|
|
||||||
}
|
|
||||||
}),
|
|
||||||
);
|
|
||||||
|
|
||||||
console.log(`🔍 Found ${deviceMap.size} devices from batch`);
|
|
||||||
|
|
||||||
// Step 2: Prepare logs and updates
|
|
||||||
for (const item of batch) {
|
for (const item of batch) {
|
||||||
const device = deviceMap.get(item.deviceTuyaUuid);
|
const device = item.device;
|
||||||
if (!device?.uuid) continue;
|
if (!device?.uuid) {
|
||||||
|
console.log(`⛔ Skipped unknown device: ${item.deviceTuyaUuid}`);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
const logs = item.log.properties.map((property) =>
|
const logs = item.log.properties.map((property) =>
|
||||||
this.deviceStatusLogRepository.create({
|
this.deviceStatusLogRepository.create({
|
||||||
@ -112,7 +102,7 @@ export class DeviceStatusFirebaseService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
console.log(`📝 Total logs to insert: ${allLogs.length}`);
|
console.log(`📝 Total logs to insert: ${allLogs.length}`);
|
||||||
// Step 3: Insert logs in chunks with ON CONFLICT DO NOTHING
|
|
||||||
const insertLogsPromise = (async () => {
|
const insertLogsPromise = (async () => {
|
||||||
const chunkSize = 300;
|
const chunkSize = 300;
|
||||||
let insertedCount = 0;
|
let insertedCount = 0;
|
||||||
@ -142,29 +132,29 @@ export class DeviceStatusFirebaseService {
|
|||||||
);
|
);
|
||||||
})();
|
})();
|
||||||
|
|
||||||
// Step 5: Wait for both insert and post-processing to finish
|
await insertLogsPromise;
|
||||||
await Promise.all([insertLogsPromise]);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async addDeviceStatusToFirebase(
|
async addDeviceStatusToFirebase(
|
||||||
addDeviceStatusDto: AddDeviceStatusDto,
|
addDeviceStatusDto: AddDeviceStatusDto & { device?: any },
|
||||||
): Promise<AddDeviceStatusDto | null> {
|
): Promise<AddDeviceStatusDto | null> {
|
||||||
try {
|
try {
|
||||||
const device = await this.getDeviceByDeviceTuyaUuid(
|
let device = addDeviceStatusDto.device;
|
||||||
addDeviceStatusDto.deviceTuyaUuid,
|
if (!device) {
|
||||||
);
|
device = await this.getDeviceByDeviceTuyaUuid(
|
||||||
|
addDeviceStatusDto.deviceTuyaUuid,
|
||||||
|
);
|
||||||
|
}
|
||||||
if (device?.uuid) {
|
if (device?.uuid) {
|
||||||
return await this.createDeviceStatusFirebase({
|
return await this.createDeviceStatusFirebase({
|
||||||
deviceUuid: device.uuid,
|
deviceUuid: device.uuid,
|
||||||
...addDeviceStatusDto,
|
...addDeviceStatusDto,
|
||||||
productType: device.productDevice.prodType,
|
productType: device.productDevice?.prodType,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
// Return null if device not found or no UUID
|
// Return null if device not found or no UUID
|
||||||
return null;
|
return null;
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
// Handle the error silently, perhaps log it internally or ignore it
|
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -178,6 +168,15 @@ export class DeviceStatusFirebaseService {
|
|||||||
relations: ['productDevice'],
|
relations: ['productDevice'],
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
async getAllDevices() {
|
||||||
|
return await this.deviceRepository.find({
|
||||||
|
where: {
|
||||||
|
isActive: true,
|
||||||
|
},
|
||||||
|
relations: ['productDevice'],
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
async getDevicesInstructionStatus(deviceUuid: string) {
|
async getDevicesInstructionStatus(deviceUuid: string) {
|
||||||
try {
|
try {
|
||||||
const deviceDetails = await this.getDeviceByDeviceUuid(deviceUuid);
|
const deviceDetails = await this.getDeviceByDeviceUuid(deviceUuid);
|
||||||
|
@ -8,7 +8,10 @@ import { TuyaWebSocketService } from './services/tuya.web.socket.service';
|
|||||||
import { OneSignalService } from './services/onesignal.service';
|
import { OneSignalService } from './services/onesignal.service';
|
||||||
import { DeviceMessagesService } from './services/device.messages.service';
|
import { DeviceMessagesService } from './services/device.messages.service';
|
||||||
import { DeviceRepositoryModule } from '../modules/device/device.repository.module';
|
import { DeviceRepositoryModule } from '../modules/device/device.repository.module';
|
||||||
import { DeviceNotificationRepository } from '../modules/device/repositories';
|
import {
|
||||||
|
DeviceNotificationRepository,
|
||||||
|
DeviceRepository,
|
||||||
|
} from '../modules/device/repositories';
|
||||||
import { DeviceStatusFirebaseModule } from '../firebase/devices-status/devices-status.module';
|
import { DeviceStatusFirebaseModule } from '../firebase/devices-status/devices-status.module';
|
||||||
import { CommunityPermissionService } from './services/community.permission.service';
|
import { CommunityPermissionService } from './services/community.permission.service';
|
||||||
import { CommunityRepository } from '../modules/community/repositories';
|
import { CommunityRepository } from '../modules/community/repositories';
|
||||||
@ -27,6 +30,7 @@ import { SosHandlerService } from './services/sos.handler.service';
|
|||||||
DeviceNotificationRepository,
|
DeviceNotificationRepository,
|
||||||
CommunityRepository,
|
CommunityRepository,
|
||||||
SosHandlerService,
|
SosHandlerService,
|
||||||
|
DeviceRepository,
|
||||||
],
|
],
|
||||||
exports: [
|
exports: [
|
||||||
HelperHashService,
|
HelperHashService,
|
||||||
|
@ -16,33 +16,44 @@ export class SosHandlerService {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
async handleSosEventFirebase(devId: string, logData: any): Promise<void> {
|
async handleSosEventFirebase(device: any, logData: any): Promise<void> {
|
||||||
|
const sosTrueStatus = [{ code: 'sos', value: true }];
|
||||||
|
const sosFalseStatus = [{ code: 'sos', value: false }];
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
// ✅ Send true status
|
||||||
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
|
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
|
||||||
deviceTuyaUuid: devId,
|
deviceTuyaUuid: device.deviceTuyaUuid,
|
||||||
status: [{ code: 'sos', value: true }],
|
status: sosTrueStatus,
|
||||||
log: logData,
|
log: logData,
|
||||||
|
device,
|
||||||
});
|
});
|
||||||
|
|
||||||
await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb([
|
await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb([
|
||||||
{
|
{
|
||||||
deviceTuyaUuid: devId,
|
deviceTuyaUuid: device.deviceTuyaUuid,
|
||||||
status: [{ code: 'sos', value: true }],
|
status: sosTrueStatus,
|
||||||
log: logData,
|
log: logData,
|
||||||
|
device,
|
||||||
},
|
},
|
||||||
]);
|
]);
|
||||||
|
|
||||||
|
// ✅ Schedule false status
|
||||||
setTimeout(async () => {
|
setTimeout(async () => {
|
||||||
try {
|
try {
|
||||||
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
|
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
|
||||||
deviceTuyaUuid: devId,
|
deviceTuyaUuid: device.deviceTuyaUuid,
|
||||||
status: [{ code: 'sos', value: false }],
|
status: sosFalseStatus,
|
||||||
log: logData,
|
log: logData,
|
||||||
|
device,
|
||||||
});
|
});
|
||||||
|
|
||||||
await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb([
|
await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb([
|
||||||
{
|
{
|
||||||
deviceTuyaUuid: devId,
|
deviceTuyaUuid: device.deviceTuyaUuid,
|
||||||
status: [{ code: 'sos', value: false }],
|
status: sosFalseStatus,
|
||||||
log: logData,
|
log: logData,
|
||||||
|
device,
|
||||||
},
|
},
|
||||||
]);
|
]);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
|
@ -1,18 +1,21 @@
|
|||||||
import { Injectable } from '@nestjs/common';
|
import { Injectable, OnModuleInit } from '@nestjs/common';
|
||||||
import TuyaWebsocket from '../../config/tuya-web-socket-config';
|
import TuyaWebsocket from '../../config/tuya-web-socket-config';
|
||||||
import { ConfigService } from '@nestjs/config';
|
import { ConfigService } from '@nestjs/config';
|
||||||
import { DeviceStatusFirebaseService } from '@app/common/firebase/devices-status/services/devices-status.service';
|
import { DeviceStatusFirebaseService } from '@app/common/firebase/devices-status/services/devices-status.service';
|
||||||
import { SosHandlerService } from './sos.handler.service';
|
import { SosHandlerService } from './sos.handler.service';
|
||||||
|
import * as NodeCache from 'node-cache';
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class TuyaWebSocketService {
|
export class TuyaWebSocketService implements OnModuleInit {
|
||||||
private client: any;
|
private client: any;
|
||||||
private readonly isDevEnv: boolean;
|
private readonly isDevEnv: boolean;
|
||||||
|
private readonly deviceCache = new NodeCache({ stdTTL: 7200 }); // TTL = 2 hour
|
||||||
|
|
||||||
private messageQueue: {
|
private messageQueue: {
|
||||||
devId: string;
|
devId: string;
|
||||||
status: any;
|
status: any;
|
||||||
logData: any;
|
logData: any;
|
||||||
|
device: any;
|
||||||
}[] = [];
|
}[] = [];
|
||||||
|
|
||||||
private isProcessing = false;
|
private isProcessing = false;
|
||||||
@ -38,12 +41,32 @@ export class TuyaWebSocketService {
|
|||||||
this.client.start();
|
this.client.start();
|
||||||
}
|
}
|
||||||
|
|
||||||
// Trigger the queue processor every 15 seconds
|
// Run the queue processor every 15 seconds
|
||||||
setInterval(() => this.processQueue(), 15000);
|
setInterval(() => this.processQueue(), 15000);
|
||||||
|
|
||||||
|
// Refresh the cache every 1 hour
|
||||||
|
setInterval(() => this.initializeDeviceCache(), 30 * 60 * 1000); // 30 minutes
|
||||||
|
}
|
||||||
|
|
||||||
|
async onModuleInit() {
|
||||||
|
await this.initializeDeviceCache();
|
||||||
|
}
|
||||||
|
|
||||||
|
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() {
|
private setupEventHandlers() {
|
||||||
// Event handlers
|
|
||||||
this.client.open(() => {
|
this.client.open(() => {
|
||||||
console.log('open');
|
console.log('open');
|
||||||
});
|
});
|
||||||
@ -51,27 +74,38 @@ export class TuyaWebSocketService {
|
|||||||
this.client.message(async (ws: WebSocket, message: any) => {
|
this.client.message(async (ws: WebSocket, message: any) => {
|
||||||
try {
|
try {
|
||||||
const { devId, status, logData } = this.extractMessageData(message);
|
const { devId, status, logData } = this.extractMessageData(message);
|
||||||
|
if (!Array.isArray(logData?.properties)) {
|
||||||
|
this.client.ackMessage(message.messageId);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const device = this.deviceCache.get(devId);
|
||||||
|
if (!device) {
|
||||||
|
// console.log(⛔ Unknown device: ${devId}, message ignored.);
|
||||||
|
this.client.ackMessage(message.messageId);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if (this.sosHandlerService.isSosTriggered(status)) {
|
if (this.sosHandlerService.isSosTriggered(status)) {
|
||||||
await this.sosHandlerService.handleSosEventFirebase(devId, logData);
|
await this.sosHandlerService.handleSosEventFirebase(devId, logData);
|
||||||
} else {
|
} else {
|
||||||
// Firebase real-time update
|
|
||||||
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
|
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
|
||||||
deviceTuyaUuid: devId,
|
deviceTuyaUuid: devId,
|
||||||
status: status,
|
status,
|
||||||
log: logData,
|
log: logData,
|
||||||
|
device,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
// Push to internal queue
|
// Push to internal queue
|
||||||
this.messageQueue.push({ devId, status, logData });
|
this.messageQueue.push({ devId, status, logData, device });
|
||||||
|
|
||||||
// Acknowledge the message
|
// Acknowledge the message
|
||||||
this.client.ackMessage(message.messageId);
|
this.client.ackMessage(message.messageId);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
console.error('Error receiving message:', error);
|
console.error('❌ Error receiving message:', error);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
this.client.reconnect(() => {
|
this.client.reconnect(() => {
|
||||||
console.log('reconnect');
|
console.log('reconnect');
|
||||||
});
|
});
|
||||||
@ -108,10 +142,11 @@ export class TuyaWebSocketService {
|
|||||||
|
|
||||||
try {
|
try {
|
||||||
await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb(
|
await this.deviceStatusFirebaseService.addBatchDeviceStatusToOurDb(
|
||||||
batch?.map((item) => ({
|
batch.map((item) => ({
|
||||||
deviceTuyaUuid: item.devId,
|
deviceTuyaUuid: item.devId,
|
||||||
status: item.status,
|
status: item.status,
|
||||||
log: item.logData,
|
log: item.logData,
|
||||||
|
device: item.device,
|
||||||
})),
|
})),
|
||||||
);
|
);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
22
package-lock.json
generated
22
package-lock.json
generated
@ -39,6 +39,7 @@
|
|||||||
"ioredis": "^5.3.2",
|
"ioredis": "^5.3.2",
|
||||||
"morgan": "^1.10.0",
|
"morgan": "^1.10.0",
|
||||||
"nest-winston": "^1.10.2",
|
"nest-winston": "^1.10.2",
|
||||||
|
"node-cache": "^5.1.2",
|
||||||
"nodemailer": "^6.9.10",
|
"nodemailer": "^6.9.10",
|
||||||
"onesignal-node": "^3.4.0",
|
"onesignal-node": "^3.4.0",
|
||||||
"passport-jwt": "^4.0.1",
|
"passport-jwt": "^4.0.1",
|
||||||
@ -10184,6 +10185,27 @@
|
|||||||
"node": "^18 || ^20 || >= 21"
|
"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": {
|
"node_modules/node-emoji": {
|
||||||
"version": "1.11.0",
|
"version": "1.11.0",
|
||||||
"resolved": "https://registry.npmjs.org/node-emoji/-/node-emoji-1.11.0.tgz",
|
"resolved": "https://registry.npmjs.org/node-emoji/-/node-emoji-1.11.0.tgz",
|
||||||
|
@ -51,6 +51,7 @@
|
|||||||
"ioredis": "^5.3.2",
|
"ioredis": "^5.3.2",
|
||||||
"morgan": "^1.10.0",
|
"morgan": "^1.10.0",
|
||||||
"nest-winston": "^1.10.2",
|
"nest-winston": "^1.10.2",
|
||||||
|
"node-cache": "^5.1.2",
|
||||||
"nodemailer": "^6.9.10",
|
"nodemailer": "^6.9.10",
|
||||||
"onesignal-node": "^3.4.0",
|
"onesignal-node": "^3.4.0",
|
||||||
"passport-jwt": "^4.0.1",
|
"passport-jwt": "^4.0.1",
|
||||||
|
@ -50,7 +50,7 @@ import { SchedulerModule } from './scheduler/scheduler.module';
|
|||||||
load: config,
|
load: config,
|
||||||
}),
|
}),
|
||||||
ThrottlerModule.forRoot({
|
ThrottlerModule.forRoot({
|
||||||
throttlers: [{ ttl: 60000, limit: 30 }],
|
throttlers: [{ ttl: 60000, limit: 100 }],
|
||||||
generateKey: (context) => {
|
generateKey: (context) => {
|
||||||
const req = context.switchToHttp().getRequest();
|
const req = context.switchToHttp().getRequest();
|
||||||
console.log('Real IP:', req.headers['x-forwarded-for']);
|
console.log('Real IP:', req.headers['x-forwarded-for']);
|
||||||
|
Reference in New Issue
Block a user