feat: initial sqs report calculation support

nest-cli.json

@@ -1,5 +1,9 @@
 {
   "collection": "@nestjs/schematics",
   "sourceRoot": "src",
-  "compilerOptions": {}
+  "compilerOptions": {
+    "assets": [
+      { "include": "./config/*.yaml", "outDir": "./dist", "watchAssets": true }
+    ]
+  }
 }

package.json

@@ -29,6 +29,7 @@
     "@nestjs/schedule": "4.0.0",
     "@ntegral/nestjs-sentry": "^4.0.0",
     "@sentry/node": "^7.38.0",
+    "js-yaml": "4.1.0",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
     "rxjs": "^7.8.0",
@@ -40,6 +41,7 @@
     "@nestjs/testing": "^9.3.9",
     "@types/express": "^4.17.17",
     "@types/jest": "^29.4.0",
+    "@types/js-yaml": "4.0.9",
     "@types/node": "^18.13.0",
     "@types/supertest": "^2.0.12",
     "@types/uuid": "9.0.8",

src/app.module.ts

@@ -6,6 +6,7 @@ import { ConfigModule, ConfigService } from '@nestjs/config';
 import { HttpModule } from '@nestjs/axios';
 import { ReportModule } from './report/report.module';
 import { ScheduleModule } from '@nestjs/schedule';
+import { AppConfiguration } from './config/configuration';
 import { ReportChangesModule } from './report-changes/report-changes.module';
 import { NotificationModule } from './notification/notification.module';
 
