Add S3/MinIO support for application settings

docs/application-settings.md

@@ -208,6 +208,50 @@ Here's an example YAML file containing account-specific settings:
         default-coop-sync: true
 ```
 
+## Setting Account Configuration in S3
+
+This is identical to the account configuration in a file system, with the main difference that your file system is
+[AWS S3](https://aws.amazon.com/de/s3/) or any S3 compatible storage, such as [MinIO](https://min.io/).
+
+
+The general idea is that you'll place all the account-specific settings in a separate YAML file and point to that file.
+
+```yaml
+settings:
+  s3:
+      accessKeyId: <S3 access key>
+      secretAccessKey: <S3 access key>
+      endpoint: <endpoint> # http://s3.storage.com
+      bucket: <bucket name> # prebid-application-settings
+      accounts-dir: accounts
+      stored-imps-dir: stored-impressions
+      stored-requests-dir: stored-requests
+      stored-responses-dir: stored-responses
+
+  # recommended to configure an in memory cache, but this is optional
+  in-memory-cache:
+      # example settings, tailor to your needs
+      cache-size: 100000
+      ttl-seconds: 1200 # 20 minutes
+      # recommended to configure
+      s3-update:
+          refresh-rate: 900000 # Refresh every 15 minutes
+          timeout: 5000
+```
+
+### File format
+
+We recommend using the `json` format for your account configuration. A minimal configuration may look like this.
+
+```json
+{
+  "id" : "979c7116-1f5a-43d4-9a87-5da3ccc4f52c",
+  "status" : "active"
+}
+```
+
+This pairs nicely if you have a default configuration defined in your prebid server config under `settings.default-account-config`.
+
 ## Setting Account Configuration in the Database
 
 In database approach account properties are stored in database table(s).

pom.xml

@@ -107,6 +107,13 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>software.amazon.awssdk</groupId>
+                <artifactId>bom</artifactId>
+                <version>2.17.274</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
@@ -275,6 +282,10 @@
             <artifactId>postgresql</artifactId>
             <version>${postgresql.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>s3</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>
             <artifactId>caffeine</artifactId>

src/main/java/org/prebid/server/settings/S3ApplicationSettings.java

@@ -0,0 +1,205 @@
+package org.prebid.server.settings;
+
+import io.vertx.core.CompositeFuture;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import org.prebid.server.auction.model.Tuple2;
+import org.prebid.server.exception.PreBidException;
+import org.prebid.server.execution.Timeout;
+import org.prebid.server.json.DecodeException;
+import org.prebid.server.json.JacksonMapper;
+import org.prebid.server.settings.model.Account;
+import org.prebid.server.settings.model.StoredDataResult;
+import org.prebid.server.settings.model.StoredResponseDataResult;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Implementation of {@link ApplicationSettings}.
+ * <p>
+ * Reads an application settings from JSON file in an s3 bucket, stores and serves them in and from the memory.
+ * <p>
+ * Immediately loads stored request data from local files. These are stored in memory for low-latency reads.
+ * This expects each file in the directory to be named "{config_id}.json".
+ */
+public class S3ApplicationSettings implements ApplicationSettings {
+
+    private static final String JSON_SUFFIX = ".json";
+
+    final S3AsyncClient asyncClient;
+    final String bucket;
+    final String accountsDirectory;
+    final String storedImpressionsDirectory;
+    final String storedRequestsDirectory;
+    final String storedResponsesDirectory;
+    final JacksonMapper jacksonMapper;
+    final Vertx vertx;
+
+    public S3ApplicationSettings(
+            S3AsyncClient asyncClient,
+            String bucket,
+            String accountsDirectory,
+            String storedImpressionsDirectory,
+            String storedRequestsDirectory,
+            String storedResponsesDirectory,
+            JacksonMapper jacksonMapper,
+            Vertx vertx) {
+        this.asyncClient = asyncClient;
+        this.bucket = bucket;
+        this.accountsDirectory = accountsDirectory;
+        this.storedImpressionsDirectory = storedImpressionsDirectory;
+        this.storedRequestsDirectory = storedRequestsDirectory;
+        this.storedResponsesDirectory = storedResponsesDirectory;
+        this.jacksonMapper = jacksonMapper;
+        this.vertx = vertx;
+    }
+
+    @Override
+    public Future<Account> getAccountById(String accountId, Timeout timeout) {
+        return downloadFile(accountsDirectory + "/" + accountId + JSON_SUFFIX)
+                .map(fileContentOpt ->
+                        fileContentOpt.map(fileContent -> jacksonMapper.decodeValue(fileContent, Account.class)))
+                .compose(accountOpt -> {
+                    if (accountOpt.isPresent()) {
+                        return Future.succeededFuture(accountOpt.get());
+                    } else {
+                        return Future
+                                .failedFuture(new PreBidException("Account with id %s not found".formatted(accountId)));
+                    }
+                })
+                .recover(ex -> {
+                    if (ex instanceof DecodeException) {
+                        return Future
+                                .failedFuture(
+                                        new PreBidException(
+                                                "Invalid json for account with id %s".formatted(accountId)));
+                    }
+                    return Future
+                            .failedFuture(new PreBidException("Account with id %s not found".formatted(accountId)));
+                });
+    }
+
+    @Override
+    public Future<StoredDataResult> getStoredData(
+            String accountId,
+            Set<String> requestIds,
+            Set<String> impIds,
+            Timeout timeout) {
+
+        return getFileContents(storedRequestsDirectory, requestIds).compose(storedIdToRequest ->
+                getFileContents(storedImpressionsDirectory, impIds)
+                        .map(storedIdToImp -> {
+                                    final List<String> missingStoredRequestIds =
+                                            getMissingStoredDataIds(storedIdToRequest).stream()
+                                                    .map("No stored request found for id: %s"::formatted).toList();
+                                    final List<String> missingStoredImpressionIds =
+                                            getMissingStoredDataIds(storedIdToImp).stream()
+                                                    .map("No stored impression found for id: %s"::formatted).toList();
+
+                                    return StoredDataResult.of(
+                                            filterOptionalFileContent(storedIdToRequest),
+                                            filterOptionalFileContent(storedIdToImp),
+                                            Stream.concat(
+                                                    missingStoredImpressionIds.stream(),
+                                                    missingStoredRequestIds.stream()).toList());
+                                }
+                        ));
+    }
+
+    private Map<String, String> filterOptionalFileContent(Map<String, Optional<String>> fileContents) {
+        return fileContents
+                .entrySet()
+                .stream()
+                .filter(e -> e.getValue().isPresent())
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().get()));
+    }
+
+    private List<String> getMissingStoredDataIds(Map<String, Optional<String>> fileContents) {
+        return fileContents.entrySet().stream().filter(e -> e.getValue().isEmpty()).map(Map.Entry::getKey).toList();
+    }
+
+    @Override
+    public Future<StoredDataResult> getAmpStoredData(
+            String accountId,
+            Set<String> requestIds,
+            Set<String> impIds,
+            Timeout timeout) {
+        return getStoredData(accountId, requestIds, Collections.emptySet(), timeout);
+    }
+
+    @Override
+    public Future<StoredDataResult> getVideoStoredData(
+            String accountId,
+            Set<String> requestIds,
+            Set<String> impIds,
+            Timeout timeout) {
+        return getStoredData(accountId, requestIds, impIds, timeout);
+    }
+
+    @Override
+    public Future<StoredResponseDataResult> getStoredResponses(Set<String> responseIds, Timeout timeout) {
+        return getFileContents(storedResponsesDirectory, responseIds).map(storedIdToResponse -> {
+            final List<String> missingStoredResponseIds =
+                    getMissingStoredDataIds(storedIdToResponse).stream()
+                            .map("No stored response found for id: %s"::formatted).toList();
+
+            return StoredResponseDataResult.of(
+                    filterOptionalFileContent(storedIdToResponse),
+                    missingStoredResponseIds
+            );
+        });
+    }
+
+    @Override
+    public Future<Map<String, String>> getCategories(String primaryAdServer, String publisher, Timeout timeout) {
+        return Future.succeededFuture(Collections.emptyMap());
+    }
+
+    private Future<Map<String, Optional<String>>> getFileContents(String directory, Set<String> ids) {
+        final List<Future<Tuple2<String, Optional<String>>>> futureListContents = ids.stream()
+                .map(impressionId ->
+                        downloadFile(directory + withInitialSlash(impressionId) + JSON_SUFFIX)
+                                .map(fileContent -> Tuple2.of(impressionId, fileContent)))
+                .collect(Collectors.toCollection(ArrayList::new));
+
+        final Future<List<Tuple2<String, Optional<String>>>> composedFutures = CompositeFuture
+                .all(new ArrayList<>(futureListContents))
+                .map(CompositeFuture::list);
+
+        return composedFutures.map(one -> one.stream().collect(Collectors.toMap(Tuple2::getLeft, Tuple2::getRight)));
+    }
+
+    /**
+     * When the impression id is the ad unit path it may already start with a slash and there's no need to add
+     * another one.
+     *
+     * @param impressionId from the bid request
+     * @return impression id with only a single slash at the beginning
+     */
+    private String withInitialSlash(String impressionId) {
+        if (impressionId.startsWith("/")) {
+            return impressionId;
+        }
+        return "/" + impressionId;
+    }
+
+    private Future<Optional<String>> downloadFile(String key) {
+        final GetObjectRequest request = GetObjectRequest.builder().bucket(bucket).key(key).build();
+
+        return Future.fromCompletionStage(
+                        asyncClient.getObject(request, AsyncResponseTransformer.toBytes()),
+                        vertx.getOrCreateContext())
+                .map(test -> Optional.of(test.asUtf8String())).recover(ex -> Future.succeededFuture(Optional.empty()));
+    }
+
+}