Add S3/MinIO support for application settings

docs/application-settings.md

@@ -259,6 +259,51 @@ Here's an example YAML file containing account-specific settings:
           default: 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
+      region: <region name> # if not provided AWS_GLOBAL will be used. Example value: 'eu-central-1'
+      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).

extra/pom.xml

@@ -51,6 +51,7 @@
         <protobuf.version>3.21.7</protobuf.version>
         <protoc.version>3.17.3</protoc.version>
         <json-logic.version>1.0.7</json-logic.version>
+        <aws.awssdk.version>2.17.274</aws.awssdk.version>
 
         <!-- Project test dependency versions -->
         <wiremock.version>3.5.4</wiremock.version>
@@ -206,6 +207,11 @@
                 <artifactId>geoip2</artifactId>
                 <version>${maxmind-client.version}</version>
             </dependency>
+            <dependency>
+                <groupId>software.amazon.awssdk</groupId>
+                <artifactId>s3</artifactId>
+                <version>${aws.awssdk.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.google.protobuf</groupId>
                 <artifactId>protobuf-java</artifactId>

pom.xml

@@ -173,6 +173,10 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
         </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,210 @@
+package org.prebid.server.settings;
+
+import io.vertx.core.CompositeFuture;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import org.apache.commons.collections4.SetUtils;
+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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+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 a 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 = Objects.requireNonNull(asyncClient);
+        this.bucket = Objects.requireNonNull(bucket);
+        this.accountsDirectory = Objects.requireNonNull(accountsDirectory);
+        this.storedImpressionsDirectory = Objects.requireNonNull(storedImpressionsDirectory);
+        this.storedRequestsDirectory = Objects.requireNonNull(storedRequestsDirectory);
+        this.storedResponsesDirectory = Objects.requireNonNull(storedResponsesDirectory);
+        this.jacksonMapper = Objects.requireNonNull(jacksonMapper);
+        this.vertx = Objects.requireNonNull(vertx);
+    }
+
+    @Override
+    public Future<Account> getAccountById(String accountId, Timeout timeout) {
+        return downloadFile(accountsDirectory + "/" + accountId + JSON_SUFFIX)
+                .flatMap(fileContentOpt -> fileContentOpt.map(Future::succeededFuture)
+                        .orElseGet(() -> Future.failedFuture(
+                                new PreBidException("Account with id %s not found".formatted(accountId)))
+                        )
+                )
+                .map(fileContent -> jacksonMapper.decodeValue(fileContent, Account.class))
+                .flatMap(account -> {
+                    if (!Objects.equals(account.getId(), accountId)) {
+                        return Future.failedFuture(new PreBidException(
+                                "Account with id %s does not match id %s in file".formatted(accountId, account.getId()))
+                        );
+                    }
+                    return Future.succeededFuture(account);
+                })
+                .recover(ex -> {
+                    if (ex instanceof DecodeException) {
+                        return Future
+                                .failedFuture(
+                                        new PreBidException(
+                                                "Invalid json for account with id %s".formatted(accountId)));
+                    }
+                    // if a previous validation already yielded a PreBidException, just return it
+                    if (ex instanceof PreBidException) {
+                        return Future.failedFuture(ex);
+                    }
+                    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 ->
+                                buildStoredDataResult(storedIdToRequest, storedIdToImp, requestIds, impIds))
+                );
+    }
+
+    private StoredDataResult buildStoredDataResult(
+            Map<String, String> storedIdToRequest,
+            Map<String, String> storedIdToImp,
+            Set<String> requestIds,
+            Set<String> impIds
+    ) {
+
+        final Stream<String> missingStoredRequestIds =
+                getMissingStoredDataIds(storedIdToRequest, requestIds).stream()
+                        .map("No stored request found for id: %s"::formatted);
+        final Stream<String> missingStoredImpressionIds =
+                getMissingStoredDataIds(storedIdToImp, impIds).stream()
+                        .map("No stored impression found for id: %s"::formatted);
+
+        return StoredDataResult.of(
+                storedIdToRequest,
+                storedIdToImp,
+                Stream.concat(
+                        missingStoredImpressionIds,
+                        missingStoredRequestIds).toList());
+    }
+
+    private Set<String> getMissingStoredDataIds(Map<String, String> fileContents, Set<String> responseIds) {
+        return SetUtils.difference(responseIds, fileContents.keySet());
+    }
+
+    @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, responseIds).stream()
+                            .map("No stored response found for id: %s"::formatted).toList();
+
+            return StoredResponseDataResult.of(storedIdToResponse, missingStoredResponseIds);
+        });
+    }
+
+    @Override
+    public Future<Map<String, String>> getCategories(String primaryAdServer, String publisher, Timeout timeout) {
+        return Future.succeededFuture(Collections.emptyMap());
+    }
+
+    private Future<Map<String, String>> getFileContents(String directory, Set<String> ids) {
+        return CompositeFuture.join(ids.stream()
+                        .<Future>map(impId -> downloadFile(directory + withInitialSlash(impId) + JSON_SUFFIX)
+                                .map(fileContentOpt -> fileContentOpt
+                                        .map(fileContent -> Tuple2.of(impId, fileContent))))
+                        .toList())
+                .map(CompositeFuture::<Optional<Tuple2<String, String>>>list)
+                .map(impIdToFileContent -> impIdToFileContent.stream()
+                        .flatMap(Optional::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 static String withInitialSlash(String impressionId) {
+        return impressionId.startsWith("/") ? impressionId : "/" + 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()));
+    }
+
+}