Process Stored Request/Imp IDs unique per account (prebid#488)

* Process Stored Request/Imp IDs unique per account

* Process Stored Request/Imp IDs unique per account - update docs

* Updated for AMP+Account

* Clean the code

* Update docs

* Update docs

* Small refactor

* Add account support for CachingApplicationSettings

* refactored after PR review

* refactored after PR review - video fixed

* Fixes after review

Author: rpanchyk

docs/config.md

@@ -24,8 +24,8 @@ metrics:
 settings:
   database:
     type: mysql
-    stored-requests-query: SELECT reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%) UNION ALL SELECT impid, impData, 'imp' as dataType FROM stored_imps WHERE impid IN (%IMP_ID_LIST%)
-    amp-stored-requests-query: SELECT reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%)
+    stored-requests-query: SELECT accountId, reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%) UNION ALL SELECT accountId, impid, impData, 'imp' as dataType FROM stored_imps WHERE impid IN (%IMP_ID_LIST%)
+    amp-stored-requests-query: SELECT accountId, reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%)
     stored-responses-query: SELECT resid, responseData FROM stored_responses WHERE resid IN (%RESPONSE_ID_LIST%)
 ```
 If some property is missed in `prebid-config.yaml` application will look for it in `src/main/resources/application.yaml` file.

docs/developers/stored-requests.md

@@ -137,7 +137,7 @@ For example, assume the following `stored-requests/stored-request.json`:
   }
 ```
 
-Then an HTTP request like:
+Then HTTP request like:
 
 ```json
 {
@@ -178,8 +178,7 @@ Prebid Server does allow Stored BidRequests and Stored Imps in the same HTTP Req
 The Stored BidRequest will be applied first, and then the Stored Imps after.
 
 **Beware**: Stored Request data will not be applied recursively.
-If a Stored BidRequest includes Imps with their own Stored Request IDs,
-then the data for those Stored Imps not be resolved.
+If a Stored BidRequest includes Imps with their own Stored Request IDs, then the data for those Stored Imps will not be resolved.
 
 ## Alternate backends
 
@@ -193,13 +192,6 @@ For PostgreSQL:
 settings:
   database:
     type: postgres
-    host: localhost
-    port: 5432
-    dbname: database-name
-    user: username
-    password: password
-    stored-requests-query: SELECT reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%) UNION ALL SELECT impid, impData, 'imp' as dataType FROM stored_imps WHERE impid IN (%IMP_ID_LIST%)
-    amp-stored-requests-query: SELECT reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%)
 ```
 
 For MySQL:
