Compare commits

..

1 Commits

Author SHA1 Message Date
bf9294a4ef fix: increase DB max pool size 2025-06-23 15:19:11 +03:00
5 changed files with 31 additions and 122 deletions

View File

@ -76,28 +76,6 @@ export class DeviceStatusFirebaseService {
); );
} }
} }
async addDeviceStatusToOurDb(
addDeviceStatusDto: AddDeviceStatusDto,
): Promise<AddDeviceStatusDto | null> {
try {
const device = await this.getDeviceByDeviceTuyaUuid(
addDeviceStatusDto.deviceTuyaUuid,
);
if (device?.uuid) {
return await this.createDeviceStatusInOurDb({
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 addDeviceStatusToFirebase( async addDeviceStatusToFirebase(
addDeviceStatusDto: AddDeviceStatusDto, addDeviceStatusDto: AddDeviceStatusDto,
): Promise<AddDeviceStatusDto | null> { ): Promise<AddDeviceStatusDto | null> {
@ -233,13 +211,6 @@ export class DeviceStatusFirebaseService {
return existingData; return existingData;
}); });
// Return the updated data
const snapshot: DataSnapshot = await get(dataRef);
return snapshot.val();
}
async createDeviceStatusInOurDb(
addDeviceStatusDto: AddDeviceStatusDto,
): Promise<any> {
// Save logs to your repository // Save logs to your repository
const newLogs = addDeviceStatusDto.log.properties.map((property) => { const newLogs = addDeviceStatusDto.log.properties.map((property) => {
return this.deviceStatusLogRepository.create({ return this.deviceStatusLogRepository.create({
@ -298,5 +269,8 @@ export class DeviceStatusFirebaseService {
addDeviceStatusDto.deviceUuid, addDeviceStatusDto.deviceUuid,
); );
} }
// Return the updated data
const snapshot: DataSnapshot = await get(dataRef);
return snapshot.val();
} }
} }

View File

@ -16,7 +16,7 @@ export class SosHandlerService {
); );
} }
async handleSosEventFirebase(devId: string, logData: any): Promise<void> { async handleSosEvent(devId: string, logData: any): Promise<void> {
try { try {
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({ await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
deviceTuyaUuid: devId, deviceTuyaUuid: devId,
@ -39,28 +39,4 @@ export class SosHandlerService {
this.logger.error('Failed to send SOS true value', err); 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);
}
}
} }

View File

@ -9,14 +9,6 @@ export class TuyaWebSocketService {
private client: any; private client: any;
private readonly isDevEnv: boolean; private readonly isDevEnv: boolean;
private messageQueue: {
devId: string;
status: any;
logData: any;
}[] = [];
private isProcessing = false;
constructor( constructor(
private readonly configService: ConfigService, private readonly configService: ConfigService,
private readonly deviceStatusFirebaseService: DeviceStatusFirebaseService, private readonly deviceStatusFirebaseService: DeviceStatusFirebaseService,
@ -34,12 +26,12 @@ export class TuyaWebSocketService {
}); });
if (this.configService.get<string>('tuya-config.TRUN_ON_TUYA_SOCKET')) { if (this.configService.get<string>('tuya-config.TRUN_ON_TUYA_SOCKET')) {
// Set up event handlers
this.setupEventHandlers(); this.setupEventHandlers();
// Start receiving messages
this.client.start(); this.client.start();
} }
// Trigger the queue processor every 2 seconds
setInterval(() => this.processQueue(), 10000);
} }
private setupEventHandlers() { private setupEventHandlers() {
@ -51,10 +43,10 @@ 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 (this.sosHandlerService.isSosTriggered(status)) { if (this.sosHandlerService.isSosTriggered(status)) {
await this.sosHandlerService.handleSosEventFirebase(devId, logData); await this.sosHandlerService.handleSosEvent(devId, logData);
} else { } else {
// Firebase real-time update
await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({ await this.deviceStatusFirebaseService.addDeviceStatusToFirebase({
deviceTuyaUuid: devId, deviceTuyaUuid: devId,
status: status, status: status,
@ -62,13 +54,9 @@ export class TuyaWebSocketService {
}); });
} }
// Push to internal queue
this.messageQueue.push({ devId, status, logData });
// 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 processing message:', error);
} }
}); });
@ -92,38 +80,6 @@ export class TuyaWebSocketService {
console.error('WebSocket error:', error); 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): { private extractMessageData(message: any): {
devId: string; devId: string;
status: any; status: any;

View File

@ -1,7 +1,7 @@
import { SeederModule } from '@app/common/seed/seeder.module'; import { SeederModule } from '@app/common/seed/seeder.module';
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config'; import { ConfigModule } from '@nestjs/config';
import { APP_GUARD, APP_INTERCEPTOR } from '@nestjs/core'; import { APP_INTERCEPTOR } from '@nestjs/core';
import { WinstonModule } from 'nest-winston'; import { WinstonModule } from 'nest-winston';
import { AuthenticationModule } from './auth/auth.module'; import { AuthenticationModule } from './auth/auth.module';
import { AutomationModule } from './automation/automation.module'; import { AutomationModule } from './automation/automation.module';
@ -35,9 +35,6 @@ import { UserNotificationModule } from './user-notification/user-notification.mo
import { UserModule } from './users/user.module'; import { UserModule } from './users/user.module';
import { VisitorPasswordModule } from './vistor-password/visitor-password.module'; import { VisitorPasswordModule } from './vistor-password/visitor-password.module';
import { ThrottlerGuard } from '@nestjs/throttler';
import { ThrottlerModule } from '@nestjs/throttler/dist/throttler.module';
import { isArray } from 'class-validator';
import { winstonLoggerOptions } from '../libs/common/src/logger/services/winston.logger'; import { winstonLoggerOptions } from '../libs/common/src/logger/services/winston.logger';
import { AqiModule } from './aqi/aqi.module'; import { AqiModule } from './aqi/aqi.module';
import { OccupancyModule } from './occupancy/occupancy.module'; import { OccupancyModule } from './occupancy/occupancy.module';
@ -47,18 +44,9 @@ import { WeatherModule } from './weather/weather.module';
ConfigModule.forRoot({ ConfigModule.forRoot({
load: config, load: config,
}), }),
ThrottlerModule.forRoot({ /* ThrottlerModule.forRoot({
throttlers: [{ ttl: 60000, limit: 30 }], throttlers: [{ ttl: 100000, limit: 30 }],
generateKey: (context) => { }), */
const req = context.switchToHttp().getRequest();
console.log('Real IP:', req.headers['x-forwarded-for']);
return req.headers['x-forwarded-for']
? isArray(req.headers['x-forwarded-for'])
? req.headers['x-forwarded-for'][0].split(':')[0]
: req.headers['x-forwarded-for'].split(':')[0]
: req.ip;
},
}),
WinstonModule.forRoot(winstonLoggerOptions), WinstonModule.forRoot(winstonLoggerOptions),
ClientModule, ClientModule,
AuthenticationModule, AuthenticationModule,
@ -100,10 +88,10 @@ import { WeatherModule } from './weather/weather.module';
provide: APP_INTERCEPTOR, provide: APP_INTERCEPTOR,
useClass: LoggingInterceptor, useClass: LoggingInterceptor,
}, },
{ /* {
provide: APP_GUARD, provide: APP_GUARD,
useClass: ThrottlerGuard, useClass: ThrottlerGuard,
}, }, */
], ],
}) })
export class AppModule {} export class AppModule {}

