Add Greenbids Real Time Data Module

extra/bundle/pom.xml

@@ -45,6 +45,11 @@
             <artifactId>pb-response-correction</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.prebid.server.hooks.modules</groupId>
+            <artifactId>greenbids-real-time-data</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

extra/modules/greenbids-real-time-data/pom.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.prebid.server.hooks.modules</groupId>
+        <artifactId>all-modules</artifactId>
+        <version>3.14.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>greenbids-real-time-data</artifactId>
+
+    <name>greenbids-real-time-data</name>
+    <description>Greenbids Real Time Data</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.github.ua-parser</groupId>
+            <artifactId>uap-java</artifactId>
+            <version>1.6.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.microsoft.onnxruntime</groupId>
+            <artifactId>onnxruntime</artifactId>
+            <version>1.16.1</version> <!-- Use the latest available version -->
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.cloud</groupId>
+            <artifactId>google-cloud-storage</artifactId>
+            <version>2.41.0</version>
+        </dependency>
+    </dependencies>
+
+</project>

extra/modules/greenbids-real-time-data/src/lombok.config

@@ -0,0 +1 @@
+lombok.anyConstructor.addConstructorProperties = true

extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/GreenbidsRealTimeDataConfiguration.java

@@ -0,0 +1,142 @@
+package org.prebid.server.hooks.modules.greenbids.real.time.data.config;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.maxmind.geoip2.DatabaseReader;
+import io.vertx.core.Vertx;
+import org.prebid.server.exception.PreBidException;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.model.filter.ThrottlingThresholds;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.ThrottlingThresholdsFactory;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.GreenbidsInferenceDataService;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.FilterService;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.ModelCache;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.OnnxModelRunner;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.OnnxModelRunnerFactory;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.OnnxModelRunnerWithThresholds;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.ThresholdCache;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.core.GreenbidsInvocationService;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.v1.GreenbidsRealTimeDataProcessedAuctionRequestHook;
+import org.prebid.server.json.ObjectMapperProvider;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+@ConditionalOnProperty(prefix = "hooks." + GreenbidsRealTimeDataModule.CODE, name = "enabled", havingValue = "true")
+@Configuration
+@EnableConfigurationProperties(GreenbidsRealTimeDataProperties.class)
+public class GreenbidsRealTimeDataConfiguration {
+
+    @Bean
+    DatabaseReader dbReader(GreenbidsRealTimeDataProperties properties) {
+        final File database = new File(properties.getGeoLiteCountryPath());
+        try {
+            return new DatabaseReader.Builder(database).build();
+        } catch (IOException e) {
+            throw new PreBidException("Failed build DatabaseReader from DB", e);
+        }
+    }
+
+    @Bean
+    GreenbidsInferenceDataService greenbidsInferenceDataService(DatabaseReader dbReader) {
+        return new GreenbidsInferenceDataService(dbReader, ObjectMapperProvider.mapper());
+    }
+
+    @Bean
+    GreenbidsRealTimeDataModule greenbidsRealTimeDataModule(
+            FilterService filterService,
+            OnnxModelRunnerWithThresholds onnxModelRunnerWithThresholds,
+            GreenbidsInferenceDataService greenbidsInferenceDataService,
+            GreenbidsInvocationService greenbidsInvocationService) {
+
+        return new GreenbidsRealTimeDataModule(List.of(
+                new GreenbidsRealTimeDataProcessedAuctionRequestHook(
+                        ObjectMapperProvider.mapper(),
+                        filterService,
+                        onnxModelRunnerWithThresholds,
+                        greenbidsInferenceDataService,
+                        greenbidsInvocationService)));
+    }
+
+    @Bean
+    FilterService filterService() {
+        return new FilterService();
+    }
+
+    @Bean
+    Storage storage(GreenbidsRealTimeDataProperties properties) {
+        return StorageOptions.newBuilder()
+                .setProjectId(properties.getGoogleCloudGreenbidsProject()).build().getService();
+    }
+
+    @Bean
+    OnnxModelRunnerFactory onnxModelRunnerFactory() {
+        return new OnnxModelRunnerFactory();
+    }
+
+    @Bean
+    ThrottlingThresholdsFactory throttlingThresholdsFactory() {
+        return new ThrottlingThresholdsFactory();
+    }
+
+    @Bean
+    ModelCache modelCache(
+            GreenbidsRealTimeDataProperties properties,
+            Vertx vertx,
+            Storage storage,
+            OnnxModelRunnerFactory onnxModelRunnerFactory) {
+
+        final Cache<String, OnnxModelRunner> modelCacheWithExpiration = Caffeine.newBuilder()
+                .expireAfterWrite(properties.getCacheExpirationMinutes(), TimeUnit.MINUTES)
+                .build();
+
+        return new ModelCache(
+                storage,
+                properties.getGcsBucketName(),
+                modelCacheWithExpiration,
+                properties.getOnnxModelCacheKeyPrefix(),
+                vertx,
+                onnxModelRunnerFactory);
+    }
+
+    @Bean
+    ThresholdCache thresholdCache(
+            GreenbidsRealTimeDataProperties properties,
+            Vertx vertx,
+            Storage storage,
+            ThrottlingThresholdsFactory throttlingThresholdsFactory) {
+
+        final Cache<String, ThrottlingThresholds> thresholdsCacheWithExpiration = Caffeine.newBuilder()
+                .expireAfterWrite(properties.getCacheExpirationMinutes(), TimeUnit.MINUTES)
+                .build();
+
+        return new ThresholdCache(
+                storage,
+                properties.getGcsBucketName(),
+                ObjectMapperProvider.mapper(),
+                thresholdsCacheWithExpiration,
+                properties.getThresholdsCacheKeyPrefix(),
+                vertx,
+                throttlingThresholdsFactory);
+    }
+
+    @Bean
+    OnnxModelRunnerWithThresholds onnxModelRunnerWithThresholds(
+            ModelCache modelCache,
+            ThresholdCache thresholdCache) {
+
+        return new OnnxModelRunnerWithThresholds(modelCache, thresholdCache);
+    }
+
+    @Bean
+    GreenbidsInvocationService greenbidsInvocationService() {
+        return new GreenbidsInvocationService();
+    }
+}

extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/GreenbidsRealTimeDataModule.java

@@ -0,0 +1,29 @@
+package org.prebid.server.hooks.modules.greenbids.real.time.data.config;
+
+import org.prebid.server.hooks.v1.Hook;
+import org.prebid.server.hooks.v1.InvocationContext;
+import org.prebid.server.hooks.v1.Module;
+
+import java.util.Collection;
+import java.util.List;
+
+public class GreenbidsRealTimeDataModule implements Module {
+
+    public static final String CODE = "greenbids-real-time-data";
+
+    private final List<? extends Hook<?, ? extends InvocationContext>> hooks;
+
+    public GreenbidsRealTimeDataModule(List<? extends Hook<?, ? extends InvocationContext>> hooks) {
+        this.hooks = hooks;
+    }
+
+    @Override
+    public String code() {
+        return CODE;
+    }
+
+    @Override
+    public Collection<? extends Hook<?, ? extends InvocationContext>> hooks() {
+        return hooks;
+    }
+}

extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/GreenbidsRealTimeDataProperties.java

@@ -0,0 +1,21 @@
+package org.prebid.server.hooks.modules.greenbids.real.time.data.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "hooks.modules." + GreenbidsRealTimeDataModule.CODE)
+@Data
+public class GreenbidsRealTimeDataProperties {
+
+    String googleCloudGreenbidsProject;
+
+    String geoLiteCountryPath;
+
+    String gcsBucketName;
+
+    Integer cacheExpirationMinutes;
+
+    String onnxModelCacheKeyPrefix;
+
+    String thresholdsCacheKeyPrefix;
+}

extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/core/FilterService.java

