feat: Live Notifications

core/src/main/kotlin/io/customer/sdk/core/network/CustomerIOHttpClient.kt

@@ -10,23 +10,41 @@ import java.net.HttpURLConnection
 import java.net.MalformedURLException
 import java.net.URL
 
+@InternalCustomerIOApi
+enum class HttpMethod { GET, POST, PUT, DELETE }
+
 @InternalCustomerIOApi
 data class HttpRequestParams(
     val path: String,
+    val method: HttpMethod = HttpMethod.POST,
     val headers: Map<String, String> = emptyMap(),
     val body: String? = null
 )
 
+/**
+ * IOException carrying the HTTP [statusCode] for non-2xx responses ([statusCode]
+ * is null for transport-level failures). Lets callers apply status-aware retry
+ * (e.g. retry 5xx, skip 4xx) without parsing the message.
+ */
+@InternalCustomerIOApi
+class HttpRequestException(
+    val statusCode: Int?,
+    message: String,
+    cause: Throwable? = null
+) : IOException(message, cause)
+
 /** HTTP client for Customer.io API calls with SDK authentication. */
 @InternalCustomerIOApi
 interface CustomerIOHttpClient {
     /**
-     * Performs a POST request to [params.path] with [params.headers] and [params.body].
+     * Performs an HTTP request to [params.path] using [params.method], with
+     * [params.headers] and [params.body].
      *
-     * @param params The request parameters (path, headers, body).
+     * @param params The request parameters (path, method, headers, body).
      * @return A `Result<String>`:
      *   - `Result.success(responseBody)` for 2xx response codes
      *   - `Result.failure(exception)` for network errors or non-2xx codes
+     *     (an [HttpRequestException] for non-2xx, carrying the status code)
      */
     suspend fun request(params: HttpRequestParams): Result<String>
 }
