feat: add LocationPlugin for identify enrichment and event filtering

- Enrich identify events with last known lat/lng in context so
  Customer.io knows user location at identification time.
- Block consumer-sent "Location Update" track events to prevent
  backend flooding. Only SDK-internal location events pass through
  using an internal marker that is stripped before delivery.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Shahroz16

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

@@ -0,0 +1,74 @@
+package io.customer.datapipelines.plugins
+
+import com.segment.analytics.kotlin.core.Analytics
+import com.segment.analytics.kotlin.core.BaseEvent
+import com.segment.analytics.kotlin.core.IdentifyEvent
+import com.segment.analytics.kotlin.core.TrackEvent
+import com.segment.analytics.kotlin.core.platform.EventPlugin
+import com.segment.analytics.kotlin.core.platform.Plugin
+import com.segment.analytics.kotlin.core.utilities.putInContext
+import io.customer.sdk.communication.Event
+import io.customer.sdk.core.util.Logger
+import io.customer.sdk.util.EventNames
+import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.buildJsonObject
+
+/**
+ * Plugin that handles location-related event processing:
+ *
+ * 1. Enriches identify events with the last known location in context,
+ *    so Customer.io knows where the user is when their profile is identified.
+ *
+ * 2. Blocks consumer-sent "Location Update" track events to prevent flooding
+ *    the backend. Only SDK-internal location events (marked with [INTERNAL_LOCATION_KEY])
+ *    are allowed through; the marker is stripped before the event reaches the destination.
+ */
+internal class LocationPlugin(private val logger: Logger) : EventPlugin {
+    override val type: Plugin.Type = Plugin.Type.Enrichment
+    override lateinit var analytics: Analytics
+
+    @Volatile
+    internal var lastLocation: Event.LocationData? = null
+
+    override fun identify(payload: IdentifyEvent): BaseEvent {
+        val location = lastLocation ?: return payload
+        payload.putInContext(
+            "location",
+            buildJsonObject {
+                put("latitude", JsonPrimitive(location.latitude))
+                put("longitude", JsonPrimitive(location.longitude))
+            }
+        )
+        return payload
+    }
+
+    override fun track(payload: TrackEvent): BaseEvent? {
+        if (payload.event != EventNames.LOCATION_UPDATE) {
+            return payload
+        }
+
+        // Check for the internal marker that only the SDK sets
+        val isInternal = payload.properties[INTERNAL_LOCATION_KEY]?.let {
+            (it as? JsonPrimitive)?.content?.toBooleanStrictOrNull() == true
+        } ?: false
+
+        if (!isInternal) {
+            logger.debug("Blocking consumer-sent \"${EventNames.LOCATION_UPDATE}\" event. Location events are managed by the SDK.")
+            return null
+        }
+
+        // Strip the internal marker before sending to destination
+        payload.properties = buildJsonObject {
+            payload.properties.forEach { (key, value) ->
+                if (key != INTERNAL_LOCATION_KEY) {
+                    put(key, value)
+                }
+            }
+        }
+        return payload
+    }
+
+    companion object {
+        internal const val INTERNAL_LOCATION_KEY = "_cio_internal_location"
+    }
+}

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

@@ -25,6 +25,7 @@ import io.customer.datapipelines.plugins.AutomaticActivityScreenTrackingPlugin
 import io.customer.datapipelines.plugins.AutomaticApplicationLifecycleTrackingPlugin
 import io.customer.datapipelines.plugins.ContextPlugin
 import io.customer.datapipelines.plugins.CustomerIODestination
+import io.customer.datapipelines.plugins.LocationPlugin
 import io.customer.datapipelines.plugins.ScreenFilterPlugin
 import io.customer.sdk.communication.Event
 import io.customer.sdk.communication.subscribe
@@ -107,6 +108,7 @@ class CustomerIO private constructor(
     )
 
     private val contextPlugin: ContextPlugin = ContextPlugin(deviceStore)
+    private val locationPlugin: LocationPlugin = LocationPlugin(logger)
 
     init {
         // Set analytics logger and debug logs based on SDK logger configuration
@@ -127,6 +129,7 @@ class CustomerIO private constructor(
 
         // Add plugin to filter events based on SDK configuration
         analytics.add(ScreenFilterPlugin(moduleConfig.screenViewUse))
+        analytics.add(locationPlugin)
         analytics.add(ApplicationLifecyclePlugin())
 
         // subscribe to journey events emitted from push/in-app module to send them via data pipelines
@@ -159,11 +162,16 @@ class CustomerIO private constructor(
     private fun trackLocation(event: Event.TrackLocationEvent) {
         val location = event.location
         logger.debug("tracking location update: lat=${location.latitude}, lng=${location.longitude}")
+
+        // Cache location for enriching future identify events
+        locationPlugin.lastLocation = location
+
         track(
             name = EventNames.LOCATION_UPDATE,
             properties = mapOf(
                 "lat" to location.latitude,
-                "lng" to location.longitude
+                "lng" to location.longitude,
+                LocationPlugin.INTERNAL_LOCATION_KEY to true
             )
         )
     }