wip: influx queries

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

@@ -1,6 +1,6 @@
 import { FastifyPluginAsync, FastifyPluginOptions } from 'fastify';
 import fp from 'fastify-plugin';
-import { InfluxDB } from '@influxdata/influxdb-client';
+import { InfluxDB, WriteApi } from '@influxdata/influxdb-client';
 import { Plugins } from './plugins';
 import { Organization, OrgsAPI } from '@influxdata/influxdb-client-apis';
 
@@ -21,6 +21,12 @@ const influxDbPlugin: FastifyPluginAsync<InfluxdbPluginOptions> = async (fastify
     const influxClient = new InfluxDB({
         token: options.token,
         url: options.host,
+        writeOptions: {
+            batchSize: 10,
+            writeSuccess: (lines: string[]) => {
+                fastify.log.info(lines, 'Influx data written');
+            },
+        },
     });
 
     const orgsApi = new OrgsAPI(influxClient);

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

@@ -32,26 +32,16 @@ export class CarsService extends InfluxDbServiceBase {
         this.writeAPi.writePoint(point);
     }
 
-    public getCarsStatus(callsigns: string[]) {
+    public async getCarsStatus(callsigns: string[]) {
         console.log(callsigns);
         const query = `from(bucket: "cars")
         |> range(start: -48h)
         |> filter(fn: (r) => r._measurement == "car_status")
-        |> filter(fn: (r) => r.callsign == "cesilko-test")
-        |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
+        |> filter(fn: (r) => [${callsigns}].includes(r.callsign))
         |> last()`;
 
-        this.queryAPi.queryRows(query, {
-            next(row, tableMeta) {
-                const obj = tableMeta.toObject(row);
-                console.log(obj);
-            },
-            error(error) {
-                console.error(error);
-            },
-            complete() {
-                console.log('Query completed');
-            },
-        });
+        const data = await this.queryAPi.collectRows(query);
+
+        console.log(data);
     }
 }