@@ -29,7 +30,11 @@ const lowSeverityLevels: SeverityLevel[] = ['log', 'info'];
   imports: [
     HttpModule,
     ScheduleModule.forRoot(),
-    ConfigModule.forRoot({ isGlobal: true }),
+    ConfigModule.forRoot({
+      isGlobal: true,
+      ignoreEnvFile: false,
+      load: [AppConfiguration],
+    }),
     SentryModule.forRootAsync({
       imports: [ConfigModule],
       inject: [ConfigService],

src/config/app.yaml

@@ -0,0 +1,2 @@
+COUCH_SQS_CLIENT_CONFIG:
+  BASE_URL: http://localhost:4984

src/config/configuration.ts

@@ -0,0 +1,27 @@
+import { readFileSync } from 'fs';
+import * as yaml from 'js-yaml';
+import { join } from 'path';
+
+const CONFIG_FILENAME = 'app.yaml';
+
+export function AppConfiguration() {
+  return flatten(
+    yaml.load(readFileSync(join(__dirname, CONFIG_FILENAME), 'utf8')) as Record<
+      string,
+      string
+    >,
+  );
+}
+
+function flatten(obj: any, prefix = '', delimiter = '_') {
+  return Object.keys(obj).reduce((acc: any, k: string) => {
+    const pre = prefix.length ? prefix + delimiter : '';
+
+    if (typeof obj[k] === 'object')
+      Object.assign(acc, flatten(obj[k], pre + k));
+    else {
+      acc[pre + k] = obj[k];
+    }
+    return acc;
+  }, {});
+}

src/couchdb/couch-sqs-client.service.spec.ts

@@ -0,0 +1,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { CouchSqsClient } from './couch-sqs.client';
+
+describe('CouchSqsClientService', () => {
+  let service: CouchSqsClient;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [CouchSqsClient],
+    }).compile();
+
+    service = module.get<CouchSqsClient>(CouchSqsClient);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

src/couchdb/couch-sqs.client.ts

@@ -0,0 +1,36 @@
+import { Logger } from '@nestjs/common';
+import { HttpService } from '@nestjs/axios';
+import { catchError, map, Observable } from 'rxjs';
+
+export class CouchSqsClientConfig {
+  BASE_URL = '';
+  BASIC_AUTH_USER = '';
+  BASIC_AUTH_PASSWORD = '';
+}
+
+export interface QueryRequest {
+  query: string;
+  args?: string[];
+}
+
+export class CouchSqsClient {
+  private readonly logger: Logger = new Logger(CouchSqsClient.name);
+
+  constructor(private httpService: HttpService) {}
+
+  executeQuery(path: string, query: QueryRequest): Observable<string> {
+    return this.httpService.post(path, query).pipe(
+      map((response) => response.data),
+      catchError((err) => {
+        this.logger.error(err);
+        this.logger.debug(
+          '[CouchSqsClient] Could not execute Query: ',
+          this.httpService.axiosRef.defaults.url,
+          path,
+          query,
+        );
+        throw err;
+      }),
+    );
+  }
+}

src/domain/report.ts

@@ -16,10 +16,11 @@ export class Report {
   schema: ReportSchema | undefined;
   queries: string[];
 
-  constructor(id: string, name: string, queries: string[]) {
+  constructor(id: string, name: string, queries: string[], mode: string) {
     this.id = id;
     this.name = name;
     this.queries = queries;
+    this.mode = mode;
   }
 
   setId(id: string): Report {

src/report/core/sqs-report-calculator.service.ts

@@ -1,23 +1,57 @@
-import { Injectable } from '@nestjs/common';
+import {
+  BadRequestException,
+  Injectable,
+  NotFoundException,
+} from '@nestjs/common';
 import { ReportCalculator } from './report-calculator';
 import { ReportData } from '../../domain/report-data';
-import { delay, Observable, of } from 'rxjs';
+import { map, mergeAll, Observable, switchMap } from 'rxjs';
 import { ReportCalculation } from '../../domain/report-calculation';
-import { Reference } from '../../domain/reference';
+import { DefaultReportStorage } from '../storage/report-storage.service';
+import { CouchSqsClient } from '../../couchdb/couch-sqs.client';
 import { v4 as uuidv4 } from 'uuid';
+import { Reference } from '../../domain/reference';
 
 @Injectable()
 export class SqsReportCalculator implements ReportCalculator {
+  constructor(
+    private sqsClient: CouchSqsClient,
+    private reportStorage: DefaultReportStorage,
+  ) {}
+
   calculate(reportCalculation: ReportCalculation): Observable<ReportData> {
-    return of(
-      new ReportData(
-        `ReportData:${uuidv4()}`,
-        reportCalculation.report,
-        new Reference(reportCalculation.id),
-      ).setData({
-        foo: 'bar',
-        dummyReportData: 'foo',
+    return this.reportStorage.fetchReport(reportCalculation.report).pipe(
+      switchMap((report) => {
+        if (!report) {
+          throw new NotFoundException();
+        }
+
+        if (report.mode !== 'sql') {
+          throw new BadRequestException();
+        }
+
+        if (report.queries.length === 0) {
+          throw new BadRequestException();
+        }
+
+        return report.queries.flatMap((query) => {
+          return this.sqsClient
+            .executeQuery('/app/_design/sqlite:config', {
+              query: query,
+              args: [], // TODO pass args here
+            })
+            .pipe(
+              map((rawResponse) => {
+                return new ReportData(
+                  `ReportData:${uuidv4()}`,
+                  reportCalculation.report,
+                  new Reference(reportCalculation.id),
+                ).setData(rawResponse);
+              }),
+            );
+        });
       }),
-    ).pipe(delay(5000));
+      mergeAll(),
+    );
   }
 }

src/report/di/couchdb-sqs-configuration.ts

@@ -0,0 +1,28 @@
+import {
+  CouchSqsClient,
+  CouchSqsClientConfig,
+} from '../../couchdb/couch-sqs.client';
+import { HttpService } from '@nestjs/axios';
+import { ConfigService } from '@nestjs/config';
+
+export const CouchSqsClientFactory = (
+  httpService: HttpService,
+  configService: ConfigService,
+): CouchSqsClient => {
+  const CONFIG_BASE = 'COUCH_SQS_CLIENT_CONFIG_';
+
+  const config: CouchSqsClientConfig = {
+    BASE_URL: configService.getOrThrow(CONFIG_BASE + 'BASE_URL'),
+    BASIC_AUTH_USER: configService.getOrThrow(CONFIG_BASE + 'BASIC_AUTH_USER'),
+    BASIC_AUTH_PASSWORD: configService.getOrThrow(
+      CONFIG_BASE + 'BASIC_AUTH_PASSWORD',
+    ),
+  };
+
+  httpService.axiosRef.defaults.baseURL = config.BASE_URL;
+  httpService.axiosRef.defaults.headers['Authorization'] = `Basic ${Buffer.from(
+    `${config.BASIC_AUTH_USER}:${config.BASIC_AUTH_PASSWORD}`,
+  ).toString('base64')}`;
+
+  return new CouchSqsClient(httpService);
+};

src/report/storage/report-storage.service.ts

@@ -37,6 +37,7 @@ export class DefaultReportStorage implements ReportStorage {
               reportEntity.id,
               reportEntity.doc.title,
               reportEntity.doc.aggregationDefinitions,
+              reportEntity.doc.mode,
             ).setSchema({
               fields: reportEntity.doc.aggregationDefinitions, // todo generate actual fields here
             }),
@@ -55,6 +56,7 @@ export class DefaultReportStorage implements ReportStorage {
           report._id,
           report.title,
           report.aggregationDefinitions,
+          report.mode,
         ).setSchema({
           fields: report.aggregationDefinitions, // todo generate actual fields here
         });

src/report/tasks/report-calculation-processor.service.ts

@@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 import { catchError, map, Observable, of, switchMap } from 'rxjs';
 import {
   ReportCalculation,
@@ -10,6 +10,8 @@ import { ReportData } from '../../domain/report-data';
 
 @Injectable()
 export class ReportCalculationProcessor {
+  private readonly logger = new Logger(ReportCalculationProcessor.name);
+
   constructor(
     private reportStorage: DefaultReportStorage,
     private reportCalculator: SqsReportCalculator,
@@ -82,12 +84,15 @@ export class ReportCalculationProcessor {
     reportCalculation: ReportCalculation,
     err: any,
   ): Observable<ReportCalculation> {
+    this.logger.error('CALCULATION_FAILED', err, {
+      reportCalculation: reportCalculation,
+    });
     return this.reportStorage.storeCalculation(
       reportCalculation
         .setStatus(ReportCalculationStatus.FINISHED_ERROR)
         .setOutcome({
           errorCode: 'CALCULATION_FAILED',
-          errorMessage: err,
+          errorMessage: 'Something went wrong.',
         })
         .setEndDate(new Date().toISOString()),
     );