@@ -0,0 +1,123 @@
+package org.prebid.server.hooks.modules.greenbids.real.time.data.core;
+
+import ai.onnxruntime.OnnxTensor;
+import ai.onnxruntime.OnnxValue;
+import ai.onnxruntime.OrtException;
+import ai.onnxruntime.OrtSession;
+import org.prebid.server.exception.PreBidException;
+import org.prebid.server.hooks.modules.greenbids.real.time.data.model.data.ThrottlingMessage;
+import org.springframework.util.CollectionUtils;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+public class FilterService {
+
+    public Map<String, Map<String, Boolean>> filterBidders(
+            OnnxModelRunner onnxModelRunner,
+            List<ThrottlingMessage> throttlingMessages,
+            Double threshold) {
+
+        final OrtSession.Result results;
+        try {
+            final String[][] throttlingInferenceRows = convertToArray(throttlingMessages);
+            results = onnxModelRunner.runModel(throttlingInferenceRows);
+            return processModelResults(results, throttlingMessages, threshold);
+        } catch (OrtException e) {
+            throw new PreBidException("Exception during model inference: ", e);
+        }
+    }
+
+    private static String[][] convertToArray(List<ThrottlingMessage> messages) {
+        return messages.stream()
+                .map(message -> new String[]{
+                        message.getBrowser(),
+                        message.getBidder(),
+                        message.getAdUnitCode(),
+                        message.getCountry(),
+                        message.getHostname(),
+                        message.getDevice(),
+                        message.getHourBucket(),
+                        message.getMinuteQuadrant()})
+                .toArray(String[][]::new);
+    }
+
+    private Map<String, Map<String, Boolean>> processModelResults(
+            OrtSession.Result results,
+            List<ThrottlingMessage> throttlingMessages,
+            Double threshold) {
+
+        validateThrottlingMessages(throttlingMessages);
+
+        return StreamSupport.stream(results.spliterator(), false)
+                .peek(FilterService::validateOnnxTensor)
+                .filter(onnxItem -> Objects.equals(onnxItem.getKey(), "probabilities"))
+                .map(Map.Entry::getValue)
+                .map(OnnxTensor.class::cast)
+                .peek(tensor -> validateTensorSize(tensor, throttlingMessages.size()))
+                .map(tensor -> extractAndProcessProbabilities(tensor, throttlingMessages, threshold))
+                .map(Map::entrySet)
+                .flatMap(Collection::stream)
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
+    private static void validateThrottlingMessages(List<ThrottlingMessage> throttlingMessages) {
+        if (throttlingMessages == null || CollectionUtils.isEmpty(throttlingMessages)) {
+            throw new PreBidException("throttlingMessages cannot be null or empty");
+        }
+    }
+
+    private static void validateOnnxTensor(Map.Entry<String, OnnxValue> onnxItem) {
+        if (!(onnxItem.getValue() instanceof OnnxTensor)) {
+            throw new PreBidException("Expected OnnxTensor for 'probabilities', but found: "
+                    + onnxItem.getValue().getClass().getName());
+        }
+    }
+
+    private static void validateTensorSize(OnnxTensor tensor, int expectedSize) {
+        final long[] tensorShape = tensor.getInfo().getShape();
+        if (tensorShape.length == 0 || tensorShape[0] != expectedSize) {
+            throw new PreBidException("Mismatch between tensor size and throttlingMessages size");
+        }
+    }
+
+    private Map<String, Map<String, Boolean>> extractAndProcessProbabilities(
+            OnnxTensor tensor,
+            List<ThrottlingMessage> throttlingMessages,
+            Double threshold) {
+
+        try {
+            final float[][] probabilities = extractProbabilitiesValues(tensor);
+            return processProbabilities(probabilities, throttlingMessages, threshold);
+        } catch (OrtException e) {
+            throw new PreBidException("Exception when extracting proba from OnnxTensor: ", e);
+        }
+    }
+
+    private float[][] extractProbabilitiesValues(OnnxTensor tensor) throws OrtException {
+        return (float[][]) tensor.getValue();
+    }
+
+    private Map<String, Map<String, Boolean>> processProbabilities(
+            float[][] probabilities,
+            List<ThrottlingMessage> throttlingMessages,
+            Double threshold) {
+
+        final Map<String, Map<String, Boolean>> result = new HashMap<>();
+
+        for (int i = 0; i < probabilities.length; i++) {
+            final ThrottlingMessage message = throttlingMessages.get(i);
+            final String impId = message.getAdUnitCode();
+            final String bidder = message.getBidder();
+            final boolean isKeptInAuction = probabilities[i][1] > threshold;
+            result.computeIfAbsent(impId, k -> new HashMap<>()).put(bidder, isKeptInAuction);
+        }
+
+        return result;
+    }
+}