Send wide events to a dedicated POST endpoint

statistics/statistics-impl/src/main/java/com/duckduckgo/app/statistics/wideevents/ApiWideEventSender.kt

@@ -0,0 +1,108 @@
+/*
+ * Copyright (c) 2025 DuckDuckGo
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.duckduckgo.app.statistics.wideevents
+
+import com.duckduckgo.app.statistics.wideevents.api.AppSection
+import com.duckduckgo.app.statistics.wideevents.api.ContextSection
+import com.duckduckgo.app.statistics.wideevents.api.FeatureData
+import com.duckduckgo.app.statistics.wideevents.api.FeatureSection
+import com.duckduckgo.app.statistics.wideevents.api.GlobalSection
+import com.duckduckgo.app.statistics.wideevents.api.WideEventRequest
+import com.duckduckgo.app.statistics.wideevents.api.WideEventService
+import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository
+import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository.WideEventStatus.CANCELLED
+import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository.WideEventStatus.FAILURE
+import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository.WideEventStatus.SUCCESS
+import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository.WideEventStatus.UNKNOWN
+import com.duckduckgo.appbuildconfig.api.AppBuildConfig
+import com.duckduckgo.common.utils.device.DeviceInfo
+import com.duckduckgo.di.scopes.AppScope
+import com.squareup.anvil.annotations.ContributesBinding
+import javax.inject.Inject
+import javax.inject.Named
+
+@ContributesBinding(AppScope::class)
+@Named("api")
+class ApiWideEventSender @Inject constructor(
+    private val wideEventService: WideEventService,
+    private val appBuildConfig: AppBuildConfig,
+    private val deviceInfo: DeviceInfo,
+) : WideEventSender {
+
+    override suspend fun sendWideEvent(event: WideEventRepository.WideEvent) {
+        requireNotNull(event.status) { "Attempting to send wide event with null status" }
+
+        val request = WideEventRequest(
+            global = GlobalSection(
+                platform = PLATFORM,
+                type = TYPE,
+                sampleRate = SAMPLE_RATE,
+            ),
+            app = AppSection(
+                name = APP_NAME,
+                version = appBuildConfig.versionName,
+                formFactor = deviceInfo.formFactor().description,
+                devMode = appBuildConfig.isDebug.toString(),
+            ),
+            feature = FeatureSection(
+                name = event.name,
+                status = event.status.toParamValue(),
+                data = buildFeatureData(event),
+            ),
+            context = event.flowEntryPoint?.let { ContextSection(name = it) },
+        )
+
+        wideEventService.sendWideEvent(request)
+    }
+
+    private fun buildFeatureData(event: WideEventRepository.WideEvent): FeatureData? {
+        val extData = mutableMapOf<String, String>()
+
+        // Add metadata as ext data
+        event.metadata
+            .filterValues { it != null }
+            .forEach { (key, value) ->
+                extData[key] = value!!
+            }
+
+        // Add steps as ext data with "step." prefix
+        event.steps.forEach { (name, success) ->
+            extData["step.$name"] = success.toString()
+        }
+
+        return if (extData.isNotEmpty()) {
+            FeatureData(ext = extData)
+        } else {
+            null
+        }
+    }
+
+    private companion object {
+        const val PLATFORM = "Android"
+        const val TYPE = "app"
+        const val SAMPLE_RATE = 1
+        const val APP_NAME = "DuckDuckGo Android"
+    }
+}
+
+private fun WideEventRepository.WideEventStatus.toParamValue(): String =
+    when (this) {
+        SUCCESS -> "SUCCESS"
+        FAILURE -> "FAILURE"
+        CANCELLED -> "CANCELLED"
+        UNKNOWN -> "UNKNOWN"
+    }

statistics/statistics-impl/src/main/java/com/duckduckgo/app/statistics/wideevents/CompletedWideEventsProcessor.kt

@@ -16,7 +16,16 @@
 
 package com.duckduckgo.app.statistics.wideevents
 
+import android.content.Context
 import androidx.lifecycle.LifecycleOwner
+import androidx.work.Constraints
+import androidx.work.CoroutineWorker
+import androidx.work.ExistingWorkPolicy
+import androidx.work.NetworkType
+import androidx.work.OneTimeWorkRequestBuilder
+import androidx.work.WorkManager
+import androidx.work.WorkerParameters
+import com.duckduckgo.anvil.annotations.ContributesWorker
 import com.duckduckgo.app.di.AppCoroutineScope
 import com.duckduckgo.app.lifecycle.MainProcessLifecycleObserver
 import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository
@@ -25,10 +34,15 @@ import com.duckduckgo.di.scopes.AppScope
 import com.squareup.anvil.annotations.ContributesMultibinding
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.flow.conflate
+import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
+import logcat.logcat
 import javax.inject.Inject
+import kotlin.collections.chunked
+import kotlin.collections.toSet
 
 @OptIn(ExperimentalCoroutinesApi::class)
 @ContributesMultibinding(AppScope::class)
@@ -38,34 +52,80 @@ class CompletedWideEventsProcessor @Inject constructor(
     @AppCoroutineScope private val appCoroutineScope: CoroutineScope,
     private val wideEventFeature: WideEventFeature,
     private val dispatcherProvider: DispatcherProvider,
+    private val workManager: WorkManager,
 ) : MainProcessLifecycleObserver {
+
+    private val mutex = Mutex()
+
     override fun onCreate(owner: LifecycleOwner) {
         appCoroutineScope.launch {
             runCatching {
                 if (!isFeatureEnabled()) return@runCatching
 
                 wideEventRepository
-                    .getCompletedWideEventIdsFlow()
-                    .conflate()
-                    .collect { ids ->
-                        // Process events in chunks to avoid querying too many events at once.
-                        ids.chunked(100).forEach { idsChunk ->
-                            processCompletedWideEvents(idsChunk.toSet())
+                    .hasCompletedWideEvents()
+                    .filter { it }
+                    .collect {
+                        try {
+                            processCompletedWideEvents()
+                        } catch (e: Exception) {
+                            logcat { "Failed to process completed wide events: ${e.stackTraceToString()}" }
+                            scheduleRetry()
                         }
                     }
             }
         }
     }
 
-    private suspend fun processCompletedWideEvents(wideEventIds: Set<Long>) {
-        wideEventRepository.getWideEvents(wideEventIds).forEach { event ->
-            wideEventSender.sendWideEvent(event)
-            wideEventRepository.deleteWideEvent(event.id)
+    suspend fun processCompletedWideEvents() {
+        if (!isFeatureEnabled()) return
+
+        mutex.withLock {
+            // Process events in chunks to avoid querying too many events at once.
+            wideEventRepository.getCompletedWideEventIds().chunked(100).forEach { idsChunk ->
+                wideEventRepository.getWideEvents(idsChunk.toSet()).forEach { event ->
+                    wideEventSender.sendWideEvent(event)
+                    wideEventRepository.deleteWideEvent(event.id)
+                }
+            }
         }
     }
 
     private suspend fun isFeatureEnabled(): Boolean =
         withContext(dispatcherProvider.io()) {
             wideEventFeature.self().isEnabled()
         }
+
+    private fun scheduleRetry() {
+        workManager.enqueueUniqueWork(
+            TAG_WORKER_COMPLETED_WIDE_EVENTS,
+            ExistingWorkPolicy.KEEP,
+            OneTimeWorkRequestBuilder<CompletedWideEventsWorker>()
+                .setConstraints(Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build())
+                .build(),
+        )
+    }
+
+    companion object {
+        const val TAG_WORKER_COMPLETED_WIDE_EVENTS = "TAG_WORKER_COMPLETED_WIDE_EVENTS"
+    }
+}
+
+@ContributesWorker(AppScope::class)
+class CompletedWideEventsWorker(
+    context: Context,
+    params: WorkerParameters,
+) : CoroutineWorker(context, params) {
+
+    @Inject
+    lateinit var completedWideEventsProcessor: CompletedWideEventsProcessor
+
+    override suspend fun doWork(): Result {
+        return try {
+            completedWideEventsProcessor.processCompletedWideEvents()
+            Result.success()
+        } catch (_: Exception) {
+            Result.retry()
+        }
+    }
 }

statistics/statistics-impl/src/main/java/com/duckduckgo/app/statistics/wideevents/DelegatingWideEventSender.kt

@@ -0,0 +1,51 @@
+/*
+ * Copyright (c) 2025 DuckDuckGo
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.duckduckgo.app.statistics.wideevents
+
+import com.duckduckgo.app.statistics.wideevents.db.WideEventRepository
+import com.duckduckgo.common.utils.DispatcherProvider
+import com.duckduckgo.di.scopes.AppScope
+import com.squareup.anvil.annotations.ContributesBinding
+import kotlinx.coroutines.withContext
+import javax.inject.Inject
+import javax.inject.Named
+
+@ContributesBinding(AppScope::class)
+class DelegatingWideEventSender @Inject constructor(
+    @Named("pixel") private val pixelWideEventSender: WideEventSender,
+    @Named("api") private val apiWideEventSender: WideEventSender,
+    private val wideEventFeature: WideEventFeature,
+    private val dispatcherProvider: DispatcherProvider,
+) : WideEventSender {
+
+    override suspend fun sendWideEvent(event: WideEventRepository.WideEvent) {
+        if (shouldUseApiSender()) {
+            apiWideEventSender.sendWideEvent(event)
+        }
+        if (shouldUsePixelSender()) {
+            pixelWideEventSender.sendWideEvent(event)
+        }
+    }
+
+    private suspend fun shouldUsePixelSender(): Boolean = withContext(dispatcherProvider.io()) {
+        wideEventFeature.sendWideEventsViaPixels().isEnabled()
+    }
+
+    private suspend fun shouldUseApiSender(): Boolean = withContext(dispatcherProvider.io()) {
+        wideEventFeature.sendWideEventsViaPost().isEnabled()
+    }
+}