@@ -66,7 +84,7 @@ internal class CustomerIOHttpClientImpl : CustomerIOHttpClient {
             // Configure the connection
             connection.connectTimeout = connectTimeoutMs
             connection.readTimeout = readTimeoutMs
-            connection.requestMethod = "POST"
+            connection.requestMethod = params.method.name
             connection.setRequestProperty("User-Agent", client.toString())
 
             // Authorization: Basic <base64("writeKey:")>
@@ -101,7 +119,7 @@ internal class CustomerIOHttpClientImpl : CustomerIOHttpClient {
             if (responseCode in 200..299) {
                 Result.success(responseBody)
             } else {
-                Result.failure(IOException("HTTP $responseCode: $responseBody"))
+                Result.failure(HttpRequestException(responseCode, "HTTP $responseCode: $responseBody"))
             }
         } catch (e: IOException) {
             Result.failure(e)

core/src/main/kotlin/io/customer/sdk/data/store/GlobalPreferenceStore.kt

@@ -12,8 +12,10 @@ import kotlinx.serialization.json.Json
 interface GlobalPreferenceStore {
     fun saveDeviceToken(token: String)
     fun saveSettings(value: Settings)
+    fun saveInstallationId(value: String)
     fun getDeviceToken(): String?
     fun getSettings(): Settings?
+    fun getInstallationId(): String?
     fun removeDeviceToken()
     fun clear(key: String)
     fun clearAll()
@@ -35,6 +37,10 @@ internal class GlobalPreferenceStoreImpl(
         putString(KEY_CONFIG_SETTINGS, Json.encodeToString(Settings.serializer(), value))
     }
 
+    override fun saveInstallationId(value: String) = prefs.edit {
+        putString(KEY_INSTALLATION_ID, value)
+    }
+
     override fun getDeviceToken(): String? = prefs.read {
         getString(KEY_DEVICE_TOKEN, null)
     }
@@ -48,10 +54,15 @@ internal class GlobalPreferenceStoreImpl(
         }.getOrNull()
     }
 
+    override fun getInstallationId(): String? = prefs.read {
+        getString(KEY_INSTALLATION_ID, null)
+    }
+
     override fun removeDeviceToken() = clear(KEY_DEVICE_TOKEN)
 
     companion object {
         private const val KEY_DEVICE_TOKEN = "device_token"
         private const val KEY_CONFIG_SETTINGS = "config_settings"
+        private const val KEY_INSTALLATION_ID = "installation_id"
     }
 }

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

@@ -14,6 +14,7 @@ import io.customer.sdk.data.store.DeviceStore
  */
 internal class ContextPlugin(
     private val deviceStore: DeviceStore,
+    private val installationId: String,
     private val eventProcessor: ContextPluginEventProcessor = DefaultContextPluginEventProcessor()
 ) : Plugin {
     override val type: Plugin.Type = Plugin.Type.Before
@@ -23,7 +24,7 @@ internal class ContextPlugin(
     internal var deviceToken: String? = null
 
     override fun execute(event: BaseEvent): BaseEvent {
-        return eventProcessor.execute(event, deviceStore) { deviceToken }
+        return eventProcessor.execute(event, deviceStore, installationId) { deviceToken }
     }
 }
 
@@ -32,15 +33,25 @@ internal class ContextPlugin(
  * Allows custom logic to be injected for testing or extension.
  */
 internal interface ContextPluginEventProcessor {
-    fun execute(event: BaseEvent, deviceStore: DeviceStore, deviceTokenProvider: () -> String?): BaseEvent
+    fun execute(
+        event: BaseEvent,
+        deviceStore: DeviceStore,
+        installationId: String,
+        deviceTokenProvider: () -> String?
+    ): BaseEvent
 }
 
 /**
  * Default implementation of [ContextPluginEventProcessor] that sets the user agent
  * in the context and ensures the device token is added if not already present.
  */
 internal class DefaultContextPluginEventProcessor : ContextPluginEventProcessor {
-    override fun execute(event: BaseEvent, deviceStore: DeviceStore, deviceTokenProvider: () -> String?): BaseEvent {
+    override fun execute(
+        event: BaseEvent,
+        deviceStore: DeviceStore,
+        installationId: String,
+        deviceTokenProvider: () -> String?
+    ): BaseEvent {
         // Set user agent in context as it is required by Customer.io Data Pipelines
         event.putInContext("userAgent", deviceStore.buildUserAgent())
         // Remove analytics library information from context as Customer.io
@@ -55,6 +66,10 @@ internal class DefaultContextPluginEventProcessor : ContextPluginEventProcessor
             event.putInContextUnderKey("device", "token", token)
         }
 
+        // Attach installation id to device in context so backend can correlate
+        // events to a single install across identify/clearIdentify cycles.
+        event.putInContextUnderKey("device", "installationId", installationId)
+
         return event
     }
 }

datapipelines/src/main/kotlin/io/customer/sdk/CustomerIO.kt

@@ -41,6 +41,7 @@ import io.customer.sdk.data.model.Settings
 import io.customer.sdk.events.TrackMetric
 import io.customer.sdk.util.EventNames
 import io.customer.tracking.migration.MigrationProcessor
+import java.util.UUID
 import kotlinx.serialization.SerializationStrategy
 import kotlinx.serialization.serializer
 
@@ -109,7 +110,7 @@ class CustomerIO private constructor(
         )
     )
 
-    private val contextPlugin: ContextPlugin = ContextPlugin(deviceStore)
+    private val contextPlugin: ContextPlugin
 
     // Tracks the last userId successfully identified in this SDK session. Used to dedup
     // back-to-back identify(userId) calls with no traits, which are no-ops server-side.
@@ -121,6 +122,14 @@ class CustomerIO private constructor(
         Analytics.debugLogsEnabled = logger.logLevel == CioLogLevel.DEBUG
         Analytics.setLogger(segmentLogger)
 
+        // Resolve installation id: load if present, otherwise generate once and persist.
+        // Survives clearIdentify and SDK re-init; only OS app-data clear wipes it.
+        val installationId = globalPreferenceStore.getInstallationId()
+            ?: UUID.randomUUID().toString().also { id ->
+                globalPreferenceStore.saveInstallationId(id)
+            }
+        contextPlugin = ContextPlugin(deviceStore = deviceStore, installationId = installationId)
+
         // Add required plugins to analytics instance
         analytics.add(contextPlugin)
 

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

@@ -65,7 +65,9 @@ class ContextPluginBehaviorTest : JUnitTest() {
     @Test
     fun execute_whenDeviceTokenIsSetFromAnotherThread_thenAddsCorrectTokenToEvent() {
         val rounds = ROUND_COUNT
-        val contextPlugin = ContextPlugin(deviceStore)
+        // #689 added a required `installationId` to ContextPlugin; the deterministic test
+        // only asserts on deviceToken propagation, so a fixed stub id is sufficient.
+        val contextPlugin = ContextPlugin(deviceStore, installationId = "test-installation-id")
         // We exercise contextPlugin.execute(event) in isolation — no need to
         // attach to analytics. ContextPlugin.execute does not touch the
         // `analytics` lateinit; attaching it under JUnitTest's StandardTestDispatcher

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

@@ -12,14 +12,19 @@ import io.customer.datapipelines.testutils.core.DataPipelinesTestConfig
 import io.customer.datapipelines.testutils.core.JUnitTest
 import io.customer.datapipelines.testutils.core.testConfiguration
 import io.customer.datapipelines.testutils.extensions.deviceToken
+import io.customer.datapipelines.testutils.extensions.installationId
 import io.customer.datapipelines.testutils.utils.OutputReaderPlugin
 import io.customer.datapipelines.testutils.utils.trackEvents
 import io.customer.sdk.core.di.SDKComponent
 import io.customer.sdk.data.store.GlobalPreferenceStore
 import io.mockk.every
+import io.mockk.slot
+import io.mockk.verify
+import java.util.UUID
 import org.amshove.kluent.shouldBeEqualTo
 import org.amshove.kluent.shouldBeNull
 import org.amshove.kluent.shouldHaveSingleItem
+import org.amshove.kluent.shouldNotBeNull
 import org.junit.jupiter.api.Test
 
 class ContextPluginTest : JUnitTest() {
@@ -97,6 +102,84 @@ class ContextPluginTest : JUnitTest() {
         result.event shouldBeEqualTo givenEventName
         result.context.deviceToken.shouldBeNull()
     }
+
+    @Test
+    fun process_givenInstallationIdExists_expectAttachedToDeviceContext() {
+        val givenInstallationId = UUID.randomUUID().toString()
+        setupWithConfig(
+            testConfiguration {
+                diGraph {
+                    android {
+                        every { globalPreferenceStore.getInstallationId() } returns givenInstallationId
+                    }
+                }
+            }
+        )
+
+        val givenEventName = String.random
+        analytics.process(TrackEvent(emptyJsonObject, givenEventName))
+
+        val result = outputReaderPlugin.trackEvents.shouldHaveSingleItem()
+        result.event shouldBeEqualTo givenEventName
+        result.context.installationId shouldBeEqualTo givenInstallationId
+    }
+
+    @Test
+    fun process_givenNoInstallationIdStored_expectGeneratedAndPersisted() {
+        val savedIdSlot = slot<String>()
+        setupWithConfig(
+            testConfiguration {
+                diGraph {
+                    android {
+                        every { globalPreferenceStore.getInstallationId() } returns null
+                        every { globalPreferenceStore.saveInstallationId(capture(savedIdSlot)) } returns Unit
+                    }
+                }
+            }
+        )
+
+        verify(exactly = 1) { globalPreferenceStore.saveInstallationId(any()) }
+        val persistedId = savedIdSlot.captured
+        // Ensure it parses as a valid UUID
+        UUID.fromString(persistedId).shouldNotBeNull()
+
+        val firstEventName = String.random
+        analytics.process(TrackEvent(emptyJsonObject, firstEventName))
+        val secondEventName = String.random
+        analytics.process(TrackEvent(emptyJsonObject, secondEventName))
+
+        val events = outputReaderPlugin.trackEvents
+        events[0].context.installationId shouldBeEqualTo persistedId
+        events[1].context.installationId shouldBeEqualTo persistedId
+    }
+
+    @Test
+    fun process_acrossClearIdentify_expectInstallationIdUnchanged() {
+        val givenInstallationId = UUID.randomUUID().toString()
+        setupWithConfig(
+            testConfiguration {
+                diGraph {
+                    android {
+                        every { globalPreferenceStore.getInstallationId() } returns givenInstallationId
+                    }
+                }
+            }
+        )
+
+        val firstEventName = String.random
+        analytics.process(TrackEvent(emptyJsonObject, firstEventName))
+        val installationIdBefore = outputReaderPlugin.trackEvents.last().context.installationId
+        outputReaderPlugin.reset()
+
+        sdkInstance.clearIdentify()
+
+        val secondEventName = String.random
+        analytics.process(TrackEvent(emptyJsonObject, secondEventName))
+        val installationIdAfter = outputReaderPlugin.trackEvents.last().context.installationId
+
+        installationIdBefore shouldBeEqualTo givenInstallationId
+        installationIdAfter shouldBeEqualTo installationIdBefore
+    }
 }
 
 private class MigrationTokenPlugin : Plugin {

datapipelines/src/test/java/io/customer/datapipelines/testutils/core/UnitTestDelegate.kt

@@ -81,6 +81,7 @@ class UnitTestDelegate(
             androidSDKComponent.overrideDependency<GlobalPreferenceStore>(instance)
         }
         every { globalPreferenceStore.getDeviceToken() } returns null
+        every { globalPreferenceStore.getInstallationId() } returns null
         // Mock device store to avoid reading/writing to device store
         // Spy on the stub to provide custom implementation for the test
         val deviceStoreStub = DeviceStoreStub().getDeviceStore(androidSDKComponent.client)

datapipelines/src/test/java/io/customer/datapipelines/testutils/extensions/JsonExtensions.kt

@@ -85,6 +85,9 @@ private fun <T> Json.encode(value: T?): JsonElement = when (value) {
 internal val JsonObject.deviceToken: String?
     get() = this.getStringAtPath("device.token")
 
+internal val JsonObject.installationId: String?
+    get() = this.getStringAtPath("device.installationId")
+
 fun JsonObject.getStringAtPath(path: String): String? {
     return findAtPath(path).firstOrNull()?.content
 }