feat: use password with encoding prefix

feat #18: initial support for authentication

.gitignore

@@ -36,3 +36,4 @@ out/
 ### VS Code ###
 .vscode/
 /dev/gpas*
+/deploy/.env

README.md

@@ -2,44 +2,110 @@
 
 Diese Anwendung versendet ein bwHC-MTB-File an das bwHC-Backend und pseudonymisiert die Patienten-ID.
 
-## Pseudonymisierung der Patienten-ID
+## Einordnung innerhalb einer DNPM-ETL-Strecke
 
-Wenn eine URI zu einer gPAS-Instanz angegeben ist, wird diese verwendet.
+Diese Anwendung erlaubt das Entgegennehmen von HTTP/REST-Anfragen aus dem Onkostar-Plugin **[onkostar-plugin-dnpmexport](https://github.com/CCC-MF/onkostar-plugin-dnpmexport)**.
+
+Der Inhalt einer Anfrage, wenn ein bwHC-MTBFile, wird pseudonymisiert und auf Duplikate geprüft.
+Duplikate werden verworfen, Änderungen werden weitergeleitet.
+
+Löschanfragen werden immer als Löschanfrage an das bwHC-backend weitergeleitet.
+
+Zudem ist eine minimalistische Weboberfläche integriert, die einen Einblick in den aktuellen Zustand der Anwendung gewährt.
+
+![Modell DNPM-ETL-Strecke](docs/etl.png)
+
+### Datenübermittlung über HTTP/REST
+
+Anfragen werden, wenn nicht als Duplikat behandelt, nach der Pseudonymisierung direkt an das bwHC-Backend gesendet.
+
+### Datenübermittlung mit Apache Kafka
+
+Anfragen werden, wenn nicht als Duplikat behandelt, nach der Pseudonymisierung an Apache Kafka übergeben.
+Eine Antwort wird dabei ebenfalls mithilfe von Apache Kafka übermittelt und nach der Entgegennahme verarbeitet.
+
+Siehe hierzu auch: https://github.com/CCC-MF/kafka-to-bwhc
+
+## Konfiguration
+
+### Pseudonymisierung der Patienten-ID
+
+Wenn eine URI zu einer gPAS-Instanz (Version >= 2023.1.0) angegeben ist, wird diese verwendet.
 Ist diese nicht gesetzt. wird intern eine Anonymisierung der Patienten-ID vorgenommen.
 
 * `APP_PSEUDONYMIZE_PREFIX`: Standortbezogenes Prefix - `UNKNOWN`, wenn nicht gesetzt
 * `APP_PSEUDONYMIZE_GENERATOR`: `BUILDIN` oder `GPAS` - `BUILDIN`, wenn nicht gesetzt
 
-### Eingebaute Pseudonymisierung
+#### Eingebaute Anonymisierung
 
-Wurde keine oder die Verwendung der eingebauten Pseudonymisierung konfiguriert, so wird für die Patienten-ID der
+Wurde keine oder die Verwendung der eingebauten Anonymisierung konfiguriert, so wird für die Patienten-ID der
 entsprechende SHA-256-Hash gebildet und Base64-codiert - hier ohne endende "=" - zuzüglich des konfigurierten Prefixes
 als Patienten-Pseudonym verwendet.
 
-### Pseudonymisierung mit gPAS
+#### Pseudonymisierung mit gPAS
 
 Wurde die Verwendung von gPAS konfiguriert, so sind weitere Angaben zu konfigurieren.
 
-* `APP_PSEUDONYMIZE_GPAS_URI`: URI der gPAS-Instanz inklusive Endpoint (z.B. `http://localhost:8080/ttp-fhir/fhir/gpas/$pseudonymizeAllowCreate`)
+* `APP_PSEUDONYMIZE_GPAS_URI`: URI der gPAS-Instanz inklusive Endpoint (z.B. `http://localhost:8080/ttp-fhir/fhir/gpas/$$pseudonymizeAllowCreate`)
 * `APP_PSEUDONYMIZE_GPAS_TARGET`: gPas Domänenname
 * `APP_PSEUDONYMIZE_GPAS_USERNAME`: gPas Basic-Auth Benutzername
 * `APP_PSEUDONYMIZE_GPAS_PASSWORD`: gPas Basic-Auth Passwort
 * `APP_PSEUDONYMIZE_GPAS_SSLCALOCATION`: Root Zertifikat für gPas, falls es dediziert hinzugefügt werden muss.
 
-## Mögliche Endpunkte
+### Anmeldung mit einem Passwort
+
+Ein initialer Administrator-Account kann optional konfiguriert werden und sorgt dafür, dass bestimmte Bereiche nur nach
+einem erfolgreichen Login erreichbar sind.
+
+* `APP_SECURITY_ADMIN_USER`: Muss angegeben werden zur Aktivierung der Zugriffsbeschränkung.
+* `APP_SECURITY_ADMIN_PASSWORD`: Das Passwort für den Administrator (Empfohlen).
+
+Ein Administrator-Passwort muss inklusive des Encoding-Prefixes vorliegen.
+
+Hier Beispiele für das Beispielpasswort `very-secret`:
+
+* `{noop}very-secret` (Das Passwort liegt im Klartext vor - nicht empfohlen!)
+* `{bcrypt}$2y$05$CCkfsMr/wbTleMyjVIK8g.Aa3RCvrvoLXVAsL.f6KeouS88vXD9b6`
+* `{sha256}9a34717f0646b5e9cfcba70055de62edb026ff4f68671ba3db96aa29297d2df5f1a037d58c745657`
+
+Wird kein Administrator-Passwort angegeben, wird ein zufälliger Wert generiert und beim Start der Anwendung in den Logs
+angezeigt.
+
+#### Auswirkungen auf den dargestellten Inhalt
+
+Nur Administratoren haben Zugriff auf den Konfigurationsbereich, nur angemeldete Benutzer können die anonymisierte oder
+pseudonymisierte Patienten-ID einsehen.
+
+Wurde kein Administrator-Account konfiguriert, sind diese Inhalte generell nicht verfügbar.
+
+### Transformation von Werten
+
+In Onkostar kann es vorkommen, dass ein Wert eines Merkmalskatalogs an einem Standort angepasst wurde und dadurch nicht dem Wert entspricht,
+der vom bwHC-Backend akzeptiert wird.
+
+Diese Anwendung bietet daher die Möglichkeit, eine Transformation vorzunehmen. Hierzu muss der "Pfad" innerhalb des JSON-MTB-Files angegeben werden und
+welcher Wert wie ersetzt werden soll.
+
+Hier ein Beispiel für die erste (Index 0 - weitere dann mit 1,2,...) Transformationsregel:
+
+* `APP_TRANSFORMATIONS_0_PATH`: Pfad zum Wert in der JSON-MTB-Datei. Beispiel: `diagnoses[*].icd10.version` für **alle** Diagnosen
+* `APP_TRANSFORMATIONS_0_FROM`: Angabe des Werts, der ersetzt werden soll. Andere Werte bleiben dabei unverändert.
+* `APP_TRANSFORMATIONS_0_TO`: Angabe des neuen Werts.
+
+### Mögliche Endpunkte zur Datenübermittlung
 
 Für REST-Requests als auch zur Nutzung von Kafka-Topics können Endpunkte konfiguriert werden.
 
 Es ist dabei nur die Konfiguration eines Endpunkts zulässig.
 Werden sowohl REST als auch Kafka-Endpunkt konfiguriert, wird nur der REST-Endpunkt verwendet.
 
-### REST
+#### REST
 
 Folgende Umgebungsvariablen müssen gesetzt sein, damit ein bwHC-MTB-File an das bwHC-Backend gesendet wird:
 
 * `APP_REST_URI`: URI der zu benutzenden API der bwHC-Backend-Instanz. z.B.: `http://localhost:9000/bwhc/etl/api`
 
-### Kafka-Topics
+#### Kafka-Topics
 
 Folgende Umgebungsvariablen müssen gesetzt sein, damit ein bwHC-MTB-File an ein Kafka-Topic übermittelt wird:
 
@@ -55,6 +121,128 @@ Weitere Einstellungen können über die Parameter von Spring Kafka konfiguriert
 Lässt sich keine Verbindung zu dem bwHC-Backend aufbauen, wird eine Rückantwort mit Status-Code `900` erwartet, welchen es
 für HTTP nicht gibt.
 
+##### Retention Time
+
+Generell werden in Apache Kafka alle Records entsprechend der Konfiguration vorgehalten.
+So wird ohne spezielle Konfiguration ein Record für 7 Tage in Apache Kafka gespeichert.
+Es sind innerhalb dieses Zeitraums auch alte Informationen weiterhin enthalten, wenn der Consent später abgelehnt wurde.
+
+Durch eine entsprechende Konfiguration des Topics kann dies verhindert werden.
+
+Beispiel - auszuführen innerhalb des Kafka-Containers: Löschen alter Records nach einem Tag
+```
+kafka-configs.sh --bootstrap-server localhost:9092 --alter --topic test --add-config retention.ms=86400000
+```
+
+##### Key based Retention
+
+Möchten Sie hingegen immer nur die letzte Meldung für einen Patienten und eine Erkrankung in Apache Kafka vorhalten,
+so ist die nachfolgend genannte Konfiguration der Kafka-Topics hilfreich.
+
+
+* `retention.ms`: Möglichst kurze Zeit in der alte Records noch erhalten bleiben, z.B. 10 Sekunden 10000
+* `cleanup.policy`: Löschen alter Records und Beibehalten des letzten Records zu einem Key [delete,compact]
+
+Beispiele für ein Topic `test`, hier bitte an die verwendeten Topics anpassen.
+
+```
+kafka-configs.sh --bootstrap-server localhost:9092 --alter --topic test --add-config retention.ms=10000
+kafka-configs.sh --bootstrap-server localhost:9092 --alter --topic test --add-config cleanup.policy=[delete,compact]
+```
+
+Da als Key eines Records die (pseudonymisierte) Patienten-ID und die (anonymisierte) Erkrankungs-ID verwendet wird,
+stehen mit obiger Konfiguration der Kafka-Topics nach 10 Sekunden nur noch der jeweils letzte Eintrag für den entsprechenden
+Key zur Verfügung.
+
+Da der Key sowohl für die Records in Richtung bwHC-Backend für die Rückantwort identisch aufgebaut ist, lassen sich so
+auch im Falle eines Consent-Widerspruchs die enthaltenen Daten als auch die Offenlegung durch Verifikationsdaten in der
+Antwort effektiv verhindern, da diese nach 10 Sekunden gelöscht werden.
+Es steht dann nur noch die jeweils letzten Information zur Verfügung, dass für einen Patienten/eine Erkrankung
+ein Consent-Widerspruch erfolgte.
+
 ## Docker-Images
 
-Diese Anwendung ist auch als Docker-Image verfügbar: https://github.com/CCC-MF/etl-processor/pkgs/container/etl-processor
+Diese Anwendung ist auch als Docker-Image verfügbar: https://github.com/CCC-MF/etl-processor/pkgs/container/etl-processor
+
+### Images lokal bauen 
+
+```bash
+./gradlew bootBuildImage
+```
+
+## Deployment
+*Ausführen als Docker Conatiner:*
+
+```bash
+cd ./deploy
+cp env-sample.env .env
+```
+Wenn gewünscht, Änderungen in der `.env` vornehmen.
+
+```bash
+docker compose up -d
+```
+
+### Einfaches Beispiel für ein eigenes Docker-Compose-File
+
+Die Datei [`docs/docker-compose.yml`](docs/docker-compose.yml) zeigt eine einfache Konfiguration für REST-Requests basierend 
+auf Docker-Compose mit der gestartet werden kann.
+
+### Betrieb hinter einem Reverse-Proxy
+
+Die Anwendung verarbeitet `X-Forwarded`-HTTP-Header und kann daher auch hinter einem Reverse-Proxy betrieben werden.
+
+Dabei werden, je nachdem welche Header durch den Reverse-Proxy gesendet werden auch Protokoll, Host oder auch Path-Prefix
+automatisch erkannt und verwendet werden. Dadurch ist z.B. eine abweichende Angabe des Pfads problemlos möglich.
+
+#### Beispiel *Traefik* (mit Docker-Labels):
+
+Das folgende Beispiel zeigt die Konfiguration in einer Docker-Compose-Datei mit Service-Labels.
+
+```
+...
+  deploy:
+    labels:
+      - "traefik.http.routers.etl.rule=PathPrefix(`/etl-processor`)"
+      - "traefik.http.routers.etl.middlewares=etl-path-strip"
+      - "traefik.http.middlewares.etl-path-strip.stripprefix.prefixes=/etl-processor"
+...
+```
+
+#### Beispiel *nginx*
+
+Das folgende Beispiel zeigt die Konfiguration einer _location_ in einer nginx-Konfigurationsdatei.
+
+```
+...
+  location /etl-processor {
+    set              $upstream http://<beispiel:8080>/;
+    proxy_set_header Host $host;
+    proxy_set_header X-Forwarded-Scheme $scheme;
+    proxy_set_header X-Forwarded-Proto  $scheme;
+    proxy_set_header X-Forwarded-For    $remote_addr;
+    proxy_set_header X-Real-IP          $remote_addr;
+    proxy_pass       $upstream;
+  }
+...
+```
+
+## Entwicklungssetup
+
+Zum Starten einer lokalen Entwicklungs- und Testumgebung kann die beiliegende Datei `dev-compose.yml` verwendet werden.
+Diese kann zur Nutzung der Datenbanken **MariaDB** als auch **PostgreSQL** angepasst werden.
+
+Zur Nutzung von Apache Kafka muss dazu ein Eintrag im hosts-File vorgenommen werden und der Hostname `kafka` auf die lokale
+IP-Adresse verweisen. Ohne diese Einstellung ist eine Nutzung von Apache Kafka außerhalb der Docker-Umgebung nicht möglich.
+
+Beim Start der Anwendung mit dem Profil `dev` wird die in `dev-compose.yml` definierte Umgebung beim Start der
+Anwendung mit gestartet:
+
+```
+SPRING_PROFILES_ACTIVE=dev ./gradlew bootRun
+```
+
+Die Datei `application-dev.yml` enthält hierzu die Konfiguration für das Profil `dev`.
+
+Beim Ausführen der Integrationstests wird eine Testdatenbank in einem Docker-Container gestartet.
+Siehe hier auch die Klasse `AbstractTestcontainerTest` unter `src/integrationTest`.

build.gradle.kts

@@ -4,14 +4,21 @@ import org.springframework.boot.gradle.tasks.bundling.BootBuildImage
 
 plugins {
     war
-    id("org.springframework.boot") version "3.1.2"
-    id("io.spring.dependency-management") version "1.1.0"
-    kotlin("jvm") version "1.9.0"
-    kotlin("plugin.spring") version "1.9.0"
+    id("org.springframework.boot") version "3.2.1"
+    id("io.spring.dependency-management") version "1.1.4"
+    kotlin("jvm") version "1.9.22"
+    kotlin("plugin.spring") version "1.9.22"
 }
 
 group = "de.ukw.ccc"
-version = "0.1.1"
+version = "0.4.0"
+
+var versions = mapOf(
+    "bwhc-dto-java" to "0.2.0",
+    "hapi-fhir" to "6.10.2",
+    "httpclient5" to "5.2.1",
+    "mockito-kotlin" to "5.2.1"
+)
 
 java {
     sourceCompatibility = JavaVersion.VERSION_17
@@ -47,15 +54,18 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-thymeleaf")
     implementation("org.springframework.boot:spring-boot-starter-web")
     implementation("org.springframework.boot:spring-boot-starter-data-jdbc")
+    implementation("org.springframework.boot:spring-boot-starter-security")
+    implementation("org.thymeleaf.extras:thymeleaf-extras-springsecurity6")
     implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
     implementation("org.springframework.kafka:spring-kafka")
     implementation("org.flywaydb:flyway-mysql")
     implementation("commons-codec:commons-codec")
     implementation("io.projectreactor.kotlin:reactor-kotlin-extensions")
-    implementation("de.ukw.ccc:bwhc-dto-java:0.2.0")
-    implementation("ca.uhn.hapi.fhir:hapi-fhir-base:6.6.2")
-    implementation("ca.uhn.hapi.fhir:hapi-fhir-structures-r4:6.6.2")
-    implementation("org.apache.httpcomponents.client5:httpclient5:5.2.1")
+    implementation("de.ukw.ccc:bwhc-dto-java:${versions["bwhc-dto-java"]}")
+    implementation("ca.uhn.hapi.fhir:hapi-fhir-base:${versions["hapi-fhir"]}")
+    implementation("ca.uhn.hapi.fhir:hapi-fhir-structures-r4:${versions["hapi-fhir"]}")
+    implementation("org.apache.httpcomponents.client5:httpclient5:${versions["httpclient5"]}")
+    implementation("com.jayway.jsonpath:json-path")
     runtimeOnly("org.mariadb.jdbc:mariadb-java-client")
     runtimeOnly("org.postgresql:postgresql")
     developmentOnly("org.springframework.boot:spring-boot-devtools")
@@ -64,7 +74,7 @@ dependencies {
     providedRuntime("org.springframework.boot:spring-boot-starter-tomcat")
     testImplementation("org.springframework.boot:spring-boot-starter-test")
     testImplementation("io.projectreactor:reactor-test")
-    testImplementation("org.mockito.kotlin:mockito-kotlin:5.0.0")
+    testImplementation("org.mockito.kotlin:mockito-kotlin:${versions["mockito-kotlin"]}")
     integrationTestImplementation("org.testcontainers:junit-jupiter")
     integrationTestImplementation("org.testcontainers:postgresql")
 }

deploy/docker-compose.yaml

@@ -0,0 +1,55 @@
+
+
+services:
+  dnpm-etl-processor:
+    image: ghcr.io/ccc-mf/etl-processor:latest
+    environment:
+      LOGGING_LEVEL_DEV: ${DNPM_LOG_LEVEL:-INFO}
+      SPRING_KAFKA_SECURITY_PROTOCOL: ${DNPM_KAFKA_SECURITY_PROTOCOL:-SSL}
+      SPRING_KAFKA_SSL_TRUST-STORE-TYPE: PKCS12
+      SPRING_KAFKA_SSL_TRUST-STORE-LOCATION: /opt/dnpm-processor/ssl/truststore.jks
+      SPRING_KAFKA_SSL_TRUST-STORE-PASSWORD: ${KAFKA_TRUST_STORE_PASSWORD}
+      SPRING_KAFKA_SSL_KEY-STORE-TYPE: PKCS12
+      SPRING_KAFKA_SSL_KEY-STORE-LOCATION: /opt/dnpm-processor/ssl/keystore.jks
+      SPRING_KAFKA_SSL_KEY-STORE-PASSWORD: ${DNPM_PROCESSOR_KEY_STORE_PASSWORD}
+      SPRING_KAFKA_PRODUCER_COMPRESSION-TYPE: gzip
+      APP_KAFKA_TOPIC: ${DNPM_KAFKA_TOPIC}
+      APP_KAFKA_SERVERS: ${KAFKA_BROKERS}
+      APP_KAFKA_GROUP_ID: ${DNPM_KAFKA_GROUP_ID}
+      APP_KAFKA_RESPONSE_TOPIC: ${DNPM_KAFKA_RESPONSE_TOPIC}
+      APP_REST_URI: ${DNPM_BWHC_REST_URI}
+      SPRING_DATASOURCE_URL: ${DNPM_DATASOURCE_URL}
+      SPRING_DATASOURCE_PASSWORD: ${DNPM_MARIADB_USER_PW}
+      SPRING_DATASOURCE_USERNAME: ${DNPM_MARIADB_DB}
+      APP_PSEUDONYMIZE_GPAS_SSLCALOCATION: /workspace/opt/dnpm-processor/ssl/mosaic.crt
+      APP_PSEUDONYMIZE_GPAS_PASSWORD: ${DNPM_PSEUDONYMIZE_GPAS_PASSWORD}
+      APP_PSEUDONYMIZE_GPAS_USERNAME: ${DNPM_PSEUDONYMIZE_GPAS_USERNAME}
+      APP_PSEUDONYMIZE_GPAS_TARGET: ${DNPM_PSEUDONYMIZE_GPAS_TARGET}
+      APP_PSEUDONYMIZE_GPAS_URI: ${DNPM_PSEUDONYMIZE_GPAS_URI}
+      APP_PSEUDONYMIZE_PREFIX: ${DNPM_APP_PSEUDONYMIZE_PREFIX}
+      APP_PSEUDONYMIZER: ${DNPM_PSEUDONYMIZE_GENERATOR}
+    volumes:
+      - /etc/localtime:/etc/localtime:ro
+      - /etc/timezone:/etc/timezone:ro
+      #- ${DNPM_TO_SSL_KEYSTORE_LOCATION}:/workspace/opt/dnpm-processor/ssl/keystore.jks:ro
+      #- ${KAFKA_TRUST_STORE_LOCATION}:/workspace/opt/dnpm-processor/ssl/truststore.jks:ro
+      #- ${DNPM_PSEUDONYMIZE_GPAS_SSLCALOCATION}:/workspace/opt/dnpm-processor/ssl/mosaic.crt
+
+    depends_on:
+      - dnpm-monitor-db
+    ports:
+      - "${DNPM_MONITORING_HTTP_PORT:-8080}:8080"
+
+  # todo add volume
+  dnpm-monitor-db:
+    image: mariadb:10
+    environment:
+      MARIADB_DATABASE: ${DNPM_MARIADB_DB}
+      MARIADB_USER: ${DNPM_MARIADB_USER}
+      MARIADB_PASSWORD: ${DNPM_MARIADB_USER_PW}
+      MARIADB_ROOT_PASSWORD: ${DNPM_MARIADB_ROOT_PW}
+    expose:
+      - "3306"
+
+
+

deploy/env-sample.env

@@ -0,0 +1,40 @@
+# monitoring access port
+DNPM_MONITORING_HTTP_PORT=8088
+DNPM_LOG_LEVEL=INFO
+
+# GPAS or BUILDIN
+DNPM_PSEUDONYMIZE_GENERATOR=BUILDIN
+DNPM_APP_PSEUDONYMIZE_PREFIX=ANONYM
+DNPM_PSEUDONYMIZE_GPAS_URI=
+DNPM_PSEUDONYMIZE_GPAS_TARGET=
+DNPM_PSEUDONYMIZE_GPAS_USERNAME=
+DNPM_PSEUDONYMIZE_GPAS_PASSWORD=
+
+# path to ca root cert if needed
+DNPM_PSEUDONYMIZE_GPAS_SSLCALOCATION=
+
+DNPM_MARIADB_DB=dnpm_monitoring
+DNPM_MARIADB_USER=$DNPM_MARIADB_DB
+DNPM_MARIADB_USER_PW=MySuperSecurePassword111
+DNPM_MARIADB_ROOT_PW=MySuperDuperSecurePassword111
+
+# monitoring data db
+DNPM_DATASOURCE_URL=jdbc:mariadb://dnpm-monitor-db:3306/$DNPM_MARIADB_DB
+
+## TARGET SYSTEMS CONFIG
+# in case of direct access to bwhc enter endpoint url here
+DNPM_BWHC_REST_URI=
+
+# produce mtb files to this topic - values 'false' disabling kafka processing
+DNPM_KAFKA_TOPIC=false
+KAFKA_BROKERS=false
+DNPM_KAFKA_SECURITY_PROTOCOL=PLAINTEXT
+
+# here we receive responses from bwhc
+DNPM_KAFKA_RESPONSE_TOPIC=dnpm-response
+DNPM_KAFKA_GROUP_ID=dnpm
+
+# SSL or PLAINTEXT
+DNPM_PROCESSOR_KEY_STORE_PASSWORD=
+DNPM_TO_SSL_KEYSTORE_LOCATION=
+

dev-compose.yml

@@ -4,8 +4,33 @@ services:
     hostname: kafka
     ports:
       - "9092:9092"
+      - "9094:9094"
     environment:
       ALLOW_PLAINTEXT_LISTENER: "yes"
+      KAFKA_CFG_NODE_ID: "0"
+      KAFKA_CFG_PROCESS_ROLES: "controller,broker"
+      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_CFG_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true
+      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
+      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
+
+  akhq:
+    image: tchiotludo/akhq:0.21.0
+    environment:
+      AKHQ_CONFIGURATION: |
+        akhq:
+          connections:
+            docker-kafka-server:
+              properties:
+                bootstrap.servers: "kafka:9092"
+              connect:
+                - name: "kafka-connect"
+                  url: "http://kafka-connect:8083"
+    ports:
+      - "8084:8080"
 
   mariadb:
     image: mariadb:10
@@ -16,6 +41,7 @@ services:
       MARIADB_USER: dev
       MARIADB_PASSWORD: dev
       MARIADB_ROOT_PASSWORD: dev
+
 #  postgres:
 #    image: postgres:alpine
 #    ports:
@@ -23,4 +49,4 @@ services:
 #    environment:
 #      POSTGRES_DB: dev
 #      POSTGRES_USER: dev
-#      POSTGRES_PASSWORD: dev
+#      POSTGRES_PASSWORD: dev

docs/docker-compose.yml

@@ -0,0 +1,26 @@
+### Example for docker-compose
+version: '3.7'
+
+volumes:
+  data:
+
+services:
+
+  ### ETL-Processor
+  etl-processor:
+    image: ghcr.io/ccc-mf/etl-processor:latest
+    environment:
+      APP_REST_URI: http://bwhc-backend/bwhc/etl/api
+      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres/etl
+      SPRING_DATASOURCE_USERNAME: etl
+      SPRING_DATASOURCE_PASSWORD: etl-password
+
+  ### Database
+  postgres:
+    image: postgres:alpine
+    environment:
+      POSTGRES_DB: etl
+      POSTGRES_USER: etl
+      POSTGRES_PASSWORD: etl-password
+    volumes:
+      - data:/var/lib/postgresql/data

src/integrationTest/kotlin/dev/dnpm/etl/processor/EtlProcessorApplicationTests.kt

@@ -19,22 +19,125 @@
 
 package dev.dnpm.etl.processor
 
+import com.fasterxml.jackson.databind.ObjectMapper
+import de.ukw.ccc.bwhc.dto.*
+import dev.dnpm.etl.processor.monitoring.RequestRepository
+import dev.dnpm.etl.processor.monitoring.RequestStatus
 import dev.dnpm.etl.processor.output.MtbFileSender
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Nested
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.extension.ExtendWith
+import org.mockito.kotlin.*
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.boot.test.mock.mockito.MockBean
+import org.springframework.context.ApplicationContext
+import org.springframework.http.MediaType
+import org.springframework.test.context.TestPropertySource
 import org.springframework.test.context.junit.jupiter.SpringExtension
+import org.springframework.test.web.servlet.MockMvc
+import org.springframework.test.web.servlet.post
 import org.testcontainers.junit.jupiter.Testcontainers
 
 @Testcontainers
 @ExtendWith(SpringExtension::class)
 @SpringBootTest
 @MockBean(MtbFileSender::class)
+@TestPropertySource(
+    properties = [
+        "app.rest.uri=http://example.com"
+    ]
+)
 class EtlProcessorApplicationTests : AbstractTestcontainerTest() {
 
     @Test
-    fun contextLoadsIfMtbFileSenderConfigured() {
+    fun contextLoadsIfMtbFileSenderConfigured(@Autowired context: ApplicationContext) {
+        // Simply check bean configuration
+        assertThat(context).isNotNull
+    }
+
+    @Nested
+    @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.MOCK)
+    @AutoConfigureMockMvc
+    @TestPropertySource(
+        properties = [
+            "app.transformations[0].path=diagnoses[*].icd10.version",
+            "app.transformations[0].from=2013",
+            "app.transformations[0].to=2014",
+        ]
+    )
+    inner class TransformationTest {
+
+        @MockBean
+        private lateinit var mtbFileSender: MtbFileSender
+
+        @Autowired
+        private lateinit var mockMvc: MockMvc
+
+        @Autowired
+        private lateinit var objectMapper: ObjectMapper
+
+        @BeforeEach
+        fun setup(@Autowired requestRepository: RequestRepository) {
+            requestRepository.deleteAll()
+        }
+
+        @Test
+        fun mtbFileIsTransformed() {
+            doAnswer {
+                MtbFileSender.Response(RequestStatus.SUCCESS)
+            }.whenever(mtbFileSender).send(any<MtbFileSender.MtbFileRequest>())
+
+            val mtbFile = MtbFile.builder()
+                .withPatient(
+                    Patient.builder()
+                        .withId("TEST_12345678")
+                        .withBirthDate("2000-08-08")
+                        .withGender(Patient.Gender.MALE)
+                        .build()
+                )
+                .withConsent(
+                    Consent.builder()
+                        .withId("1")
+                        .withStatus(Consent.Status.ACTIVE)
+                        .withPatient("TEST_12345678")
+                        .build()
+                )
+                .withEpisode(
+                    Episode.builder()
+                        .withId("1")
+                        .withPatient("TEST_12345678")
+                        .withPeriod(PeriodStart("2023-08-08"))
+                        .build()
+                )
+                .withDiagnoses(
+                    listOf(
+                        Diagnosis.builder()
+                            .withId("1234")
+                            .withIcd10(Icd10.builder().withCode("F79.9").withVersion("2013").build())
+                            .build()
+                    )
+                )
+                .build()
+
+            mockMvc.post("/mtbfile") {
+                content = objectMapper.writeValueAsString(mtbFile)
+                contentType = MediaType.APPLICATION_JSON
+            }.andExpect {
+                status {
+                    isAccepted()
+                }
+            }
+
+            val captor = argumentCaptor<MtbFileSender.MtbFileRequest>()
+            verify(mtbFileSender).send(captor.capture())
+            assertThat(captor.firstValue.mtbFile.diagnoses).hasSize(1).allMatch { diagnosis ->
+                diagnosis.icd10.version == "2014"
+            }
+        }
     }
 
 }

src/integrationTest/kotlin/dev/dnpm/etl/processor/config/AppConfigurationTest.kt

@@ -31,13 +31,13 @@ import org.springframework.beans.factory.NoSuchBeanDefinitionException
 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.boot.test.mock.mockito.MockBean
-import org.springframework.boot.test.mock.mockito.MockBeans
 import org.springframework.context.ApplicationContext
 import org.springframework.test.context.ContextConfiguration
 import org.springframework.test.context.TestPropertySource
 
 @SpringBootTest
-@ContextConfiguration(classes = [KafkaAutoConfiguration::class, AppKafkaConfiguration::class, AppRestConfiguration::class])
+@ContextConfiguration(classes = [AppConfiguration::class, KafkaAutoConfiguration::class, AppKafkaConfiguration::class, AppRestConfiguration::class])
+@MockBean(ObjectMapper::class)
 class AppConfigurationTest {
 
     @Nested
@@ -65,10 +65,7 @@ class AppConfigurationTest {
             "app.kafka.group-id=test"
         ]
     )
-    @MockBeans(value = [
-        MockBean(ObjectMapper::class),
-        MockBean(RequestRepository::class)
-    ])
+    @MockBean(RequestRepository::class)
     inner class AppConfigurationKafkaTest(private val context: ApplicationContext) {
 
         @Test
@@ -99,4 +96,24 @@ class AppConfigurationTest {
 
     }
 
+    @Nested
+    @TestPropertySource(
+        properties = [
+            "app.transformations[0].path=consent.status",
+            "app.transformations[0].from=rejected",
+            "app.transformations[0].to=accept",
+        ]
+    )
+    inner class AppConfigurationTransformationTest(private val context: ApplicationContext) {
+
+        @Test
+        fun shouldRecognizeTransformations() {
+            val appConfigProperties = context.getBean(AppConfigProperties::class.java)
+
+            assertThat(appConfigProperties).isNotNull
+            assertThat(appConfigProperties.transformations).hasSize(1)
+        }
+
+    }
+
 }

src/integrationTest/kotlin/dev/dnpm/etl/processor/services/RequestServiceIntegrationTest.kt

@@ -32,6 +32,7 @@ import org.junit.jupiter.api.extension.ExtendWith
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.context.SpringBootTest
 import org.springframework.boot.test.mock.mockito.MockBean
+import org.springframework.test.context.TestPropertySource
 import org.springframework.test.context.junit.jupiter.SpringExtension
 import org.springframework.transaction.annotation.Transactional
 import org.testcontainers.junit.jupiter.Testcontainers
@@ -43,6 +44,11 @@ import java.util.*
 @SpringBootTest
 @Transactional
 @MockBean(MtbFileSender::class)
+@TestPropertySource(
+    properties = [
+        "app.rest.uri=http://example.com"
+    ]
+)
 class RequestServiceIntegrationTest : AbstractTestcontainerTest() {
 
     private lateinit var requestRepository: RequestRepository

src/main/java/dev/dnpm/etl/processor/pseudonym/GpasPseudonymGenerator.java

@@ -22,6 +22,21 @@ package dev.dnpm.etl.processor.pseudonym;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.parser.IParser;
 import dev.dnpm.etl.processor.config.GPasConfigProperties;
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.util.Base64;
+import java.util.HashMap;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
@@ -39,7 +54,11 @@ import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.http.*;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
 import org.springframework.retry.RetryCallback;
 import org.springframework.retry.RetryContext;
@@ -51,31 +70,13 @@ import org.springframework.retry.support.RetryTemplate;
 import org.springframework.web.client.RestClientException;
 import org.springframework.web.client.RestTemplate;
 
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.TrustManagerFactory;
-import java.io.BufferedInputStream;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.net.ConnectException;
-import java.security.KeyManagementException;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
-import java.security.NoSuchAlgorithmException;
-import java.security.cert.CertificateException;
-import java.security.cert.CertificateFactory;
-import java.security.cert.X509Certificate;
-import java.util.Base64;
-import java.util.HashMap;
-
 public class GpasPseudonymGenerator implements Generator {
 
+    private final static FhirContext r4Context = FhirContext.forR4();
     private final String gPasUrl;
     private final String psnTargetDomain;
-    private static FhirContext r4Context = FhirContext.forR4();
     private final HttpHeaders httpHeader;
-
     private final RetryTemplate retryTemplate = defaultTemplate();
-
     private final Logger log = LoggerFactory.getLogger(GpasPseudonymGenerator.class);
 
     private SSLContext customSslContext;
@@ -90,12 +91,16 @@ public class GpasPseudonymGenerator implements Generator {
         try {
             if (StringUtils.isNotBlank(gpasCfg.getSslCaLocation())) {
                 customSslContext = getSslContext(gpasCfg.getSslCaLocation());
+                log.debug(String.format("%s has been initialized with SSL certificate %s",
+                    this.getClass().getName(), gpasCfg.getSslCaLocation()));
             }
         } catch (IOException | KeyManagementException | KeyStoreException | CertificateException |
                  NoSuchAlgorithmException e) {
             throw new RuntimeException(e);
         }
 
+        log.debug(String.format("%s has been initialized", this.getClass().getName()));
+
     }
 
     @Override
@@ -110,12 +115,33 @@ public class GpasPseudonymGenerator implements Generator {
 
     @NotNull
     public static String unwrapPseudonym(Parameters gPasPseudonymResult) {
-        Identifier pseudonym = (Identifier) gPasPseudonymResult.getParameter().stream().findFirst()
-            .get().getPart().stream().filter(a -> a.getName().equals("pseudonym")).findFirst()
+        final var parameters = gPasPseudonymResult.getParameter().stream().findFirst();
+
+        if (parameters.isEmpty()) {
+            throw new PseudonymRequestFailed("Empty HL7 parameters, cannot find first one");
+        }
+
+        final var identifier = (Identifier) parameters.get().getPart().stream()
+            .filter(a -> a.getName().equals("pseudonym"))
+            .findFirst()
             .orElseGet(ParametersParameterComponent::new).getValue();
 
         // pseudonym
-        return pseudonym.getSystem() + "|" + pseudonym.getValue();
+        return sanitizeValue(identifier.getValue());
+    }
+
+    /**
+     * Allow only filename friendly values
+     *
+     * @param psnValue GAPS pseudonym value
+     * @return cleaned up value
+     */
+    public static String sanitizeValue(String psnValue) {
+        // pattern to match forbidden characters
+        String forbiddenCharsRegex = "[\\\\/:*?\"<>|;]";
+
+        // Replace all forbidden characters with underscores
+        return psnValue.replaceAll(forbiddenCharsRegex, "_");
     }
 
 

src/main/kotlin/dev/dnpm/etl/processor/EtlProcessorApplication.kt

@@ -20,9 +20,10 @@
 package dev.dnpm.etl.processor
 
 import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration
 import org.springframework.boot.runApplication
 
-@SpringBootApplication
+@SpringBootApplication(exclude = [SecurityAutoConfiguration::class])
 class EtlProcessorApplication
 
 fun main(args: Array<String>) {

src/main/kotlin/dev/dnpm/etl/processor/config/AppConfigProperties.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -23,8 +23,9 @@ import org.springframework.boot.context.properties.ConfigurationProperties
 
 @ConfigurationProperties(AppConfigProperties.NAME)
 data class AppConfigProperties(
-    var bwhc_uri: String?,
-    var generator: PseudonymGenerator = PseudonymGenerator.BUILDIN
+    var bwhcUri: String?,
+    var generator: PseudonymGenerator = PseudonymGenerator.BUILDIN,
+    var transformations: List<TransformationProperties> = listOf()
 ) {
     companion object {
         const val NAME = "app"
@@ -75,7 +76,23 @@ data class KafkaTargetProperties(
     }
 }
 
+@ConfigurationProperties(SecurityConfigProperties.NAME)
+data class SecurityConfigProperties(
+    val adminUser: String?,
+    val adminPassword: String?,
+) {
+    companion object {
+        const val NAME = "app.security"
+    }
+}
+
 enum class PseudonymGenerator {
     BUILDIN,
     GPAS
-}
+}
+
+data class TransformationProperties(
+    val path: String,
+    val from: String,
+    val to: String
+)

src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -25,11 +25,21 @@ import dev.dnpm.etl.processor.pseudonym.AnonymizingGenerator
 import dev.dnpm.etl.processor.pseudonym.Generator
 import dev.dnpm.etl.processor.pseudonym.GpasPseudonymGenerator
 import dev.dnpm.etl.processor.pseudonym.PseudonymizeService
+import dev.dnpm.etl.processor.services.Transformation
+import dev.dnpm.etl.processor.services.TransformationService
+import org.slf4j.LoggerFactory
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.retry.policy.SimpleRetryPolicy
+import org.springframework.retry.support.RetryTemplate
+import org.springframework.retry.support.RetryTemplateBuilder
+import org.springframework.scheduling.annotation.EnableScheduling
 import reactor.core.publisher.Sinks
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.toJavaDuration
+
 
 @Configuration
 @EnableConfigurationProperties(
@@ -39,8 +49,11 @@ import reactor.core.publisher.Sinks
         GPasConfigProperties::class
     ]
 )
+@EnableScheduling
 class AppConfiguration {
 
+    private val logger = LoggerFactory.getLogger(AppConfiguration::class.java)
+
     @ConditionalOnProperty(value = ["app.pseudonymizer"], havingValue = "GPAS")
     @Bean
     fun gpasPseudonymGenerator(configProperties: GPasConfigProperties): Generator {
@@ -71,5 +84,25 @@ class AppConfiguration {
         return Sinks.many().multicast().directBestEffort()
     }
 
+    @Bean
+    fun transformationService(
+        objectMapper: ObjectMapper,
+        configProperties: AppConfigProperties
+    ): TransformationService {
+        logger.info("Apply ${configProperties.transformations.size} transformation rules")
+        return TransformationService(objectMapper, configProperties.transformations.map {
+            Transformation.of(it.path) from it.from to it.to
+        })
+    }
+
+    @Bean
+    fun retryTemplate(): RetryTemplate {
+        return RetryTemplateBuilder()
+            .notRetryOn(IllegalArgumentException::class.java)
+            .fixedBackoff(5.seconds.toJavaDuration())
+            .customPolicy(SimpleRetryPolicy(3))
+            .build()
+    }
+
 }
 

src/main/kotlin/dev/dnpm/etl/processor/config/AppKafkaConfiguration.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -20,6 +20,8 @@
 package dev.dnpm.etl.processor.config
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import dev.dnpm.etl.processor.monitoring.ConnectionCheckService
+import dev.dnpm.etl.processor.monitoring.KafkaConnectionCheckService
 import dev.dnpm.etl.processor.output.KafkaMtbFileSender
 import dev.dnpm.etl.processor.output.MtbFileSender
 import dev.dnpm.etl.processor.services.kafka.KafkaResponseProcessor
@@ -35,6 +37,7 @@ import org.springframework.kafka.core.ConsumerFactory
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.listener.ContainerProperties
 import org.springframework.kafka.listener.KafkaMessageListenerContainer
+import org.springframework.retry.support.RetryTemplate
 
 @Configuration
 @EnableConfigurationProperties(
@@ -51,10 +54,11 @@ class AppKafkaConfiguration {
     fun kafkaMtbFileSender(
         kafkaTemplate: KafkaTemplate<String, String>,
         kafkaTargetProperties: KafkaTargetProperties,
+        retryTemplate: RetryTemplate,
         objectMapper: ObjectMapper
     ): MtbFileSender {
         logger.info("Selected 'KafkaMtbFileSender'")
-        return KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, objectMapper)
+        return KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, retryTemplate, objectMapper)
     }
 
     @Bean
@@ -76,4 +80,9 @@ class AppKafkaConfiguration {
         return KafkaResponseProcessor(applicationEventPublisher, objectMapper)
     }
 
+    @Bean
+    fun connectionCheckService(consumerFactory: ConsumerFactory<String, String>): ConnectionCheckService {
+        return KafkaConnectionCheckService(consumerFactory.createConsumer())
+    }
+
 }

src/main/kotlin/dev/dnpm/etl/processor/config/AppRestConfiguration.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -19,6 +19,8 @@
 
 package dev.dnpm.etl.processor.config
 
+import dev.dnpm.etl.processor.monitoring.ConnectionCheckService
+import dev.dnpm.etl.processor.monitoring.RestConnectionCheckService
 import dev.dnpm.etl.processor.output.MtbFileSender
 import dev.dnpm.etl.processor.output.RestMtbFileSender
 import org.slf4j.LoggerFactory
@@ -28,6 +30,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.core.annotation.Order
+import org.springframework.retry.support.RetryTemplate
 import org.springframework.web.client.RestTemplate
 
 @Configuration
@@ -49,9 +52,21 @@ class AppRestConfiguration {
     }
 
     @Bean
-    fun restMtbFileSender(restTemplate: RestTemplate, restTargetProperties: RestTargetProperties): MtbFileSender {
+    fun restMtbFileSender(
+        restTemplate: RestTemplate,
+        restTargetProperties: RestTargetProperties,
+        retryTemplate: RetryTemplate
+    ): MtbFileSender {
         logger.info("Selected 'RestMtbFileSender'")
-        return RestMtbFileSender(restTemplate, restTargetProperties)
+        return RestMtbFileSender(restTemplate, restTargetProperties, retryTemplate)
+    }
+
+    @Bean
+    fun connectionCheckService(
+        restTemplate: RestTemplate,
+        restTargetProperties: RestTargetProperties
+    ): ConnectionCheckService {
+        return RestConnectionCheckService(restTemplate, restTargetProperties)
     }
 
 }

src/main/kotlin/dev/dnpm/etl/processor/config/AppSecurityConfiguration.kt

@@ -0,0 +1,99 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.config
+
+import org.slf4j.LoggerFactory
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.boot.context.properties.EnableConfigurationProperties
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.security.config.annotation.web.builders.HttpSecurity
+import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity
+import org.springframework.security.config.annotation.web.invoke
+import org.springframework.security.core.userdetails.User
+import org.springframework.security.core.userdetails.UserDetails
+import org.springframework.security.crypto.factory.PasswordEncoderFactories
+import org.springframework.security.crypto.password.PasswordEncoder
+import org.springframework.security.provisioning.InMemoryUserDetailsManager
+import org.springframework.security.web.SecurityFilterChain
+import java.util.*
+
+
+@Configuration
+@EnableConfigurationProperties(
+    value = [
+        SecurityConfigProperties::class
+    ]
+)
+@ConditionalOnProperty(value = ["app.security.admin-user"])
+@EnableWebSecurity
+class AppSecurityConfiguration(
+    private val securityConfigProperties: SecurityConfigProperties
+) {
+
+    private val logger = LoggerFactory.getLogger(AppSecurityConfiguration::class.java)
+
+    @Bean
+    fun userDetailsService(passwordEncoder: PasswordEncoder): InMemoryUserDetailsManager {
+        val adminUser = if (securityConfigProperties.adminUser.isNullOrBlank()) {
+            logger.warn("Using random Admin User: admin")
+            "admin"
+        } else {
+            securityConfigProperties.adminUser
+        }
+
+        val adminPassword = if (securityConfigProperties.adminPassword.isNullOrBlank()) {
+            val random = UUID.randomUUID().toString()
+            logger.warn("Using random Admin Passwort: {}", random)
+            passwordEncoder.encode(random)
+        } else {
+            securityConfigProperties.adminPassword
+        }
+
+        val user: UserDetails = User.withUsername(adminUser)
+            .password(adminPassword)
+            .roles("ADMIN")
+            .build()
+
+        return InMemoryUserDetailsManager(user)
+    }
+
+    @Bean
+    fun filterChain(http: HttpSecurity): SecurityFilterChain {
+        http {
+            authorizeRequests {
+                authorize("/configs/**", hasRole("ADMIN"))
+                authorize(anyRequest, permitAll)
+            }
+            formLogin {
+                loginPage = "/login"
+            }
+            csrf { disable() }
+        }
+        return http.build()
+    }
+
+    @Bean
+    fun passwordEncoder(): PasswordEncoder {
+        return PasswordEncoderFactories.createDelegatingPasswordEncoder()
+    }
+
+}
+

src/main/kotlin/dev/dnpm/etl/processor/monitoring/ConnectionCheckService.kt

@@ -0,0 +1,85 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+
+package dev.dnpm.etl.processor.monitoring
+
+import dev.dnpm.etl.processor.config.RestTargetProperties
+import jakarta.annotation.PostConstruct
+import org.apache.kafka.clients.consumer.Consumer
+import org.apache.kafka.common.errors.TimeoutException
+import org.springframework.http.HttpStatus
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.web.client.RestTemplate
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.toJavaDuration
+
+interface ConnectionCheckService {
+
+    fun connectionAvailable(): Boolean
+
+}
+
+class KafkaConnectionCheckService(
+    private val consumer: Consumer<String, String>
+) : ConnectionCheckService {
+
+    private var connectionAvailable: Boolean = false
+
+
+    @PostConstruct
+    @Scheduled(cron = "0 * * * * *")
+    fun check() {
+        connectionAvailable = try {
+            null != consumer.listTopics(5.seconds.toJavaDuration())
+        } catch (e: TimeoutException) {
+            false
+        }
+    }
+
+    override fun connectionAvailable(): Boolean {
+        return this.connectionAvailable
+    }
+
+}
+
+class RestConnectionCheckService(
+    private val restTemplate: RestTemplate,
+    private val restTargetProperties: RestTargetProperties
+) : ConnectionCheckService {
+
+    private var connectionAvailable: Boolean = false
+
+    @PostConstruct
+    @Scheduled(cron = "0 * * * * *")
+    fun check() {
+        connectionAvailable = try {
+            restTemplate.getForEntity(
+                restTargetProperties.uri?.replace("/etl/api", "").toString(),
+                String::class.java
+            ).statusCode == HttpStatus.OK
+        } catch (e: Exception) {
+            false
+        }
+    }
+
+    override fun connectionAvailable(): Boolean {
+        return this.connectionAvailable
+    }
+}

src/main/kotlin/dev/dnpm/etl/processor/monitoring/ReportService.kt

@@ -34,7 +34,10 @@ class ReportService(
             return listOf()
         }
         return try {
-            objectMapper.readValue(dataQualityReport, DataQualityReport::class.java).issues
+            objectMapper
+                .readValue(dataQualityReport, DataQualityReport::class.java)
+                .issues
+                .sortedBy { it.severity }
         } catch (e: Exception) {
             val otherIssue =
                 Issue(Severity.ERROR, "Not parsable data quality report '$dataQualityReport'")
@@ -56,5 +59,6 @@ class ReportService(
     enum class Severity(@JsonValue val value: String) {
         ERROR("error"),
         WARNING("warning"),
+        INFO("info")
     }
 }

src/main/kotlin/dev/dnpm/etl/processor/monitoring/Request.kt

@@ -24,6 +24,7 @@ import org.springframework.data.jdbc.repository.query.Query
 import org.springframework.data.relational.core.mapping.Embedded
 import org.springframework.data.relational.core.mapping.Table
 import org.springframework.data.repository.CrudRepository
+import org.springframework.data.repository.PagingAndSortingRepository
 import java.time.Instant
 import java.util.*
 
@@ -52,7 +53,7 @@ data class CountedState(
     val status: RequestStatus,
 )
 
-interface RequestRepository : CrudRepository<Request, Long> {
+interface RequestRepository : CrudRepository<Request, Long>, PagingAndSortingRepository<Request, Long> {
 
     fun findAllByPatientIdOrderByProcessedAtDesc(patientId: String): List<Request>
 

src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -26,10 +26,12 @@ import dev.dnpm.etl.processor.config.KafkaTargetProperties
 import dev.dnpm.etl.processor.monitoring.RequestStatus
 import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.retry.support.RetryTemplate
 
 class KafkaMtbFileSender(
     private val kafkaTemplate: KafkaTemplate<String, String>,
     private val kafkaTargetProperties: KafkaTargetProperties,
+    private val retryTemplate: RetryTemplate,
     private val objectMapper: ObjectMapper
 ) : MtbFileSender {
 
@@ -37,16 +39,18 @@ class KafkaMtbFileSender(
 
     override fun send(request: MtbFileSender.MtbFileRequest): MtbFileSender.Response {
         return try {
-            val result = kafkaTemplate.send(
-                kafkaTargetProperties.topic,
-                key(request),
-                objectMapper.writeValueAsString(request.mtbFile)
-            )
-            if (result.get() != null) {
-                logger.debug("Sent file via KafkaMtbFileSender")
-                MtbFileSender.Response(RequestStatus.UNKNOWN)
-            } else {
-                MtbFileSender.Response(RequestStatus.ERROR)
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
+                val result = kafkaTemplate.send(
+                    kafkaTargetProperties.topic,
+                    key(request),
+                    objectMapper.writeValueAsString(Data(request.requestId, request.mtbFile))
+                )
+                if (result.get() != null) {
+                    logger.debug("Sent file via KafkaMtbFileSender")
+                    MtbFileSender.Response(RequestStatus.UNKNOWN)
+                } else {
+                    MtbFileSender.Response(RequestStatus.ERROR)
+                }
             }
         } catch (e: Exception) {
             logger.error("An error occurred sending to kafka", e)
@@ -65,17 +69,19 @@ class KafkaMtbFileSender(
             .build()
 
         return try {
-            val result = kafkaTemplate.send(
-                kafkaTargetProperties.topic,
-                key(request),
-                objectMapper.writeValueAsString(dummyMtbFile)
-            )
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
+                val result = kafkaTemplate.send(
+                    kafkaTargetProperties.topic,
+                    key(request),
+                    objectMapper.writeValueAsString(Data(request.requestId, dummyMtbFile))
+                )
 
-            if (result.get() != null) {
-                logger.debug("Sent deletion request via KafkaMtbFileSender")
-                MtbFileSender.Response(RequestStatus.UNKNOWN)
-            } else {
-                MtbFileSender.Response(RequestStatus.ERROR)
+                if (result.get() != null) {
+                    logger.debug("Sent deletion request via KafkaMtbFileSender")
+                    MtbFileSender.Response(RequestStatus.UNKNOWN)
+                } else {
+                    MtbFileSender.Response(RequestStatus.ERROR)
+                }
             }
         } catch (e: Exception) {
             logger.error("An error occurred sending to kafka", e)
@@ -83,14 +89,18 @@ class KafkaMtbFileSender(
         }
     }
 
+    override fun endpoint(): String {
+        return "${this.kafkaTargetProperties.servers} (${this.kafkaTargetProperties.topic}/${this.kafkaTargetProperties.responseTopic})"
+    }
+
     private fun key(request: MtbFileSender.MtbFileRequest): String {
         return "{\"pid\": \"${request.mtbFile.patient.id}\", " +
-                "\"eid\": \"${request.mtbFile.episode.id}\", " +
-                "\"requestId\": \"${request.requestId}\"}"
+                "\"eid\": \"${request.mtbFile.episode.id}\"}"
     }
 
     private fun key(request: MtbFileSender.DeleteRequest): String {
-        return "{\"pid\": \"${request.patientId}\", " +
-                "\"requestId\": \"${request.requestId}\"}"
+        return "{\"pid\": \"${request.patientId}\"}"
     }
+
+    data class Data(val requestId: String, val content: MtbFile)
 }

src/main/kotlin/dev/dnpm/etl/processor/output/MtbFileSender.kt

@@ -28,6 +28,8 @@ interface MtbFileSender {
 
     fun send(request: DeleteRequest): Response
 
+    fun endpoint(): String
+
     data class Response(val status: RequestStatus, val body: String = "")
 
     data class MtbFileRequest(val requestId: String, val mtbFile: MtbFile)

src/main/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSender.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -25,32 +25,39 @@ import org.slf4j.LoggerFactory
 import org.springframework.http.HttpEntity
 import org.springframework.http.HttpHeaders
 import org.springframework.http.MediaType
+import org.springframework.retry.support.RetryTemplate
 import org.springframework.web.client.RestClientException
 import org.springframework.web.client.RestTemplate
 
 class RestMtbFileSender(
     private val restTemplate: RestTemplate,
-    private val restTargetProperties: RestTargetProperties
+    private val restTargetProperties: RestTargetProperties,
+    private val retryTemplate: RetryTemplate
 ) : MtbFileSender {
 
     private val logger = LoggerFactory.getLogger(RestMtbFileSender::class.java)
 
     override fun send(request: MtbFileSender.MtbFileRequest): MtbFileSender.Response {
         try {
-            val headers = HttpHeaders()
-            headers.contentType = MediaType.APPLICATION_JSON
-            val entityReq = HttpEntity(request.mtbFile, headers)
-            val response = restTemplate.postForEntity(
-                "${restTargetProperties.uri}/MTBFile",
-                entityReq,
-                String::class.java
-            )
-            if (!response.statusCode.is2xxSuccessful) {
-                logger.warn("Error sending to remote system: {}", response.body)
-                return MtbFileSender.Response(response.statusCode.asRequestStatus(), "Status-Code: ${response.statusCode.value()}")
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
+                val headers = HttpHeaders()
+                headers.contentType = MediaType.APPLICATION_JSON
+                val entityReq = HttpEntity(request.mtbFile, headers)
+                val response = restTemplate.postForEntity(
+                    "${restTargetProperties.uri}/MTBFile",
+                    entityReq,
+                    String::class.java
+                )
+                if (!response.statusCode.is2xxSuccessful) {
+                    logger.warn("Error sending to remote system: {}", response.body)
+                    return@execute MtbFileSender.Response(
+                        response.statusCode.asRequestStatus(),
+                        "Status-Code: ${response.statusCode.value()}"
+                    )
+                }
+                logger.debug("Sent file via RestMtbFileSender")
+                return@execute MtbFileSender.Response(response.statusCode.asRequestStatus(), response.body.orEmpty())
             }
-            logger.debug("Sent file via RestMtbFileSender")
-            return MtbFileSender.Response(response.statusCode.asRequestStatus(), response.body.orEmpty())
         } catch (e: IllegalArgumentException) {
             logger.error("Not a valid URI to export to: '{}'", restTargetProperties.uri!!)
         } catch (e: RestClientException) {
@@ -62,16 +69,18 @@ class RestMtbFileSender(
 
     override fun send(request: MtbFileSender.DeleteRequest): MtbFileSender.Response {
         try {
-            val headers = HttpHeaders()
-            headers.contentType = MediaType.APPLICATION_JSON
-            val entityReq = HttpEntity(null, headers)
-            restTemplate.delete(
-                "${restTargetProperties.uri}/Patient/${request.patientId}",
-                entityReq,
-                String::class.java
-            )
-            logger.debug("Sent file via RestMtbFileSender")
-            return MtbFileSender.Response(RequestStatus.SUCCESS)
+            return retryTemplate.execute<MtbFileSender.Response, Exception> {
+                val headers = HttpHeaders()
+                headers.contentType = MediaType.APPLICATION_JSON
+                val entityReq = HttpEntity(null, headers)
+                restTemplate.delete(
+                    "${restTargetProperties.uri}/Patient/${request.patientId}",
+                    entityReq,
+                    String::class.java
+                )
+                logger.debug("Sent file via RestMtbFileSender")
+                return@execute MtbFileSender.Response(RequestStatus.SUCCESS)
+            }
         } catch (e: IllegalArgumentException) {
             logger.error("Not a valid URI to export to: '{}'", restTargetProperties.uri!!)
         } catch (e: RestClientException) {
@@ -81,4 +90,8 @@ class RestMtbFileSender(
         return MtbFileSender.Response(RequestStatus.ERROR, "Sonstiger Fehler bei der Übertragung")
     }
 
+    override fun endpoint(): String {
+        return this.restTargetProperties.uri.orEmpty()
+    }
+
 }

src/main/kotlin/dev/dnpm/etl/processor/pseudonym/extensions.kt

@@ -35,16 +35,19 @@ infix fun MtbFile.pseudonymizeWith(pseudonymizeService: PseudonymizeService) {
     this.familyMemberDiagnoses.forEach { it.patient = patientPseudonym }
     this.geneticCounsellingRequests.forEach { it.patient = patientPseudonym }
     this.histologyReevaluationRequests.forEach { it.patient = patientPseudonym }
-    this.histologyReports.forEach { it.patient = patientPseudonym }
+    this.histologyReports.forEach {
+        it.patient = patientPseudonym
+        it.tumorMorphology.patient = patientPseudonym
+    }
     this.lastGuidelineTherapies.forEach { it.patient = patientPseudonym }
     this.molecularPathologyFindings.forEach { it.patient = patientPseudonym }
-    this.molecularTherapies.forEach { it.history.forEach { it.patient = patientPseudonym } }
+    this.molecularTherapies.forEach { molecularTherapy -> molecularTherapy.history.forEach { it.patient = patientPseudonym } }
     this.ngsReports.forEach { it.patient = patientPseudonym }
     this.previousGuidelineTherapies.forEach { it.patient = patientPseudonym }
     this.rebiopsyRequests.forEach { it.patient = patientPseudonym }
     this.recommendations.forEach { it.patient = patientPseudonym }
     this.recommendations.forEach { it.patient = patientPseudonym }
     this.responses.forEach { it.patient = patientPseudonym }
-    this.specimens.forEach { it.patient = patientPseudonym }
+    this.studyInclusionRequests.forEach { it.patient = patientPseudonym }
     this.specimens.forEach { it.patient = patientPseudonym }
 }

src/main/kotlin/dev/dnpm/etl/processor/services/RequestProcessor.kt

@@ -30,7 +30,6 @@ import dev.dnpm.etl.processor.pseudonym.PseudonymizeService
 import dev.dnpm.etl.processor.pseudonym.pseudonymizeWith
 import org.apache.commons.codec.binary.Base32
 import org.apache.commons.codec.digest.DigestUtils
-import org.slf4j.LoggerFactory
 import org.springframework.context.ApplicationEventPublisher
 import org.springframework.stereotype.Service
 import java.time.Instant
@@ -39,21 +38,20 @@ import java.util.*
 @Service
 class RequestProcessor(
     private val pseudonymizeService: PseudonymizeService,
+    private val transformationService: TransformationService,
     private val sender: MtbFileSender,
     private val requestService: RequestService,
     private val objectMapper: ObjectMapper,
     private val applicationEventPublisher: ApplicationEventPublisher
 ) {
 
-    private val logger = LoggerFactory.getLogger(RequestProcessor::class.java)
-
     fun processMtbFile(mtbFile: MtbFile) {
         val requestId = UUID.randomUUID().toString()
         val pid = mtbFile.patient.id
 
         mtbFile pseudonymizeWith pseudonymizeService
 
-        val request = MtbFileSender.MtbFileRequest(requestId, mtbFile)
+        val request = MtbFileSender.MtbFileRequest(requestId, transformationService.transform(mtbFile))
 
         requestService.save(
             Request(

src/main/kotlin/dev/dnpm/etl/processor/services/ResponseProcessor.kt

@@ -19,7 +19,6 @@
 
 package dev.dnpm.etl.processor.services
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import dev.dnpm.etl.processor.monitoring.Report
 import dev.dnpm.etl.processor.monitoring.RequestRepository
 import dev.dnpm.etl.processor.monitoring.RequestStatus
@@ -33,8 +32,7 @@ import java.util.*
 @Service
 class ResponseProcessor(
     private val requestRepository: RequestRepository,
-    private val statisticsUpdateProducer: Sinks.Many<Any>,
-    private val objectMapper: ObjectMapper
+    private val statisticsUpdateProducer: Sinks.Many<Any>
 ) {
 
     private val logger = LoggerFactory.getLogger(ResponseProcessor::class.java)
@@ -73,7 +71,7 @@ class ResponseProcessor(
                 }
 
                 else -> {
-                    logger.error("Cannot process response: Unknown response code!")
+                    logger.error("Cannot process response: Unknown response!")
                     return@ifPresentOrElse
                 }
             }

src/main/kotlin/dev/dnpm/etl/processor/services/TransformationService.kt

@@ -0,0 +1,85 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.services
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.jayway.jsonpath.JsonPath
+import com.jayway.jsonpath.PathNotFoundException
+import de.ukw.ccc.bwhc.dto.MtbFile
+
+class TransformationService(private val objectMapper: ObjectMapper, private val transformations: List<Transformation>) {
+    fun transform(mtbFile: MtbFile): MtbFile {
+        var json = objectMapper.writeValueAsString(mtbFile)
+
+        transformations.forEach { transformation ->
+            val jsonPath = JsonPath.parse(json)
+
+            try {
+                val before = transformation.path.substringBeforeLast(".")
+                val last = transformation.path.substringAfterLast(".")
+
+                val existingValue = if (transformation.existingValue is Number) transformation.existingValue else transformation.existingValue.toString()
+                val newValue = if (transformation.newValue is Number) transformation.newValue else transformation.newValue.toString()
+
+                jsonPath.set("$.$before.[?]$last", newValue, {
+                    it.item(HashMap::class.java)[last] == existingValue
+                })
+            } catch (e: PathNotFoundException) {
+                // Ignore
+            }
+
+            json = jsonPath.jsonString()
+        }
+
+        return objectMapper.readValue(json, MtbFile::class.java)
+    }
+
+    fun getTransformations(): List<Transformation> {
+        return this.transformations
+    }
+
+}
+
+class Transformation private constructor(val path: String) {
+
+    lateinit var existingValue: Any
+        private set
+    lateinit var newValue: Any
+        private set
+
+    infix fun from(value: Any): Transformation {
+        this.existingValue = value
+        return this
+    }
+
+    infix fun to(value: Any): Transformation {
+        this.newValue = value
+        return this
+    }
+
+    companion object {
+
+        fun of(path: String): Transformation {
+            return Transformation(path)
+        }
+
+    }
+
+}

src/main/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessor.kt

@@ -41,50 +41,40 @@ class KafkaResponseProcessor(
 
     override fun onMessage(data: ConsumerRecord<String, String>) {
         try {
-            Optional.of(objectMapper.readValue(data.key(), ResponseKey::class.java))
+            Optional.of(objectMapper.readValue(data.value(), ResponseBody::class.java))
         } catch (e: Exception) {
+            logger.error("Cannot process Kafka response", e)
             Optional.empty()
-        }.ifPresentOrElse({ responseKey ->
-            val event = try {
-                val responseBody = objectMapper.readValue(data.value(), ResponseBody::class.java)
-                ResponseEvent(
-                    responseKey.requestId,
-                    Instant.ofEpochMilli(data.timestamp()),
-                    responseBody.statusCode.asRequestStatus(),
-                    when (responseBody.statusCode.asRequestStatus()) {
-                        RequestStatus.SUCCESS -> {
-                            Optional.empty()
-                        }
-
-                        RequestStatus.WARNING, RequestStatus.ERROR -> {
-                            Optional.of(objectMapper.writeValueAsString(responseBody.statusBody))
-                        }
-
-                        else -> {
-                            logger.error("Kafka response: Unknown response code!")
-                            Optional.empty()
-                        }
+        }.ifPresentOrElse({ responseBody ->
+            val event = ResponseEvent(
+                responseBody.requestId,
+                Instant.ofEpochMilli(data.timestamp()),
+                responseBody.statusCode.asRequestStatus(),
+                when (responseBody.statusCode.asRequestStatus()) {
+                    RequestStatus.SUCCESS -> {
+                        Optional.empty()
                     }
-                )
-            } catch (e: Exception) {
-                logger.error("Cannot process Kafka response", e)
-                ResponseEvent(
-                    responseKey.requestId,
-                    Instant.ofEpochMilli(data.timestamp()),
-                    RequestStatus.ERROR,
-                    Optional.of("Cannot process Kafka response")
-                )
-            }
+
+                    RequestStatus.WARNING, RequestStatus.ERROR -> {
+                        Optional.of(objectMapper.writeValueAsString(responseBody.statusBody))
+                    }
+
+                    else -> {
+                        logger.error("Kafka response: Unknown response code '{}'!", responseBody.statusCode)
+                        Optional.empty()
+                    }
+                }
+            )
             eventPublisher.publishEvent(event)
         }, {
-            logger.error("No response key in Kafka response")
+            logger.error("No requestId in Kafka response")
         })
     }
 
-    data class ResponseKey(val requestId: String)
-
     data class ResponseBody(
-        @JsonProperty("status_code") @JsonAlias("status code") val statusCode: Int,
-        @JsonProperty("status_body") val statusBody: Map<String, Any>
+        @JsonProperty("request_id") @JsonAlias("requestId") val requestId: String,
+        @JsonProperty("status_code") @JsonAlias("statusCode") val statusCode: Int,
+        @JsonProperty("status_body") @JsonAlias("statusBody") val statusBody: Map<String, Any>
     )
+
 }

src/main/kotlin/dev/dnpm/etl/processor/web/ConfigController.kt

@@ -0,0 +1,52 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.web
+
+import dev.dnpm.etl.processor.monitoring.ConnectionCheckService
+import dev.dnpm.etl.processor.output.MtbFileSender
+import dev.dnpm.etl.processor.pseudonym.Generator
+import dev.dnpm.etl.processor.services.TransformationService
+import org.springframework.stereotype.Controller
+import org.springframework.ui.Model
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.RequestMapping
+
+@Controller
+@RequestMapping(path = ["configs"])
+class ConfigController(
+    private val transformationService: TransformationService,
+    private val pseudonymGenerator: Generator,
+    private val mtbFileSender: MtbFileSender,
+    private val connectionCheckService: ConnectionCheckService
+
+) {
+
+    @GetMapping
+    fun index(model: Model): String {
+        model.addAttribute("pseudonymGenerator", pseudonymGenerator.javaClass.simpleName)
+        model.addAttribute("mtbFileSender", mtbFileSender.javaClass.simpleName)
+        model.addAttribute("mtbFileEndpoint", mtbFileSender.endpoint())
+        model.addAttribute("connectionAvailable", connectionCheckService.connectionAvailable())
+        model.addAttribute("transformations", transformationService.getTransformations())
+
+        return "configs"
+    }
+
+}

src/main/kotlin/dev/dnpm/etl/processor/web/HomeController.kt

@@ -23,6 +23,9 @@ import dev.dnpm.etl.processor.NotFoundException
 import dev.dnpm.etl.processor.monitoring.ReportService
 import dev.dnpm.etl.processor.monitoring.RequestId
 import dev.dnpm.etl.processor.monitoring.RequestRepository
+import org.springframework.data.domain.Pageable
+import org.springframework.data.domain.Sort
+import org.springframework.data.web.PageableDefault
 import org.springframework.stereotype.Controller
 import org.springframework.ui.Model
 import org.springframework.web.bind.annotation.GetMapping
@@ -37,8 +40,8 @@ class HomeController(
 ) {
 
     @GetMapping
-    fun index(model: Model): String {
-        val requests = requestRepository.findAll().sortedByDescending { it.processedAt }.take(25)
+    fun index(@PageableDefault(page = 0, size = 20, sort = ["processedAt"], direction = Sort.Direction.DESC) pageable: Pageable, model: Model): String {
+        val requests = requestRepository.findAll(pageable)
         model.addAttribute("requests", requests)
 
         return "index"

src/main/kotlin/dev/dnpm/etl/processor/web/LoginController.kt

@@ -0,0 +1,33 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.web
+
+import org.springframework.stereotype.Controller
+import org.springframework.web.bind.annotation.GetMapping
+
+@Controller
+class LoginController {
+
+    @GetMapping(path = ["/login"])
+    fun login(): String {
+        return "login"
+    }
+
+}

src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsRestController.kt

@@ -83,9 +83,9 @@ class StatisticsRestController(
             .groupBy { formatter.format(it.processedAt) }
             .map {
                 val requestList = it.value
-                    .groupBy { it.status }
-                    .map {
-                        Pair(it.key, it.value.size)
+                    .groupBy { request -> request.status }
+                    .map { request ->
+                        Pair(request.key, request.value.size)
                     }
                     .toMap()
                 Pair(

src/main/resources/application-dev.yml

@@ -4,12 +4,15 @@ spring:
       file: ./dev-compose.yml
 
 app:
-  rest:
-    uri: http://localhost:9000/bwhc/etl/api
+  #rest:
+  #  uri: http://localhost:9000/bwhc/etl/api
   kafka:
     topic: test
-    response-topic: test-response
-    servers: kafka:9092
+    response-topic: test_response
+    servers: localhost:9094
+  #security:
+  #  admin-user: admin
+  #  admin-password: "{noop}very-secret"
 
 server:
   port: 8000

src/main/resources/application.yml

@@ -4,4 +4,7 @@ spring:
     consumer:
       group-id: ${app.kafka.group-id}
   flyway:
-    locations: "classpath:db/migration/{vendor}"
+    locations: "classpath:db/migration/{vendor}"
+
+server:
+  forward-headers-strategy: framework

src/main/resources/static/icon.svg

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!-- Created with Inkscape (http://www.inkscape.org/) -->
+
+<svg
+   width="256"
+   height="256"
+   viewBox="0 0 67.733332 67.733335"
+   version="1.1"
+   id="svg5"
+   xmlns="http://www.w3.org/2000/svg"
+   xmlns:svg="http://www.w3.org/2000/svg">
+  <defs
+     id="defs2" />
+  <g
+     id="layer1">
+    <g
+       id="g26002"
+       transform="matrix(1.5,0,0,1.5,-16.933333,-1.8487648)">
+      <path
+         id="path12437"
+         transform="matrix(0.21771408,0,0,0.21771408,73.025692,24.874779)"
+         style="fill:#f59e00;fill-opacity:1"
+         d="m -110.41995,43.223174 -55.55561,-2e-6 -27.7778,-48.1125685 27.77781,-48.1125655 55.5556,3e-6 27.777803,48.1125679 z" />
+      <path
+         id="path13446"
+         transform="matrix(0.21771408,0,0,0.21771408,54.882836,14.399994)"
+         style="fill:#004d6e;fill-opacity:1"
+         d="m -110.41995,43.223174 -55.55561,-2e-6 -27.7778,-48.1125685 27.77781,-48.1125655 55.5556,3e-6 27.777803,48.1125679 z" />
+      <path
+         id="path13448"
+         transform="matrix(0.21771408,0,0,0.21771408,54.882835,35.349561)"
+         style="fill:#706f6f;fill-opacity:1"
+         d="m -110.41995,43.223174 -55.55561,-2e-6 -27.7778,-48.1125685 27.77781,-48.1125655 55.5556,3e-6 27.777803,48.1125679 z" />
+      <path
+         id="path25844"
+         transform="matrix(0.21771408,0,0,0.21771408,60.930454,24.874778)"
+         style="fill:#ffffff;fill-opacity:1"
+         d="m -110.41995,43.223174 -55.55561,-2e-6 -27.7778,-48.1125685 27.77781,-48.1125655 55.5556,3e-6 27.777803,48.1125679 z" />
+    </g>
+  </g>
+</svg>

src/main/resources/static/scripts.js

@@ -4,14 +4,14 @@ const dateFormat = new Intl.DateTimeFormat('de-DE', dateFormatOptions);
 const dateTimeFormatOptions = { year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: 'numeric', second: 'numeric' };
 const dateTimeFormat = new Intl.DateTimeFormat('de-DE', dateTimeFormatOptions);
 
-window.onload = () => {
+window.addEventListener('load', () => {
     Array.from(document.getElementsByTagName('time')).forEach((timeTag) => {
         let date = Date.parse(timeTag.getAttribute('datetime'));
         if (! isNaN(date)) {
             timeTag.innerText = dateTimeFormat.format(date);
         }
     });
-};
+});
 
 function drawPieChart(url, elemId, title, data) {
     if (data) {

src/main/resources/static/style.css

@@ -1,44 +1,104 @@
+:root {
+    --text: #333;
+    --table-border: rgba(16, 24, 40, .1);
+
+    --bg-blue: rgb(0, 74, 157);
+    --bg-blue-op: rgba(0, 74, 157, .35);
+
+    --bg-green: rgb(0, 128, 0);
+    --bg-green-op: rgba(0, 128, 0, .35);
+
+
+    --bg-yellow: rgb(255, 140, 0);
+    --bg-yellow-op: rgba(255, 140, 0, .35);
+
+
+    --bg-red: rgb(255, 0, 0);
+    --bg-red-op: rgba(255, 0, 0, .35);
+
+    --bg-gray: rgb(112, 128, 144);
+    --bg-gray-op: rgba(112, 128, 144, .35);
+}
+
+html {
+    background: linear-gradient(-5deg, var(--bg-blue-op), transparent 10em);
+    min-height: 100vh;
+    overflow-y: scroll;
+}
+
 body {
-    margin: 0;
+    margin: 0 0 5em 0;
     font-family: sans-serif;
     font-size: .8rem;
-    color: #333;
+    color: var(--text);
+
+    min-height: 100vh;
+
+    background: url(bg.jpeg) no-repeat;
+    background-size: contain;
 }
 
 nav {
     margin: 0 auto;
-    background: #d5dad5;
-    height: 3rem;
+    padding: 2em 0;
+
+    line-height: 1.5rem;
     max-width: 1140px;
+
+    border-bottom: 1px solid var(--table-border);
 }
 
-nav a {
-    color: #004a8f;
-    text-transform: uppercase;
+nav > a.nav-home {
+    float: left;
+
+    color: var(--text);
+    line-height: 1.5em;
     text-decoration: none;
-    line-height: 2rem;
-    font-weight: 700;
+
+    font-size: 1.5em;
+    font-weight: bold;
 }
 
-nav a:hover {
-    text-decoration: underline;
+nav > a.nav-home > img {
+    width: 1.5em;
+    vertical-align: middle;
 }
 
 nav > ul {
-    margin: 0 3rem;
+    margin: 0 0 0 auto;
     padding: 0;
+
+    width: max-content;
 }
 
 nav > ul > li {
-    background: #fbfbfb;
-    display: block;
-    float: left;
-    padding: 2px 1rem;
-    border-left: 1px solid #d5dad5;
+    display: inline-block;
+    padding: 0 1rem;
 }
 
-nav > ul > li:first-of-type {
-    border-left: none;
+nav > ul > li.login {
+    margin: 0 0 0 1em;
+    padding: 0 0 0 2em;
+    border-left: 1px solid var(--table-border);
+}
+
+nav li a {
+    color: var(--bg-blue);
+    text-transform: uppercase;
+    text-decoration: none;
+    font-weight: 700;
+}
+
+nav li.login a {
+    color: var(--bg-red);
+}
+
+nav li a:hover {
+    text-decoration: underline;
+}
+
+a {
+    color: var(--bg-blue);
 }
 
 .breadcrumps {
@@ -57,22 +117,30 @@ nav > ul > li:first-of-type {
     display: inline;
 }
 
-.breadcrumps ul li+li:before {
+.breadcrumps ul li + li:before {
     padding: .4rem;
     color: gray;
     content: "/\00a0";
 }
 
 .breadcrumps ul li a {
-    color: #333333;
+    color: var(--text);
     text-decoration: none;
 }
 
+.centered {
+    text-align: center;
+}
+
 main {
     margin: 0 auto;
     max-width: 1140px;
 }
 
+section {
+    margin: 3em 0;
+}
+
 form {
     margin: 1rem 0;
     padding: 1rem;
@@ -114,16 +182,91 @@ form.samplecode-input input:focus-visible {
     background: lightgreen;
 }
 
-table {
-    border-top: 1px solid lightgray;
-    border-left: 1px solid lightgray;
-    border-spacing: 0;
-    border-radius: 3px;
+.login-form {
+    width: fit-content;
+    margin: 3em auto;
+    padding: 2em 5em;
 
+    border: 1px solid var(--table-border);
+    border-radius: .5em;
+    background: white;
+}
+
+.login-form form {
+    width: 20em;
+    margin: 0 auto;
+    display: grid;
+    grid-gap: .5em;
+
+    border: none;
+    background: none;
+}
+
+.login-form form * {
+    padding: 0.5em;
+    border: 1px solid var(--table-border);
+    border-radius: 3px;
+}
+
+.login-form button {
+    margin: 1em 0;
+    background: var(--bg-blue);
+    color: white;
+    border: none;
+}
+
+.border {
+    padding: 1.5em;
+    border: 1px solid var(--table-border);
+    border-radius: .5em;
+    background: white;
+}
+
+table, .chart {
+    border: 1px solid var(--table-border);
+    padding: 1.5em;
+
+    border-spacing: 0;
+    border-radius: .5em;
+
+    background: white;
+}
+
+table {
     min-width: 100%;
     font-family: sans-serif;
 }
 
+.border > table {
+    padding: 0;
+    border: none;
+    background: transparent;
+}
+
+.page-control {
+    border-radius: .5em;
+    padding: 1em 2em;
+    text-align: center;
+
+    line-height: 1.75em;
+}
+
+.page-control a {
+    padding: 0 .25em;
+    font-size: 1.75em;
+    color: var(--bg-gray);
+    text-decoration: none;
+}
+
+.page-control a[href] {
+    color: var(--bg-blue);
+}
+
+.page-control span {
+    padding: 0 .5em;
+    vertical-align: text-bottom;
+}
+
 #samples-table.max {
     width: 100vw;
     position: fixed;
@@ -140,43 +283,97 @@ table.samples {
     display: block;
 }
 
-th {
-    background: #eee;
-}
+th, td {
+    padding: 0.4rem .2rem;
 
-td, th {
-    padding: .2rem;
-
-    border-right: 1px solid lightgray;
-    border-bottom: 1px solid lightgray;
+    line-height: 2em;
 
     text-align: left;
     white-space: nowrap;
     vertical-align: top;
 }
 
+th {
+    border-bottom: 1px solid var(--bg-gray);
+}
+
 td {
     font-family: monospace;
+    border-bottom: 1px solid var(--bg-gray-op);
 }
 
-td.bg-green, th.bg-green {
-    background: green;
-    color: white;
+tr:last-of-type > td {
+    border-bottom: none;
 }
 
-td.bg-yellow, th.bg-yellow {
-    background: darkorange;
-    color: white;
+td > small {
+    display: block;
+    text-align: center;
 }
 
-td.bg-red, th.bg-red {
-    background: red;
-    color: white;
+td.patient-id {
+    width: 32em;
+    text-overflow: ellipsis;
+    overflow: hidden;
+    display: block;
 }
 
-td.bg-gray, th.bg-gray {
-    background: slategray;
+td.bg-blue, th.bg-blue,
+td.bg-green, th.bg-green,
+td.bg-yellow, th.bg-yellow,
+td.bg-red, th.bg-red,
+td.bg-gray, th.bg-gray
+{
+    width: 8em;
+}
+
+td.bg-blue > small, th.bg-blue > small {
+    background: var(--bg-blue);
     color: white;
+    border-radius: 0.4em;
+}
+
+td.bg-green > small, th.bg-green > small {
+    background: var(--bg-green);
+    color: white;
+    border-radius: 0.4em;
+}
+
+td.bg-yellow > small, th.bg-yellow > small {
+    background: var(--bg-yellow);
+    color: white;
+    border-radius: 0.4em;
+}
+
+td.bg-red > small, th.bg-red > small {
+    background: var(--bg-red);
+    color: white;
+    border-radius: 0.4em;
+}
+
+td.bg-gray > small, th.bg-gray > small {
+    background: var(--bg-gray);
+    color: white;
+    border-radius: 0.4em;
+}
+
+.bg-path {
+    background: var(--bg-gray-op);
+}
+
+.bg-from {
+    background: var(--bg-red-op);
+}
+
+.bg-to {
+    background: var(--bg-green-op);
+}
+
+.bg-path, .bg-from, .bg-to {
+    padding: 0.25rem 0.5rem;
+    border-radius: 3px;
+
+    font-family: monospace;
 }
 
 td.bg-shaded, th.bg-shaded {
@@ -276,12 +473,6 @@ input.inline:focus-visible {
 }
 
 .chart {
-    padding: 1rem;
-    margin: .2rem;
-
-    border: 1px solid lightgray;
-    border-radius: 3px;
-
     width: calc(100% - 2.4rem - 4px);
     height: 320px;
 
@@ -290,4 +481,43 @@ input.inline:focus-visible {
 
 .chart-50pc {
     width: calc(50% - 2.4rem - 4px);
+}
+
+.connection-display {
+    display: grid;
+    grid-template-columns: 10em 16em 10em;
+    place-items: center;
+    width: fit-content;
+    margin: 1em 0;
+}
+
+.connection-display > * {
+    text-align: center;
+    margin: auto 0;
+}
+
+.connection-display .connection {
+    display: block;
+    width: 100%;
+    height: 4px;
+    background: repeating-linear-gradient(to left, white, white 2px, transparent 2px, transparent 8px, white 8px) var(--bg-red);
+}
+
+.connection-display .connection.available {
+    background: var(--bg-green);
+}
+
+.notification {
+    margin: 1em;
+    padding: .5em;
+    border-radius: 3px;
+    text-align: center;
+}
+
+.notification.success {
+    color: var(--bg-green);
+}
+
+.notification.error {
+    color: var(--bg-red);
 }

src/main/resources/templates/configs.html

@@ -0,0 +1,104 @@
+<!DOCTYPE html>
+<html lang="de" xmlns:th="http://www.thymeleaf.org">
+<head>
+    <meta charset="UTF-8">
+    <title>ETL-Prozessor</title>
+    <link rel="stylesheet" th:href="@{/style.css}" />
+</head>
+<body>
+    <div th:replace="~{fragments.html :: nav}"></div>
+    <main>
+        <h1>Konfiguration</h1>
+
+        <section>
+            <h2>🔧 Allgemeine Konfiguration</h2>
+            <table>
+                <thead>
+                <tr>
+                    <th>Name</th>
+                    <th>Wert</th>
+                </tr>
+                </thead>
+                <tbody>
+                    <tr>
+                        <td>Pseudonym erzeugt über</td>
+                        <td>[[ ${pseudonymGenerator} ]]</td>
+                    </tr>
+                    <tr>
+                        <td>MTBFile-Sender</td>
+                        <td>[[ ${mtbFileSender} ]]</td>
+                    </tr>
+                    <tr>
+                        <td th:if="${mtbFileSender.startsWith('Rest')}">REST-Endpunkt</td>
+                        <td th:if="${mtbFileSender.startsWith('Kafka')}">Kafka-Broker und Topics</td>
+                        <td>[[ ${mtbFileEndpoint} ]]</td>
+                    </tr>
+                </tbody>
+            </table>
+        </section>
+
+        <section>
+            <h2><span th:if="${connectionAvailable}">✅</span><span th:if="${not(connectionAvailable)}">⚡</span> Verbindung zum bwHC-Backend</h2>
+            <div>
+                Verbindung über <code>[[ ${mtbFileSender} ]]</code>. Die Verbindung ist aktuell
+                <strong th:if="${connectionAvailable}" style="color: green">verfügbar.</strong>
+                <strong th:if="${not(connectionAvailable)}" style="color: red">nicht verfügbar.</strong>
+            </div>
+            <div class="connection-display border">
+                <img th:src="@{/server.png}" alt="ETL-Processor" />
+                <span class="connection" th:styleappend="${connectionAvailable ? 'available' : ''}"></span>
+                <img th:if="${mtbFileSender.startsWith('Rest')}" th:src="@{/server.png}" alt="bwHC-Backend" />
+                <img th:if="${mtbFileSender.startsWith('Kafka')}" th:src="@{/kafka.png}" alt="Kafka-Broker" />
+                <span>ETL-Processor</span>
+                <span></span>
+                <span th:if="${mtbFileSender.startsWith('Rest')}">bwHC-Backend</span>
+                <span th:if="${mtbFileSender.startsWith('Kafka')}">Kafka-Broker</span>
+            </div>
+        </section>
+
+        <section>
+            <h2><span th:if="${not transformations.isEmpty()}">✅</span><span th:if="${transformations.isEmpty()}">⛔</span> Transformationen</h2>
+
+            <h3>Syntax</h3>
+            Hier einige Beispiele zum Syntax des JSON-Path
+            <ul>
+                <li style="padding: 0.6rem 0;"><span class="bg-path">diagnoses[*].icdO3T.version</span>: Ersetze die ICD-O3T-Version in allen Diagnosen, z.B. zur Version der deutschen Übersetzung</li>
+                <li style="padding: 0.6rem 0;"><span class="bg-path">patient.gender</span>: Ersetze das Geschlecht des Patienten, z.B. in das von bwHC verlangte Format</li>
+            </ul>
+
+            <h3>Konfigurierte Transformationen</h3>
+            <th:block th:if="${transformations.isEmpty()}">
+            <p>
+                Keine konfigurierten Transformationen.
+            </p>
+            </th:block>
+            <th:block th:if="${not transformations.isEmpty()}">
+            <p>
+                Hier sehen Sie eine Übersicht der konfigurierten Transformationen.
+            </p>
+
+            <table>
+                <thead>
+                <tr>
+                    <th>JSON-Path</th>
+                    <th>Transformation von &rArr; nach</th>
+                </tr>
+                </thead>
+                <tbody>
+                <tr th:each="transformation : ${transformations}">
+                    <td>
+                        <span class="bg-path" title="Ersetze Wert(e) an dieser Stelle im MTB-File">[[ ${transformation.path} ]]</span>
+                    </td>
+                    <td>
+                        <span class="bg-from" title="Ersetze immer dann, wenn dieser Wert enthalten ist">[[ ${transformation.existingValue} ]]</span>
+                        <strong>&rArr;</strong>
+                        <span class="bg-to" title="Ersetze durch diesen Wert">[[ ${transformation.newValue} ]]</span>
+                    </td>
+                </tr>
+                </tbody>
+            </table>
+            </th:block>
+        </section>
+    </main>
+</body>
+</html>

src/main/resources/templates/fragments.html

@@ -1,5 +1,5 @@
 <!DOCTYPE html>
-<html xmlns:th="http://www.thymeleaf.org">
+<html xmlns:th="http://www.thymeleaf.org" xmlns:sec="http://www.thymeleaf.org">
 <head>
     <meta charset="UTF-8">
     <link rel="stylesheet" th:href="@{/style.css}" />
@@ -7,9 +7,22 @@
 <body>
     <div th:fragment="nav">
         <nav>
+            <a class="nav-home" th:href="@{/}">
+                <img th:src="@{/icon.svg}" alt="Icon" />
+                <span>ETL-Processor</span>
+            </a>
             <ul>
                 <li><a th:href="@{/}">Übersicht</a></li>
                 <li><a th:href="@{/statistics}">Statistiken</a></li>
+                <li sec:authorize="hasRole('ADMIN')">
+                    <a th:href="@{/configs}">Konfiguration</a>
+                </li>
+                <li class="login" sec:authorize="not isAuthenticated()">
+                    <a th:href="@{/login}">Login</a>
+                </li>
+                <li class="login" sec:authorize="isAuthenticated()">
+                    <a th:href="@{/logout}">Abmelden</a>
+                </li>
             </ul>
         </nav>
     </div>

src/main/resources/templates/index.html

@@ -1,5 +1,5 @@
 <!DOCTYPE html>
-<html lang="de" xmlns:th="http://www.thymeleaf.org">
+<html lang="de" xmlns:th="http://www.thymeleaf.org" xmlns:sec="http://www.thymeleaf.org">
 <head>
     <meta charset="UTF-8">
     <title>ETL-Prozessor</title>
@@ -11,35 +11,63 @@
 
         <h1>Letzte Anfragen</h1>
 
-        <table>
-            <thead>
-                <tr>
-                    <th>Status</th>
-                    <th>Typ</th>
-                    <th>ID</th>
-                    <th>Datum</th>
-                    <th>Patienten-ID</th>
-                </tr>
-            </thead>
-            <tbody>
-                <tr th:each="request : ${requests}">
-                    <td th:if="${request.status.value.contains('success')}" class="bg-green"><small>[[ ${request.status} ]]</small></td>
-                    <td th:if="${request.status.value.contains('warning')}" class="bg-yellow"><small>[[ ${request.status} ]]</small></td>
-                    <td th:if="${request.status.value.contains('error')}" class="bg-red"><small>[[ ${request.status} ]]</small></td>
-                    <td th:if="${request.status.value == 'unknown'}" class="bg-gray"><small>[[ ${request.status} ]]</small></td>
-                    <td th:if="${request.status.value == 'duplication'}" class="bg-gray"><small>[[ ${request.status} ]]</small></td>
-                    <td th:style="${request.type.value == 'delete'} ? 'color: red;'"><small>[[ ${request.type} ]]</small></td>
-                    <td th:if="not ${request.report}">[[ ${request.uuid} ]]</td>
-                    <td th:if="${request.report}">
-                        <a th:href="@{/report/{id}(id=${request.uuid})}">[[ ${request.uuid} ]]</a>
-                    </td>
-                    <td><time th:datetime="${request.processedAt}">[[ ${request.processedAt} ]]</time></td>
-                    <td>[[ ${request.patientId} ]]</td>
-                </tr>
-            </tbody>
-        </table>
+        <div class="border">
+            <div class="page-control">
+                <a id="first-page-link" th:href="@{/(page=${0})}" title="Zum Anfang: Taste W" th:if="${not requests.isFirst()}">&larrb;</a><a th:if="${requests.isFirst()}">&larrb;</a>
+                <a id="prev-page-link" th:href="@{/(page=${requests.getNumber() - 1})}" title="Seite zurück: Taste A" th:if="${not requests.isFirst()}">&larr;</a><a th:if="${requests.isFirst()}">&larr;</a>
+                <span>Seite [[ ${requests.getNumber() + 1} ]] von [[ ${requests.getTotalPages()} ]]</span>
+                <a id="next-page-link" th:href="@{/(page=${requests.getNumber() + 1})}" title="Seite vor: Taste D" th:if="${not requests.isLast()}">&rarr;</a><a th:if="${requests.isLast()}">&rarr;</a>
+                <a id="last-page-link" th:href="@{/(page=${requests.getTotalPages() - 1})}" title="Zum Ende: Taste S" th:if="${not requests.isLast()}">&rarrb;</a><a th:if="${requests.isLast()}">&rarrb;</a>
+            </div>
+            <table class="paged">
+                <thead>
+                    <tr>
+                        <th>Status</th>
+                        <th>Typ</th>
+                        <th>ID</th>
+                        <th>Datum</th>
+                        <th>Patienten-ID</th>
+                    </tr>
+                </thead>
+                <tbody>
+                    <tr th:each="request : ${requests}">
+                        <td th:if="${request.status.value.contains('success')}" class="bg-green"><small>[[ ${request.status} ]]</small></td>
+                        <td th:if="${request.status.value.contains('warning')}" class="bg-yellow"><small>[[ ${request.status} ]]</small></td>
+                        <td th:if="${request.status.value.contains('error')}" class="bg-red"><small>[[ ${request.status} ]]</small></td>
+                        <td th:if="${request.status.value == 'unknown'}" class="bg-gray"><small>[[ ${request.status} ]]</small></td>
+                        <td th:if="${request.status.value == 'duplication'}" class="bg-gray"><small>[[ ${request.status} ]]</small></td>
+                        <td th:style="${request.type.value == 'delete'} ? 'color: red;'"><small>[[ ${request.type} ]]</small></td>
+                        <td th:if="not ${request.report}">[[ ${request.uuid} ]]</td>
+                        <td th:if="${request.report}">
+                            <a th:href="@{/report/{id}(id=${request.uuid})}">[[ ${request.uuid} ]]</a>
+                        </td>
+                        <td><time th:datetime="${request.processedAt}">[[ ${request.processedAt} ]]</time></td>
+                        <td class="patient-id" sec:authorize="authenticated">[[ ${request.patientId} ]]</td>
+                        <td class="patient-id" sec:authorize="not authenticated">***</td>
+                    </tr>
+                </tbody>
+            </table>
+        </div>
 
     </main>
     <script th:src="@{/scripts.js}"></script>
+    <script>
+        window.addEventListener('load', () => {
+            let keyBindings = {
+                'w': 'first-page-link',
+                'a': 'prev-page-link',
+                'd': 'next-page-link',
+                's': 'last-page-link'
+            };
+            window.onkeydown = (event) => {
+                for (const [key, elemId] of Object.entries(keyBindings)) {
+                    if (event.key === key && document.getElementById(elemId)) {
+                        document.getElementById(elemId).style.background = 'yellow';
+                        document.getElementById(elemId).click();
+                    }
+                }
+            };
+        });
+    </script>
 </body>
 </html>

src/main/resources/templates/login.html

@@ -0,0 +1,23 @@
+<!DOCTYPE html>
+<html lang="de" xmlns:th="http://www.thymeleaf.org">
+<head>
+    <meta charset="UTF-8">
+    <title>ETL-Prozessor</title>
+    <link rel="stylesheet" th:href="@{/style.css}" />
+</head>
+<body>
+    <div th:replace="~{fragments.html :: nav}"></div>
+    <main>
+        <div class="login-form">
+            <h2 class="centered">Anmelden</h2>
+            <div class="centered notification error" th:if="${param.error}">Anmeldung nicht erfolgreich</div>
+            <div class="centered notification success" th:if="${param.logout}">Sie haben sich abgemeldet</div>
+            <form method="post" th:action="@{/login}">
+                <input type="text" id="username" name="username" class="form-control" placeholder="Username" required="" autofocus="">
+                <input type="password" id="password" name="password" class="form-control" placeholder="Password" required="">
+                <button class="" type="submit">Anmelden</button>
+            </form>
+        </div>
+    </main>
+</body>
+</html>

src/main/resources/templates/report.html

@@ -1,5 +1,5 @@
 <!DOCTYPE html>
-<html lang="de" xmlns:th="http://www.thymeleaf.org">
+<html lang="de" xmlns:th="http://www.thymeleaf.org" xmlns:sec="http://www.thymeleaf.org">
 <head>
     <meta charset="UTF-8">
     <title>ETL-Prozessor</title>
@@ -15,6 +15,7 @@
             <thead>
                 <tr>
                     <th>Status</th>
+                    <th>Typ</th>
                     <th>ID</th>
                     <th>Datum</th>
                     <th>Patienten-ID</th>
@@ -27,24 +28,31 @@
                     <td th:if="${request.status.value == 'error'}" class="bg-red"><small>[[ ${request.status} ]]</small></td>
                     <td th:if="${request.status.value == 'unknown'}" class="bg-gray"><small>[[ ${request.status} ]]</small></td>
                     <td th:if="${request.status.value == 'duplication'}" class="bg-gray"><small>[[ ${request.status} ]]</small></td>
+                    <td th:style="${request.type.value == 'delete'} ? 'color: red;'"><small>[[ ${request.type} ]]</small></td>
                     <td>[[ ${request.uuid} ]]</td>
                     <td><time th:datetime="${request.processedAt}">[[ ${request.processedAt} ]]</time></td>
-                    <td>[[ ${request.patientId} ]]</td>
+                    <td class="patient-id" sec:authorize="authenticated">[[ ${request.patientId} ]]</td>
+                    <td class="patient-id" sec:authorize="not authenticated">***</td>
                 </tr>
             </tbody>
         </table>
 
         <h2 th:text="${request.report.description}"></h2>
 
-        <table th:if="not ${issues.isEmpty()}">
+        <p th:if="${issues.isEmpty()}">
+            Keine weiteren Angaben.
+        </p>
+
+        <table th:if="${not issues.isEmpty()}">
             <thead>
-            <tr>
-                <th>Schweregrad</th>
-                <th>Beschreibung</th>
-            </tr>
+                <tr>
+                    <th>Schweregrad</th>
+                    <th>Beschreibung</th>
+                </tr>
             </thead>
             <tbody>
             <tr th:each="issue : ${issues}">
+                <td th:if="${issue.severity.value == 'info'}" class="bg-blue"><small>[[ ${issue.severity} ]]</small></td>
                 <td th:if="${issue.severity.value == 'warning'}" class="bg-yellow"><small>[[ ${issue.severity} ]]</small></td>
                 <td th:if="${issue.severity.value == 'error'}" class="bg-red"><small>[[ ${issue.severity} ]]</small></td>
                 <td>[[ ${issue.message} ]]</td>

src/main/resources/templates/statistics.html

@@ -13,25 +13,29 @@
             Hier sehen Sie eine Übersicht über eingegangene Anfragen.
         </p>
 
-        <h2>MTB-File-Anfragen</h2>
-        <p>
-            Anfragen zur Aktualisierung von Patientendaten durch Übermittlung eines MTB-Files.
-        </p>
-        <div>
-            <div id="piechart1" class="chart chart-50pc"></div>
-            <div id="piechart2" class="chart chart-50pc"></div>
-        </div>
-        <div id="barchart" class="chart"></div>
+        <section>
+            <h2>MTB-File-Anfragen</h2>
+            <p>
+                Anfragen zur Aktualisierung von Patientendaten durch Übermittlung eines MTB-Files.
+            </p>
+            <div>
+                <div id="piechart1" class="chart chart-50pc"></div>
+                <div id="piechart2" class="chart chart-50pc"></div>
+            </div>
+            <div id="barchart" class="chart"></div>
+        </section>
 
-        <h2>Löschanfragen</h2>
-        <p>
-            Anfragen zur Löschung von Patientendaten, wenn kein Consent vorliegt.
-        </p>
-        <div>
-            <div id="piechartdel1" class="chart chart-50pc"></div>
-            <div id="piechartdel2" class="chart chart-50pc"></div>
-        </div>
-        <div id="barchartdel" class="chart"></div>
+        <section>
+            <h2>Löschanfragen</h2>
+            <p>
+                Anfragen zur Löschung von Patientendaten, wenn kein Consent vorliegt.
+            </p>
+            <div>
+                <div id="piechartdel1" class="chart chart-50pc"></div>
+                <div id="piechartdel2" class="chart chart-50pc"></div>
+            </div>
+            <div id="barchartdel" class="chart"></div>
+        </section>
 
     </main>
     <script th:src="@{/echarts.min.js}"></script>

src/test/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSenderTest.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -35,6 +35,8 @@ import org.mockito.junit.jupiter.MockitoExtension
 import org.mockito.kotlin.*
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.support.SendResult
+import org.springframework.retry.policy.SimpleRetryPolicy
+import org.springframework.retry.support.RetryTemplateBuilder
 import java.util.concurrent.CompletableFuture.completedFuture
 import java.util.concurrent.ExecutionException
 
@@ -52,10 +54,12 @@ class KafkaMtbFileSenderTest {
         @Mock kafkaTemplate: KafkaTemplate<String, String>
     ) {
         val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(1)).build()
+
         this.objectMapper = ObjectMapper()
         this.kafkaTemplate = kafkaTemplate
 
-        this.kafkaMtbFileSender = KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, objectMapper)
+        this.kafkaMtbFileSender = KafkaMtbFileSender(kafkaTemplate, kafkaTargetProperties, retryTemplate, objectMapper)
     }
 
     @ParameterizedTest
@@ -97,9 +101,9 @@ class KafkaMtbFileSenderTest {
         val captor = argumentCaptor<String>()
         verify(kafkaTemplate, times(1)).send(anyString(), captor.capture(), captor.capture())
         assertThat(captor.firstValue).isNotNull
-        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\", \"eid\": \"1\", \"requestId\": \"TestID\"}")
+        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\", \"eid\": \"1\"}")
         assertThat(captor.secondValue).isNotNull
-        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(mtbFile(Consent.Status.ACTIVE)))
+        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(kafkaRecordData("TestID", Consent.Status.ACTIVE)))
     }
 
     @Test
@@ -113,9 +117,61 @@ class KafkaMtbFileSenderTest {
         val captor = argumentCaptor<String>()
         verify(kafkaTemplate, times(1)).send(anyString(), captor.capture(), captor.capture())
         assertThat(captor.firstValue).isNotNull
-        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\", \"requestId\": \"TestID\"}")
+        assertThat(captor.firstValue).isEqualTo("{\"pid\": \"PID\"}")
         assertThat(captor.secondValue).isNotNull
-        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(mtbFile(Consent.Status.REJECTED)))
+        assertThat(captor.secondValue).isEqualTo(objectMapper.writeValueAsString(kafkaRecordData("TestID", Consent.Status.REJECTED)))
+    }
+
+    @ParameterizedTest
+    @MethodSource("requestWithResponseSource")
+    fun shouldRetryOnMtbFileKafkaSendError(testData: TestData) {
+        val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(3)).build()
+        this.kafkaMtbFileSender = KafkaMtbFileSender(this.kafkaTemplate, kafkaTargetProperties, retryTemplate, this.objectMapper)
+
+        doAnswer {
+            if (null != testData.exception) {
+                throw testData.exception
+            }
+            completedFuture(SendResult<String, String>(null, null))
+        }.whenever(kafkaTemplate).send(anyString(), anyString(), anyString())
+
+        kafkaMtbFileSender.send(MtbFileSender.MtbFileRequest("TestID", mtbFile(Consent.Status.ACTIVE)))
+
+        val expectedCount = when (testData.exception) {
+            // OK - No Retry
+            null -> times(1)
+            // Request failed - Retry max 3 times
+            else -> times(3)
+        }
+
+        verify(kafkaTemplate, expectedCount).send(anyString(), anyString(), anyString())
+    }
+
+    @ParameterizedTest
+    @MethodSource("requestWithResponseSource")
+    fun shouldRetryOnDeleteKafkaSendError(testData: TestData) {
+        val kafkaTargetProperties = KafkaTargetProperties("testtopic")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(3)).build()
+        this.kafkaMtbFileSender = KafkaMtbFileSender(this.kafkaTemplate, kafkaTargetProperties, retryTemplate, this.objectMapper)
+
+        doAnswer {
+            if (null != testData.exception) {
+                throw testData.exception
+            }
+            completedFuture(SendResult<String, String>(null, null))
+        }.whenever(kafkaTemplate).send(anyString(), anyString(), anyString())
+
+        kafkaMtbFileSender.send(MtbFileSender.DeleteRequest("TestID", "PID"))
+
+        val expectedCount = when (testData.exception) {
+            // OK - No Retry
+            null -> times(1)
+            // Request failed - Retry max 3 times
+            else -> times(3)
+        }
+
+        verify(kafkaTemplate, expectedCount).send(anyString(), anyString(), anyString())
     }
 
     companion object {
@@ -154,6 +210,10 @@ class KafkaMtbFileSenderTest {
             }.build()
         }
 
+        fun kafkaRecordData(requestId: String, consentStatus: Consent.Status): KafkaMtbFileSender.Data {
+            return KafkaMtbFileSender.Data(requestId, mtbFile(consentStatus))
+        }
+
         data class TestData(val requestStatus: RequestStatus, val exception: Throwable? = null)
 
         @JvmStatic

src/test/kotlin/dev/dnpm/etl/processor/output/RestMtbFileSenderTest.kt

@@ -1,7 +1,7 @@
 /*
  * This file is part of ETL-Processor
  *
- * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ * Copyright (c) 2024  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published
@@ -28,6 +28,9 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.MethodSource
 import org.springframework.http.HttpMethod
 import org.springframework.http.HttpStatus
+import org.springframework.retry.policy.SimpleRetryPolicy
+import org.springframework.retry.support.RetryTemplateBuilder
+import org.springframework.test.web.client.ExpectedCount
 import org.springframework.test.web.client.MockRestServiceServer
 import org.springframework.test.web.client.match.MockRestRequestMatchers.method
 import org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo
@@ -44,10 +47,11 @@ class RestMtbFileSenderTest {
     fun setup() {
         val restTemplate = RestTemplate()
         val restTargetProperties = RestTargetProperties("http://localhost:9000/mtbfile")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(1)).build()
 
         this.mockRestServiceServer = MockRestServiceServer.createServer(restTemplate)
 
-        this.restMtbFileSender = RestMtbFileSender(restTemplate, restTargetProperties)
+        this.restMtbFileSender = RestMtbFileSender(restTemplate, restTargetProperties, retryTemplate)
     }
 
     @ParameterizedTest
@@ -80,6 +84,64 @@ class RestMtbFileSenderTest {
         assertThat(response.body).isEqualTo(requestWithResponse.response.body)
     }
 
+    @ParameterizedTest
+    @MethodSource("mtbFileRequestWithResponseSource")
+    fun shouldRetryOnMtbFileHttpRequestError(requestWithResponse: RequestWithResponse) {
+        val restTemplate = RestTemplate()
+        val restTargetProperties = RestTargetProperties("http://localhost:9000/mtbfile")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(3)).build()
+
+        this.mockRestServiceServer = MockRestServiceServer.createServer(restTemplate)
+        this.restMtbFileSender = RestMtbFileSender(restTemplate, restTargetProperties, retryTemplate)
+
+        val expectedCount = when (requestWithResponse.httpStatus) {
+            // OK - No Retry
+            HttpStatus.OK, HttpStatus.CREATED -> ExpectedCount.max(1)
+            // Request failed - Retry max 3 times
+            else -> ExpectedCount.max(3)
+        }
+
+        this.mockRestServiceServer.expect(expectedCount) {
+            method(HttpMethod.POST)
+            requestTo("/mtbfile")
+        }.andRespond {
+            withStatus(requestWithResponse.httpStatus).body(requestWithResponse.body).createResponse(it)
+        }
+
+        val response = restMtbFileSender.send(MtbFileSender.MtbFileRequest("TestID", mtbFile))
+        assertThat(response.status).isEqualTo(requestWithResponse.response.status)
+        assertThat(response.body).isEqualTo(requestWithResponse.response.body)
+    }
+
+    @ParameterizedTest
+    @MethodSource("deleteRequestWithResponseSource")
+    fun shouldRetryOnDeleteHttpRequestError(requestWithResponse: RequestWithResponse) {
+        val restTemplate = RestTemplate()
+        val restTargetProperties = RestTargetProperties("http://localhost:9000/mtbfile")
+        val retryTemplate = RetryTemplateBuilder().customPolicy(SimpleRetryPolicy(3)).build()
+
+        this.mockRestServiceServer = MockRestServiceServer.createServer(restTemplate)
+        this.restMtbFileSender = RestMtbFileSender(restTemplate, restTargetProperties, retryTemplate)
+
+        val expectedCount = when (requestWithResponse.httpStatus) {
+            // OK - No Retry
+            HttpStatus.OK, HttpStatus.CREATED -> ExpectedCount.max(1)
+            // Request failed - Retry max 3 times
+            else -> ExpectedCount.max(3)
+        }
+
+        this.mockRestServiceServer.expect(expectedCount) {
+            method(HttpMethod.DELETE)
+            requestTo("/mtbfile")
+        }.andRespond {
+            withStatus(requestWithResponse.httpStatus).body(requestWithResponse.body).createResponse(it)
+        }
+
+        val response = restMtbFileSender.send(MtbFileSender.DeleteRequest("TestID", "PID"))
+        assertThat(response.status).isEqualTo(requestWithResponse.response.status)
+        assertThat(response.body).isEqualTo(requestWithResponse.response.body)
+    }
+
     companion object {
         data class RequestWithResponse(
             val httpStatus: HttpStatus,
@@ -105,7 +167,7 @@ class RestMtbFileSenderTest {
                 }
             """.trimIndent()
 
-        val mtbFile = MtbFile.builder()
+        val mtbFile: MtbFile = MtbFile.builder()
             .withPatient(
                 Patient.builder()
                     .withId("PID")
@@ -129,7 +191,7 @@ class RestMtbFileSenderTest {
             )
             .build()
 
-        private val errorResponseBody = "Sonstiger Fehler bei der Übertragung"
+        private const val ERROR_RESPONSE_BODY = "Sonstiger Fehler bei der Übertragung"
 
         /**
          * Synthetic http responses with related request status
@@ -147,23 +209,23 @@ class RestMtbFileSenderTest {
                 RequestWithResponse(
                     HttpStatus.BAD_REQUEST,
                     "??",
-                    MtbFileSender.Response(RequestStatus.ERROR, errorResponseBody)
+                    MtbFileSender.Response(RequestStatus.ERROR, ERROR_RESPONSE_BODY)
                 ),
                 RequestWithResponse(
                     HttpStatus.UNPROCESSABLE_ENTITY,
                     errorBody,
-                    MtbFileSender.Response(RequestStatus.ERROR, errorResponseBody)
+                    MtbFileSender.Response(RequestStatus.ERROR, ERROR_RESPONSE_BODY)
                 ),
                 // Some more errors not mentioned in documentation
                 RequestWithResponse(
                     HttpStatus.NOT_FOUND,
                     "what????",
-                    MtbFileSender.Response(RequestStatus.ERROR, errorResponseBody)
+                    MtbFileSender.Response(RequestStatus.ERROR, ERROR_RESPONSE_BODY)
                 ),
                 RequestWithResponse(
                     HttpStatus.INTERNAL_SERVER_ERROR,
                     "what????",
-                    MtbFileSender.Response(RequestStatus.ERROR, errorResponseBody)
+                    MtbFileSender.Response(RequestStatus.ERROR, ERROR_RESPONSE_BODY)
                 )
             )
         }
@@ -180,12 +242,12 @@ class RestMtbFileSenderTest {
                 RequestWithResponse(
                     HttpStatus.NOT_FOUND,
                     "what????",
-                    MtbFileSender.Response(RequestStatus.ERROR, errorResponseBody)
+                    MtbFileSender.Response(RequestStatus.ERROR, ERROR_RESPONSE_BODY)
                 ),
                 RequestWithResponse(
                     HttpStatus.INTERNAL_SERVER_ERROR,
                     "what????",
-                    MtbFileSender.Response(RequestStatus.ERROR, errorResponseBody)
+                    MtbFileSender.Response(RequestStatus.ERROR, ERROR_RESPONSE_BODY)
                 )
             )
         }

src/test/kotlin/dev/dnpm/etl/processor/pseudonym/ExtensionsTest.kt

@@ -0,0 +1,64 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.pseudonym
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import de.ukw.ccc.bwhc.dto.MtbFile
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.mockito.ArgumentMatchers
+import org.mockito.Mock
+import org.mockito.junit.jupiter.MockitoExtension
+import org.mockito.kotlin.doAnswer
+import org.mockito.kotlin.whenever
+import org.springframework.core.io.ClassPathResource
+
+const val FAKE_MTB_FILE_PATH = "fake_MTBFile.json"
+const val CLEAN_PATIENT_ID = "5dad2f0b-49c6-47d8-a952-7b9e9e0f7549"
+
+@ExtendWith(MockitoExtension::class)
+class ExtensionsTest {
+
+    private fun fakeMtbFile(): MtbFile {
+        val mtbFile = ClassPathResource(FAKE_MTB_FILE_PATH).inputStream
+        return ObjectMapper().readValue(mtbFile, MtbFile::class.java)
+    }
+
+    private fun MtbFile.serialized(): String {
+        return ObjectMapper().writeValueAsString(this)
+    }
+
+    @Test
+    fun shouldNotContainCleanPatientId(@Mock pseudonymizeService: PseudonymizeService) {
+        doAnswer {
+            it.arguments[0]
+            "PSEUDO-ID"
+        }.whenever(pseudonymizeService).patientPseudonym(ArgumentMatchers.anyString())
+
+        val mtbFile = fakeMtbFile()
+
+        mtbFile.pseudonymizeWith(pseudonymizeService)
+
+        assertThat(mtbFile.patient.id).isEqualTo("PSEUDO-ID")
+        assertThat(mtbFile.serialized()).doesNotContain(CLEAN_PATIENT_ID)
+    }
+
+}

src/test/kotlin/dev/dnpm/etl/processor/pseudonym/PseudonymizeServiceTest.kt

@@ -70,6 +70,13 @@ class PseudonymizeServiceTest {
         assertThat(mtbFile.patient.id).isEqualTo("123")
     }
 
+    @Test
+    fun sanitizeFileName(@Mock generator: GpasPseudonymGenerator) {
+        val result= GpasPseudonymGenerator.sanitizeValue("l://a\\bs;1*2?3>")
+
+        assertThat(result).isEqualTo("l___a_bs_1_2_3_")
+    }
+
     @Test
     fun shouldUsePseudonymPrefixForBuiltin(@Mock generator: AnonymizingGenerator) {
         doAnswer {

src/test/kotlin/dev/dnpm/etl/processor/services/ReportServiceTest.kt

@@ -41,6 +41,7 @@ class ReportServiceTest {
             {
                 "patient": "4711",
                 "issues": [
+                    { "severity": "info", "message": "Info Message" },
                     { "severity": "warning", "message": "Warning Message" },
                     { "severity": "error", "message": "Error Message" }
                 ]
@@ -49,11 +50,13 @@ class ReportServiceTest {
 
         val actual = this.reportService.deserialize(json)
 
-        assertThat(actual).hasSize(2)
-        assertThat(actual[0].severity).isEqualTo(ReportService.Severity.WARNING)
-        assertThat(actual[0].message).isEqualTo("Warning Message")
-        assertThat(actual[1].severity).isEqualTo(ReportService.Severity.ERROR)
-        assertThat(actual[1].message).isEqualTo("Error Message")
+        assertThat(actual).hasSize(3)
+        assertThat(actual[0].severity).isEqualTo(ReportService.Severity.ERROR)
+        assertThat(actual[0].message).isEqualTo("Error Message")
+        assertThat(actual[1].severity).isEqualTo(ReportService.Severity.WARNING)
+        assertThat(actual[1].message).isEqualTo("Warning Message")
+        assertThat(actual[2].severity).isEqualTo(ReportService.Severity.INFO)
+        assertThat(actual[2].message).isEqualTo("Info Message")
     }
 
     @Test

src/test/kotlin/dev/dnpm/etl/processor/services/RequestProcessorTest.kt

@@ -37,6 +37,7 @@ import org.mockito.Mockito.*
 import org.mockito.junit.jupiter.MockitoExtension
 import org.mockito.kotlin.any
 import org.mockito.kotlin.argumentCaptor
+import org.mockito.kotlin.whenever
 import org.springframework.context.ApplicationEventPublisher
 import java.time.Instant
 import java.util.*
@@ -46,6 +47,7 @@ import java.util.*
 class RequestProcessorTest {
 
     private lateinit var pseudonymizeService: PseudonymizeService
+    private lateinit var transformationService: TransformationService
     private lateinit var sender: MtbFileSender
     private lateinit var requestService: RequestService
     private lateinit var applicationEventPublisher: ApplicationEventPublisher
@@ -55,11 +57,13 @@ class RequestProcessorTest {
     @BeforeEach
     fun setup(
         @Mock pseudonymizeService: PseudonymizeService,
+        @Mock transformationService: TransformationService,
         @Mock sender: RestMtbFileSender,
         @Mock requestService: RequestService,
         @Mock applicationEventPublisher: ApplicationEventPublisher
     ) {
         this.pseudonymizeService = pseudonymizeService
+        this.transformationService = transformationService
         this.sender = sender
         this.requestService = requestService
         this.applicationEventPublisher = applicationEventPublisher
@@ -68,6 +72,7 @@ class RequestProcessorTest {
 
         requestProcessor = RequestProcessor(
             pseudonymizeService,
+            transformationService,
             sender,
             requestService,
             objectMapper,
@@ -98,6 +103,10 @@ class RequestProcessorTest {
             it.arguments[0] as String
         }.`when`(pseudonymizeService).patientPseudonym(any())
 
+        doAnswer {
+            it.arguments[0]
+        }.whenever(transformationService).transform(any())
+
         val mtbFile = MtbFile.builder()
             .withPatient(
                 Patient.builder()
@@ -153,6 +162,10 @@ class RequestProcessorTest {
             it.arguments[0] as String
         }.`when`(pseudonymizeService).patientPseudonym(any())
 
+        doAnswer {
+            it.arguments[0]
+        }.whenever(transformationService).transform(any())
+
         val mtbFile = MtbFile.builder()
             .withPatient(
                 Patient.builder()
@@ -212,6 +225,10 @@ class RequestProcessorTest {
             it.arguments[0] as String
         }.`when`(pseudonymizeService).patientPseudonym(any())
 
+        doAnswer {
+            it.arguments[0]
+        }.whenever(transformationService).transform(any())
+
         val mtbFile = MtbFile.builder()
             .withPatient(
                 Patient.builder()
@@ -271,6 +288,10 @@ class RequestProcessorTest {
             it.arguments[0] as String
         }.`when`(pseudonymizeService).patientPseudonym(any())
 
+        doAnswer {
+            it.arguments[0]
+        }.whenever(transformationService).transform(any())
+
         val mtbFile = MtbFile.builder()
             .withPatient(
                 Patient.builder()

src/test/kotlin/dev/dnpm/etl/processor/services/ResponseProcessorTest.kt

@@ -19,8 +19,6 @@
 
 package dev.dnpm.etl.processor.services
 
-import com.fasterxml.jackson.databind.ObjectMapper
-import com.fasterxml.jackson.module.kotlin.KotlinModule
 import dev.dnpm.etl.processor.monitoring.Request
 import dev.dnpm.etl.processor.monitoring.RequestRepository
 import dev.dnpm.etl.processor.monitoring.RequestStatus
@@ -62,12 +60,10 @@ class ResponseProcessorTest {
         @Mock requestRepository: RequestRepository,
         @Mock statisticsUpdateProducer: Sinks.Many<Any>
     ) {
-        val objectMapper = ObjectMapper().registerModule(KotlinModule.Builder().build())
-
         this.requestRepository = requestRepository
         this.statisticsUpdateProducer = statisticsUpdateProducer
 
-        this.responseProcessor = ResponseProcessor(requestRepository, statisticsUpdateProducer, objectMapper)
+        this.responseProcessor = ResponseProcessor(requestRepository, statisticsUpdateProducer)
     }
 
     @Test

src/test/kotlin/dev/dnpm/etl/processor/services/TransformationServiceTest.kt

@@ -0,0 +1,95 @@
+/*
+ * This file is part of ETL-Processor
+ *
+ * Copyright (c) 2023  Comprehensive Cancer Center Mainfranken, Datenintegrationszentrum Philipps-Universität Marburg and Contributors
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package dev.dnpm.etl.processor.services
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import de.ukw.ccc.bwhc.dto.Consent
+import de.ukw.ccc.bwhc.dto.Diagnosis
+import de.ukw.ccc.bwhc.dto.Icd10
+import de.ukw.ccc.bwhc.dto.MtbFile
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+
+class TransformationServiceTest {
+
+    private lateinit var service: TransformationService
+
+    @BeforeEach
+    fun setup() {
+        this.service = TransformationService(
+            ObjectMapper(), listOf(
+                Transformation.of("consent.status") from Consent.Status.ACTIVE to Consent.Status.REJECTED,
+                Transformation.of("diagnoses[*].icd10.version") from "2013" to "2014",
+            )
+        )
+    }
+
+    @Test
+    fun shouldTransformMtbFile() {
+        val mtbFile = MtbFile.builder().withDiagnoses(
+            listOf(
+                Diagnosis.builder().withId("1234").withIcd10(Icd10("F79.9").also {
+                    it.version = "2013"
+                }).build()
+            )
+        ).build()
+
+        val actual = this.service.transform(mtbFile)
+
+        assertThat(actual).isNotNull
+        assertThat(actual.diagnoses[0].icd10.version).isEqualTo("2014")
+    }
+
+    @Test
+    fun shouldOnlyTransformGivenValues() {
+        val mtbFile = MtbFile.builder().withDiagnoses(
+            listOf(
+                Diagnosis.builder().withId("1234").withIcd10(Icd10("F79.9").also {
+                    it.version = "2013"
+                }).build(),
+                Diagnosis.builder().withId("5678").withIcd10(Icd10("F79.8").also {
+                    it.version = "2019"
+                }).build()
+            )
+        ).build()
+
+        val actual = this.service.transform(mtbFile)
+
+        assertThat(actual).isNotNull
+        assertThat(actual.diagnoses[0].icd10.code).isEqualTo("F79.9")
+        assertThat(actual.diagnoses[0].icd10.version).isEqualTo("2014")
+        assertThat(actual.diagnoses[1].icd10.code).isEqualTo("F79.8")
+        assertThat(actual.diagnoses[1].icd10.version).isEqualTo("2019")
+    }
+
+    @Test
+    fun shouldTransformMtbFileWithConsentEnum() {
+        val mtbFile = MtbFile.builder().withConsent(
+            Consent("123", "456", Consent.Status.ACTIVE)
+        ).build()
+
+        val actual = this.service.transform(mtbFile)
+
+        assertThat(actual.consent).isNotNull
+        assertThat(actual.consent.status).isEqualTo(Consent.Status.REJECTED)
+    }
+
+}

src/test/kotlin/dev/dnpm/etl/processor/services/kafka/KafkaResponseProcessorTest.kt

@@ -45,8 +45,8 @@ class KafkaResponseProcessorTest {
 
     private lateinit var kafkaResponseProcessor: KafkaResponseProcessor
 
-    private fun createkafkaRecord(
-        requestId: String? = null,
+    private fun createKafkaRecord(
+        requestId: String,
         statusCode: Int = 200,
         statusBody: Map<String, Any>? = mapOf()
     ): ConsumerRecord<String, String> {
@@ -54,15 +54,11 @@ class KafkaResponseProcessorTest {
             "test-topic",
             0,
             0,
-            if (requestId == null) {
-                null
-            } else {
-                this.objectMapper.writeValueAsString(KafkaResponseProcessor.ResponseKey(requestId))
-            },
+            null,
             if (statusBody == null) {
                 ""
             } else {
-                this.objectMapper.writeValueAsString(KafkaResponseProcessor.ResponseBody(statusCode, statusBody))
+                this.objectMapper.writeValueAsString(KafkaResponseProcessor.ResponseBody(requestId, statusCode, statusBody))
             }
         )
     }
@@ -78,23 +74,57 @@ class KafkaResponseProcessorTest {
     }
 
     @Test
-    fun shouldNotProcessRecordsWithoutValidKey() {
-        this.kafkaResponseProcessor.onMessage(createkafkaRecord(null, 200))
+    fun shouldNotProcessRecordsWithoutRequestIdInBody() {
+        val record = ConsumerRecord<String, String>(
+            "test-topic",
+            0,
+            0,
+            null,
+            """
+                {
+                    "statusCode": 200,
+                    "statusBody": {}
+                }
+            """.trimIndent()
+        )
 
-        verify(eventPublisher, never()).publishEvent(any())
+        this.kafkaResponseProcessor.onMessage(record)
+
+        verify(eventPublisher, never()).publishEvent(any<ResponseEvent>())
     }
 
     @Test
-    fun shouldNotProcessRecordsWithoutValidBody() {
-        this.kafkaResponseProcessor.onMessage(createkafkaRecord(requestId = "TestID1234", statusBody = null))
+    fun shouldProcessRecordsWithAliasNames() {
+        val record = ConsumerRecord<String, String>(
+            "test-topic",
+            0,
+            0,
+            null,
+            """
+                {
+                    "request_id": "test0123456789",
+                    "status_code": 200,
+                    "status_body": {}
+                }
+            """.trimIndent()
+        )
 
-        verify(eventPublisher, never()).publishEvent(any())
+        this.kafkaResponseProcessor.onMessage(record)
+
+        verify(eventPublisher, times(1)).publishEvent(any<ResponseEvent>())
+    }
+
+    @Test
+    fun shouldNotProcessRecordsWithoutValidStatusBody() {
+        this.kafkaResponseProcessor.onMessage(createKafkaRecord(requestId = "TestID1234", statusBody = null))
+
+        verify(eventPublisher, never()).publishEvent(any<ResponseEvent>())
     }
 
     @ParameterizedTest
     @MethodSource("statusCodeSource")
     fun shouldProcessValidRecordsWithStatusCode(statusCode: Int) {
-        this.kafkaResponseProcessor.onMessage(createkafkaRecord("TestID1234", statusCode))
+        this.kafkaResponseProcessor.onMessage(createKafkaRecord("TestID1234", statusCode))
         verify(eventPublisher, times(1)).publishEvent(any<ResponseEvent>())
     }