feat(dispatch): multi s3 tenant

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/adapter/in/streaming/StreamingInAdapter.java

@@ -5,11 +5,11 @@
 import de.muenchen.oss.swim.dispatcher.domain.exception.PresignedUrlException;
 import de.muenchen.oss.swim.dispatcher.domain.exception.UseCaseException;
 import de.muenchen.oss.swim.dispatcher.domain.model.ErrorDetails;
+import de.muenchen.oss.swim.dispatcher.domain.model.FileEvent;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.function.Consumer;
 import lombok.RequiredArgsConstructor;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.context.annotation.Bean;
 import org.springframework.messaging.Message;
 import org.springframework.stereotype.Service;
@@ -21,26 +21,23 @@ public class StreamingInAdapter {
     private final ErrorHandlerInPort errorHandlerInPort;
 
     @Bean
-    protected Consumer<Message<FileEventDTO>> finished() {
+    protected Consumer<Message<FileEvent>> finished() {
         return fileFinishedEventDTOMessage -> {
-            final FileEventDTO fileFinishedDTO = fileFinishedEventDTOMessage.getPayload();
+            final FileEvent fileFinishedDTO = fileFinishedEventDTOMessage.getPayload();
             try {
-                markFileFinishedInPort.markFileFinished(fileFinishedDTO.useCase(), fileFinishedDTO.presignedUrl());
-                if (StringUtils.isNotBlank(fileFinishedDTO.metadataPresignedUrl())) {
-                    markFileFinishedInPort.markFileFinished(fileFinishedDTO.useCase(), fileFinishedDTO.metadataPresignedUrl());
-                }
+                markFileFinishedInPort.markFileFinished(fileFinishedDTO);
             } catch (PresignedUrlException | UseCaseException e) {
                 throw new RuntimeException(e);
             }
         };
     }
 
     @Bean
-    protected Consumer<Message<FileEventDTO>> dlq() {
+    protected Consumer<Message<FileEvent>> dlq() {
         return message -> {
-            final FileEventDTO fileFinishedDTO = message.getPayload();
+            final FileEvent fileFinishedDTO = message.getPayload();
             final ErrorDetails error = this.errorDetailsFromHeaders(message.getHeaders());
-            errorHandlerInPort.handleError(fileFinishedDTO.useCase(), fileFinishedDTO.presignedUrl(), fileFinishedDTO.metadataPresignedUrl(), error);
+            errorHandlerInPort.handleError(fileFinishedDTO, error);
         };
     }
 

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/adapter/out/s3/S3Properties.java

@@ -1,23 +1,78 @@
 package de.muenchen.oss.swim.dispatcher.adapter.out.s3;
 
+import de.muenchen.oss.swim.dispatcher.domain.model.UseCase;
+import io.minio.MinioClient;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
-import lombok.Data;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
 
-@Data
+@Setter
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@ToString
 @ConfigurationProperties("swim.s3")
 class S3Properties {
-    @NotBlank
-    private String url;
-    @NotBlank
-    private String accessKey;
-    @NotBlank
-    private String secretKey;
+    /**
+     * Internal mapping of tenant names to already created Clients.
+     * Ensures only one MinoClient per tenant is created and reused.
+     */
+    private static final Map<String, MinioClient> CLIENTS = new ConcurrentHashMap<>();
+    /**
+     * List of S3 tenants which can be used as {@link UseCase#getTenant()}.
+     */
+    @NestedConfigurationProperty
+    @NotNull
+    private Map<String, ConnectionOptions> tenants;
     /**
      * Time in seconds after which the created presigned urls expire.
      * Default: 7d
      */
     @NotNull
     private int presignedUrlExpiry = 7 * 24 * 60 * 60;
+
+    /**
+     * Configuration for a single S3 tenant.
+     */
+    @NoArgsConstructor
+    @AllArgsConstructor
+    @Setter
+    @Getter
+    @ToString(exclude = { "secretKey", "accessKey" })
+    protected static class ConnectionOptions {
+        @NotBlank
+        private String url;
+        @NotBlank
+        private String accessKey;
+        @NotBlank
+        private String secretKey;
+    }
+
+    /**
+     * Get a MinioClient for a specific tenant name.
+     * See {@link #tenants}.
+     *
+     * @param tenant The name of the tenant to get the client for.
+     * @return A MinioClient for the specified tenant.
+     */
+    protected MinioClient getClient(final String tenant) {
+        return CLIENTS.computeIfAbsent(tenant, key -> {
+            if (!this.tenants.containsKey(key)) {
+                throw new IllegalArgumentException("Tenant doesn't exist: " + tenant);
+            }
+            final ConnectionOptions options = this.tenants.get(key);
+            return MinioClient.builder()
+                    .endpoint(options.url)
+                    .credentials(options.accessKey, options.secretKey)
+                    .build();
+        });
+    }
 }

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/adapter/out/streaming/StreamingOutAdapter.java

@@ -1,6 +1,7 @@
 package de.muenchen.oss.swim.dispatcher.adapter.out.streaming;
 
 import de.muenchen.oss.swim.dispatcher.application.port.out.FileDispatchingOutPort;
+import de.muenchen.oss.swim.dispatcher.domain.model.FileEvent;
 import lombok.RequiredArgsConstructor;
 import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.stereotype.Service;
@@ -13,7 +14,7 @@ public class StreamingOutAdapter implements FileDispatchingOutPort {
     @Override
     @SuppressWarnings("PMD.UseObjectForClearerAPI")
     public void dispatchFile(final String bindingName, final String useCase, final String presignedUrl, final String metadataPresignedUrl) {
-        final FileEventDTO event = new FileEventDTO(useCase, presignedUrl, metadataPresignedUrl);
+        final FileEvent event = new FileEvent(useCase, presignedUrl, metadataPresignedUrl);
         streamBridge.send(bindingName, event);
     }
 }

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/application/port/in/ErrorHandlerInPort.java

@@ -1,6 +1,7 @@
 package de.muenchen.oss.swim.dispatcher.application.port.in;
 
 import de.muenchen.oss.swim.dispatcher.domain.model.ErrorDetails;
+import de.muenchen.oss.swim.dispatcher.domain.model.FileEvent;
 import jakarta.validation.constraints.NotNull;
 import org.springframework.validation.annotation.Validated;
 
@@ -10,10 +11,8 @@ public interface ErrorHandlerInPort {
      * Handle error which was thrown while processing dispatched message.
      * Could either be in external service or while marking file as finished.
      *
-     * @param useCaseName Name of the useCase for which the Exception occurred.
-     * @param presignedUrl PresignedUrl of the file for which the Exception occurred.
-     * @param metadataPresignedUrl PresignedUrl of the metadata file for which the Exception occurred.
+     * @param event The event for the file the error occurred for.
      * @param cause The Exception which occurred.
      */
-    void handleError(String useCaseName, String presignedUrl, String metadataPresignedUrl, @NotNull ErrorDetails cause);
+    void handleError(@NotNull FileEvent event, @NotNull ErrorDetails cause);
 }

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/application/port/in/MarkFileFinishedInPort.java

@@ -2,16 +2,17 @@
 
 import de.muenchen.oss.swim.dispatcher.domain.exception.PresignedUrlException;
 import de.muenchen.oss.swim.dispatcher.domain.exception.UseCaseException;
-import jakarta.validation.constraints.NotBlank;
+import de.muenchen.oss.swim.dispatcher.domain.model.FileEvent;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
 import org.springframework.validation.annotation.Validated;
 
 @Validated
 public interface MarkFileFinishedInPort {
     /**
      * Mark a file as finished processing.
      *
-     * @param useCase The name of the use case the file was found for.
-     * @param presignedUrl The presigned url of the file.
+     * @param event The event for the file to marks as finished.
      */
-    void markFileFinished(@NotBlank String useCase, @NotBlank String presignedUrl) throws PresignedUrlException, UseCaseException;
+    void markFileFinished(@NotNull @Valid FileEvent event) throws PresignedUrlException, UseCaseException;
 }

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/application/port/out/FileSystemOutPort.java

@@ -2,6 +2,7 @@
 
 import de.muenchen.oss.swim.dispatcher.domain.exception.PresignedUrlException;
 import de.muenchen.oss.swim.dispatcher.domain.model.File;
+import de.muenchen.oss.swim.dispatcher.domain.model.UseCase;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import java.io.InputStream;
@@ -22,7 +23,9 @@ public interface FileSystemOutPort {
      * @param excludeTags Tag entries where none should be on the file.
      * @return Map of files (with all tags) having required and not having any exclude tags.
      */
+    @SuppressWarnings("PMD.UseObjectForClearerAPI")
     Map<File, Map<String, String>> getMatchingFilesWithTags(
+            @NotBlank String tenant,
             @NotBlank String bucket,
             @NotNull String pathPrefix,
             @NotNull boolean recursive,
@@ -37,7 +40,7 @@ Map<File, Map<String, String>> getMatchingFilesWithTags(
      * @param pathPrefix The path to look under.
      * @return The names of the subdirectories.
      */
-    List<String> getSubDirectories(
+    List<String> getSubDirectories(@NotBlank String tenant,
             @NotBlank String bucket,
             @NotNull String pathPrefix);
 
@@ -48,7 +51,7 @@ List<String> getSubDirectories(
      * @param path The path to the file.
      * @param tags The tags to add.
      */
-    void tagFile(@NotBlank String bucket, @NotBlank String path, @NotNull Map<String, String> tags);
+    void tagFile(@NotBlank String tenant, @NotBlank String bucket, @NotBlank String path, @NotNull Map<String, String> tags);
 
     /**
      * Check if a file exists.
@@ -57,7 +60,7 @@ List<String> getSubDirectories(
      * @param path The path to the file.
      * @return If the file exists.
      */
-    boolean fileExists(@NotBlank String bucket, @NotBlank String path);
+    boolean fileExists(@NotBlank String tenant, @NotBlank String bucket, @NotBlank String path);
 
     /**
      * Get content of a file.
@@ -66,7 +69,7 @@ List<String> getSubDirectories(
      * @param path The path of the file.
      * @return The content of the file.
      */
-    InputStream readFile(@NotBlank String bucket, @NotBlank String path);
+    InputStream readFile(@NotBlank String tenant, @NotBlank String bucket, @NotBlank String path);
 
     /**
      * Get presigned url for downloading a file.
@@ -75,14 +78,16 @@ List<String> getSubDirectories(
      * @param path The path of the file.
      * @return The presigned url for the file.
      */
-    String getPresignedUrl(@NotBlank String bucket, @NotBlank String path);
+    String getPresignedUrl(@NotBlank String tenant, @NotBlank String bucket, @NotBlank String path);
 
     /**
-     * Verify a presigned url for downloading a file.
+     * Verify and extract File from a presigned URL for downloading a file.
      *
+     * @param useCase The resolved use case of the presigned url.
      * @param presignedUrl The presigned url.
+     * @return The File extracted from the presigned URL.
      */
-    boolean verifyPresignedUrl(@NotBlank String presignedUrl) throws PresignedUrlException;
+    File verifyAndResolvePresignedUrl(@NotNull UseCase useCase, @NotBlank String presignedUrl) throws PresignedUrlException;
 
     /**
      * Move a file from one place to another.
@@ -91,7 +96,8 @@ List<String> getSubDirectories(
      * @param srcPath The source path of the file.
      * @param destPath The destination path of the file.
      */
-    void moveFile(@NotBlank String bucket, @NotBlank String srcPath, @NotBlank String destPath);
+    @SuppressWarnings("PMD.UseObjectForClearerAPI")
+    void moveFile(@NotBlank String tenant, @NotBlank String bucket, @NotBlank String srcPath, @NotBlank String destPath);
 
     /**
      * Copy a file from one place to another.
@@ -103,5 +109,6 @@ List<String> getSubDirectories(
      * @param clearTags If the existing tags should be removed.
      */
     @SuppressWarnings("PMD.UseObjectForClearerAPI")
-    void copyFile(@NotBlank String srcBucket, @NotBlank String srcPath, @NotBlank String destBucket, @NotBlank String destPath, boolean clearTags);
+    void copyFile(@NotBlank String tenant, @NotBlank String srcBucket, @NotBlank String srcPath, @NotBlank String destBucket, @NotBlank String destPath,
+            boolean clearTags);
 }

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/application/port/out/ReadProtocolOutPort.java

@@ -14,5 +14,5 @@ public interface ReadProtocolOutPort {
      * @param path The path of the protocol file.
      * @return The parsed protocol entries.
      */
-    List<ProtocolEntry> loadProtocol(@NotBlank String bucket, @NotBlank String path);
+    List<ProtocolEntry> loadProtocol(@NotBlank String tenant, @NotBlank String bucket, @NotBlank String path);
 }

dispatch-service/src/main/java/de/muenchen/oss/swim/dispatcher/application/usecase/DispatcherUseCase.java

@@ -54,7 +54,7 @@ public void triggerDispatching() {
             // handle recursive by directory
             if (useCase.isRecursive()) {
                 // get folders
-                final List<String> folders = fileSystemOutPort.getSubDirectories(useCase.getBucket(), dispatchPath);
+                final List<String> folders = fileSystemOutPort.getSubDirectories(useCase.getTenant(), useCase.getBucket(), dispatchPath);
                 // dispatch files per folder if not in finished folder
                 for (final String folder : folders) {
                     if (!folder.contains(swimDispatcherProperties.getFinishedFolder())) {
@@ -81,6 +81,7 @@ public void triggerDispatching() {
     private @NotNull
     Map<String, Throwable> processDirectory(final UseCase useCase, final String folder, final boolean recursive) {
         final Map<File, Map<String, String>> readyFiles = fileSystemOutPort.getMatchingFilesWithTags(
+                useCase.getTenant(),
                 useCase.getBucket(),
                 folder,
                 recursive,
@@ -185,19 +186,19 @@ protected void dispatchFile(final UseCase useCase, final File file, final String
             // build metadata file path
             final String metadataPath = file.getMetadataFilePath();
             // continue if not existing
-            if (!fileSystemOutPort.fileExists(file.bucket(), metadataPath)) {
+            if (!fileSystemOutPort.fileExists(file.tenant(), file.bucket(), metadataPath)) {
                 final String message = String.format("Metadata file %s missing", metadataPath);
                 throw new MetadataException(message);
             }
-            metadataPresignedUrl = fileSystemOutPort.getPresignedUrl(file.bucket(), metadataPath);
+            metadataPresignedUrl = fileSystemOutPort.getPresignedUrl(file.tenant(), file.bucket(), metadataPath);
         } else {
             metadataPresignedUrl = null;
         }
         // dispatch file
-        final String presignedUrl = fileSystemOutPort.getPresignedUrl(file.bucket(), file.path());
+        final String presignedUrl = fileSystemOutPort.getPresignedUrl(file.tenant(), file.bucket(), file.path());
         fileDispatchingOutPort.dispatchFile(destination, useCase.getName(), presignedUrl, metadataPresignedUrl);
         // mark file as dispatched
-        fileSystemOutPort.tagFile(file.bucket(), file.path(), Map.of(
+        fileSystemOutPort.tagFile(file.tenant(), file.bucket(), file.path(), Map.of(
                 swimDispatcherProperties.getDispatchStateTagKey(),
                 swimDispatcherProperties.getDispatchedStateTagValue()));
     }
@@ -215,7 +216,7 @@ protected void dispatchFile(final UseCase useCase, final File file, final String
     protected String resolveDestinationBinding(final UseCase useCase, final File file) throws MetadataException {
         // resolve via metadata file if enabled
         if (useCase.isOverwriteDestinationViaMetadata()) {
-            try (InputStream metadataFileStream = this.fileSystemOutPort.readFile(file.bucket(), file.getMetadataFilePath())) {
+            try (InputStream metadataFileStream = this.fileSystemOutPort.readFile(file.tenant(), file.bucket(), file.getMetadataFilePath())) {
                 final Metadata metadata = metadataHelper.parseMetadataFile(metadataFileStream);
                 final String value = metadata.indexFields().get(swimDispatcherProperties.getMetadataDispatchBindingKey());
                 if (StringUtils.isNotBlank(value)) {
@@ -237,11 +238,11 @@ protected String resolveDestinationBinding(final UseCase useCase, final File fil
      */
     protected void finishFile(final UseCase useCase, final File file) {
         // finish metadata file if required and exists
-        if (useCase.isRequiresMetadata() && this.fileSystemOutPort.fileExists(file.bucket(), file.getMetadataFilePath())) {
-            this.fileHandlingHelper.finishFile(useCase, file.bucket(), file.getMetadataFilePath());
+        if (useCase.isRequiresMetadata() && this.fileSystemOutPort.fileExists(useCase.getTenant(), file.bucket(), file.getMetadataFilePath())) {
+            this.fileHandlingHelper.finishFile(useCase, file.tenant(), file.bucket(), file.getMetadataFilePath());
         }
         // finish file
-        this.fileHandlingHelper.finishFile(useCase, file.bucket(), file.path());
+        this.fileHandlingHelper.finishFile(useCase, file.tenant(), file.bucket(), file.path());
     }
 
     /**
@@ -275,7 +276,7 @@ protected void rerouteFileToUseCase(final UseCase useCase, final File file, fina
         // copy file to target use case
         final String rawPath = useCase.getRawPath(swimDispatcherProperties, file.path());
         final String destPath = String.format("%s/from_%s/%s", targetUseCase.getDispatchPath(swimDispatcherProperties), useCase.getName(), rawPath);
-        fileSystemOutPort.copyFile(file.bucket(), file.path(), targetUseCase.getBucket(), destPath, true);
+        fileSystemOutPort.copyFile(file.tenant(), file.bucket(), file.path(), targetUseCase.getBucket(), destPath, true);
         // finish file
         this.finishFile(useCase, file);
         log.info("File {} in bucket {} rerouted from use case {} to use case {}", file.path(), file.bucket(), useCase.getName(), targetUseCase.getName());