feat: connect to InfluxDB and store telemetry + car data

Author: raska-vilem

GAPP/apps/gapp-server/src/app.ts

@@ -6,6 +6,8 @@ import swaggerUi from '@fastify/swagger-ui';
 import { carsController } from './controllers/cars.controller';
 import sondehubPlugin from './plugins/sondehub';
 import { sondesController } from './controllers/sondes.controller';
+import carsServicePlugin from './plugins/cars-service';
+import telemetryServicePlugin from './plugins/telemetry-service';
 
 interface AppOptions extends FastifyPluginOptions {
     influxDbToken: string;
@@ -23,12 +25,12 @@ export const app = async (fastify: FastifyInstance, opts: AppOptions) => {
         token: opts.influxDbToken,
         org: opts.influxDbOrg,
     });
-    await fastify.register(sondehubPlugin, { dev: true });
+    await fastify.register(sondehubPlugin, { dev: false });
+    await fastify.register(carsServicePlugin);
+    await fastify.register(telemetryServicePlugin);
 
     await fastify.register(swagger);
-    await fastify.register(swaggerUi, {
-        routePrefix: '/docs',
-    });
+    await fastify.register(swaggerUi, { routePrefix: '/docs' });
 
     // ROUTES
     fastify.register(carsController, { prefix: '/cars' });

GAPP/apps/gapp-server/src/controllers/cars.controller.ts

