remove write method, fix tests

Author: typotter

features/dd-sdk-android-flags/src/main/kotlin/com/datadog/android/flags/internal/EvaluationEventWriter.kt

@@ -22,19 +22,7 @@ import com.datadog.tools.annotation.NoOpImplementation
 @NoOpImplementation
 internal interface EvaluationEventWriter {
     /**
-     * Writes a single flag evaluation event to persistent storage.
-     *
-     * The event is serialized and stored as an individual record. Multiple events
-     * will be collected and batched together at upload time.
-     *
-     * Thread Safety: This method must be thread-safe and handle concurrent calls correctly.
-     *
-     * @param event the evaluation event to write to storage
-     */
-    fun write(event: BatchedFlagEvaluations.FlagEvaluation)
-
-    /**
-     * Writes a list of flag evaluation events to persistent storage.
+     * Writes flag evaluation events to persistent storage.
      *
      * The events are serialized and stored as individual records. Multiple events
      * will be collected and batched together at upload time.

features/dd-sdk-android-flags/src/main/kotlin/com/datadog/android/flags/internal/EvaluationEventsProcessor.kt

@@ -40,8 +40,8 @@ internal class EvaluationEventsProcessor(
     private val timeProvider: TimeProvider,
     private val scheduledExecutor: ScheduledExecutorService,
     private val internalLogger: InternalLogger,
-    private val flushIntervalMs: Long = DEFAULT_FLUSH_INTERVAL_MS,
-    private val maxAggregations: Int = MAX_AGGREGATIONS_BEFORE_FLUSH
+    private val flushIntervalMs: Long,
+    private val maxAggregations: Int
 ) {
 
     @Volatile
@@ -88,7 +88,7 @@ internal class EvaluationEventsProcessor(
             variantKey = variantKey,
             allocationKey = allocationKey,
             targetingKey = context.targetingKey,
-            rumViewId = ddContext.viewId,
+            viewName = ddContext.viewName,
             errorCode = errorCode
         )
 
@@ -208,11 +208,4 @@ internal class EvaluationEventsProcessor(
             Thread.currentThread().interrupt()
         }
     }
-
-    companion object {
-        const val DEFAULT_FLUSH_INTERVAL_MS = 10_000L // 10 seconds
-        const val MIN_FLUSH_INTERVAL_MS = 1_000L
-        const val MAX_FLUSH_INTERVAL_MS = 60_000L
-        const val MAX_AGGREGATIONS_BEFORE_FLUSH = 1000
-    }
 }

features/dd-sdk-android-flags/src/main/kotlin/com/datadog/android/flags/internal/aggregation/AggregationKey.kt

@@ -14,7 +14,7 @@ package com.datadog.android.flags.internal.aggregation
  * - variant key (or null when no variant is assigned)
  * - allocation key (or null when no allocation is assigned)
  * - targeting key
- * - RUM view ID (or null when RUM is not active)
+ * - RUM view name (or null when RUM is not active or no view available)
  * - error code (ErrorCode enum name for aggregation, e.g., "FLAG_NOT_FOUND")
  *
  * Note: Error messages are stored in AggregationStats but not used for aggregation
@@ -27,6 +27,6 @@ internal data class AggregationKey(
     val variantKey: String? = null,
     val allocationKey: String? = null,
     val targetingKey: String?,
-    val rumViewId: String? = null,
+    val viewName: String? = null,
     val errorCode: String? = null
 )

features/dd-sdk-android-flags/src/main/kotlin/com/datadog/android/flags/internal/aggregation/AggregationStats.kt

@@ -8,6 +8,7 @@ package com.datadog.android.flags.internal.aggregation
 
 import com.datadog.android.flags.model.BatchedFlagEvaluations
 import com.datadog.android.flags.model.EvaluationContext
+import com.datadog.android.flags.model.ResolutionReason
 import com.datadog.android.flags.model.BatchedFlagEvaluations.Context1 as EvaluationEventContext
 
 /**
@@ -37,16 +38,10 @@ internal class AggregationStats(
     private val reason: String?,
     errorMessage: String?
 ) {
-    @Volatile
+    // All field access is synchronized - see recordEvaluation() and toEvaluationEvent()
     private var count: Int = 1
-
-    @Volatile
     private var firstEvaluation: Long = firstTimestamp
-
-    @Volatile
     private var lastEvaluation: Long = firstTimestamp
-
-    @Volatile
     private var lastErrorMessage: String? = errorMessage
 
     /**
@@ -122,7 +117,9 @@ internal class AggregationStats(
             evaluationCount = snapshotCount.toLong(),
             firstEvaluation = snapshotFirst,
             lastEvaluation = snapshotLast,
-            runtimeDefaultUsed = reason == null || reason == "DEFAULT" || reason == "ERROR"
+            runtimeDefaultUsed = reason == null ||
+                reason == ResolutionReason.DEFAULT.name ||
+                reason == ResolutionReason.ERROR.name
         )
     }
 }

features/dd-sdk-android-flags/src/test/kotlin/com/datadog/android/flags/internal/EvaluationEventsProcessorTest.kt

@@ -91,15 +91,31 @@ internal class EvaluationEventsProcessorTest {
     private lateinit var fakeData: PrecomputedFlag
     private lateinit var fakeDDContext: DDContext
 
+    fun createEvaluationEventsProcessor(
+        writer: EvaluationEventWriter,
+        timeProvider: TimeProvider,
+        scheduledExecutor: ScheduledExecutorService,
+        internalLogger: InternalLogger,
+        flushIntervalMs: Long = TEST_FLUSH_INTERVAL_MS,
+        maxAggregations: Int = TEST_MAX_AGGREGATIONS
+    ): EvaluationEventsProcessor {
+        return EvaluationEventsProcessor(
+            writer = writer,
+            timeProvider = timeProvider,
+            scheduledExecutor = scheduledExecutor,
+            internalLogger = internalLogger,
+            flushIntervalMs = flushIntervalMs,
+            maxAggregations = maxAggregations
+        )
+    }
+
     @BeforeEach
     fun `set up`(forge: Forge) {
-        testedProcessor = EvaluationEventsProcessor(
+        testedProcessor = createEvaluationEventsProcessor(
             writer = mockWriter,
             timeProvider = mockTimeProvider,
             scheduledExecutor = mockScheduledExecutor,
-            internalLogger = mockInternalLogger,
-            flushIntervalMs = TEST_FLUSH_INTERVAL_MS,
-            maxAggregations = TEST_MAX_AGGREGATIONS
+            internalLogger = mockInternalLogger
         )
 
         fakeContext = EvaluationContext(targetingKey = fakeTargetingKey)
@@ -361,12 +377,12 @@ internal class EvaluationEventsProcessorTest {
     }
 
     @Test
-    fun `M create separate aggregations W processEvaluation() { different error codes }`() {
+    fun `M create separate aggregations W processEvaluation() { different error codes }`(forge: Forge) {
         // Given
         val errorCode1 = ErrorCode.FLAG_NOT_FOUND.name
         val errorCode2 = ErrorCode.PROVIDER_NOT_READY.name
-        val errorMessage1 = "Flag not found"
-        val errorMessage2 = "Provider not ready"
+        val errorMessage1 = forge.anAlphabeticalString()
+        val errorMessage2 = forge.anAlphabeticalString()
 
         // When
         testedProcessor.processEvaluation(
@@ -493,7 +509,7 @@ internal class EvaluationEventsProcessorTest {
     fun `M flush automatically W processEvaluation() { size limit reached }`() {
         // Given
         val maxAggregations = 5
-        val testProcessor = EvaluationEventsProcessor(
+        val testProcessor = createEvaluationEventsProcessor(
             writer = mockWriter,
             timeProvider = mockTimeProvider,
             scheduledExecutor = mockScheduledExecutor,
@@ -551,13 +567,14 @@ internal class EvaluationEventsProcessorTest {
         )
         testedProcessor.flush()
 
-        // Then
+        // Then - should have called writeAll twice, once per flush
         val eventCaptor = argumentCaptor<List<BatchedFlagEvaluations.FlagEvaluation>>()
-        verify(mockWriter).writeAll(eventCaptor.capture())
+        verify(mockWriter, times(2)).writeAll(eventCaptor.capture())
 
-        val events = eventCaptor.firstValue
-        assertThat(events).hasSize(2)
-        assertThat(events.map { it.evaluationCount }).containsOnly(1L)
+        // Flatten all events from both writeAll calls
+        val allEvents = eventCaptor.allValues.flatten()
+        assertThat(allEvents).hasSize(2)
+        assertThat(allEvents.map { it.evaluationCount }).containsOnly(1L)
     }
 
     @Test
@@ -676,7 +693,7 @@ internal class EvaluationEventsProcessorTest {
         testedProcessor.schedulePeriodicFlush() // Initial schedule
         assertThat(scheduleCount).isEqualTo(1)
 
-        taskRunnable?.run() // Execute the scheduled task
+        checkNotNull(taskRunnable).run() // Execute the scheduled task
 
         // Then - should have rescheduled itself
         assertThat(scheduleCount).isEqualTo(2)
@@ -698,7 +715,7 @@ internal class EvaluationEventsProcessorTest {
 
         // When
         testedProcessor.schedulePeriodicFlush()
-        taskRunnable?.run() // Execute with no evaluations to flush
+        checkNotNull(taskRunnable).run() // Execute with no evaluations to flush
 
         // Then - should not write but should reschedule
         verify(mockWriter, never()).writeAll(any())
@@ -825,7 +842,7 @@ internal class EvaluationEventsProcessorTest {
         stopThread.start()
 
         // Execute the scheduled task (which will try to reschedule)
-        taskRunnable?.run()
+        checkNotNull(taskRunnable).run()
 
         stopThread.join()
 
@@ -1011,16 +1028,13 @@ internal class EvaluationEventsProcessorTest {
         val slowWriteLatch = CountDownLatch(1)
         val writeStartedLatch = CountDownLatch(1)
         val slowWriter = object : EvaluationEventWriter {
-            override fun write(event: BatchedFlagEvaluations.FlagEvaluation) {
-                writeAll(listOf(event))
-            }
             override fun writeAll(events: List<BatchedFlagEvaluations.FlagEvaluation>) {
                 writeStartedLatch.countDown()
                 slowWriteLatch.await() // Block until released
             }
         }
 
-        val slowProcessor = EvaluationEventsProcessor(
+        val slowProcessor = createEvaluationEventsProcessor(
             writer = slowWriter,
             timeProvider = mockTimeProvider,
             scheduledExecutor = mockScheduledExecutor,
@@ -1122,40 +1136,27 @@ internal class EvaluationEventsProcessorTest {
 
     @Test
     fun `M continue accepting evaluations W flush() { during flush operation }`() {
-        // Given - populate initial evaluations
-        repeat(50) { index ->
-            val context = EvaluationContext(targetingKey = "initial-$index")
-            testedProcessor.processEvaluation(
-                fakeFlagKey,
-                context,
-                fakeDDContext,
-                fakeData.variationKey,
-                fakeData.allocationKey,
-                fakeData.reason,
-                null,
-                null
-            )
-        }
+        // Given - stub scheduler to avoid reschedule interactions
+        whenever(mockScheduledExecutor.schedule(any<Runnable>(), any(), any())) doReturn mockScheduledFuture
 
         // Create a slow writer to simulate long flush
         val flushInProgress = CountDownLatch(1)
         val continueFlush = CountDownLatch(1)
         val writeCount = java.util.concurrent.atomic.AtomicInteger(0)
+        val firstCallFlag = java.util.concurrent.atomic.AtomicBoolean(true)
 
         val slowWriter = object : EvaluationEventWriter {
-            override fun write(event: BatchedFlagEvaluations.FlagEvaluation) {
-                writeAll(listOf(event))
-            }
             override fun writeAll(events: List<BatchedFlagEvaluations.FlagEvaluation>) {
+                val isFirstCall = firstCallFlag.compareAndSet(true, false)
                 writeCount.addAndGet(events.size)
-                if (writeCount.get() == 1) {
+                if (isFirstCall) {
                     flushInProgress.countDown() // Signal first write started
                     continueFlush.await() // Block until signaled
                 }
             }
         }
 
-        val slowProcessor = EvaluationEventsProcessor(
+        val slowProcessor = createEvaluationEventsProcessor(
             writer = slowWriter,
             timeProvider = mockTimeProvider,
             scheduledExecutor = mockScheduledExecutor,