@@ -209,10 +201,11 @@ settings:
     type: mysql
 ```
 
-The select query columns of `stored-data-query` and `amp-stored-data-query` properties should correspond to the specific format:
-- first column: ID of stored data item
-- second column: value of stored data item
-- third column: type of stored data item. Can be `request` for stored requests or `imp` for stored impressions.
+The select query columns of `stored-requests-query` and `amp-stored-requests-query` properties should correspond to the specific format:
+- first column: account ID which is searched by.
+- second column: ID of stored data item which is searched by.
+- third column: value of stored data item.
+- forth column: type of stored data item. Can be `request` for stored requests or `imp` for stored impressions.
 
 ### HTTP backend
 
@@ -261,8 +254,8 @@ settings:
     dbname: database-name
     user: username
     password: password
-    stored-requests-query: SELECT reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%) UNION ALL SELECT impid, impData, 'imp' as dataType FROM stored_imps WHERE impid IN (%IMP_ID_LIST%)
-    amp-stored-requests-query: SELECT reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%)
+    stored-requests-query: SELECT accountId, reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%) UNION ALL SELECT accountId, impid, impData, 'imp' as dataType FROM stored_imps WHERE impid IN (%IMP_ID_LIST%)
+    amp-stored-requests-query: SELECT accountId, reqid, requestData, 'request' as dataType FROM stored_requests WHERE reqid IN (%REQUEST_ID_LIST%)
   http:
     endpoint: http://stored-requests.prebid.com
     amp-endpoint: http://stored-requests.prebid.com?amp=true

src/main/java/org/prebid/server/auction/AmpRequestFactory.java

@@ -121,7 +121,7 @@ public Future<AuctionContext> fromRequest(RoutingContext routingContext, long st
      */
     private Future<Tuple2<BidRequest, List<String>>> createBidRequest(RoutingContext context, String tagId) {
         final List<String> errors = new ArrayList<>();
-        return storedRequestProcessor.processAmpRequest(tagId)
+        return storedRequestProcessor.processAmpRequest(context.request().getParam(ACCOUNT_REQUEST_PARAM), tagId)
                 .map(bidRequest -> validateStoredBidRequest(tagId, bidRequest))
                 .map(bidRequest -> fillExplicitParameters(bidRequest, context))
                 .map(bidRequest -> overrideParameters(bidRequest, context.request(), errors))

src/main/java/org/prebid/server/auction/AuctionRequestFactory.java

@@ -253,7 +253,7 @@ private BidRequest parseRequest(RoutingContext context, List<String> errors) {
      * updated by values derived from headers and other request attributes.
      */
     private Future<BidRequest> updateBidRequest(RoutingContext context, BidRequest bidRequest) {
-        return storedRequestProcessor.processStoredRequests(bidRequest)
+        return storedRequestProcessor.processStoredRequests(accountIdFrom(bidRequest), bidRequest)
                 .map(resolvedBidRequest -> fillImplicitParameters(resolvedBidRequest, context, timeoutResolver))
                 .map(this::validateRequest)
                 .map(interstitialProcessor::process);

src/main/java/org/prebid/server/auction/BidResponseCreator.java

@@ -222,7 +222,7 @@ private Future<BidResponse> cacheBidsAndCreateResponse(List<BidderResponse> bidd
                 auctionContext,
                 cacheInfo,
                 eventsContext)
-                .compose(cacheResult -> videoStoredDataResult(bidRequest.getImp(), auctionContext.getTimeout())
+                .compose(cacheResult -> videoStoredDataResult(auctionContext)
                         .map(videoStoredDataResult -> toBidResponse(
                                 bidderResponses,
                                 auctionContext,
@@ -698,9 +698,14 @@ private BidResponse toBidResponse(List<BidderResponse> bidderResponses,
                 .build();
     }
 
-    private Future<VideoStoredDataResult> videoStoredDataResult(List<Imp> imps, Timeout timeout) {
+    private Future<VideoStoredDataResult> videoStoredDataResult(AuctionContext auctionContext) {
+        final List<Imp> imps = auctionContext.getBidRequest().getImp();
+        final String accountId = auctionContext.getAccount().getId();
+        final Timeout timeout = auctionContext.getTimeout();
+
         final List<String> errors = new ArrayList<>();
         final List<Imp> storedVideoInjectableImps = new ArrayList<>();
+
         for (Imp imp : imps) {
             try {
                 if (checkEchoVideoAttrs(imp)) {
@@ -711,7 +716,7 @@ private Future<VideoStoredDataResult> videoStoredDataResult(List<Imp> imps, Time
             }
         }
 
-        return storedRequestProcessor.videoStoredDataResult(storedVideoInjectableImps, errors, timeout)
+        return storedRequestProcessor.videoStoredDataResult(accountId, storedVideoInjectableImps, errors, timeout)
                 .otherwise(throwable -> VideoStoredDataResult.of(Collections.emptyMap(),
                         Collections.singletonList(throwable.getMessage())));
     }

src/main/java/org/prebid/server/auction/StoredRequestProcessor.java

@@ -66,7 +66,7 @@ public StoredRequestProcessor(long defaultTimeout,
      * fetched jsons from source. In case any error happen during the process, returns failedFuture with
      * InvalidRequestException {@link InvalidRequestException} as cause.
      */
-    Future<BidRequest> processStoredRequests(BidRequest bidRequest) {
+    Future<BidRequest> processStoredRequests(String accountId, BidRequest bidRequest) {
         final Map<BidRequest, String> bidRequestToStoredRequestId;
         final Map<Imp, String> impToStoredRequestId;
         try {
@@ -86,7 +86,7 @@ Future<BidRequest> processStoredRequests(BidRequest bidRequest) {
         }
 
         final Future<StoredDataResult> storedDataFuture =
-                applicationSettings.getStoredData(requestIds, impIds, timeout(bidRequest))
+                applicationSettings.getStoredData(accountId, requestIds, impIds, timeout(bidRequest))
                         .compose(storedDataResult -> updateMetrics(storedDataResult, requestIds, impIds));
 
         return storedRequestsToBidRequest(storedDataFuture, bidRequest,
@@ -97,19 +97,20 @@ private Future<StoredDataResult> updateMetrics(StoredDataResult storedDataResult
                                                    Set<String> impIds) {
         requestIds.forEach(
                 id -> metrics.updateStoredRequestMetric(storedDataResult.getStoredIdToRequest().containsKey(id)));
-        impIds.forEach(id -> metrics.updateStoredImpsMetric(storedDataResult.getStoredIdToImp().containsKey(id)));
+        impIds.forEach(
+                id -> metrics.updateStoredImpsMetric(storedDataResult.getStoredIdToImp().containsKey(id)));
 
         return Future.succeededFuture(storedDataResult);
     }
 
     /**
      * Fetches AMP request from the source.
      */
-    Future<BidRequest> processAmpRequest(String ampRequestId) {
+    Future<BidRequest> processAmpRequest(String accountId, String ampRequestId) {
         final BidRequest bidRequest = BidRequest.builder().build();
         final Future<StoredDataResult> ampStoredDataFuture =
                 applicationSettings.getAmpStoredData(
-                        Collections.singleton(ampRequestId), Collections.emptySet(), timeout(bidRequest))
+                        accountId, Collections.singleton(ampRequestId), Collections.emptySet(), timeout(bidRequest))
                         .compose(storedDataResult -> updateMetrics(
                                 storedDataResult, Collections.singleton(ampRequestId), Collections.emptySet()));
 
@@ -119,14 +120,15 @@ Future<BidRequest> processAmpRequest(String ampRequestId) {
     /**
      * Fetches stored request.video and map existing values to imp.id.
      */
-    Future<VideoStoredDataResult> videoStoredDataResult(List<Imp> imps, List<String> errors, Timeout timeout) {
+    Future<VideoStoredDataResult> videoStoredDataResult(String accountId, List<Imp> imps, List<String> errors,
+                                                        Timeout timeout) {
         final Map<String, String> storedIdToImpId =
                 mapStoredRequestHolderToStoredRequestId(imps, this::getStoredRequestFromImp)
                         .entrySet().stream()
                         .collect(Collectors.toMap(Map.Entry::getValue,
                                 impIdToStoredId -> impIdToStoredId.getKey().getId()));
 
-        return applicationSettings.getStoredData(Collections.emptySet(), storedIdToImpId.keySet(), timeout)
+        return applicationSettings.getStoredData(accountId, Collections.emptySet(), storedIdToImpId.keySet(), timeout)
                 .map(storedDataResult -> makeVideoStoredDataResult(storedDataResult, storedIdToImpId, errors));
     }
 

src/main/java/org/prebid/server/auction/VideoRequestFactory.java

@@ -1,7 +1,10 @@
 package org.prebid.server.auction;
 
+import com.iab.openrtb.request.App;
 import com.iab.openrtb.request.BidRequest;
 import com.iab.openrtb.request.Device;
+import com.iab.openrtb.request.Publisher;
+import com.iab.openrtb.request.Site;
 import com.iab.openrtb.request.video.BidRequestVideo;
 import com.iab.openrtb.request.video.Pod;
 import com.iab.openrtb.request.video.PodError;
@@ -10,13 +13,16 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.RoutingContext;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.prebid.server.auction.model.AuctionContext;
 import org.prebid.server.auction.model.WithPodErrors;
 import org.prebid.server.exception.InvalidRequestException;
 import org.prebid.server.json.DecodeException;
 import org.prebid.server.json.JacksonMapper;
 import org.prebid.server.metric.MetricName;
+import org.prebid.server.proto.openrtb.ext.request.ExtPublisher;
+import org.prebid.server.proto.openrtb.ext.request.ExtPublisherPrebid;
 import org.prebid.server.util.HttpUtil;
 
 import java.util.ArrayList;
@@ -77,6 +83,38 @@ public Future<WithPodErrors<AuctionContext>> fromRequest(RoutingContext routingC
                         .map(auctionContext -> WithPodErrors.of(auctionContext, bidRequestToPodError.getPodErrors())));
     }
 
+    /**
+     * Extracts publisher id either from {@link BidRequestVideo}.app.publisher
+     * or {@link BidRequestVideo}.site.publisher. If neither is present returns empty string.
+     */
+    private String accountIdFrom(BidRequestVideo bidRequestVideo) {
+        final App app = bidRequestVideo.getApp();
+        final Publisher appPublisher = app != null ? app.getPublisher() : null;
+        final Site site = bidRequestVideo.getSite();
+        final Publisher sitePublisher = site != null ? site.getPublisher() : null;
+
+        final Publisher publisher = ObjectUtils.defaultIfNull(appPublisher, sitePublisher);
+        final String publisherId = publisher != null ? resolvePublisherId(publisher) : null;
+        return ObjectUtils.defaultIfNull(publisherId, StringUtils.EMPTY);
+    }
+
+    /**
+     * Resolves what value should be used as a publisher id - either taken from publisher.ext.parentAccount
+     * or publisher.id in this respective priority.
+     */
+    private String resolvePublisherId(Publisher publisher) {
+        final String parentAccountId = parentAccountIdFromExtPublisher(publisher.getExt());
+        return ObjectUtils.defaultIfNull(parentAccountId, publisher.getId());
+    }
+
+    /**
+     * Parses publisher.ext and returns parentAccount value from it. Returns null if any parsing error occurs.
+     */
+    private String parentAccountIdFromExtPublisher(ExtPublisher extPublisher) {
+        final ExtPublisherPrebid extPublisherPrebid = extPublisher != null ? extPublisher.getPrebid() : null;
+        return extPublisherPrebid != null ? StringUtils.stripToNull(extPublisherPrebid.getParentAccount()) : null;
+    }
+
     /**
      * Parses request body to {@link BidRequestVideo}.
      * <p>
@@ -137,7 +175,8 @@ private Future<WithPodErrors<BidRequest>> createBidRequest(RoutingContext routin
                                                                BidRequestVideo bidRequestVideo,
                                                                String storedVideoId,
                                                                Set<String> podConfigIds) {
-        return storedRequestProcessor.processVideoRequest(storedVideoId, podConfigIds, bidRequestVideo)
+        return storedRequestProcessor.processVideoRequest(accountIdFrom(bidRequestVideo), storedVideoId, podConfigIds,
+                bidRequestVideo)
                 .map(bidRequestToErrors -> fillImplicitParameters(routingContext, bidRequestToErrors))
                 .map(this::validateRequest);
     }

src/main/java/org/prebid/server/auction/VideoStoredRequestProcessor.java

@@ -42,7 +42,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -95,13 +94,15 @@ public VideoStoredRequestProcessor(ApplicationSettings applicationSettings, Vide
     /**
      * Fetches ParsedStoredDataResult<BidRequestVideo, Imp> from stored request.
      */
-    Future<WithPodErrors<BidRequest>> processVideoRequest(String storedBidRequestId, Set<String> podIds,
+    Future<WithPodErrors<BidRequest>> processVideoRequest(String accountId, String storedBidRequestId,
+                                                          Set<String> podIds,
                                                           BidRequestVideo receivedRequest) {
-        final Set<String> storedRequestIds = new HashSet<>();
-        if (StringUtils.isNotBlank(storedBidRequestId)) {
-            storedRequestIds.add(storedBidRequestId);
-        }
-        return applicationSettings.getVideoStoredData(storedRequestIds, podIds, timeoutFactory.create(defaultTimeout))
+        final Set<String> storedRequestIds = StringUtils.isNotBlank(storedBidRequestId)
+                ? Collections.singleton(storedBidRequestId)
+                : Collections.emptySet();
+
+        return applicationSettings.getVideoStoredData(accountId, storedRequestIds, podIds,
+                timeoutFactory.create(defaultTimeout))
                 .compose(storedDataResult -> updateMetrics(storedDataResult, storedRequestIds, podIds))
                 .map(storedData -> mergeToBidRequest(storedData, receivedRequest, storedBidRequestId))
                 .recover(exception -> Future.failedFuture(new InvalidRequestException(

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

@@ -21,32 +21,35 @@
 public interface ApplicationSettings {
 
     /**
-     * Returns {@link Account} info for given a accountId
+     * Returns {@link Account} for the given account ID.
      */
     Future<Account> getAccountById(String accountId, Timeout timeout);
 
     /**
-     * Returns AddUnitConfig info for a given adUnitConfigId
+     * Returns configuration for the given adUnit config ID.
      */
     Future<String> getAdUnitConfigById(String adUnitConfigId, Timeout timeout);
 
     /**
-     * Fetches stored requests and imps
+     * Fetches stored requests and imps by IDs.
      */
-    Future<StoredDataResult> getStoredData(Set<String> requestIds, Set<String> impIds, Timeout timeout);
+    Future<StoredDataResult> getStoredData(String accountId, Set<String> requestIds, Set<String> impIds,
+                                           Timeout timeout);
 
     /**
-     * Fetches stored response
+     * Fetches AMP stored requests and imps by IDs.
      */
-    Future<StoredResponseDataResult> getStoredResponses(Set<String> responseIds, Timeout timeout);
+    Future<StoredDataResult> getAmpStoredData(String accountId, Set<String> requestIds, Set<String> impIds,
+                                              Timeout timeout);
 
     /**
-     * Fetches AMP stored requests and imps
+     * Fetches Video stored requests and imps by IDs.
      */
-    Future<StoredDataResult> getAmpStoredData(Set<String> requestIds, Set<String> impIds, Timeout timeout);
+    Future<StoredDataResult> getVideoStoredData(String accountId, Set<String> requestIds, Set<String> impIds,
+                                                Timeout timeout);
 
     /**
-     * Fetches Video stored requests and imps
+     * Fetches stored response by IDs.
      */
-    Future<StoredDataResult> getVideoStoredData(Set<String> requestIds, Set<String> impIds, Timeout timeout);
+    Future<StoredResponseDataResult> getStoredResponses(Set<String> responseIds, Timeout timeout);
 }