traffic calculator with hardcoded baseline

src/main/java/com/uid2/optout/vertx/OptOutTrafficCalculator.java

@@ -179,6 +179,20 @@ List<List<Long>> parseAllowlistRanges(JsonObject config) throws MalformedTraffic
 
             ranges.sort(Comparator.comparing(range -> range.get(0)));
 
+            // Validate no overlapping ranges
+            for (int i = 0; i < ranges.size() - 1; i++) {
+                long currentEnd = ranges.get(i).get(1);
+                long nextStart = ranges.get(i + 1).get(0);
+                if (currentEnd >= nextStart) {
+                    LOGGER.error("Overlapping allowlist ranges detected: [{}, {}] overlaps with [{}, {}]",
+                            ranges.get(i).get(0), currentEnd, nextStart, ranges.get(i + 1).get(1));
+                    throw new MalformedTrafficCalcConfigException(
+                            "Overlapping allowlist ranges detected at indices " + i + " and " + (i + 1));
+                }
+            }
+
+        } catch (MalformedTrafficCalcConfigException e) {
+            throw e;
         } catch (Exception e) {
             LOGGER.error("Failed to parse allowlist ranges", e);
             throw new MalformedTrafficCalcConfigException("Failed to parse allowlist ranges: " + e.getMessage());
@@ -215,8 +229,10 @@ public TrafficStatus calculateStatus(List<Message> sqsMessages) {
             long oldestQueueTs = findOldestQueueTimestamp(sqsMessages);
             LOGGER.info("Traffic calculation: oldestQueueTs={}", oldestQueueTs);
 
-            // Define start time of the delta evaluation window [newestDeltaTs - 24h, newestDeltaTs]
-            long deltaWindowStart = newestDeltaTs - this.evaluationWindowSeconds - getAllowlistDuration(newestDeltaTs, newestDeltaTs - this.evaluationWindowSeconds);
+            // Define start time of the delta evaluation window
+            // We need evaluationWindowSeconds of non-allowlisted time, so we iteratively extend
+            // the window to account for any allowlist ranges in the extended portion
+            long deltaWindowStart = calculateWindowStartWithAllowlist(newestDeltaTs, this.evaluationWindowSeconds);
 
             // Evict old cache entries (older than delta window start)
             evictOldCacheEntries(deltaWindowStart);
@@ -391,6 +407,31 @@ long getAllowlistDuration(long t, long windowStart) {
         return totalDuration;
     }
 
+    /**
+     * Calculate the window start time that provides evaluationWindowSeconds of non-allowlisted time.
+     * Iteratively extends the window to account for allowlist ranges that may fall in extended portions.
+     */
+    long calculateWindowStartWithAllowlist(long newestDeltaTs, int evaluationWindowSeconds) {
+        long allowlistDuration = getAllowlistDuration(newestDeltaTs, newestDeltaTs - evaluationWindowSeconds);
+
+        // Each iteration discovers at least one new allowlist range, so max iterations = number of ranges
+        int maxIterations = this.allowlistRanges.size() + 1;
+
+        for (int i = 0; i < maxIterations && allowlistDuration > 0; i++) {
+            long newWindowStart = newestDeltaTs - evaluationWindowSeconds - allowlistDuration;
+            long newAllowlistDuration = getAllowlistDuration(newestDeltaTs, newWindowStart);
+
+            if (newAllowlistDuration == allowlistDuration) {
+                // No new allowlist time in extended portion, we've converged
+                break;
+            }
+
+            allowlistDuration = newAllowlistDuration;
+        }
+
+        return newestDeltaTs - evaluationWindowSeconds - allowlistDuration;
+    }
+
     /**
      * Find the oldest SQS queue message timestamp
      */

src/main/java/com/uid2/optout/vertx/OptOutTrafficFilter.java

@@ -0,0 +1,167 @@
+package com.uid2.optout.vertx;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Collections;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.charset.StandardCharsets;
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.json.JsonArray;
+
+public class OptOutTrafficFilter {
+    private static final Logger LOGGER = LoggerFactory.getLogger(OptOutTrafficFilter.class);
+
+    private final String trafficFilterConfigPath;
+    List<TrafficFilterRule> filterRules;
+
+    /**
+     * Traffic filter rule defining a time range and a list of IP addresses to exclude
+     */
+    private static class TrafficFilterRule {
+        private final List<Long> range;
+        private final List<String> ipAddresses;
+
+        TrafficFilterRule(List<Long> range, List<String> ipAddresses) {
+            this.range = range;
+            this.ipAddresses = ipAddresses;
+        }
+
+        public long getRangeStart() {
+            return range.get(0);
+        }
+        public long getRangeEnd() {
+            return range.get(1);
+        }
+        public List<String> getIpAddresses() {
+            return ipAddresses;
+        }
+    }
+
+    public static class MalformedTrafficFilterConfigException extends Exception {
+        public MalformedTrafficFilterConfigException(String message) {
+            super(message);
+        }
+    }
+
+    /**
+     * Constructor for OptOutTrafficFilter
+     * 
+     * @param trafficFilterConfigPath S3 path for traffic filter config
+     * @throws MalformedTrafficFilterConfigException if the traffic filter config is invalid
+     */
+    public OptOutTrafficFilter(String trafficFilterConfigPath) throws MalformedTrafficFilterConfigException {
+        this.trafficFilterConfigPath = trafficFilterConfigPath;
+        // Initial filter rules load
+        this.filterRules = Collections.emptyList(); // start empty
+        reloadTrafficFilterConfig(); // load ConfigMap
+
+        LOGGER.info("OptOutTrafficFilter initialized: filterRules={}", 
+        filterRules.size());
+    }
+
+    /**
+     * Reload traffic filter config from ConfigMap.
+     * Expected format:
+	 * {
+	 *   "blacklist_requests": [
+	 *     {range: [startTimestamp, endTimestamp], IPs: ["ip1"]},
+	 *     {range: [startTimestamp, endTimestamp], IPs: ["ip1", "ip2"]},
+	 *     {range: [startTimestamp, endTimestamp], IPs: ["ip1", "ip3"]},
+	 *   ]
+	 * }
+     * 
+     * Can be called periodically to pick up config changes without restarting.
+     */
+    public void reloadTrafficFilterConfig() throws MalformedTrafficFilterConfigException {
+        LOGGER.info("Loading traffic filter config from ConfigMap");
+        try (InputStream is = Files.newInputStream(Paths.get(trafficFilterConfigPath))) {
+            String content = new String(is.readAllBytes(), StandardCharsets.UTF_8);
+            JsonObject filterConfigJson = new JsonObject(content);
+
+            this.filterRules = parseFilterRules(filterConfigJson);
+
+            LOGGER.info("Successfully loaded traffic filter config from ConfigMap: filterRules={}",
+                       filterRules.size());
+
+        } catch (Exception e) {
+            LOGGER.warn("No traffic filter config found at: {}", trafficFilterConfigPath, e);
+            throw new MalformedTrafficFilterConfigException(e.getMessage());
+        } 
+    }
+
+    /**
+     * Parse request filtering rules from JSON config
+     */
+    List<TrafficFilterRule> parseFilterRules(JsonObject config) throws MalformedTrafficFilterConfigException {
+        List<TrafficFilterRule> rules = new ArrayList<>();
+        try {
+            JsonArray blacklistRequests = config.getJsonArray("blacklist_requests");
+            if (blacklistRequests == null) {
+                LOGGER.error("Invalid traffic filter config: blacklist_requests is null");
+                throw new MalformedTrafficFilterConfigException("Invalid traffic filter config: blacklist_requests is null");
+            }
+            for (int i = 0; i < blacklistRequests.size(); i++) {
+                JsonObject ruleJson = blacklistRequests.getJsonObject(i);
+
+                // parse range
+                var rangeJson = ruleJson.getJsonArray("range");
+                List<Long> range = new ArrayList<>();
+                if (rangeJson != null && rangeJson.size() == 2) {
+                    long start = rangeJson.getLong(0);
+                    long end = rangeJson.getLong(1);
+
+                    if (start >= end) {
+                        LOGGER.error("Invalid traffic filter rule: range start must be less than end: {}", ruleJson.encode());
+                        throw new MalformedTrafficFilterConfigException("Invalid traffic filter rule: range start must be less than end");
+                    }
+                    range.add(start);
+                    range.add(end);
+                }
+
+                // parse IPs
+                var ipAddressesJson = ruleJson.getJsonArray("IPs");
+                List<String> ipAddresses = new ArrayList<>();
+                if (ipAddressesJson != null) {
+                    for (int j = 0; j < ipAddressesJson.size(); j++) {
+                        ipAddresses.add(ipAddressesJson.getString(j));
+                    }
+                }
+
+                // log error and throw exception if rule is invalid
+                if (range.size() != 2 || ipAddresses.size() == 0 || range.get(1) - range.get(0) > 86400) { // range must be 24 hours or less
+                    LOGGER.error("Invalid traffic filter rule: {}", ruleJson.encode());
+                    throw new MalformedTrafficFilterConfigException("Invalid traffic filter rule");
+                }
+
+                TrafficFilterRule rule = new TrafficFilterRule(range, ipAddresses);
+
+                LOGGER.info("Loaded traffic filter rule: range=[{}, {}], IPs={}", rule.getRangeStart(), rule.getRangeEnd(), rule.getIpAddresses());
+                rules.add(rule);
+            }
+            return rules;
+        } catch (Exception e) {
+            LOGGER.error("Failed to parse traffic filter rules: config={}, error={}", config.encode(), e.getMessage());
+            throw new MalformedTrafficFilterConfigException(e.getMessage());
+        }
+    }
+
+    public boolean isBlacklisted(SqsParsedMessage message) {
+        long timestamp = message.getTimestamp();
+        String clientIp = message.getClientIp();
+
+        for (TrafficFilterRule rule : filterRules) {
+            if(timestamp >= rule.getRangeStart() && timestamp <= rule.getRangeEnd()) {
+                if(rule.getIpAddresses().contains(clientIp)) {
+                    return true;
+                }
+            };
+        }
+        return false;
+    }
+
+}

src/test/java/com/uid2/optout/vertx/OptOutTrafficCalculatorTest.java

@@ -319,6 +319,61 @@ void testParseTrafficCalcConfigRanges_nullArray() throws Exception {
         assertTrue(result.isEmpty());
     }
 
+    @Test
+    void testParseTrafficCalcConfigRanges_overlappingRanges() throws Exception {
+        // Setup - overlapping ranges
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        JsonObject configWithRanges = new JsonObject();
+        JsonArray ranges = new JsonArray()
+            .add(new JsonArray().add(1000L).add(2000L))
+            .add(new JsonArray().add(1500L).add(2500L)); // Overlaps with first range
+        configWithRanges.put("traffic_calc_allowlist_ranges", ranges);
+
+        // Act & Assert - should throw exception due to overlap
+        assertThrows(MalformedTrafficCalcConfigException.class, () -> {
+            calculator.parseAllowlistRanges(configWithRanges);
+        });
+    }
+
+    @Test
+    void testParseTrafficCalcConfigRanges_adjacentRangesWithSameBoundary() throws Exception {
+        // Setup - ranges where end of first equals start of second (touching but not overlapping semantically, but we treat as overlap)
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        JsonObject configWithRanges = new JsonObject();
+        JsonArray ranges = new JsonArray()
+            .add(new JsonArray().add(1000L).add(2000L))
+            .add(new JsonArray().add(2000L).add(3000L)); // Starts exactly where first ends
+        configWithRanges.put("traffic_calc_allowlist_ranges", ranges);
+
+        // Act & Assert - should throw exception because ranges touch at boundary
+        assertThrows(MalformedTrafficCalcConfigException.class, () -> {
+            calculator.parseAllowlistRanges(configWithRanges);
+        });
+    }
+
+    @Test
+    void testParseTrafficCalcConfigRanges_nonOverlappingRanges() throws Exception {
+        // Setup - ranges that don't overlap (with gap between them)
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        JsonObject configWithRanges = new JsonObject();
+        JsonArray ranges = new JsonArray()
+            .add(new JsonArray().add(1000L).add(2000L))
+            .add(new JsonArray().add(2001L).add(3000L)); // Starts after first ends
+        configWithRanges.put("traffic_calc_allowlist_ranges", ranges);
+
+        // Act
+        List<List<Long>> result = calculator.parseAllowlistRanges(configWithRanges);
+
+        // Assert - should succeed with 2 ranges
+        assertEquals(2, result.size());
+    }
+
     // ============================================================================
     // SECTION 3: isInTrafficCalcConfig()
     // ============================================================================
@@ -657,6 +712,101 @@ void testGetTrafficCalcConfigDuration_rangeSpansEntireWindow() throws Exception
         assertEquals(5000L, duration);
     }
 
+    // ============================================================================
+    // SECTION 4.5: calculateWindowStartWithAllowlist()
+    // ============================================================================
+
+    @Test
+    void testCalculateWindowStartWithAllowlist_noAllowlist() throws Exception {
+        // Setup - no allowlist ranges
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        // Act - window should be [3, 8] with no extension
+        long windowStart = calculator.calculateWindowStartWithAllowlist(8L, 5);
+
+        // Assert - no allowlist, so window start is simply newestDeltaTs - evaluationWindowSeconds
+        assertEquals(3L, windowStart);
+    }
+
+    @Test
+    void testCalculateWindowStartWithAllowlist_allowlistInOriginalWindowOnly() throws Exception {
+        // Setup - allowlist range only in original window, not in extended portion
+        String trafficCalcConfigJson = """
+            {
+                "traffic_calc_allowlist_ranges": [
+                    [6, 7]
+                ]
+            }
+            """;
+        createConfigFromPartialJson(trafficCalcConfigJson);
+
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        // Act - newestDeltaTs=8, evaluationWindow=5
+        // Original window [3, 8] has [6,7] allowlisted (1 hour)
+        // Extended portion [2, 3] has no allowlist
+        // So window start should be 8 - 5 - 1 = 2
+        long windowStart = calculator.calculateWindowStartWithAllowlist(8L, 5);
+
+        assertEquals(2L, windowStart);
+    }
+
+    @Test
+    void testCalculateWindowStartWithAllowlist_allowlistInExtendedPortion() throws Exception {
+        // Setup - allowlist ranges in both original window AND extended portion
+        // This is the user's example: evaluationWindow=5, newestDeltaTs=8, allowlist={[2,3], [6,7]}
+        String trafficCalcConfigJson = """
+            {
+                "traffic_calc_allowlist_ranges": [
+                    [2, 3],
+                    [6, 7]
+                ]
+            }
+            """;
+        createConfigFromPartialJson(trafficCalcConfigJson);
+
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        // Act
+        // Original window [3, 8]: [6,7] allowlisted = 1 hour
+        // First extension to [2, 8]: [2,3] and [6,7] allowlisted = 2 hours total
+        // Second extension to [1, 8]: still [2,3] and [6,7] = 2 hours (no new allowlist)
+        // Final: windowStart = 8 - 5 - 2 = 1
+        long windowStart = calculator.calculateWindowStartWithAllowlist(8L, 5);
+
+        assertEquals(1L, windowStart);
+    }
+
+    @Test
+    void testCalculateWindowStartWithAllowlist_allowlistBeforeWindow() throws Exception {
+        // Setup - allowlist range entirely before the initial window
+        // This tests that we don't over-extend when allowlist is old
+        // evaluationWindow=5, newestDeltaTs=20, allowlist=[10,13]
+        String trafficCalcConfigJson = """
+            {
+                "traffic_calc_allowlist_ranges": [
+                    [10, 13]
+                ]
+            }
+            """;
+        createConfigFromPartialJson(trafficCalcConfigJson);
+
+        OptOutTrafficCalculator calculator = new OptOutTrafficCalculator(
+            cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH);
+
+        // Act
+        // Initial window [15, 20]: no allowlist overlap, allowlistDuration = 0
+        // No extension needed
+        // Final: windowStart = 20 - 5 - 0 = 15
+        long windowStart = calculator.calculateWindowStartWithAllowlist(20L, 5);
+
+        // Verify: window [15, 20] has 5 hours, 0 allowlisted = 5 non-allowlisted
+        assertEquals(15L, windowStart);
+    }
+
     // ============================================================================
     // SECTION 5: determineStatus()
     // ============================================================================