Storage: use JSON schema to structure data and spawn GraphQL interface

.github/workflows/ods.yml

@@ -97,6 +97,7 @@ jobs:
           docker-compose -f docker-compose.yml build storage
           docker-compose -f docker-compose.yml build storage-db-liquibase
           docker-compose -f docker-compose.yml build storage-mq
+          docker-compose -f docker-compose.yml build graphql
 
       - name: Build Integration-test
         run: |
@@ -117,10 +118,12 @@ jobs:
           IMAGE_ID_STORAGE=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage
           IMAGE_ID_LIQUIBASE=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage-db-liquibase
           IMAGE_ID_STORAGEMQ=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage-mq
+          IMAGE_ID_STORAGE_GRAPHQL=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage-graphql
 
           docker save $IMAGE_ID_STORAGE > storage_postgrest.tar
           docker save $IMAGE_ID_LIQUIBASE > storage_liquibase.tar
           docker save $IMAGE_ID_STORAGEMQ > storage_mq.tar
+          docker save $IMAGE_ID_STORAGE_GRAPHQL > storage_graphql.tar
 
       - name: Upload Storage Docker image as artifact
         uses: actions/upload-artifact@v1
@@ -140,6 +143,12 @@ jobs:
           name: storagemq-artifact
           path: storage_mq.tar
 
+      - name: Upload Storage-GraphQL Docker image as artifact
+        uses: actions/upload-artifact@v1
+        with:
+          name: storagegraphql-artifact
+          path: storage_graphql.tar
+
       # The CDC-Test builds a cut-down image that would override the previously built production image
       # Therefore it is executed after uploading the production image
       - name: Consumer-side CDC-Test
@@ -477,13 +486,18 @@ jobs:
         uses: actions/download-artifact@v1
         with:
           name: storagemq-artifact
+      - name: Download storage-graphql artifact
+        uses: actions/download-artifact@v1
+        with:
+          name: storagegraphql-artifact
 
       - name: Load Docker Images from artifacts
         run: |
 
           docker load -i ./storage-artifact/storage_postgrest.tar
           docker load -i ./liquibase-artifact/storage_liquibase.tar
           docker load -i ./storagemq-artifact/storage_mq.tar
+          docker load -i ./storagegraphql-artifact/storage_graphql.tar
 
       - name: Storage Push to registry
         run: |
@@ -498,6 +512,7 @@ jobs:
           IMAGE_ID_STORAGE=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage
           IMAGE_ID_LIQUIBASE=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage-db-liquibase
           IMAGE_ID_STORAGE_MQ=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage-mq
+          IMAGE_ID_STORAGE_GRAPHQL=$(sed -n 's/^DOCKER_REGISTRY=//p' .env)/storage-graphql
 
           docker tag $IMAGE_ID_STORAGE $IMAGE_ID_STORAGE:$STORAGE_VERSION
           docker tag $IMAGE_ID_STORAGE $IMAGE_ID_STORAGE:latest
@@ -508,6 +523,9 @@ jobs:
           docker tag $IMAGE_ID_STORAGE_MQ $IMAGE_ID_STORAGE_MQ:$STORAGE_VERSION
           docker tag $IMAGE_ID_STORAGE_MQ $IMAGE_ID_STORAGE_MQ:latest
 
+          docker tag $IMAGE_ID_STORAGE_GRAPHQL $IMAGE_ID_STORAGE_GRAPHQL:$STORAGE_VERSION
+          docker tag $IMAGE_ID_STORAGE_GRAPHQL $IMAGE_ID_STORAGE_GRAPHQL:latest
+
           docker push $IMAGE_ID_STORAGE:$STORAGE_VERSION
           docker push $IMAGE_ID_STORAGE:latest
 
@@ -517,6 +535,9 @@ jobs:
           docker push $IMAGE_ID_STORAGE_MQ:$STORAGE_VERSION
           docker push $IMAGE_ID_STORAGE_MQ:latest
 
+          docker push $IMAGE_ID_STORAGE_GRAPHQL:$STORAGE_VERSION
+          docker push $IMAGE_ID_STORAGE_GRAPHQL:latest
+
   pipeline_upload:
     name: Pipeline Publish
     runs-on: ubuntu-18.04