View File

@ -3,6 +3,7 @@ import { SeederService } from '@app/common/seed/services/seeder.service';
import { Logger, ValidationPipe } from '@nestjs/common'; import { Logger, ValidationPipe } from '@nestjs/common';
import { NestFactory } from '@nestjs/core'; import { NestFactory } from '@nestjs/core';
import { json, urlencoded } from 'body-parser'; import { json, urlencoded } from 'body-parser';
import rateLimit from 'express-rate-limit';
import helmet from 'helmet'; import helmet from 'helmet';
import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston'; import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston';
import { setupSwaggerAuthentication } from '../libs/common/src/util/user-auth.swagger.utils'; import { setupSwaggerAuthentication } from '../libs/common/src/util/user-auth.swagger.utils';
@ -21,6 +22,20 @@ async function bootstrap() {
app.use(new RequestContextMiddleware().use); app.use(new RequestContextMiddleware().use);
app.use(
rateLimit({
windowMs: 5 * 60 * 1000,
max: 500,
}),
);
app.use((req, res, next) => {
console.log('Real IP:', req.ip);
next();
});
// app.getHttpAdapter().getInstance().set('trust proxy', 1);
app.use( app.use(
helmet({ helmet({
contentSecurityPolicy: false, contentSecurityPolicy: false,