chore: wire three-layer geofence transition delivery

mrehan27 · claude · mrehan27 · commit 801492e38eb9 · 2026-05-25T21:02:32.000+05:00
GeofenceEventTracker drives the three-layer delivery flow: cooldown
deduplication → persist to PendingGeofenceMetricStore → direct HTTP
via GeofenceDeliveryTracker, with flushPending() replay on init and
on every ProfileIdentifiedEvent.

Each delivery attempt picks one path:
- No HttpClient → EventBus, drain row
- No userId → retain row for the next flush
- HTTP attempt → success drains; failure retains for retry

GeofenceDeliveryTrackerImpl reads apiHost from BackgroundDeliveryContextStore
so customer-configured overrides apply. Delivery fails if no host is
persisted yet (cold-start before any foreground SDK init).

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/Sources/Location/Geofence/Event/GeofenceDeliveryTracker.swift b/Sources/Location/Geofence/Event/GeofenceDeliveryTracker.swift
@@ -16,12 +16,12 @@ protocol GeofenceDeliveryTracker: AutoMockable {
 
 final class GeofenceDeliveryTrackerImpl: GeofenceDeliveryTracker {
     private let httpClient: HttpClient
-    private let region: Region
+    private let contextStore: BackgroundDeliveryContextStore
     private let logger: Logger
 
-    init(httpClient: HttpClient, region: Region, logger: Logger) {
+    init(httpClient: HttpClient, contextStore: BackgroundDeliveryContextStore, logger: Logger) {
         self.httpClient = httpClient
-        self.region = region
+        self.contextStore = contextStore
         self.logger = logger
     }
 
@@ -34,6 +34,10 @@ final class GeofenceDeliveryTrackerImpl: GeofenceDeliveryTracker {
             logger.error("cannot deliver geofence metric without a userId")
             return onComplete(.failure(.noRequestMade(nil)))
         }
+        guard let apiHost = contextStore.currentApiHost, !apiHost.isEmpty else {
+            logger.error("cannot deliver geofence metric without a persisted apiHost")
+            return onComplete(.failure(.noRequestMade(nil)))
+        }
 
         var properties: [String: Any] = [
             "geofence_id": metric.geofenceId,
@@ -52,7 +56,7 @@ final class GeofenceDeliveryTrackerImpl: GeofenceDeliveryTracker {
         let endpoint: CIOApiEndpoint = .trackPushMetricsCdp
         guard let httpParams = HttpRequestParams(
             endpoint: endpoint,
-            baseUrl: Self.apiHost(for: region),
+            baseUrl: Self.absoluteUrl(host: apiHost),
             headers: nil,
             body: try? JSONSerialization.data(withJSONObject: body)
         ) else {
@@ -70,10 +74,13 @@ final class GeofenceDeliveryTrackerImpl: GeofenceDeliveryTracker {
         }
     }
 
-    private static func apiHost(for region: Region) -> String {
-        switch region {
-        case .US: return "https://cdp.customer.io/v1"
-        case .EU: return "https://cdp-eu.customer.io/v1"
+    /// `BackgroundDeliveryContextStore.currentApiHost` is host-only (no scheme); the request
+    /// builder needs the full base URL, so prepend `https://` unless the caller has already
+    /// qualified it.
+    private static func absoluteUrl(host: String) -> String {
+        if host.hasPrefix("http://") || host.hasPrefix("https://") {
+            return host
         }
+        return "https://" + host
     }
 }
diff --git a/Sources/Location/Geofence/Event/GeofenceEventTracker.swift b/Sources/Location/Geofence/Event/GeofenceEventTracker.swift
@@ -1,38 +1,56 @@
 import CioInternalCommon
 import Foundation
 
-/// Sends geofence transition events to the data pipeline with cooldown-based deduplication.
+/// Sends geofence transition events through a three-layer delivery flow:
+/// 1. Cooldown-based deduplication (keyed by "geofenceId:transitionType")
+/// 2. Persist to `PendingGeofenceMetricStore` before any send attempt
+/// 3. Dispatch in `deliver`:
+///    - No HttpClient → EventBus, drain row
+///    - No userId yet → retain row for the next flush
+///    - HTTP attempt → success drains; failure retains for retry
 ///
-/// Events are keyed by "geofenceId:transitionType" and suppressed if the same event
-/// was emitted within the cooldown interval. Cooldowns are persisted to survive app restarts
-/// via the `GeofenceStorage` actor.
-///
-/// Communicates with DataPipeline via `EventBusHandler` (matching the in-app and push
-/// metric pattern) so the geofence module has no direct dependency on the DataPipeline module.
+/// `flushPending()` replays queued rows on module init and on every `ProfileIdentifiedEvent`.
+/// Concurrent deliveries for the same row are deduplicated in-process via the active-delivery
+/// ID set; on app kill the row stays on disk and is retried in the next process.
 final class GeofenceEventTracker {
     private let storage: GeofenceStorage
+    private let pendingStore: PendingGeofenceMetricStore
+    private let deliveryTracker: GeofenceDeliveryTracker?
+    private let contextStore: BackgroundDeliveryContextStore
     private let eventBusHandler: EventBusHandler
     private let dateUtil: DateUtil
     private let logger: Logger
     private let cooldownInterval: TimeInterval
+    private let activeDeliveryIds: Synchronized<Set<UUID>> = Synchronized([])
 
     init(
         storage: GeofenceStorage,
+        pendingStore: PendingGeofenceMetricStore,
+        deliveryTracker: GeofenceDeliveryTracker?,
+        contextStore: BackgroundDeliveryContextStore,
         eventBusHandler: EventBusHandler,
         dateUtil: DateUtil,
         logger: Logger,
         cooldownInterval: TimeInterval = GeofenceConstants.eventCooldownInterval
     ) {
         self.storage = storage
+        self.pendingStore = pendingStore
+        self.deliveryTracker = deliveryTracker
+        self.contextStore = contextStore
         self.eventBusHandler = eventBusHandler
         self.dateUtil = dateUtil
         self.logger = logger
         self.cooldownInterval = cooldownInterval
     }
 
     /// Tracks a geofence transition event, suppressing duplicates within the cooldown window.
-    /// Also purges expired cooldown entries opportunistically.
-    func trackTransition(geofenceId: String, transition: GeofenceTransition) async {
+    /// Persists the metric, then dispatches to one of the three delivery paths
+    /// (EventBus drain / queue-retain / direct HTTP) per the rules in `deliver`.
+    func trackTransition(
+        geofenceId: String,
+        transition: GeofenceTransition,
+        location: LocationData? = nil
+    ) async {
         let cooldownKey = "\(geofenceId):\(transition.rawValue)"
         let now = dateUtil.now
 
@@ -41,12 +59,78 @@ final class GeofenceEventTracker {
             return
         }
 
-        eventBusHandler.postEvent(TrackGeofenceMetricEvent(
+        let metric = PendingGeofenceMetric(
             geofenceId: geofenceId,
-            transition: transition
-        ))
-        logger.geofenceEventTracked(geofenceId: geofenceId, transition: transition)
+            transition: transition,
+            latitude: location?.latitude,
+            longitude: location?.longitude,
+            timestamp: now
+        )
+        _ = await pendingStore.append(metric)
 
+        await deliver(metric: metric)
         await storage.purgeExpiredCooldowns(now: now, interval: cooldownInterval)
     }
+
+    /// Replays every queued metric through `deliver`.
+    func flushPending() async {
+        let pending = await pendingStore.loadAll()
+        for metric in pending {
+            await deliver(metric: metric)
+        }
+    }
+
+    // MARK: - Private
+
+    private func deliver(metric: PendingGeofenceMetric) async {
+        // Claim before delivering so two concurrent callers (e.g. trackTransition and a
+        // ProfileIdentifiedEvent-triggered flush, or two flushes) can't both send the
+        // same row. The claim set is in-memory only — on app kill the row stays on
+        // disk and is retried by flushPending in the next process.
+        guard activeDeliveryIds.mutating({ $0.insert(metric.id).inserted }) else { return }
+        defer { activeDeliveryIds.mutating { _ = $0.remove(metric.id) } }
+
+        guard let deliveryTracker else {
+            // No HTTP path will ever be available in this process (MessagingPush not
+            // initialized). Deliver via EventBus and drain; nothing would recover this
+            // row otherwise.
+            postEventBus(metric: metric)
+            _ = await pendingStore.remove(id: metric.id)
+            return
+        }
+
+        guard let userId = contextStore.currentUserId, !userId.isEmpty else {
+            // No userId yet (signed out / not yet identified). Leave the row so a later
+            // ProfileIdentifiedEvent → flushPending can deliver it via direct HTTP with
+            // the right userId. No EventBus — that would attribute the event to the
+            // anonymous profile and duplicate when the flush succeeds.
+            return
+        }
+
+        let success = await withCheckedContinuation { (continuation: CheckedContinuation<Bool, Never>) in
+            deliveryTracker.deliver(metric: metric, userId: userId) { result in
+                switch result {
+                case .success:
+                    continuation.resume(returning: true)
+                case .failure:
+                    continuation.resume(returning: false)
+                }
+            }
+        }
+
+        if success {
+            _ = await pendingStore.remove(id: metric.id)
+            logger.geofenceEventTracked(geofenceId: metric.geofenceId, transition: metric.transition)
+        }
+        // HTTP failure: row stays for next flush. No EventBus — same duplicate
+        // risk if a later flush succeeds.
+    }
+
+    private func postEventBus(metric: PendingGeofenceMetric) {
+        eventBusHandler.postEvent(TrackGeofenceMetricEvent(
+            geofenceId: metric.geofenceId,
+            transition: metric.transition
+        ))
+        logger.geofenceEventTracked(geofenceId: metric.geofenceId, transition: metric.transition)
+    }
 }
diff --git a/Sources/Location/LocationModuleState.swift b/Sources/Location/LocationModuleState.swift
@@ -35,7 +35,13 @@ final class LocationModuleState {
         )
         let locationEnrichmentProvider = LocationProfileEnrichmentProvider(storage: storage, config: config)
         di.profileEnrichmentRegistry.register(locationEnrichmentProvider)
-        registerEventSubscriptions(coordinator: coordinator, eventBusHandler: di.eventBusHandler)
+        let geofenceEventTracker = makeGeofenceEventTracker(di: di)
+        registerEventSubscriptions(
+            coordinator: coordinator,
+            geofenceEventTracker: geofenceEventTracker,
+            eventBusHandler: di.eventBusHandler
+        )
+        Task { await geofenceEventTracker.flushPending() }
         let locationProvider = CoreLocationProvider(logger: di.logger)
         let implementation = LocationServicesImplementation(
             config: config,
@@ -49,10 +55,31 @@ final class LocationModuleState {
         Task { await implementation.setUpLifecycleObserver() }
     }
 
-    private func registerEventSubscriptions(coordinator: LocationSyncCoordinator, eventBusHandler: EventBusHandler) {
+    private func registerEventSubscriptions(
+        coordinator: LocationSyncCoordinator,
+        geofenceEventTracker: GeofenceEventTracker,
+        eventBusHandler: EventBusHandler
+    ) {
         eventBusHandler.addObserver(ProfileIdentifiedEvent.self) { _ in
             Task { await coordinator.syncCachedLocationIfNeeded() }
+            Task { await geofenceEventTracker.flushPending() }
+        }
+    }
+
+    private func makeGeofenceEventTracker(di: DIGraphShared) -> GeofenceEventTracker {
+        let contextStore = di.backgroundDeliveryContextStore
+        let deliveryTracker: GeofenceDeliveryTracker? = di.getOptional(HttpClient.self).map {
+            GeofenceDeliveryTrackerImpl(httpClient: $0, contextStore: contextStore, logger: di.logger)
         }
+        return GeofenceEventTracker(
+            storage: GeofenceStorage(),
+            pendingStore: PendingGeofenceMetricStore(),
+            deliveryTracker: deliveryTracker,
+            contextStore: contextStore,
+            eventBusHandler: di.eventBusHandler,
+            dateUtil: di.dateUtil,
+            logger: di.logger
+        )
     }
 
     /// The current Location services instance. Before initialization, returns an implementation that logs an error when used.
diff --git a/Tests/Location/Geofence/Event/GeofenceDeliveryTrackerTests.swift b/Tests/Location/Geofence/Event/GeofenceDeliveryTrackerTests.swift
@@ -6,13 +6,22 @@ import Testing
 
 @Suite("GeofenceDeliveryTracker")
 struct GeofenceDeliveryTrackerTests {
+    private func makeContextStore(host: String? = "cdp.customer.io/v1") -> BackgroundDeliveryContextStore {
+        let store = BackgroundDeliveryContextStore(
+            fileManager: .default,
+            directoryURL: FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
+        )
+        if let host { store.setApiHost(host) }
+        return store
+    }
+
     private func makeTracker(
-        region: Region = .US,
+        contextStore: BackgroundDeliveryContextStore? = nil,
         httpClient: HttpClientMock = HttpClientMock()
     ) -> (tracker: GeofenceDeliveryTrackerImpl, httpClient: HttpClientMock) {
         let tracker = GeofenceDeliveryTrackerImpl(
             httpClient: httpClient,
-            region: region,
+            contextStore: contextStore ?? makeContextStore(),
             logger: LoggerMock()
         )
         return (tracker, httpClient)
@@ -44,7 +53,7 @@ struct GeofenceDeliveryTrackerTests {
 
     @Test
     func deliver_givenEnterTransition_expectAndroidWireFormat() async {
-        let (tracker, httpClient) = makeTracker(region: .US)
+        let (tracker, httpClient) = makeTracker()
         httpClient.requestClosure = { _, onComplete in onComplete(.success(Data())) }
 
         await withCheckedContinuation { continuation in
@@ -99,8 +108,8 @@ struct GeofenceDeliveryTrackerTests {
     }
 
     @Test
-    func deliver_givenEURegion_expectEUHost() async {
-        let (tracker, httpClient) = makeTracker(region: .EU)
+    func deliver_givenEUApiHost_expectEUUrl() async {
+        let (tracker, httpClient) = makeTracker(contextStore: makeContextStore(host: "cdp-eu.customer.io/v1"))
         httpClient.requestClosure = { _, onComplete in onComplete(.success(Data())) }
 
         await withCheckedContinuation { continuation in
@@ -113,6 +122,21 @@ struct GeofenceDeliveryTrackerTests {
         #expect(url == "https://cdp-eu.customer.io/v1/track")
     }
 
+    @Test
+    func deliver_givenSchemeQualifiedHost_expectSchemeNotDuplicated() async {
+        let (tracker, httpClient) = makeTracker(contextStore: makeContextStore(host: "https://cdp.customer.io/v1"))
+        httpClient.requestClosure = { _, onComplete in onComplete(.success(Data())) }
+
+        await withCheckedContinuation { continuation in
+            tracker.deliver(metric: makeMetric(), userId: "user_42") { _ in
+                continuation.resume()
+            }
+        }
+
+        let url = httpClient.requestReceivedArguments?.params.url.absoluteString
+        #expect(url == "https://cdp.customer.io/v1/track")
+    }
+
     // MARK: - Guard clauses
 
     @Test
@@ -129,6 +153,20 @@ struct GeofenceDeliveryTrackerTests {
         if case .success = result { Issue.record("expected failure for empty userId") }
     }
 
+    @Test
+    func deliver_givenNoPersistedApiHost_expectFailureAndNoHttpCall() async {
+        let (tracker, httpClient) = makeTracker(contextStore: makeContextStore(host: nil))
+
+        let result: Result<Void, HttpRequestError> = await withCheckedContinuation { continuation in
+            tracker.deliver(metric: makeMetric(), userId: "user_42") { result in
+                continuation.resume(returning: result)
+            }
+        }
+
+        #expect(httpClient.requestCallsCount == 0)
+        if case .success = result { Issue.record("expected failure for missing apiHost") }
+    }
+
     // MARK: - Result propagation
 
     @Test
diff --git a/Tests/Location/Geofence/Event/GeofenceEventTrackerTests.swift b/Tests/Location/Geofence/Event/GeofenceEventTrackerTests.swift
