Logging House Messages Store

logging-house-client/build.gradle.kts

@@ -14,6 +14,8 @@ val jsonVersion: String by project
 dependencies {
     implementation("${edcGroup}:control-plane-core:${edcVersion}")
     implementation("${edcGroup}:http-spi:${edcVersion}")
+    implementation("${edcGroup}:sql-core:${edcVersion}")
+    implementation("${edcGroup}:transaction-datasource-spi:${edcVersion}")
 
     implementation("com.squareup.okhttp3:okhttp:${okHttpVersion}")
     implementation("org.json:json:${jsonVersion}")
@@ -22,6 +24,9 @@ dependencies {
     implementation("de.fraunhofer.iais.eis.ids.infomodel:infomodel-java:1.0.2-basecamp")
     implementation("de.fraunhofer.iais.eis.ids.infomodel:infomodel-util:1.0.2-basecamp")
 
+    implementation("org.postgresql:postgresql:42.4.5")
+    implementation("org.flywaydb:flyway-core:9.0.1")
+
     testImplementation("org.assertj:assertj-core:${assertj}")
     testImplementation("org.junit.jupiter:junit-jupiter-api:${jupiterVersion}")
     testImplementation("org.mockito:mockito-core:${mockitoVersion}")

logging-house-client/src/main/java/com/truzzt/extension/logginghouse/client/LoggingHouseClientExtension.java

@@ -8,13 +8,22 @@
  *  SPDX-License-Identifier: Apache-2.0
  *
  *  Contributors:
+ *       truzzt GmbH - Initial implementation
  *
  */
 
 package com.truzzt.extension.logginghouse.client;
 
+import com.truzzt.extension.logginghouse.client.flyway.FlywayService;
+import com.truzzt.extension.logginghouse.client.flyway.migration.DatabaseMigrationManager;
 import com.truzzt.extension.logginghouse.client.ids.jsonld.JsonLd;
 import com.truzzt.extension.logginghouse.client.ids.multipart.IdsMultipartSender;
+import com.truzzt.extension.logginghouse.client.messages.CreateProcessMessageSender;
+import com.truzzt.extension.logginghouse.client.messages.LogMessageSender;
+import com.truzzt.extension.logginghouse.client.multipart.IdsMultipartClearingRemoteMessageDispatcher;
+import com.truzzt.extension.logginghouse.client.multipart.MultiContextJsonLdSerializer;
+import com.truzzt.extension.logginghouse.client.store.sql.SqlLoggingHouseMessageStore;
+import com.truzzt.extension.logginghouse.client.store.sql.schema.postgres.PostgresDialectStatements;
 import de.fraunhofer.iais.eis.LogMessage;
 import de.fraunhofer.iais.eis.RequestMessage;
 import org.eclipse.edc.connector.contract.spi.event.contractnegotiation.ContractNegotiationAccepted;
@@ -42,14 +51,16 @@
 import org.eclipse.edc.spi.system.ServiceExtension;
 import org.eclipse.edc.spi.system.ServiceExtensionContext;
 import org.eclipse.edc.spi.types.TypeManager;
+import org.eclipse.edc.sql.QueryExecutor;
+import org.eclipse.edc.transaction.datasource.spi.DataSourceRegistry;
+import org.eclipse.edc.transaction.spi.TransactionContext;
 
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.Map;
 
 public class LoggingHouseClientExtension implements ServiceExtension {
 
-
     public static final String LOGGINGHOUSE_CLIENT_EXTENSION = "LoggingHouseClientExtension";
     private static final String TYPE_MANAGER_SERIALIZER_KEY = "ids-clearinghouse";
 
@@ -65,6 +76,13 @@ public class LoggingHouseClientExtension implements ServiceExtension {
     @Inject
     private Hostname hostname;
 
+    @Inject
+    private DataSourceRegistry dataSourceRegistry;
+    @Inject
+    private TransactionContext transactionContext;
+    @Inject
+    private QueryExecutor queryExecutor;
+
     @Inject
     private ContractNegotiationStore contractNegotiationStore;
     @Inject
@@ -82,11 +100,19 @@ public class LoggingHouseClientExtension implements ServiceExtension {
     @Setting
     public static final String LOGGINGHOUSE_CLIENT_EXTENSION_ENABLED = "edc.logginghouse.extension.enabled";
 
+    @Setting
+    private static final String EDC_DATASOURCE_REPAIR_SETTING = "edc.flyway.repair";
+
+    @Setting
+    private static final String FLYWAY_CLEAN_ENABLED = "edc.flyway.clean.enable";
+
+    @Setting
+    private static final String FLYWAY_CLEAN = "edc.flyway.clean";
+
     private URL loggingHouseLogUrl;
     public Monitor monitor;
     private boolean enabled;
 
-
     @Override
     public String name() {
         return LOGGINGHOUSE_CLIENT_EXTENSION;
@@ -107,10 +133,14 @@ public void initialize(ServiceExtensionContext context) {
 
         this.loggingHouseLogUrl = readUrlFromSettings(context);
 
+        runFlywayMigrations(context);
+
         registerSerializerClearingHouseMessages(context);
         registerClearingHouseMessageSenders(context);
 
-        registerEventSubscriber(context);
+        var loggingHouseMessageStore = initializeLoggingHouseMessageStore(typeManager);
+
+        registerEventSubscriber(context, loggingHouseMessageStore);
     }
 
     private URL readUrlFromSettings(ServiceExtensionContext context) {
@@ -129,13 +159,33 @@ private URL readUrlFromSettings(ServiceExtensionContext context) {
         }
     }
 
-    private void registerEventSubscriber(ServiceExtensionContext context) {
+    private void runFlywayMigrations(ServiceExtensionContext context) {
+        var flywayService = new FlywayService(
+                context.getMonitor(),
+                context.getSetting(EDC_DATASOURCE_REPAIR_SETTING, false),
+                context.getSetting(FLYWAY_CLEAN_ENABLED, false),
+                context.getSetting(FLYWAY_CLEAN, false)
+        );
+        var migrationManager = new DatabaseMigrationManager(context.getConfig(), context.getMonitor(), flywayService);
+        migrationManager.migrate();
+    }
+
+    private SqlLoggingHouseMessageStore initializeLoggingHouseMessageStore(TypeManager typeManager) {
+        return new SqlLoggingHouseMessageStore(
+                dataSourceRegistry,
+                DataSourceRegistry.DEFAULT_DATASOURCE,
+                transactionContext,
+                typeManager.getMapper(),
+                new PostgresDialectStatements(),
+                queryExecutor
+        );
+    }
+
+    private void registerEventSubscriber(ServiceExtensionContext context, SqlLoggingHouseMessageStore loggingHouseMessageStore) {
         monitor.debug("Registering event subscriber for LoggingHouseClientExtension");
 
-        var eventSubscriber = new IdsClearingHouseServiceImpl(
-                dispatcherRegistry,
-                hostname,
-                loggingHouseLogUrl,
+        var eventSubscriber = new LoggingHouseEventSubscriber(
+                loggingHouseMessageStore,
                 contractNegotiationStore,
                 transferProcessStore,
                 monitor);
@@ -150,7 +200,7 @@ private void registerEventSubscriber(ServiceExtensionContext context) {
         eventRouter.registerSync(TransferProcessCompleted.class, eventSubscriber);
         eventRouter.registerSync(TransferProcessFailed.class, eventSubscriber);
         eventRouter.registerSync(TransferProcessTerminated.class, eventSubscriber);
-        context.registerService(IdsClearingHouseServiceImpl.class, eventSubscriber);
+        context.registerService(LoggingHouseEventSubscriber.class, eventSubscriber);
 
         monitor.debug("Registered event subscriber for LoggingHouseClientExtension");
     }
@@ -192,7 +242,6 @@ private void registerClearingHouseMessageSenders(ServiceExtensionContext context
         dispatcherRegistry.register(dispatcher);
     }
 
-
     @Override
     public void start() {
         if (!enabled) {