@@ -15,6 +15,12 @@ export const carsController: FastifyPluginAsyncTypebox = async (fastify) => {
             const { callsign } = req.query;
             const { latitude, longitude, altitude } = req.body;
 
+            req.server.carsService.writeCarStatus(callsign, {
+                latitude,
+                longitude,
+                altitude,
+            });
+
             await req.server.sondehub.uploadStationPosition({
                 uploader_callsign: callsign,
                 uploader_position: [latitude, longitude, altitude],

GAPP/apps/gapp-server/src/controllers/sondes.controller.ts

@@ -15,6 +15,7 @@ export const sondesController: FastifyPluginAsyncTypebox = async (fastify) => {
         async (req, rep) => {
             const telemetryPacket = ttnPacketDto(req.body);
 
+            req.server.telemetryService.writeTelemetry(telemetryPacket);
             req.server.sondehub.addTelemetry(telemetryPacket);
 
             rep.code(200).send('OK');

GAPP/apps/gapp-server/src/plugins/cars-service.ts

@@ -0,0 +1,26 @@
+import { FastifyPluginAsync } from 'fastify';
+import fp from 'fastify-plugin';
+import { Plugins } from './plugins';
+import { CarsService } from '../services/cars.service';
+
+declare module 'fastify' {
+    interface FastifyInstance {
+        carsService: CarsService;
+    }
+}
+
+const carsServicePlugin: FastifyPluginAsync = async (fastify) => {
+    const carsService = new CarsService(fastify.influxClient, fastify.influxOrg);
+
+    await carsService.init();
+
+    fastify.decorate('carsService', carsService);
+    fastify.addHook('onClose', async () => {
+        await carsService.deinit();
+    });
+};
+
+export default fp(carsServicePlugin, {
+    name: Plugins.CARS_SERVICE,
+    dependencies: [Plugins.INFLUXDB],
+});

GAPP/apps/gapp-server/src/plugins/influxdb.ts

@@ -1,7 +1,8 @@
 import { FastifyPluginAsync, FastifyPluginOptions } from 'fastify';
 import fp from 'fastify-plugin';
-import { InfluxDB, QueryApi, WriteApi } from '@influxdata/influxdb-client';
+import { InfluxDB } from '@influxdata/influxdb-client';
 import { Plugins } from './plugins';
+import { Organization, OrgsAPI } from '@influxdata/influxdb-client-apis';
 
 interface InfluxdbPluginOptions extends FastifyPluginOptions {
     host: string;
@@ -11,8 +12,8 @@ interface InfluxdbPluginOptions extends FastifyPluginOptions {
 
 declare module 'fastify' {
     interface FastifyInstance {
-        influxWriteApi: WriteApi;
-        influxQueryApi: QueryApi;
+        influxClient: InfluxDB;
+        influxOrg: Organization;
     }
 }
 
@@ -22,16 +23,21 @@ const influxDbPlugin: FastifyPluginAsync<InfluxdbPluginOptions> = async (fastify
         url: options.host,
     });
 
-    const writeApi = influxClient.getWriteApi(options.org, 'fik');
-    const queryApi = influxClient.getQueryApi(options.org);
+    const orgsApi = new OrgsAPI(influxClient);
+    const orgs = await orgsApi.getOrgs();
+    let org = orgs.orgs.find((org) => org.name === options.org);
+    if (!org) {
+        fastify.log.info(`Creating organization ${options.org}`);
+        org = await orgsApi.postOrgs({
+            body: {
+                name: options.org,
+                description: 'Organization for storing telemetry data',
+            },
+        });
+    }
 
-    fastify.decorate('influxWriteApi', writeApi);
-    fastify.decorate('influxQueryApi', queryApi);
-    fastify.addHook('onClose', async () => {
-        fastify.log.info('Closing influxdb write api...');
-        await writeApi.close();
-        fastify.log.info('Influxdb write api closed');
-    });
+    fastify.decorate('influxOrg', org);
+    fastify.decorate('influxClient', influxClient);
 };
 
 export default fp(influxDbPlugin, {

GAPP/apps/gapp-server/src/plugins/plugins.ts

@@ -1,5 +1,6 @@
 export enum Plugins {
     INFLUXDB = 'influxdb',
     SONDEHUB = 'sondehub',
-    MQTT = 'mqtt',
+    CARS_SERVICE = 'carsService',
+    TELEMETRY_SERVICE = 'telemetryService',
 }

GAPP/apps/gapp-server/src/plugins/telemetry-service.ts

@@ -0,0 +1,26 @@
+import { FastifyPluginAsync } from 'fastify';
+import fp from 'fastify-plugin';
+import { Plugins } from './plugins';
+import { TelemetryService } from '../services/telemetry.service';
+
+declare module 'fastify' {
+    interface FastifyInstance {
+        telemetryService: TelemetryService;
+    }
+}
+
+const telemetryServicePlugin: FastifyPluginAsync = async (fastify) => {
+    const telemetryService = new TelemetryService(fastify.influxClient, fastify.influxOrg);
+
+    await telemetryService.init();
+
+    fastify.decorate('telemetryService', telemetryService);
+    fastify.addHook('onClose', async () => {
+        await telemetryService.deinit();
+    });
+};
+
+export default fp(telemetryServicePlugin, {
+    name: Plugins.TELEMETRY_SERVICE,
+    dependencies: [Plugins.INFLUXDB],
+});

GAPP/apps/gapp-server/src/schemas/index.ts

@@ -1,4 +1,4 @@
-import { Type as T } from '@sinclair/typebox';
+import { Static, Type as T } from '@sinclair/typebox';
 
 export const Q_Callsign = T.Object({
     callsign: T.String(),
@@ -11,6 +11,7 @@ export const B_CarStatus = T.Object({
     longitude: T.Number(),
     altitude: T.Number(),
 });
+export type CarStatus = Static<typeof B_CarStatus>;
 
 export const B_SondeTelemetry = T.Object({
     end_device_ids: T.Object({

GAPP/apps/gapp-server/src/services/cars.service.ts

@@ -0,0 +1,32 @@
+import { InfluxDB, Point, WriteApi } from '@influxdata/influxdb-client';
+import { InfluxDbServiceBase } from '../utils/influxdb-service-base';
+import { Organization } from '@influxdata/influxdb-client-apis';
+import { CarStatus } from '../schemas';
+
+export class CarsService extends InfluxDbServiceBase {
+    private writeAPi: WriteApi;
+
+    constructor(private client: InfluxDB, private org: Organization) {
+        super(client, org.id);
+    }
+
+    public async init() {
+        await this.ensureBucket('cars');
+        this.writeAPi = this.client.getWriteApi(this.org.id, 'cars', 'ms');
+    }
+
+    public async deinit() {
+        await this.writeAPi.close();
+    }
+
+    public writeCarStatus(callsign: string, status: CarStatus) {
+        const point = new Point('car_status')
+            .timestamp(Date.now())
+            .tag('callsign', callsign)
+            .floatField('latitude', status.latitude)
+            .floatField('longitude', status.longitude)
+            .floatField('altitude', status.altitude);
+
+        this.writeAPi.writePoint(point);
+    }
+}

GAPP/apps/gapp-server/src/services/telemetry.service.ts

@@ -0,0 +1,32 @@
+import { InfluxDB, Point, WriteApi } from '@influxdata/influxdb-client';
+import { Organization } from '@influxdata/influxdb-client-apis';
+import { InfluxDbServiceBase } from '../utils/influxdb-service-base';
+import { TelemetryPacket } from '@gapp/sondehub';
+
+export class TelemetryService extends InfluxDbServiceBase {
+    private writeAPi: WriteApi;
+
+    constructor(private client: InfluxDB, private org: Organization) {
+        super(client, org.id);
+    }
+
+    public async init() {
+        await this.ensureBucket('telemetry');
+        this.writeAPi = this.client.getWriteApi(this.org.id, 'telemetry', 'ms');
+    }
+
+    public async deinit() {
+        await this.writeAPi.close();
+    }
+
+    public writeTelemetry(telemetry: TelemetryPacket) {
+        const point = new Point('telemetry_packet')
+            .timestamp(new Date(telemetry.time_received))
+            .tag('callsign', telemetry.payload_callsign)
+            .floatField('latitude', telemetry.lat)
+            .floatField('longitude', telemetry.lon)
+            .floatField('altitude', telemetry.alt);
+
+        this.writeAPi.writePoint(point);
+    }
+}