diff --git a/src/main/java/com/uid2/optout/vertx/OptOutTrafficCalculator.java b/src/main/java/com/uid2/optout/vertx/OptOutTrafficCalculator.java new file mode 100644 index 0000000..ff32b9f --- /dev/null +++ b/src/main/java/com/uid2/optout/vertx/OptOutTrafficCalculator.java @@ -0,0 +1,499 @@ +package com.uid2.optout.vertx; + +import com.uid2.shared.cloud.ICloudStorage; +import com.uid2.shared.optout.OptOutCollection; +import com.uid2.shared.optout.OptOutEntry; +import com.uid2.shared.optout.OptOutUtils; +import io.vertx.core.json.JsonObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.sqs.model.Message; +import software.amazon.awssdk.services.sqs.model.MessageSystemAttributeName; + +import java.nio.charset.StandardCharsets; + +import java.nio.file.Files; +import java.nio.file.Paths; +import java.io.InputStream; +import java.io.IOException; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +/** + * Calculates opt-out traffic patterns to determine DEFAULT or DELAYED_PROCESSING status. + * + * Compares recent ~24h traffic (sumCurrent) against previous ~24h baseline (sumPast). + * Both sums exclude records in whitelist ranges (surge windows determined by engineers). + * + * Returns DELAYED_PROCESSING if sumCurrent >= 5 × sumPast, indicating abnormal traffic spike. + */ +public class OptOutTrafficCalculator { + private static final Logger LOGGER = LoggerFactory.getLogger(OptOutTrafficCalculator.class); + + private static final int HOURS_24 = 24 * 3600; // 24 hours in seconds + private static final int DEFAULT_THRESHOLD_MULTIPLIER = 5; + + private final Map deltaFileCache = new ConcurrentHashMap<>(); + private final ICloudStorage cloudStorage; + private final String s3DeltaPrefix; // (e.g. "optout-v2/delta/") + private final String trafficCalcConfigPath; + private int thresholdMultiplier; + private int currentEvaluationWindowSeconds; + private int previousEvaluationWindowSeconds; + private List> whitelistRanges; + + public enum TrafficStatus { + DELAYED_PROCESSING, + DEFAULT + } + + /** + * Cache entry for a delta file containing all record timestamps. + * + * Memory usage: ~8 bytes per timestamp (long) + * 1GB of memory can store ~130 million timestamps (1024^3)/8 + */ + private static class FileRecordCache { + final List timestamps; // All non-sentinel record timestamps + final long newestTimestamp; // evict delta from cache based on oldest record timestamp + + FileRecordCache(List timestamps) { + this.timestamps = timestamps; + this.newestTimestamp = timestamps.isEmpty() ? 0 : Collections.max(timestamps); + } + } + + /** + * Exception thrown by malformed traffic calculator config + */ + public static class MalformedTrafficCalcConfigException extends Exception { + public MalformedTrafficCalcConfigException(String message) { + super(message); + } + } + + /** + * Constructor for OptOutTrafficCalculator + * + * @param cloudStorage Cloud storage for reading delta files and whitelist from S3 + * @param s3DeltaPrefix S3 prefix for delta files + * @param trafficCalcConfigS3Path S3 path for traffic calc config + */ + public OptOutTrafficCalculator(ICloudStorage cloudStorage, String s3DeltaPrefix, String trafficCalcConfigPath) throws MalformedTrafficCalcConfigException { + this.cloudStorage = cloudStorage; + this.thresholdMultiplier = DEFAULT_THRESHOLD_MULTIPLIER; + this.s3DeltaPrefix = s3DeltaPrefix; + this.trafficCalcConfigPath = trafficCalcConfigPath; + this.currentEvaluationWindowSeconds = HOURS_24 - 300; //23h55m (5m queue window) + this.previousEvaluationWindowSeconds = HOURS_24; //24h + // Initial whitelist load + this.whitelistRanges = Collections.emptyList(); // Start empty + reloadTrafficCalcConfig(); // Load ConfigMap + + LOGGER.info("OptOutTrafficCalculator initialized: s3DeltaPrefix={}, threshold={}x", + s3DeltaPrefix, thresholdMultiplier); + } + + /** + * Reload traffic calc config from ConfigMap. + * Expected format: + * { + * "traffic_calc_current_evaluation_window_seconds": 86400, + * "traffic_calc_previous_evaluation_window_seconds": 86400, + * "traffic_calc_whitelist_ranges": [ + * [startTimestamp1, endTimestamp1], + * [startTimestamp2, endTimestamp2] + * ], + * "traffic_calc_threshold_multiplier": 5 + * } + * + * Can be called periodically to pick up config changes without restarting. + */ + public void reloadTrafficCalcConfig() throws MalformedTrafficCalcConfigException { + LOGGER.info("Loading traffic calc config from ConfigMap"); + try (InputStream is = Files.newInputStream(Paths.get(trafficCalcConfigPath))) { + String content = new String(is.readAllBytes(), StandardCharsets.UTF_8); + JsonObject whitelistConfig = new JsonObject(content); + + this.thresholdMultiplier = whitelistConfig.getInteger("traffic_calc_threshold_multiplier", DEFAULT_THRESHOLD_MULTIPLIER); + this.currentEvaluationWindowSeconds = whitelistConfig.getInteger("traffic_calc_current_evaluation_window_seconds", HOURS_24 - 300); + this.previousEvaluationWindowSeconds = whitelistConfig.getInteger("traffic_calc_previous_evaluation_window_seconds", HOURS_24); + + List> ranges = parseWhitelistRanges(whitelistConfig); + this.whitelistRanges = ranges; + + LOGGER.info("Successfully loaded traffic calc config from ConfigMap: currentEvaluationWindowSeconds={}, previousEvaluationWindowSeconds={}, whitelistRanges={}", + this.currentEvaluationWindowSeconds, this.previousEvaluationWindowSeconds, ranges.size()); + + } catch (Exception e) { + LOGGER.warn("Failed to load traffic calc config. Config is malformed or missing: {}", trafficCalcConfigPath, e); + throw new MalformedTrafficCalcConfigException("Failed to load traffic calc config: " + e.getMessage()); + } + } + + /** + * Parse whitelist ranges from JSON config + */ + List> parseWhitelistRanges(JsonObject config) throws MalformedTrafficCalcConfigException { + List> ranges = new ArrayList<>(); + + try { + if (config.containsKey("traffic_calc_whitelist_ranges")) { + var rangesArray = config.getJsonArray("traffic_calc_whitelist_ranges"); + if (rangesArray != null) { + for (int i = 0; i < rangesArray.size(); i++) { + var rangeArray = rangesArray.getJsonArray(i); + if (rangeArray != null && rangeArray.size() >= 2) { + long start = rangeArray.getLong(0); + long end = rangeArray.getLong(1); + + if(start >= end) { + LOGGER.error("Invalid whitelist range: start must be less than end: [{}, {}]", start, end); + throw new MalformedTrafficCalcConfigException("Invalid whitelist range at index " + i + ": start must be less than end"); + } + + if (end - start > 86400) { + LOGGER.error("Invalid whitelist range: range must be less than 24 hours: [{}, {}]", start, end); + throw new MalformedTrafficCalcConfigException("Invalid whitelist range at index " + i + ": range must be less than 24 hours"); + } + + List range = Arrays.asList(start, end); + ranges.add(range); + LOGGER.info("Loaded whitelist range: [{}, {}]", start, end); + } + } + } + } + + // Sort ranges by start time for efficient lookups + ranges.sort(Comparator.comparing(range -> range.get(0))); + + } catch (Exception e) { + LOGGER.error("Failed to parse whitelist ranges", e); + throw new MalformedTrafficCalcConfigException("Failed to parse whitelist ranges: " + e.getMessage()); + } + + return ranges; + } + + /** + * Calculate traffic status based on delta files and SQS queue messages. + * + * @param sqsMessages List of SQS messages + * @return TrafficStatus (DELAYED_PROCESSING or DEFAULT) + */ + public TrafficStatus calculateStatus(List sqsMessages) { + + try { + // Get list of delta files from S3 (sorted newest to oldest) + List deltaS3Paths = listDeltaFiles(); + + if (deltaS3Paths.isEmpty()) { + LOGGER.warn("No delta files found in S3 with prefix: {}", s3DeltaPrefix); + return TrafficStatus.DEFAULT; + } + + // Find t = oldest SQS queue message timestamp + long t = findOldestQueueTimestamp(sqsMessages); + LOGGER.info("Traffic calculation starting with t={} (oldest SQS message)", t); + + // Define time windows + long currentWindowStart = t - this.currentEvaluationWindowSeconds - getWhitelistDuration(t, t - this.currentEvaluationWindowSeconds); // for range [t-23h55m, t+5m] + long pastWindowStart = currentWindowStart - this.previousEvaluationWindowSeconds - getWhitelistDuration(currentWindowStart, currentWindowStart - this.previousEvaluationWindowSeconds); // for range [t-47h55m, t-23h55m] + + // Evict old cache entries (older than past window start) + evictOldCacheEntries(pastWindowStart); + + // Process delta files and count + int sumCurrent = 0; + int sumPast = 0; + + for (String s3Path : deltaS3Paths) { + List timestamps = getTimestampsFromFile(s3Path); + + boolean shouldStop = false; + for (long ts : timestamps) { + // Stop condition: record is older than our 48h window + if (ts < pastWindowStart) { + LOGGER.debug("Stopping delta file processing at timestamp {} (older than t-48h)", ts); + break; + } + + // skip records in whitelisted ranges + if (isInWhitelist(ts)) { + continue; + } + + // Count for sumCurrent: [t-24h, t] + if (ts >= currentWindowStart && ts <= t) { + sumCurrent++; + } + + // Count for sumPast: [t-48h, t-24h] + if (ts >= pastWindowStart && ts < currentWindowStart) { + sumPast++; + } + } + + if (shouldStop) { + break; + } + } + + // Count SQS messages in [t, t+5m] + if (sqsMessages != null && !sqsMessages.isEmpty()) { + int sqsCount = countSqsMessages( + sqsMessages, t); + sumCurrent += sqsCount; + } + + // Determine status + TrafficStatus status = determineStatus(sumCurrent, sumPast); + + LOGGER.info("Traffic calculation complete: sumCurrent={}, sumPast={}, status={}", + sumCurrent, sumPast, status); + + return status; + + } catch (Exception e) { + LOGGER.error("Error calculating traffic status", e); + return TrafficStatus.DEFAULT; + } + } + + /** + * List all delta files from S3, sorted newest to oldest + */ + private List listDeltaFiles() { + try { + // List all objects with the delta prefix + List allFiles = cloudStorage.list(s3DeltaPrefix); + + // Filter to only .dat delta files and sort newest to oldest + return allFiles.stream() + .filter(OptOutUtils::isDeltaFile) + .sorted(OptOutUtils.DeltaFilenameComparatorDescending) + .collect(Collectors.toList()); + + } catch (Exception e) { + LOGGER.error("Failed to list delta files from S3 with prefix: {}", s3DeltaPrefix, e); + return Collections.emptyList(); + } + } + + /** + * Get timestamps from a delta file (S3 path), using cache if available + */ + private List getTimestampsFromFile(String s3Path) throws IOException { + // Extract filename from S3 path for cache key + String filename = s3Path.substring(s3Path.lastIndexOf('/') + 1); + + // Check cache first + FileRecordCache cached = deltaFileCache.get(filename); + if (cached != null) { + LOGGER.debug("Using cached timestamps for file: {}", filename); + return cached.timestamps; + } + + // Cache miss - download from S3 + LOGGER.debug("Downloading and reading timestamps from S3: {}", s3Path); + List timestamps = readTimestampsFromS3(s3Path); + + // Store in cache + deltaFileCache.put(filename, new FileRecordCache(timestamps)); + + return timestamps; + } + + /** + * Read all non-sentinel record timestamps from a delta file in S3 + */ + private List readTimestampsFromS3(String s3Path) throws IOException { + try (InputStream is = cloudStorage.download(s3Path)) { + byte[] data = is.readAllBytes(); + OptOutCollection collection = new OptOutCollection(data); + + List timestamps = new ArrayList<>(); + for (int i = 0; i < collection.size(); i++) { + OptOutEntry entry = collection.get(i); + + // Skip sentinel entries + if (entry.isSpecialHash()) { + continue; + } + + timestamps.add(entry.timestamp); + } + + return timestamps; + } catch (Exception e) { + LOGGER.error("Failed to read delta file from S3: {}", s3Path, e); + throw new IOException("Failed to read delta file from S3: " + s3Path, e); + } + } + + /** + * Calculate total duration of whitelist ranges that overlap with the given time window. + */ + long getWhitelistDuration(long t, long windowStart) { + long totalDuration = 0; + for (List range : this.whitelistRanges) { + long start = range.get(0); + long end = range.get(1); + + // Clip range to window boundaries + if (start < windowStart) { + start = windowStart; + } + if (end > t) { + end = t; + } + + // Only add duration if there's actual overlap (start < end) + if (start < end) { + totalDuration += end - start; + } + } + return totalDuration; + } + + /** + * Find the oldest SQS queue message timestamp + */ + private long findOldestQueueTimestamp(List sqsMessages) throws IOException { + long oldest = System.currentTimeMillis() / 1000; + + if (sqsMessages != null && !sqsMessages.isEmpty()) { + for (Message msg : sqsMessages) { + Long ts = extractTimestampFromMessage(msg); + if (ts != null && ts < oldest) { + oldest = ts; + } + } + } + + return oldest; + } + + /** + * Extract timestamp from SQS message (from SentTimestamp attribute) + */ + private Long extractTimestampFromMessage(Message msg) { + // Get SentTimestamp attribute (milliseconds) + String sentTimestamp = msg.attributes().get(MessageSystemAttributeName.SENT_TIMESTAMP); + if (sentTimestamp != null) { + try { + return Long.parseLong(sentTimestamp) / 1000; // Convert ms to seconds + } catch (NumberFormatException e) { + LOGGER.debug("Invalid SentTimestamp: {}", sentTimestamp); + } + } + + // Fallback: use current time + return System.currentTimeMillis() / 1000; + } + + /** + * Count SQS messages from t to t+5 minutes + */ + private int countSqsMessages(List sqsMessages, long t) { + + int count = 0; + + for (Message msg : sqsMessages) { + Long ts = extractTimestampFromMessage(msg); + + if (ts < t || ts > t + 5 * 60) { + continue; + } + + if (isInWhitelist(ts)) { + continue; + } + count++; + + } + + LOGGER.info("SQS messages: {} in window [t={}, t+5(minutes)={}]", count, t, t + 5 * 60); + return count; + } + + /** + * Check if a timestamp falls within any whitelist range + */ + boolean isInWhitelist(long timestamp) { + if (whitelistRanges == null || whitelistRanges.isEmpty()) { + return false; + } + + for (List range : whitelistRanges) { + if (range.size() < 2) { + continue; + } + + long start = range.get(0); + long end = range.get(1); + + if (timestamp >= start && timestamp <= end) { + return true; + } + } + + return false; + } + + /** + * Evict cache entries with data older than the cutoff timestamp + */ + private void evictOldCacheEntries(long cutoffTimestamp) { + int beforeSize = deltaFileCache.size(); + + deltaFileCache.entrySet().removeIf(entry -> + entry.getValue().newestTimestamp < cutoffTimestamp + ); + + int afterSize = deltaFileCache.size(); + if (beforeSize != afterSize) { + LOGGER.info("Evicted {} old cache entries (before={}, after={})", + beforeSize - afterSize, beforeSize, afterSize); + } + } + + /** + * Determine traffic status based on current vs past counts + */ + TrafficStatus determineStatus(int sumCurrent, int sumPast) { + if (sumPast == 0) { + // Avoid division by zero - if no baseline traffic, return DEFAULT status + LOGGER.warn("sumPast is 0, cannot detect thresholdcrossing. Returning DEFAULT status."); + return TrafficStatus.DEFAULT; + } + + if (sumCurrent >= thresholdMultiplier * sumPast) { + LOGGER.warn("DELAYED_PROCESSING threshold breached: sumCurrent={} >= {}×sumPast={}", + sumCurrent, thresholdMultiplier, sumPast); + return TrafficStatus.DELAYED_PROCESSING; + } + + LOGGER.info("Traffic within normal range: sumCurrent={} < {}×sumPast={}", + sumCurrent, thresholdMultiplier, sumPast); + return TrafficStatus.DEFAULT; + } + + /** + * Get cache statistics for monitoring + */ + public Map getCacheStats() { + Map stats = new HashMap<>(); + stats.put("cached_files", deltaFileCache.size()); + + int totalTimestamps = deltaFileCache.values().stream() + .mapToInt(cache -> cache.timestamps.size()) + .sum(); + stats.put("total_cached_timestamps", totalTimestamps); + + return stats; + } + +} diff --git a/src/test/java/com/uid2/optout/vertx/OptOutTrafficCalculatorTest.java b/src/test/java/com/uid2/optout/vertx/OptOutTrafficCalculatorTest.java new file mode 100644 index 0000000..f35b7bf --- /dev/null +++ b/src/test/java/com/uid2/optout/vertx/OptOutTrafficCalculatorTest.java @@ -0,0 +1,1399 @@ +package com.uid2.optout.vertx; + +import com.uid2.shared.cloud.CloudStorageException; +import com.uid2.shared.cloud.ICloudStorage; +import com.uid2.shared.optout.OptOutCollection; +import com.uid2.shared.optout.OptOutEntry; +import io.vertx.core.json.JsonArray; +import io.vertx.core.json.JsonObject; +import java.nio.file.Files; +import java.nio.file.Path; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; +import software.amazon.awssdk.services.sqs.model.Message; +import software.amazon.awssdk.services.sqs.model.MessageSystemAttributeName; + +import com.uid2.optout.vertx.OptOutTrafficCalculator.MalformedTrafficCalcConfigException; +import java.io.ByteArrayInputStream; +import java.util.*; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + + +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +public class OptOutTrafficCalculatorTest { + + @Mock + private ICloudStorage cloudStorage; + + private static final String S3_DELTA_PREFIX = "optout-v2/delta/"; + private static final String TRAFFIC_CONFIG_PATH = "./traffic-config.json"; + + @BeforeEach + void setUp() { + try { + createTrafficConfigFile("{}"); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @AfterEach + void tearDown() { + if (Files.exists(Path.of(TRAFFIC_CONFIG_PATH))) { + try { + Files.delete(Path.of(TRAFFIC_CONFIG_PATH)); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + } + + private void createTrafficConfigFile(String content) { + try { + Path configPath = Path.of(TRAFFIC_CONFIG_PATH); + Files.writeString(configPath, content); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + // ============================================================================ + // SECTION 1: Constructor & Initialization Tests + // ============================================================================ + + @Test + void testConstructor_defaultThreshold() throws Exception { + // Setup - default threshold of 5 + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - DEFAULT when below threshold, DELAYED_PROCESSING when above threshold + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(10, 3); + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); // 10 < 5*3 + + status = calculator.determineStatus(15, 3); + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); // 15 >= 5*3 + } + + @Test + void testConstructor_customThreshold() throws Exception { + // Setup - custom threshold of 10 + createTrafficConfigFile("{\"traffic_calc_threshold_multiplier\": 10}"); + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - DEFAULT when below threshold, DELAYED_PROCESSING when above threshold + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(49, 5); + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); // 49 < 10*5 + status = calculator.determineStatus(50, 5); + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); // 50 >= 10*5 + } + + @Test + void testConstructor_whitelistLoadFailure() throws Exception { + // Setup - whitelist load failure + createTrafficConfigFile("Invalid JSON"); + assertThrows(MalformedTrafficCalcConfigException.class, () -> { + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + }); + + createTrafficConfigFile("{}"); + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + createTrafficConfigFile("Invalid JSON"); + assertThrows(MalformedTrafficCalcConfigException.class, () -> { + calculator.reloadTrafficCalcConfig(); + }); + } + + // ============================================================================ + // SECTION 2: parseWhitelistRanges() + // ============================================================================ + + @Test + void testParseWhitelistRanges_emptyConfig() throws Exception { + // Setup - no config + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + JsonObject emptyConfig = new JsonObject(); + + // Act + List> ranges = calculator.parseWhitelistRanges(emptyConfig); + + // Assert - empty ranges + assertTrue(ranges.isEmpty()); + } + + @Test + void testParseWhitelistRanges_singleRange() throws Exception { + // Setup - single range + 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)); + configWithRanges.put("traffic_calc_whitelist_ranges", ranges); + + // Act + List> result = calculator.parseWhitelistRanges(configWithRanges); + + // Assert - single range + assertEquals(1, result.size()); + assertEquals(1000L, result.get(0).get(0)); + assertEquals(2000L, result.get(0).get(1)); + } + + @Test + void testParseWhitelistRanges_multipleRanges() throws Exception { + // Setup - multiple 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(3000L).add(4000L)) + .add(new JsonArray().add(5000L).add(6000L)); + configWithRanges.put("traffic_calc_whitelist_ranges", ranges); + + // Act + List> result = calculator.parseWhitelistRanges(configWithRanges); + + // Assert - multiple ranges + assertEquals(3, result.size()); + assertEquals(1000L, result.get(0).get(0)); + assertEquals(3000L, result.get(1).get(0)); + assertEquals(5000L, result.get(2).get(0)); + } + + @Test + void testParseWhitelistRanges_misorderedRange() throws Exception { + // Setup - range with end < start is malformed + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + JsonObject configWithRanges = new JsonObject(); + JsonArray ranges = new JsonArray() + .add(new JsonArray().add(2000L).add(1000L)); // End before start + configWithRanges.put("traffic_calc_whitelist_ranges", ranges); + + // Act + assertThrows(MalformedTrafficCalcConfigException.class, () -> { + calculator.parseWhitelistRanges(configWithRanges); + }); + } + + @Test + void testParseWhitelistRanges_rangeTooLong() throws Exception { + // Setup - range longer than 24 hours is malformed + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + JsonObject configWithRanges = new JsonObject(); + JsonArray ranges = new JsonArray() + .add(new JsonArray().add(2000L).add(200000L)); // Longer than 24 hours + configWithRanges.put("traffic_calc_whitelist_ranges", ranges); + + // Act + assertThrows(MalformedTrafficCalcConfigException.class, () -> { + calculator.parseWhitelistRanges(configWithRanges); + }); + } + + @Test + void testParseWhitelistRanges_sortsByStartTime() throws Exception { + // Setup - ranges added out of order + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + JsonObject configWithRanges = new JsonObject(); + JsonArray ranges = new JsonArray() + .add(new JsonArray().add(5000L).add(6000L)) + .add(new JsonArray().add(1000L).add(2000L)) + .add(new JsonArray().add(3000L).add(4000L)); + configWithRanges.put("traffic_calc_whitelist_ranges", ranges); + + // Act + List> result = calculator.parseWhitelistRanges(configWithRanges); + + // Assert - should be sorted by start time + assertEquals(3, result.size()); + assertEquals(1000L, result.get(0).get(0)); + assertEquals(3000L, result.get(1).get(0)); + assertEquals(5000L, result.get(2).get(0)); + } + + @Test + void testParseWhitelistRanges_invalidRangeTooFewElements() throws Exception { + // Setup - invalid range with only 1 element; + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + JsonObject configWithRanges = new JsonObject(); + JsonArray ranges = new JsonArray() + .add(new JsonArray().add(1000L)) // Only 1 element + .add(new JsonArray().add(2000L).add(3000L)); // Valid + configWithRanges.put("traffic_calc_whitelist_ranges", ranges); + + // Act + List> result = calculator.parseWhitelistRanges(configWithRanges); + + // Assert - should skip invalid range + assertEquals(1, result.size()); + assertEquals(2000L, result.get(0).get(0)); + } + + @Test + void testParseWhitelistRanges_nullArray() throws Exception { + // Setup - null array + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + JsonObject configWithRanges = new JsonObject(); + configWithRanges.put("traffic_calc_whitelist_ranges", (JsonArray) null); + + // Act + List> result = calculator.parseWhitelistRanges(configWithRanges); + + // Assert - empty ranges + assertTrue(result.isEmpty()); + } + + // ============================================================================ + // SECTION 3: isInWhitelist() + // ============================================================================ + + @Test + void testIsInWhitelist_withinSingleRange() throws Exception { + // Setup - load whitelist with single range [1000, 2000] + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - true when within range + assertTrue(calculator.isInWhitelist(1500L)); + } + + @Test + void testIsInWhitelist_exactlyAtStart() throws Exception { + // Setup - load whitelist with single range [1000, 2000] + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - true when exactly at start of range + assertTrue(calculator.isInWhitelist(1000L)); + } + + @Test + void testIsInWhitelist_exactlyAtEnd() throws Exception { + // Setup - load whitelist with single range [1000, 2000] + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - true when exactly at end of range + assertTrue(calculator.isInWhitelist(2000L)); + } + + @Test + void testIsInWhitelist_beforeRange() throws Exception { + // Setup - load whitelist with single range [1000, 2000] + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - false when before range + assertFalse(calculator.isInWhitelist(999L)); + } + + @Test + void testIsInWhitelist_afterRange() throws Exception { + // Setup - load whitelist with single range [1000, 2000] + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - false when after range + assertFalse(calculator.isInWhitelist(2001L)); + } + + @Test + void testIsInWhitelist_betweenRanges() throws Exception { + // Setup - load whitelist with two ranges [1000, 2000] and [3000, 4000] + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000], + [3000, 4000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - false when between ranges + assertFalse(calculator.isInWhitelist(2500L)); + } + + @Test + void testIsInWhitelist_emptyRanges() throws Exception { + // Setup - no whitelist loaded + createTrafficConfigFile("{}"); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - false when empty ranges + assertFalse(calculator.isInWhitelist(1500L)); + } + + @Test + void testIsInWhitelist_nullRanges() throws Exception { + // Setup - no whitelist loaded (will fail and set empty) + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": null + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert - false when null/empty ranges + assertFalse(calculator.isInWhitelist(1500L)); + } + + @Test + void testIsInWhitelist_invalidRangeSize() throws Exception { + // Setup - load whitelist with invalid range (only 1 element) and valid range + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000], + [2000, 3000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert + assertFalse(calculator.isInWhitelist(1500L)); // Should not match invalid range + assertTrue(calculator.isInWhitelist(2500L)); // Should match valid range + } + + @Test + void testIsInWhitelist_multipleRanges() throws Exception { + // Setup - load whitelist with multiple ranges + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000], + [3000, 4000], + [5000, 6000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert + assertTrue(calculator.isInWhitelist(1500L)); // In first range + assertTrue(calculator.isInWhitelist(3500L)); // In second range + assertTrue(calculator.isInWhitelist(5500L)); // In third range + assertFalse(calculator.isInWhitelist(2500L)); // Between first and second + } + + // ============================================================================ + // SECTION 4: getWhitelistDuration() + // ============================================================================ + + @Test + void testGetWhitelistDuration_noRanges() throws Exception { + // Setup - no ranges + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Assert + assertEquals(0L, calculator.getWhitelistDuration(10000L, 5000L)); // 0 duration when no ranges + } + + @Test + void testGetWhitelistDuration_rangeFullyWithinWindow() throws Exception { + // Setup - range fully within window + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [6000, 7000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - window [5000, 10000], range [6000, 7000] + long duration = calculator.getWhitelistDuration(10000L, 5000L); + + // Assert - full range duration + assertEquals(1000L, duration); + } + + @Test + void testGetWhitelistDuration_rangePartiallyOverlapsStart() throws Exception { + // Setup - range partially overlaps start of window + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [3000, 7000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - window [5000, 10000], range [3000, 7000] + long duration = calculator.getWhitelistDuration(10000L, 5000L); + + // Assert - should clip to [5000, 7000] = 2000 + assertEquals(2000L, duration); + } + + @Test + void testGetWhitelistDuration_rangePartiallyOverlapsEnd() throws Exception { + // Setup - range partially overlaps end of window + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [8000, 12000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - window [5000, 10000], range [8000, 12000] + long duration = calculator.getWhitelistDuration(10000L, 5000L); + + // Assert - should clip to [8000, 10000] = 2000 + assertEquals(2000L, duration); + } + + @Test + void testGetWhitelistDuration_rangeCompletelyOutsideWindow() throws Exception { + // Setup - range completely outside window + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - window [5000, 10000], range [1000, 2000] + long duration = calculator.getWhitelistDuration(10000L, 5000L); + + // Assert - 0 duration when range completely outside window + assertEquals(0L, duration); + } + + @Test + void testGetWhitelistDuration_multipleRanges() throws Exception { + // Setup - multiple ranges + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [6000, 7000], + [8000, 9000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - window [5000, 10000], ranges [6000, 7000] and [8000, 9000] + long duration = calculator.getWhitelistDuration(10000L, 5000L); + + // Assert - 1000 + 1000 = 2000 + assertEquals(2000L, duration); + } + + @Test + void testGetWhitelistDuration_rangeSpansEntireWindow() throws Exception { + // Setup - range spans entire window + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [3000, 12000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - window [5000, 10000], range [3000, 12000] + long duration = calculator.getWhitelistDuration(10000L, 5000L); + + // Assert - entire window is whitelisted = 5000 + assertEquals(5000L, duration); + } + + // ============================================================================ + // SECTION 5: determineStatus() + // ============================================================================ + + @Test + void testDetermineStatus_belowThreshold() throws Exception { + // Setup - below threshold + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - 10 < 5 * 3 + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(10, 3); + + // Assert - DEFAULT when below threshold + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testDetermineStatus_atThreshold() throws Exception { + // Setup - at threshold + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - 15 == 5 * 3 + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(15, 3); + + // Assert - DELAYED_PROCESSING when at threshold + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); + } + + @Test + void testDetermineStatus_aboveThreshold() throws Exception { + // Setup - above threshold + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - 20 > 5 * 3 + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(20, 3); + + // Assert - DELAYED_PROCESSING when above threshold + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); + } + + @Test + void testDetermineStatus_sumPastZero() throws Exception { + // Setup - sumPast is 0 + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - should return DEFAULT to avoid crash + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(100, 0); + + // Assert + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testDetermineStatus_bothZero() throws Exception { + // Setup - both sumCurrent and sumPast are 0; + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - should return DEFAULT to avoid crash + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(0, 0); + + // Assert + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testDetermineStatus_sumCurrentZero() throws Exception { + // Setup - sumCurrent is 0 + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - 0 < 5 * 10 + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(0, 10); + + // Assert - DEFAULT when sumCurrent is 0 + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @ParameterizedTest + @CsvSource({ + "1, 1, 1, DELAYED_PROCESSING", // threshold=1: 1 >= 1*1 + "2, 4, 2, DELAYED_PROCESSING", // threshold=2: 4 >= 2*2 + "5, 10, 2, DELAYED_PROCESSING", // threshold=5: 10 >= 5*2 + "10, 100, 10, DELAYED_PROCESSING", // threshold=10: 100 >= 10*10 + "5, 24, 5, DEFAULT", // threshold=5: 24 < 5*5 + "100, 1000, 11, DEFAULT" // threshold=100: 1000 < 100*11 + }) + void testDetermineStatus_variousThresholds(int threshold, int sumCurrent, int sumPast, String expectedStatus) throws Exception { + // Setup - various thresholds + createTrafficConfigFile("{\"traffic_calc_threshold_multiplier\": " + threshold + "}"); + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(sumCurrent, sumPast); + + // Assert + assertEquals(OptOutTrafficCalculator.TrafficStatus.valueOf(expectedStatus), status); + } + + @Test + void testDetermineStatus_largeNumbers() throws Exception { + // Setup - test with large numbers + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + OptOutTrafficCalculator.TrafficStatus status = calculator.determineStatus(1_000_000, 200_000); + + // Assert - 1M >= 5 * 200K = 1M + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); + } + + // ============================================================================ + // SECTION 6: S3 Config Reload Tests + // ============================================================================ + + @Test + void testReloadWhitelist_success() throws Exception { + // Setup - initial whitelist + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000], + [3000, 4000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Change the whitelist to a new range + String newWhitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [5000, 6000] + ] + } + """; + createTrafficConfigFile(newWhitelistJson); + + // Act - reload the whitelist + calculator.reloadTrafficCalcConfig(); + + // Assert - verify new whitelist is loaded + assertTrue(calculator.isInWhitelist(5500L)); + } + + @Test + void testReloadWhitelist_failure() throws Exception { + // Setup - initial whitelist + String whitelistJson = """ + { + "traffic_calc_whitelist_ranges": [ + [1000, 2000] + ] + } + """; + createTrafficConfigFile(whitelistJson); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Now make it fail + createTrafficConfigFile("Invalid JSON"); + + // Act - should not throw exception + assertThrows(MalformedTrafficCalcConfigException.class, () -> { + calculator.reloadTrafficCalcConfig(); + }); + + } + + // ============================================================================ + // SECTION 7: Cache Management Tests + // ============================================================================ + + @Test + void testGetCacheStats_emptyCache() throws Exception { + // Setup + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + Map stats = calculator.getCacheStats(); + + // Assert - should return empty stats + assertEquals(0, stats.get("cached_files")); + assertEquals(0, stats.get("total_cached_timestamps")); + } + + // ============================================================================ + // SECTION 8: Helper Methods for Test Data Creation + // ============================================================================ + + /** + * Create a mock SQS message with specified timestamp + */ + private Message createSqsMessage(long timestampSeconds) { + Map attributes = new HashMap<>(); + attributes.put(MessageSystemAttributeName.SENT_TIMESTAMP, String.valueOf(timestampSeconds * 1000)); + + return Message.builder() + .messageId("test-msg-" + timestampSeconds) + .body("{\"test\": \"data\"}") + .attributes(attributes) + .build(); + } + + /** + * Create a mock SQS message without timestamp + */ + private Message createSqsMessageWithoutTimestamp() { + return Message.builder() + .messageId("test-msg-no-timestamp") + .body("{\"test\": \"data\"}") + .attributes(new HashMap<>()) + .build(); + } + + /** + * Create delta file bytes with specified timestamps + */ + private byte[] createDeltaFileBytes(List timestamps) throws Exception { + // Create OptOutEntry objects using newTestEntry + List entries = new ArrayList<>(); + + long idCounter = 1000; // Use incrementing IDs for test entries + for (long timestamp : timestamps) { + entries.add(OptOutEntry.newTestEntry(idCounter++, timestamp)); + } + + // Create OptOutCollection + OptOutCollection collection = new OptOutCollection(entries.toArray(new OptOutEntry[0])); + return collection.getStore(); + } + + + // ============================================================================ + // SECTION 9: Tests for calculateStatus() + // ============================================================================ + + @Test + void testCalculateStatus_noDeltaFiles() throws Exception { + // Setup - no delta files + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Collections.emptyList()); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(Collections.emptyList()); + + // Assert - should return DEFAULT when no delta files + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_normalTraffic() throws Exception { + // Setup - setup time: current time + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + // Create delta files with timestamps distributed over 48 hours + List timestamps = new ArrayList<>(); + + // Past window: t-47h to t-25h (add 50 entries) + for (int i = 0; i < 50; i++) { + timestamps.add(t - 47*3600 + i * 1000); + } + + // Current window: t-23h to t-1h (add 100 entries - 2x past) + for (int i = 0; i < 100; i++) { + timestamps.add(t - 23*3600 + i * 1000); + } + + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - 100+1 < 5 * 50 = 250, so should be DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_delayedProcessing() throws Exception { + // Setup - create delta files with spike in current window + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + // Create delta files with spike in current window + List timestamps = new ArrayList<>(); + + // Past window: t-47h to t-25h (add 10 entries) + for (int i = 0; i < 10; i++) { + timestamps.add(t - 47*3600 + i * 1000); + } + + // Current window: t-23h to t-1h (add 100 entries - 10x past!) + for (int i = 0; i < 100; i++) { + timestamps.add(t - 23*3600 + i * 1000); + } + + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - 100+1 >= 5 * 10 = 50, DELAYED_PROCESSING + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); + } + + @Test + void testCalculateStatus_noSqsMessages() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + List timestamps = Arrays.asList(t - 3600, t - 7200); // Some entries + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - null SQS messages + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(null); + + // Assert - should still calculate based on delta files, DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_emptySqsMessages() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + List timestamps = Arrays.asList(t - 3600); + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - empty SQS messages + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(Collections.emptyList()); + + // Assert - should still calculate based on delta files, DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_multipleSqsMessages() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + List timestamps = new ArrayList<>(); + // Past window - 10 entries + for (int i = 0; i < 10; i++) { + timestamps.add(t - 48*3600 + i * 1000); + } + // Current window - 20 entries + for (int i = 0; i < 20; i++) { + timestamps.add(t - 12*3600 + i * 1000); + } + + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - multiple SQS messages, oldest determines t + // Add enough SQS messages to push total count over DELAYED_PROCESSING threshold + List sqsMessages = new ArrayList<>(); + for (int i = 0; i < 101; i++) { + sqsMessages.add(createSqsMessage(t - i * 10)); + } + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - DELAYED_PROCESSING + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); + } + + @Test + void testCalculateStatus_withWhitelist() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + // Whitelist that covers part of current window + String whitelistJson = String.format(""" + { + "traffic_calc_whitelist_ranges": [ + [%d, %d] + ] + } + """, t - 12*3600, t - 6*3600); + + List timestamps = new ArrayList<>(); + // Past window - 20 entries + for (int i = 0; i < 20; i++) { + timestamps.add(t - 48*3600 + i * 100); + } + + // Current window - 100 entries (50 in whitelist range, 50 outside) + for (int i = 0; i < 50; i++) { + timestamps.add(t - 12*3600 + i * 100); // In whitelist + } + for (int i = 0; i < 50; i++) { + timestamps.add(t - 3600 + i * 100); // Outside whitelist + } + + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + createTrafficConfigFile(whitelistJson); + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/delta-001.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - should filter out whitelisted entries + // Only ~50 from current window count (not whitelisted) + 1 SQS = 51 + // 51 < 5 * 20 = 100, so DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_cacheUtilization() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + List timestamps = Arrays.asList(t - 3600, t - 7200); + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - first call should populate cache + List sqsMessages = Arrays.asList(createSqsMessage(t)); + calculator.calculateStatus(sqsMessages); + + Map stats = calculator.getCacheStats(); + int cachedFiles = (Integer) stats.get("cached_files"); + + // Second call should use cache (no additional S3 download) + calculator.calculateStatus(sqsMessages); + + Map stats2 = calculator.getCacheStats(); + int cachedFiles2 = (Integer) stats2.get("cached_files"); + + // Assert - cache should be populated and remain consistent + assertEquals(1, cachedFiles); + assertEquals(cachedFiles, cachedFiles2); + + // Verify S3 download was called only once per file + verify(cloudStorage, times(1)).download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat"); + } + + @Test + void testCalculateStatus_s3Exception() throws Exception { + // Setup - S3 list error + when(cloudStorage.list(S3_DELTA_PREFIX)).thenThrow(new RuntimeException("S3 error")); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - should not throw exception + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(Collections.emptyList()); + + // Assert - DEFAULT on error + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_deltaFileReadException() throws Exception { + // Setup - S3 download error + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenThrow(new CloudStorageException("Failed to download")); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - empty SQS messages + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(Collections.emptyList()); + + // Assert - DEFAULT on error + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_invalidSqsMessageTimestamp() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + List timestamps = Arrays.asList(t - 3600); + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act - SQS message without timestamp (should use current time) + List sqsMessages = Arrays.asList(createSqsMessageWithoutTimestamp()); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_multipleDeltaFiles() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + // File 1 - recent entries + List timestamps1 = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + timestamps1.add(t - 12*3600 + i * 1000); + } + byte[] deltaFileBytes1 = createDeltaFileBytes(timestamps1); + + // File 2 - older entries + List timestamps2 = new ArrayList<>(); + for (int i = 0; i < 30; i++) { + timestamps2.add(t - 36*3600 + i * 1000); + } + byte[] deltaFileBytes2 = createDeltaFileBytes(timestamps2); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList( + "optout-v2/delta/optout-delta--01_2025-11-13T02.00.00Z_bbbbbbbb.dat", + "optout-v2/delta/optout-delta--01_2025-11-13T01.00.00Z_aaaaaaaa.dat" + )); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T02.00.00Z_bbbbbbbb.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes1)); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T01.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes2)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + + // Verify cache has both files + Map stats = calculator.getCacheStats(); + assertEquals(2, stats.get("cached_files")); + } + + @Test + void testCalculateStatus_windowBoundaryTimestamps() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + long currentWindowStart = t - 24*3600 + 300; // t-23h55m + long pastWindowStart = currentWindowStart - 24*3600; // t-47h55m + + List timestamps = Arrays.asList( + t, + currentWindowStart, + pastWindowStart, + t - 24*3600, + t - 48*3600 + ); + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } + + @Test + void testCalculateStatus_timestampsCached() throws Exception { + // Setup - create delta files with some entries + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + + List timestamps = Arrays.asList(t - 3600, t - 7200); + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + + // Cache should contain the timestamps + Map stats = calculator.getCacheStats(); + assertEquals(2, stats.get("total_cached_timestamps")); + } + + @Test + void testCalculateStatus_whitelistReducesPreviousWindowBaseline_customWindows() throws Exception { + // Setup - test with custom 3-hour evaluation windows + // Whitelist in previous window reduces baseline, causing DELAYED_PROCESSING + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + long threeHours = 3 * 3600; // 10800 seconds + + // Whitelist covering most of the PREVIOUS window (t-6h to t-4h) + // This reduces the baseline count in the previous window + String whitelistJson = String.format(""" + { + "traffic_calc_current_evaluation_window_seconds": %d, + "traffic_calc_previous_evaluation_window_seconds": %d, + "traffic_calc_whitelist_ranges": [ + [%d, %d] + ] + } + """, threeHours, threeHours, t - 6*3600, t - 4*3600); + + List timestamps = new ArrayList<>(); + + // Previous window (t-6h to t-3h): Add 100 entries + // 80 of these will be whitelisted (between t-6h and t-4h) + // Only 20 will count toward baseline + for (int i = 0; i < 80; i++) { + timestamps.add(t - 6*3600 + i); // Whitelisted entries in previous window + } + for (int i = 0; i < 20; i++) { + timestamps.add(t - 4*3600 + i); // Non-whitelisted entries in previous window + } + + // Current window (t-3h to t): Add 120 entries (none whitelisted) + // This creates a spike: 120 + 1 SQS >= 5 * 20 = 100 + for (int i = 0; i < 120; i++) { + timestamps.add(t - threeHours + i); // Current window entries + } + + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + createTrafficConfigFile(whitelistJson); + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - DELAYED_PROCESSING + // With 3-hour evaluation windows: + // Previous window (t-6h to t-3h): 100 total entries, 80 whitelisted → 20 counted + // Current window (t-3h to t): 120 entries (none whitelisted) + 1 SQS = 121 + // 121 >= 5 * 20 = 100, so DELAYED_PROCESSING + assertEquals(OptOutTrafficCalculator.TrafficStatus.DELAYED_PROCESSING, status); + } + + @Test + void testCalculateStatus_whitelistReducesCurrentWindowBaseline_customWindows() throws Exception { + // Setup - test with custom 3-hour evaluation windows + // Whitelist in previous window reduces baseline, causing DELAYED_PROCESSING + long currentTime = System.currentTimeMillis() / 1000; + long t = currentTime; + long threeHours = 3 * 3600; // 10800 seconds + + // Whitelist covering most of the CURRENT window (t-3h to t+5m) + // This reduces the baseline count in the previous window + String whitelistJson = String.format(""" + { + "traffic_calc_current_evaluation_window_seconds": %d, + "traffic_calc_previous_evaluation_window_seconds": %d, + "traffic_calc_whitelist_ranges": [ + [%d, %d] + ] + } + """, threeHours, threeHours, t - 3*3600, t - 1*3600); + + List timestamps = new ArrayList<>(); + + // Current window (t-3h to t+5m): Add 100 entries + // 80 of these will be whitelisted (between t-3h and t-1h) + // Only 20 will count toward baseline + for (int i = 0; i < 80; i++) { + timestamps.add(t - 3*3600 + i); // Whitelisted entries in current window + } + for (int i = 0; i < 20; i++) { + timestamps.add(t - 1*3600 + i); // Non-whitelisted entries in current window + } + + // Previous window (t-6h to t-3h): Add 10 entries (none whitelisted) + for (int i = 0; i < 10; i++) { + timestamps.add(t - 6*3600 + i); // Non-whitelisted entries in previous window + } + + byte[] deltaFileBytes = createDeltaFileBytes(timestamps); + + createTrafficConfigFile(whitelistJson); + when(cloudStorage.list(S3_DELTA_PREFIX)).thenReturn(Arrays.asList("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")); + when(cloudStorage.download("optout-v2/delta/optout-delta--01_2025-11-13T00.00.00Z_aaaaaaaa.dat")) + .thenReturn(new ByteArrayInputStream(deltaFileBytes)); + + OptOutTrafficCalculator calculator = new OptOutTrafficCalculator( + cloudStorage, S3_DELTA_PREFIX, TRAFFIC_CONFIG_PATH); + + // Act + List sqsMessages = Arrays.asList(createSqsMessage(t)); + OptOutTrafficCalculator.TrafficStatus status = calculator.calculateStatus(sqsMessages); + + // Assert - DEFAULT + // With 3-hour evaluation windows: + // Previous window (t-6h to t-3h): 10 entries (none whitelisted) + // Current window (t-3h to t): 20 entries (non-whitelisted) + 1 SQS = 21 + // 21 < 5 * 10 = 50, so DEFAULT + assertEquals(OptOutTrafficCalculator.TrafficStatus.DEFAULT, status); + } +}