adapter/src/main/java/org/jvalue/ods/adapterservice/datasource/DatasourceManager.java

@@ -13,6 +13,7 @@
 import org.jvalue.ods.adapterservice.datasource.repository.DatasourceRepository;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.everit.json.schema.ValidationException;
 
 import org.jvalue.ods.adapterservice.datasource.validator.*;
 
@@ -100,7 +101,7 @@ public DataImport.MetaData trigger(Long id, RuntimeParameters runtimeParameters)
    */
   @Transactional
   DataImport.MetaData executeImport(Long id, RuntimeParameters runtimeParameters)
-      throws DatasourceNotFoundException, ImporterParameterException, InterpreterParameterException, IOException {
+      throws DatasourceNotFoundException, ImporterParameterException, InterpreterParameterException, IOException {    
     Datasource datasource = getDatasource(id);
     DataImport dataImport = new DataImport(datasource, "", ValidationMetaData.HealthStatus.FAILED);
     Validator validator = new JsonSchemaValidator();

adapter/src/main/java/org/jvalue/ods/adapterservice/datasource/model/types/CustomStringArrayType.java

@@ -53,7 +53,7 @@ public Object	deepCopy(Object value) {
   public Serializable	disassemble(Object value) throws HibernateException {
     return (Serializable) value;
   }
-
+  
   @Override
   public boolean equals(Object x, Object y) throws HibernateException {
     return x.equals(y);

adapter/src/main/java/org/jvalue/ods/adapterservice/datasource/validator/JsonSchemaValidator.java

@@ -1,13 +1,19 @@
 package org.jvalue.ods.adapterservice.datasource.validator;
 
+import org.jvalue.ods.adapterservice.datasource.validator.ValidationMetaData;
 import org.jvalue.ods.adapterservice.datasource.model.*;
+import org.jvalue.ods.adapterservice.datasource.model.exceptions.*;
+import java.io.IOException;
 import org.everit.json.schema.ValidationException;
 
 import org.everit.json.schema.Schema;
 import org.everit.json.schema.loader.SchemaLoader;
+
 import org.json.JSONArray;
 import org.json.JSONObject;
+import org.json.JSONTokener;
 
+import java.util.Arrays;
 import com.google.gson.Gson;
 
 public class JsonSchemaValidator implements Validator {
@@ -29,7 +35,6 @@ public ValidationMetaData validate(DataImport dataImport){
       else {
         schema.validate(new JSONObject(dataImport.getData()));
       }
-
       validationMetaData.setHealthStatus(ValidationMetaData.HealthStatus.OK);
       return validationMetaData;
     } catch ( ValidationException e) {

docker-compose.yml

@@ -163,7 +163,7 @@ services:
     build: ./storage/storage-mq/
     environment:
       CONNECTION_RETRIES: 30
-      CONNECTION_BACKOFF_IN_MS: 2000
+      CONNECTION_BACKOFF_IN_MS: 20000
 
       POSTGRES_HOST: storage-db
       POSTGRES_PORT: 5432
@@ -221,15 +221,26 @@ services:
       POSTGRES_DB: ods
       POSTGRES_USER: ods_admin
       POSTGRES_PASSWORD: ods_pw
+
+  graphql:
+    image: ${DOCKER_REGISTRY}/storage-graphql
+    build: ./storage/storage-graphql/
+    depends_on:
+        - storage-db
+        - storage-db-liquibase
+    ports:
+        - 5432:5432
+    command: ["--connection", postgres://ods_admin:ods_pw@storage-db:5432/ods, "--port", "5432", "--schema", "storage", "--append-plugins", "postgraphile-plugin-connection-filter"]
+
   # Uncomment this if you want to persist the data.
   # volumes:
   #   - "./pgdata:/var/lib/postgresql/data"
 
   # Uncomment this if you want to manage the Postgres databases with adminer
-  # adminer: # management UI for Postgres
-  #   image: adminer
-  #   ports:
-  #     - 8081:8080
+  adminer: # management UI for Postgres
+    image: adminer
+    ports:
+      - 8081:8080
 
   storage-db-liquibase: # perform database migration on start up
     image: ${DOCKER_REGISTRY}/storage-db-liquibase

pipeline/src/pipeline-config/outboxEventPublisher.ts

@@ -14,10 +14,12 @@ export async function publishCreation(
   client: ClientBase,
   pipelineId: number,
   pipelineName: string,
+  schema?: Record<string, unknown>,
 ): Promise<string> {
   const content = {
     pipelineId: pipelineId,
     pipelineName: pipelineName,
+    schema: schema != null ? schema : undefined,
   };
   return await insertEvent(client, AMQP_PIPELINE_CONFIG_CREATED_TOPIC, content);
 }

pipeline/src/pipeline-config/pipelineConfigManager.ts

@@ -23,6 +23,7 @@ export class PipelineConfigManager {
         client,
         savedConfig.id,
         savedConfig.metadata.displayName,
+        savedConfig.schema,
       );
       return savedConfig;
     });

storage/storage-graphql/Dockerfile

@@ -0,0 +1,8 @@
+FROM node:alpine
+
+# Install PostGraphile and PostGraphile connection filter plugin
+RUN npm install -g postgraphile
+RUN npm install -g postgraphile-plugin-connection-filter
+
+EXPOSE 5000
+ENTRYPOINT ["postgraphile", "-n", "0.0.0.0"]

storage/storage-mq/package.json

@@ -19,6 +19,7 @@
     "@jvalue/node-dry-amqp": "0.1.2",
     "@jvalue/node-dry-basics": "0.0.3",
     "@jvalue/node-dry-pg": "1.2.1",
+    "@types/json-schema": "^7.0.9",
     "cors": "^2.8.5",
     "express": "^4.17.1"
   },

storage/storage-mq/src/api/pipelineConfigEventHandler.ts

@@ -1,3 +1,4 @@
+import { JsonSchemaElementBase, isDefined } from '../service/sharedHelper';
 import { StorageStructureRepository } from '../storage-structure/storageStructureRepository';
 
 export class PipelineConfigEventHandler {
@@ -11,6 +12,13 @@ export class PipelineConfigEventHandler {
     await this.structureRepository.create(
       pipelineCreatedEvent.pipelineId.toString(),
     );
+    if (isDefined(pipelineCreatedEvent.schema)) {
+      await this.structureRepository.createForSchema(
+        pipelineCreatedEvent.schema,
+        pipelineCreatedEvent.pipelineName +
+          pipelineCreatedEvent.pipelineId.toString(),
+      );
+    }
   }
 
   async handleDeletion(
@@ -25,6 +33,7 @@ export class PipelineConfigEventHandler {
 export interface PipelineCreatedEvent {
   pipelineId: number;
   pipelineName: string;
+  schema?: JsonSchemaElementBase;
 }
 
 export interface PipelineDeletedEvent {

storage/storage-mq/src/api/pipelineExecutionEventHandler.ts

@@ -1,3 +1,4 @@
+import { JsonSchemaElementBase, isDefined } from '../service/sharedHelper';
 import { StorageContentRepository } from '../storage-content/storageContentRepository';
 
 export class PipelineExecutionEventHandler {
@@ -14,13 +15,25 @@ export class PipelineExecutionEventHandler {
         data: pipelineExecutedEvent.data,
       },
     );
+    if (isDefined(pipelineExecutedEvent.schema)) {
+      await this.contentRepository.saveContentForSchema(
+        pipelineExecutedEvent.pipelineName +
+          pipelineExecutedEvent.pipelineId.toString(),
+        {
+          pipelineId: pipelineExecutedEvent.pipelineId,
+          timestamp: pipelineExecutedEvent.timestamp ?? new Date(),
+          data: pipelineExecutedEvent.data,
+          schema: pipelineExecutedEvent.schema,
+        },
+      );
+    }
   }
 }
 
 export interface PipelineExecutedEvent {
   pipelineId: number;
   pipelineName: string;
   data: unknown;
-  schema?: Record<string, unknown>;
+  schema?: JsonSchemaElementBase;
   timestamp?: Date;
 }