feat: added SSE feature and NATS connection

package.json

@@ -28,10 +28,13 @@
   },
   "dependencies": {
     "@nestjs/common": "^11.0.1",
+    "@nestjs/config": "^4.0.3",
     "@nestjs/core": "^11.0.1",
     "@nestjs/platform-express": "^11.0.1",
     "@nestjs/swagger": "^11.0.0",
     "@nestjs/typeorm": "^11.0.0",
+    "class-validator": "^0.14.4",
+    "nats": "^2.29.3",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.2",
     "typeorm": "^0.3.28"

src/app.module.ts

@@ -1,52 +1,41 @@
 import { Module } from '@nestjs/common';
+import { ConfigModule, ConfigService } from '@nestjs/config';
 import { TypeOrmModule, TypeOrmModuleOptions } from '@nestjs/typeorm';
 import { AppController } from './app.controller';
 import { AppService } from './app.service';
 import { MeasureModule } from './data-api/measure.module';
 import { SensorModule } from './data-api/sensor.module';
+import { validate } from './env.validation';
 
-function parseSslFlag(value?: string): boolean {
-  return value === 'true' || value === '1';
-}
-
-function buildTypeOrmOptions(): TypeOrmModuleOptions {
-  const {
-    MEASURES_DB_HOST,
-    MEASURES_DB_PORT,
-    MEASURES_DB_USER,
-    MEASURES_DB_PASSWORD,
-    MEASURES_DB_NAME,
-    DB_SSL,
-  } = process.env;
-
-  if (
-    !MEASURES_DB_HOST ||
-    !MEASURES_DB_PORT ||
-    !MEASURES_DB_USER ||
-    !MEASURES_DB_NAME
-  ) {
-    throw new Error(
-      'Missing TypeORM database configuration. Set MEASURES_DB_HOST, MEASURES_DB_PORT, MEASURES_DB_USER and MEASURES_DB_NAME.',
-    );
-  }
-
-  return {
-    type: 'postgres' as const,
-    host: MEASURES_DB_HOST,
-    port: Number.parseInt(MEASURES_DB_PORT, 10),
-    username: MEASURES_DB_USER,
-    password: MEASURES_DB_PASSWORD,
-    database: MEASURES_DB_NAME,
-    ssl: parseSslFlag(DB_SSL),
-    autoLoadEntities: true,
-  };
-}
+const databaseImports =
+  process.env.NODE_ENV === 'test'
+    ? []
+    : [
+        TypeOrmModule.forRootAsync({
+          inject: [ConfigService],
+          useFactory: (configService: ConfigService): TypeOrmModuleOptions => {
+            return {
+              type: 'postgres',
+              host: configService.get<string>('MEASURES_DB_HOST'),
+              port: configService.get<number>('MEASURES_DB_PORT'),
+              username: configService.get<string>('MEASURES_DB_USER'),
+              password: configService.get<string>('MEASURES_DB_PASSWORD'),
+              database: configService.get<string>('MEASURES_DB_NAME'),
+              ssl: configService.get<boolean>('DB_SSL'),
+              autoLoadEntities: true,
+            };
+          },
+        }),
+      ];
 
 @Module({
   imports: [
-    TypeOrmModule.forRootAsync({
-      useFactory: () => buildTypeOrmOptions(),
+    ConfigModule.forRoot({
+      isGlobal: true,
+      validate,
+      expandVariables: true,
     }),
+    ...databaseImports,
     MeasureModule,
     SensorModule,
   ],

src/data-api/controller/measure.controller.spec.ts

@@ -7,6 +7,7 @@ import { PaginatedQueryModel } from '../models/paginated-query.model';
 import { QueryResponseDto } from '../dto/query.response.dto';
 import { EncryptedEnvelopeModel } from '../models/encrypted-envelope.model';
 import { EncryptedEnvelopeDto } from '../dto/encrypted-envelope.dto';
+import { firstValueFrom, of } from 'rxjs';
 
 describe('MeasureController', () => {
   let controller: MeasureController;
@@ -123,7 +124,7 @@ describe('MeasureController', () => {
       expect(result).toEqual([queryResponseDto]);
     });
 
-    it('should use default limit = 1000 when limit is not provided', async () => {
+    it('should use default limit = 999 when limit is not provided', async () => {
       const from = '2024-01-01T00:00:00Z';
       const to = '2024-01-02T00:00:00Z';
 
@@ -159,7 +160,7 @@ describe('MeasureController', () => {
       expect(serviceQueryMock).toHaveBeenCalledWith({
         from,
         to,
-        limit: 1000,
+        limit: 999,
         gatewayId: undefined,
         sensorId: undefined,
         sensorType: undefined,
@@ -308,4 +309,94 @@ describe('MeasureController', () => {
       });
     });
   });
+
+  describe('stream', () => {
+    it('should pass filters, since and JWT context to the stream listener', async () => {
+      const payload = Buffer.from(
+        JSON.stringify({
+          tenantId: 'tenant-1',
+          exp: Math.floor(
+            new Date('2026-03-23T10:05:00.000Z').getTime() / 1000,
+          ),
+        }),
+      ).toString('base64url');
+
+      mockStreamListenerService.stream.mockReturnValue(
+        of({
+          kind: 'data',
+          data: {
+            gatewayId: 'gw-1',
+            sensorId: 'sensor-1',
+            sensorType: 'temperature',
+            timestamp: '2026-03-23T10:00:00.000Z',
+            encryptedData: 'enc',
+            iv: 'iv',
+            authTag: 'tag',
+            keyVersion: 1,
+          },
+        }),
+      );
+
+      const result = await firstValueFrom(
+        controller.stream(
+          {
+            headers: {
+              authorization: `Bearer header.${payload}.signature`,
+            },
+          } as never,
+          'gw-1' as never,
+          'sensor-1' as never,
+          'temperature' as never,
+          '2026-03-23T09:50:00.000Z',
+        ),
+      );
+
+      expect(mockStreamListenerService.stream).toHaveBeenCalledWith({
+        gatewayId: ['gw-1'],
+        sensorId: ['sensor-1'],
+        sensorType: ['temperature'],
+        since: '2026-03-23T09:50:00.000Z',
+        tenantId: 'tenant-1',
+        tokenExpiresAt: new Date('2026-03-23T10:05:00.000Z').getTime(),
+      });
+      expect(result).toEqual({
+        data: {
+          gatewayId: 'gw-1',
+          sensorId: 'sensor-1',
+          sensorType: 'temperature',
+          timestamp: '2026-03-23T10:00:00.000Z',
+          encryptedData: 'enc',
+          iv: 'iv',
+          authTag: 'tag',
+          keyVersion: 1,
+        },
+      });
+    });
+
+    it('should map token_expired to an SSE error event', async () => {
+      mockStreamListenerService.stream.mockReturnValue(
+        of({
+          kind: 'error',
+          reason: 'token_expired',
+        }),
+      );
+
+      const result = await firstValueFrom(
+        controller.stream(
+          { headers: {} } as never,
+          undefined,
+          undefined,
+          undefined,
+          undefined,
+        ),
+      );
+
+      expect(result).toEqual({
+        data: {
+          type: 'error',
+          reason: 'token_expired',
+        },
+      });
+    });
+  });
 });