fix: offline events storage sync (#498)

Author: Shahroz16

buildSrc/src/main/kotlin/io.customer/android/Versions.kt

@@ -30,7 +30,7 @@ object Versions {
     internal const val MOCKITO = "4.8.1"
     internal const val MOCKK = "1.12.2"
     internal const val MOSHI = "1.14.0"
-    internal const val SEGMENT = "1.16.3"
+    internal const val SEGMENT = "1.19.1"
     internal const val TIMBER = "5.0.0"
     internal const val REDUX_KOTLIN = "0.6.0"
     internal const val ROBOLECTRIC = "4.9"

datapipelines/src/main/kotlin/io/customer/datapipelines/extensions/AnalyticsExt.kt

@@ -2,7 +2,11 @@ package io.customer.datapipelines.extensions
 
 import com.segment.analytics.kotlin.core.Configuration
 import com.segment.analytics.kotlin.core.ErrorHandler
+import com.segment.analytics.kotlin.core.Settings
+import com.segment.analytics.kotlin.core.emptyJsonObject
 import io.customer.datapipelines.config.DataPipelinesModuleConfig
+import io.customer.datapipelines.plugins.CUSTOMER_IO_DATA_PIPELINES
+import kotlinx.serialization.json.buildJsonObject
 
 /**
  * Updates analytics configuration using the provided [DataPipelinesModuleConfig].
@@ -17,6 +21,15 @@ internal fun updateAnalyticsConfig(
     // Force set to false as we don't need to forward events to Segment destination
     // User can disable CIO destination to achieve same results
     this.autoAddSegmentDestination = false
+    this.defaultSettings = this.defaultSettings?.copy(
+        integrations = buildJsonObject {
+            put(CUSTOMER_IO_DATA_PIPELINES, emptyJsonObject)
+        }
+    ) ?: Settings(
+        integrations = buildJsonObject {
+            put(CUSTOMER_IO_DATA_PIPELINES, emptyJsonObject)
+        }
+    )
     this.trackApplicationLifecycleEvents = moduleConfig.trackApplicationLifecycleEvents
     this.apiHost = moduleConfig.apiHost
     this.cdnHost = moduleConfig.cdnHost

datapipelines/src/main/kotlin/io/customer/datapipelines/plugins/CustomerIODestination.kt

@@ -33,6 +33,8 @@ data class CustomerIOSettings(
     var apiHost: String? = null
 )
 
+internal const val CUSTOMER_IO_DATA_PIPELINES = "Customer.io Data Pipelines"
+
 /**
  * CustomerIODestination plugin that is used to send events to Customer IO CDP api, in the choice of region.
  * How it works
@@ -44,7 +46,7 @@ class CustomerIODestination : DestinationPlugin(), VersionedPlugin, Subscriber {
 
     private var pipeline: EventPipeline? = null
     private var flushPolicies: List<FlushPolicy> = emptyList()
-    override val key: String = "Customer.io Data Pipelines"
+    override val key: String = CUSTOMER_IO_DATA_PIPELINES
 
     override fun track(payload: TrackEvent): BaseEvent {
         enqueue(payload)
@@ -79,13 +81,11 @@ class CustomerIODestination : DestinationPlugin(), VersionedPlugin, Subscriber {
         super.setup(analytics)
 
         // convert flushAt and flushIntervals into FlushPolicies
-        flushPolicies = if (analytics.configuration.flushPolicies.isEmpty()) {
+        flushPolicies = analytics.configuration.flushPolicies.ifEmpty {
             listOf(
                 CountBasedFlushPolicy(analytics.configuration.flushAt),
                 FrequencyFlushPolicy(analytics.configuration.flushInterval * 1000L)
             )
-        } else {
-            analytics.configuration.flushPolicies
         }
 
         // Add DestinationMetadata enrichment plugin

datapipelines/src/test/java/io/customer/datapipelines/DataPipelinesCompatibilityTests.kt

@@ -3,6 +3,7 @@ package io.customer.datapipelines
 import com.segment.analytics.kotlin.core.Storage
 import com.segment.analytics.kotlin.core.emptyJsonArray
 import com.segment.analytics.kotlin.core.emptyJsonObject
+import com.segment.analytics.kotlin.core.utilities.StorageImpl
 import com.segment.analytics.kotlin.core.utilities.getString
 import io.customer.commontest.config.TestConfig
 import io.customer.commontest.extensions.random
@@ -22,9 +23,12 @@ import io.customer.sdk.events.Metric
 import io.customer.sdk.events.TrackMetric
 import io.customer.sdk.util.EventNames
 import io.mockk.every
+import java.io.File
 import kotlinx.coroutines.test.runTest
+import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
 import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.jsonArray
 import kotlinx.serialization.json.jsonObject
@@ -43,7 +47,7 @@ class DataPipelinesCompatibilityTests : JUnitTest() {
 
     private lateinit var globalPreferenceStore: GlobalPreferenceStore
     private lateinit var deviceStore: DeviceStore
-    private lateinit var storage: Storage
+    private lateinit var storage: StorageImpl
 
     override fun setup(testConfig: TestConfig) {
         super.setup(
@@ -59,31 +63,25 @@ class DataPipelinesCompatibilityTests : JUnitTest() {
         globalPreferenceStore = androidSDKComponent.globalPreferenceStore
         deviceStore = androidSDKComponent.deviceStore
 
-        storage = analytics.storage
+        storage = analytics.storage as StorageImpl
     }
 
     private suspend fun getQueuedEvents(): JsonArray {
         // Rollover to ensure that analytics completes writing to the current file
         // and update file contents with valid JSON
         storage.rollover()
-        val cdpApiKey = sdkInstance.moduleConfig.cdpApiKey
         // Find all files that contain the CDP API key in the name
         // The file we are looking for is named after the CDP API key
         // /tmp/analytics-kotlin/{CDP_API_KEY}/events/{CDP_API_KEY}-{N}.tmp before the rollover
         // /tmp/analytics-kotlin/{CDP_API_KEY}/events/{CDP_API_KEY}-{N} after the rollover
-        val eventsFiles = storage.storageDirectory.walk().filter { file ->
-            file.name.contains(cdpApiKey) && file.isFile && file.extension.isBlank()
-        }
-        val result = mutableListOf<JsonElement>()
-        // Read the contents of each file and extract the JSON array of batched events
-        eventsFiles.forEach { file ->
-            val contents = file.readText()
-            val storedEvents = contents.decodeJson()
-            val jsonArray = storedEvents["batch"]?.jsonArray ?: emptyJsonArray
-            result.addAll(jsonArray)
-        }
+
+        val storagePath = storage.read(Storage.Constants.Events)?.let {
+            it.split(',')[0]
+        }.shouldNotBeNull()
+        val storageContents = File(storagePath).readText()
+        val jsonFormat = Json.decodeFromString(JsonObject.serializer(), storageContents)
         // Return the flat list of batched events
-        return JsonArray(result)
+        return jsonFormat["batch"]?.jsonArray ?: emptyJsonArray
     }
 
     //endregion

datapipelines/src/test/java/io/customer/datapipelines/plugins/CustomerIODestinationTest.kt

@@ -0,0 +1,65 @@
+package io.customer.datapipelines.plugins
+
+import com.segment.analytics.kotlin.core.HTTPClient
+import io.customer.commontest.config.TestConfig
+import io.customer.datapipelines.testutils.core.JUnitTest
+import io.customer.datapipelines.testutils.core.testConfiguration
+import io.customer.datapipelines.testutils.utils.OutputReaderPlugin
+import io.customer.datapipelines.testutils.utils.trackEvents
+import io.mockk.every
+import io.mockk.mockkConstructor
+import okio.IOException
+import org.amshove.kluent.shouldBe
+import org.amshove.kluent.shouldNotBe
+import org.junit.jupiter.api.Test
+
+/**
+ * Test for CustomerIODestination plugin, focusing on behavior when network settings fetch fails.
+ * These tests verify that even when the settings fetch fails, the plugin remains enabled
+ * and can process events.
+ */
+class CustomerIODestinationTest : JUnitTest() {
+
+    private lateinit var outputReaderPlugin: OutputReaderPlugin
+
+    override fun setup(testConfig: TestConfig) {
+        // Mock HTTP client to throw exception when fetching settings
+        mockkConstructor(HTTPClient::class)
+        every { anyConstructed<HTTPClient>().settings(any()) } throws IOException("Network error")
+
+        super.setup(
+            testConfiguration {
+                sdkConfig {
+                    autoAddCustomerIODestination(true)
+                }
+            }
+        )
+
+        outputReaderPlugin = OutputReaderPlugin()
+        analytics.add(outputReaderPlugin)
+    }
+
+    // process_givenScreenViewUseAnalytics_expectScreenEventWithoutPropertiesProcessed
+    @Test
+    fun givenFetchingSettingFails_expectCustomerIODestinationPluginStillBeEnabled() {
+        // Verify plugin is present in analytics instance
+        sdkInstance.analytics.find(CustomerIODestination::class) shouldNotBe null
+
+        // Track an event to verify it flows through the pipeline
+        sdkInstance.track("test_event")
+
+        // Verify event was processed
+        outputReaderPlugin.trackEvents.size shouldBe 1
+        outputReaderPlugin.trackEvents.first().event shouldBe "test_event"
+    }
+
+    @Test
+    fun givenDefaultSetting_expectConfigurationIntegrationIncludeCustomerIODestination() {
+        // Verify settings contain CustomerIO destination even with network failure
+        val configuration = analytics.configuration
+        configuration.defaultSettings shouldNotBe null
+        val integrations = configuration.defaultSettings?.integrations
+        integrations shouldNotBe null
+        (integrations?.contains(CUSTOMER_IO_DATA_PIPELINES) ?: false) shouldBe true
+    }
+}