feat(observability): Worker for sending observability events.

observability/dagger/api/observability-dagger.api

@@ -10,7 +10,21 @@ public final class me/proton/core/observability/dagger/BuildConfig {
 }
 
 public abstract interface class me/proton/core/observability/dagger/CoreObservabilityModule {
+	public static final field Companion Lme/proton/core/observability/dagger/CoreObservabilityModule$Companion;
 	public abstract fun bindIsObservabilityEnabled (Lme/proton/core/observability/data/IsObservabilityEnabledImpl;)Lme/proton/core/observability/domain/usecase/IsObservabilityEnabled;
+	public abstract fun bindObservabilityWorkerManager (Lme/proton/core/observability/data/worker/ObservabilityWorkerManagerImpl;)Lme/proton/core/observability/domain/ObservabilityWorkerManager;
 	public abstract fun bindSendObservabilityEvents (Lme/proton/core/observability/data/usecase/SendObservabilityEventsImpl;)Lme/proton/core/observability/domain/usecase/SendObservabilityEvents;
 }
 
+public final class me/proton/core/observability/dagger/CoreObservabilityModule$Companion {
+	public final fun provideObservabilityWorkerManagerImpl (Landroidx/work/WorkManager;)Lme/proton/core/observability/data/worker/ObservabilityWorkerManagerImpl;
+}
+
+public final class me/proton/core/observability/dagger/CoreObservabilityModule_Companion_ProvideObservabilityWorkerManagerImplFactory : dagger/internal/Factory {
+	public fun <init> (Ljavax/inject/Provider;)V
+	public static fun create (Ljavax/inject/Provider;)Lme/proton/core/observability/dagger/CoreObservabilityModule_Companion_ProvideObservabilityWorkerManagerImplFactory;
+	public synthetic fun get ()Ljava/lang/Object;
+	public fun get ()Lme/proton/core/observability/data/worker/ObservabilityWorkerManagerImpl;
+	public static fun provideObservabilityWorkerManagerImpl (Landroidx/work/WorkManager;)Lme/proton/core/observability/data/worker/ObservabilityWorkerManagerImpl;
+}
+

observability/dagger/build.gradle.kts

@@ -1,4 +1,5 @@
 import studio.forface.easygradle.dsl.*
+import studio.forface.easygradle.dsl.android.*
 
 /*
  * Copyright (c) 2022 Proton Technologies AG
@@ -32,6 +33,7 @@ android {
 dependencies {
     api(
         project(Module.observabilityData),
-        project(Module.observabilityDomain)
+        project(Module.observabilityDomain),
+        `android-work-runtime`
     )
 }

observability/dagger/src/main/kotlin/me/proton/core/observability/dagger/CoreObservabilityModule.kt

@@ -18,14 +18,20 @@
 
 package me.proton.core.observability.dagger
 
+import android.os.SystemClock
+import androidx.work.WorkManager
 import dagger.Binds
 import dagger.Module
+import dagger.Provides
 import dagger.hilt.InstallIn
 import dagger.hilt.components.SingletonComponent
 import me.proton.core.observability.data.IsObservabilityEnabledImpl
 import me.proton.core.observability.data.usecase.SendObservabilityEventsImpl
+import me.proton.core.observability.data.worker.ObservabilityWorkerManagerImpl
+import me.proton.core.observability.domain.ObservabilityWorkerManager
 import me.proton.core.observability.domain.usecase.IsObservabilityEnabled
 import me.proton.core.observability.domain.usecase.SendObservabilityEvents
+import javax.inject.Singleton
 
 @Module
 @InstallIn(SingletonComponent::class)
@@ -35,4 +41,14 @@ public interface CoreObservabilityModule {
 
     @Binds
     public fun bindIsObservabilityEnabled(impl: IsObservabilityEnabledImpl): IsObservabilityEnabled
+
+    @Binds
+    public fun bindObservabilityWorkerManager(impl: ObservabilityWorkerManagerImpl): ObservabilityWorkerManager
+
+    public companion object {
+        @Provides
+        @Singleton
+        public fun provideObservabilityWorkerManagerImpl(workManager: WorkManager): ObservabilityWorkerManagerImpl =
+            ObservabilityWorkerManagerImpl({ SystemClock.elapsedRealtime() }, workManager)
+    }
 }

observability/data/api/observability-data.api

@@ -1,3 +1,7 @@
+public class hilt_aggregated_deps/_me_proton_core_observability_data_worker_ObservabilityWorker_HiltModule {
+	public fun <init> ()V
+}
+
 public final class me/proton/core/observability/data/BuildConfig {
 	public static final field BUILD_TYPE Ljava/lang/String;
 	public static final field DEBUG Z
@@ -10,6 +14,14 @@ public final class me/proton/core/observability/data/IsObservabilityEnabledImpl
 	public fun invoke (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
+public final class me/proton/core/observability/data/IsObservabilityEnabledImpl_Factory : dagger/internal/Factory {
+	public fun <init> (Ljavax/inject/Provider;Ljavax/inject/Provider;)V
+	public static fun create (Ljavax/inject/Provider;Ljavax/inject/Provider;)Lme/proton/core/observability/data/IsObservabilityEnabledImpl_Factory;
+	public synthetic fun get ()Ljava/lang/Object;
+	public fun get ()Lme/proton/core/observability/data/IsObservabilityEnabledImpl;
+	public static fun newInstance (Landroid/content/Context;Lme/proton/core/usersettings/domain/repository/DeviceSettingsRepository;)Lme/proton/core/observability/data/IsObservabilityEnabledImpl;
+}
+
 public final class me/proton/core/observability/data/ObservabilityRepositoryImpl : me/proton/core/observability/domain/ObservabilityRepository {
 	public fun <init> (Lme/proton/core/observability/data/db/ObservabilityDatabase;)V
 	public fun addEvent (Lme/proton/core/observability/domain/entity/ObservabilityEvent;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
@@ -20,6 +32,14 @@ public final class me/proton/core/observability/data/ObservabilityRepositoryImpl
 	public fun getEvents (Ljava/lang/Integer;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
+public final class me/proton/core/observability/data/ObservabilityRepositoryImpl_Factory : dagger/internal/Factory {
+	public fun <init> (Ljavax/inject/Provider;)V
+	public static fun create (Ljavax/inject/Provider;)Lme/proton/core/observability/data/ObservabilityRepositoryImpl_Factory;
+	public synthetic fun get ()Ljava/lang/Object;
+	public fun get ()Lme/proton/core/observability/data/ObservabilityRepositoryImpl;
+	public static fun newInstance (Lme/proton/core/observability/data/db/ObservabilityDatabase;)Lme/proton/core/observability/data/ObservabilityRepositoryImpl;
+}
+
 public abstract class me/proton/core/observability/data/db/ObservabilityDao : me/proton/core/data/room/db/BaseDao {
 	public fun <init> ()V
 }
@@ -58,3 +78,39 @@ public final class me/proton/core/observability/data/usecase/SendObservabilityEv
 	public fun invoke (Ljava/util/List;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
+public final class me/proton/core/observability/data/usecase/SendObservabilityEventsImpl_Factory : dagger/internal/Factory {
+	public fun <init> (Ljavax/inject/Provider;)V
+	public static fun create (Ljavax/inject/Provider;)Lme/proton/core/observability/data/usecase/SendObservabilityEventsImpl_Factory;
+	public synthetic fun get ()Ljava/lang/Object;
+	public fun get ()Lme/proton/core/observability/data/usecase/SendObservabilityEventsImpl;
+	public static fun newInstance (Lme/proton/core/network/data/ApiProvider;)Lme/proton/core/observability/data/usecase/SendObservabilityEventsImpl;
+}
+
+public final class me/proton/core/observability/data/worker/ObservabilityWorkerManagerImpl : me/proton/core/observability/domain/ObservabilityWorkerManager {
+	public fun <init> (Lkotlin/jvm/functions/Function0;Landroidx/work/WorkManager;)V
+	public fun cancel ()V
+	public fun getDurationSinceLastShipment-LV8wdWc (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun schedule-LRDsOJo (J)V
+	public fun setLastSentNow (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+}
+
+public abstract interface class me/proton/core/observability/data/worker/ObservabilityWorker_AssistedFactory : androidx/hilt/work/WorkerAssistedFactory {
+}
+
+public final class me/proton/core/observability/data/worker/ObservabilityWorker_AssistedFactory_Impl : me/proton/core/observability/data/worker/ObservabilityWorker_AssistedFactory {
+	public synthetic fun create (Landroid/content/Context;Landroidx/work/WorkerParameters;)Landroidx/work/ListenableWorker;
+	public fun create (Landroid/content/Context;Landroidx/work/WorkerParameters;)Lme/proton/core/observability/data/worker/ObservabilityWorker;
+	public static fun create (Lme/proton/core/observability/data/worker/ObservabilityWorker_Factory;)Ljavax/inject/Provider;
+}
+
+public final class me/proton/core/observability/data/worker/ObservabilityWorker_Factory {
+	public fun <init> (Ljavax/inject/Provider;)V
+	public static fun create (Ljavax/inject/Provider;)Lme/proton/core/observability/data/worker/ObservabilityWorker_Factory;
+	public fun get (Landroid/content/Context;Landroidx/work/WorkerParameters;)Lme/proton/core/observability/data/worker/ObservabilityWorker;
+	public static fun newInstance (Landroid/content/Context;Landroidx/work/WorkerParameters;Lme/proton/core/observability/domain/usecase/ProcessObservabilityEvents;)Lme/proton/core/observability/data/worker/ObservabilityWorker;
+}
+
+public abstract interface class me/proton/core/observability/data/worker/ObservabilityWorker_HiltModule {
+	public abstract fun bind (Lme/proton/core/observability/data/worker/ObservabilityWorker_AssistedFactory;)Landroidx/hilt/work/WorkerAssistedFactory;
+}
+

observability/data/build.gradle.kts

@@ -20,11 +20,16 @@ import studio.forface.easygradle.dsl.android.*
 
 plugins {
     protonAndroidLibrary
+    protonDagger
     kotlin("plugin.serialization")
 }
 
 publishOption.shouldBePublishedAsLib = true
 
+protonDagger {
+    workManagerHiltIntegration = true
+}
+
 android {
     compileOptions {
         isCoreLibraryDesugaringEnabled = true
@@ -33,6 +38,7 @@ android {
 
 dependencies {
     api(
+        `android-work-runtime`,
         `javax-inject`,
         `hilt-android`,
         project(Module.dataRoom),
@@ -55,9 +61,15 @@ dependencies {
 
     testImplementation(
         project(Module.androidTest),
+        `android-test-core-ktx`,
+        `android-work-testing`,
         `coroutines-test`,
+        `hilt-android-testing`,
         junit,
         `kotlin-test`,
-        mockk
+        mockk,
+        robolectric
     )
+
+    kaptTest(`hilt-android-compiler`)
 }

observability/data/src/main/kotlin/me/proton/core/observability/data/worker/ObservabilityWorker.kt

@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2022 Proton Technologies AG
+ * This file is part of Proton AG and ProtonCore.
+ *
+ * ProtonCore is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * ProtonCore is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with ProtonCore.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package me.proton.core.observability.data.worker
+
+import android.content.Context
+import androidx.hilt.work.HiltWorker
+import androidx.work.CoroutineWorker
+import androidx.work.WorkerParameters
+import androidx.work.workDataOf
+import dagger.assisted.Assisted
+import dagger.assisted.AssistedInject
+import me.proton.core.network.domain.ApiException
+import me.proton.core.network.domain.isRetryable
+import me.proton.core.observability.domain.usecase.ProcessObservabilityEvents
+import me.proton.core.util.kotlin.CoreLogger
+
+@HiltWorker
+internal class ObservabilityWorker @AssistedInject constructor(
+    @Assisted appContext: Context,
+    @Assisted params: WorkerParameters,
+    private val processObservabilityEvents: ProcessObservabilityEvents
+) : CoroutineWorker(appContext, params) {
+    override suspend fun doWork(): Result {
+        return processObservabilityEvents.runCatching {
+            invoke()
+            Result.success()
+        }.recover {
+            if (it is ApiException && it.isRetryable()) {
+                Result.retry()
+            } else {
+                if (it !is ApiException) { // ApiExceptions are logged upstream already.
+                    CoreLogger.e("ObservabilityWorker", it, "Could not send observability events.")
+                }
+                Result.failure(workDataOf("errorMessage" to it.message))
+            }
+        }.getOrThrow()
+    }
+}

observability/data/src/main/kotlin/me/proton/core/observability/data/worker/ObservabilityWorkerManagerImpl.kt

@@ -0,0 +1,80 @@
+/*
+ * Copyright (c) 2022 Proton Technologies AG
+ * This file is part of Proton AG and ProtonCore.
+ *
+ * ProtonCore is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * ProtonCore is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with ProtonCore.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package me.proton.core.observability.data.worker
+
+import androidx.work.Constraints
+import androidx.work.ExistingWorkPolicy
+import androidx.work.NetworkType
+import androidx.work.OneTimeWorkRequestBuilder
+import androidx.work.WorkManager
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import me.proton.core.observability.domain.ObservabilityWorkerManager
+import java.util.concurrent.TimeUnit
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
+
+public class ObservabilityWorkerManagerImpl constructor(
+    private val clockMillis: () -> Long,
+    private val workManager: WorkManager
+) : ObservabilityWorkerManager {
+    private val lastSentAtMs = MutexValue<Long?>(null)
+
+    override fun cancel() {
+        workManager.cancelUniqueWork(WORK_NAME)
+    }
+
+    override suspend fun getDurationSinceLastShipment(): Duration? =
+        lastSentAtMs.getValue()?.let { clockMillis() - it }?.milliseconds
+
+    override suspend fun setLastSentNow() {
+        lastSentAtMs.setValue(clockMillis())
+    }
+
+    override fun schedule(delay: Duration) {
+        val request = OneTimeWorkRequestBuilder<ObservabilityWorker>()
+            .setConstraints(
+                Constraints.Builder()
+                    .setRequiredNetworkType(NetworkType.CONNECTED)
+                    .build()
+            )
+            .setInitialDelay(delay.inWholeMilliseconds, TimeUnit.MILLISECONDS)
+            .build()
+        val policy = when (delay) {
+            // If there is no delay, replace currently scheduled worker, so we can run immediately.
+            Duration.ZERO -> ExistingWorkPolicy.REPLACE
+
+            // If there is a delay, keep the currently scheduled worker, since it'll likely run earlier.
+            else -> ExistingWorkPolicy.KEEP
+        }
+        workManager.beginUniqueWork(WORK_NAME, policy, request).enqueue()
+    }
+
+    private class MutexValue<T>(initialValue: T) {
+        private val mutex = Mutex()
+        private var value: T = initialValue
+
+        suspend fun getValue(): T = mutex.withLock { value }
+        suspend fun setValue(newValue: T) = mutex.withLock { value = newValue }
+    }
+
+    private companion object {
+        private const val WORK_NAME = "me.proton.core.observability.data.worker"
+    }
+}

observability/data/src/test/kotlin/me/proton/core/observability/data/worker/ObservabilityWorkerManagerImplTest.kt

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2022 Proton Technologies AG
+ * This file is part of Proton AG and ProtonCore.
+ *
+ * ProtonCore is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * ProtonCore is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with ProtonCore.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package me.proton.core.observability.data.worker
+
+import androidx.work.ExistingWorkPolicy
+import androidx.work.OneTimeWorkRequest
+import androidx.work.WorkContinuation
+import androidx.work.WorkManager
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.slot
+import io.mockk.verify
+import kotlinx.coroutines.test.runTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.time.Duration.Companion.ZERO
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
+
+class ObservabilityWorkerManagerImplTest {
+    private lateinit var clock: FakeClock
+    private lateinit var tested: ObservabilityWorkerManagerImpl
+    private lateinit var workManager: WorkManager
+
+    @BeforeTest
+    fun setUp() {
+        clock = FakeClock()
+        workManager = mockk()
+        tested = ObservabilityWorkerManagerImpl(clock::now, workManager)
+    }
+
+    @Test
+    fun durationSinceLastShipment() = runTest {
+        assertNull(tested.getDurationSinceLastShipment())
+        tested.setLastSentNow()
+        clock.current = 1000
+        assertEquals(1000.milliseconds, tested.getDurationSinceLastShipment())
+
+        tested.setLastSentNow()
+        clock.current = 1500
+        assertEquals(500.milliseconds, tested.getDurationSinceLastShipment())
+    }
+
+    @Test
+    fun scheduling() {
+        val workNameSlot = slot<String>()
+        val workPolicySlot = slot<ExistingWorkPolicy>()
+        val requestSlot = slot<OneTimeWorkRequest>()
+        val workContinuation = mockk<WorkContinuation>(relaxed = true)
+        every {
+            workManager.beginUniqueWork(
+                capture(workNameSlot),
+                capture(workPolicySlot),
+                capture(requestSlot)
+            )
+        } returns workContinuation
+        every { workManager.cancelUniqueWork(any()) } returns mockk()
+
+        // WHEN
+        tested.schedule(ZERO)
+
+        // THEN
+        verify { workContinuation.enqueue() }
+        assertEquals(ExistingWorkPolicy.REPLACE, workPolicySlot.captured)
+        assertEquals(0, requestSlot.captured.workSpec.initialDelay)
+
+        // WHEN
+        tested.cancel()
+
+        // THEN
+        verify { workManager.cancelUniqueWork(workNameSlot.captured) }
+
+        // WHEN
+        tested.schedule(2.minutes)
+
+        // THEN
+        assertEquals(ExistingWorkPolicy.KEEP, workPolicySlot.captured)
+        assertEquals(2.minutes.inWholeMilliseconds, requestSlot.captured.workSpec.initialDelay)
+    }
+
+    private class FakeClock {
+        var current: Long = 0
+
+        fun now(): Long = current
+    }
+}

observability/data/src/test/kotlin/me/proton/core/observability/data/worker/ObservabilityWorkerTest.kt

@@ -0,0 +1,190 @@
+/*
+ * Copyright (c) 2022 Proton Technologies AG
+ * This file is part of Proton AG and ProtonCore.
+ *
+ * ProtonCore is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * ProtonCore is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with ProtonCore.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package me.proton.core.observability.data.worker
+
+import android.content.Context
+import androidx.hilt.work.HiltWorkerFactory
+import androidx.test.core.app.ApplicationProvider
+import androidx.work.ListenableWorker
+import androidx.work.testing.TestListenableWorkerBuilder
+import dagger.hilt.android.testing.BindValue
+import dagger.hilt.android.testing.HiltAndroidRule
+import dagger.hilt.android.testing.HiltAndroidTest
+import dagger.hilt.android.testing.HiltTestApplication
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.mockk
+import kotlinx.coroutines.runBlocking
+import me.proton.core.network.domain.ApiException
+import me.proton.core.network.domain.ApiResult
+import me.proton.core.network.domain.HttpResponseCodes
+import me.proton.core.observability.domain.ObservabilityRepository
+import me.proton.core.observability.domain.ObservabilityWorkerManager
+import me.proton.core.observability.domain.entity.ObservabilityEvent
+import me.proton.core.observability.domain.usecase.IsObservabilityEnabled
+import me.proton.core.observability.domain.usecase.SendObservabilityEvents
+import org.junit.Before
+import org.junit.Rule
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+import javax.inject.Inject
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertIs
+
+@HiltAndroidTest
+@Config(application = HiltTestApplication::class)
+@RunWith(RobolectricTestRunner::class)
+class ObservabilityWorkerTest {
+    @get:Rule
+    val hiltRule = HiltAndroidRule(this)
+
+    @Inject
+    internal lateinit var hiltWorkerFactory: HiltWorkerFactory
+
+    @BindValue
+    internal lateinit var isObservabilityEnabled: IsObservabilityEnabled
+
+    @BindValue
+    internal lateinit var observabilityWorkerManager: ObservabilityWorkerManager
+
+    @BindValue
+    internal lateinit var repository: ObservabilityRepository
+
+    @BindValue
+    internal lateinit var sendObservabilityEvents: SendObservabilityEvents
+
+    private lateinit var context: Context
+
+    @Before
+    fun setUp() {
+        hiltRule.inject()
+        context = ApplicationProvider.getApplicationContext()
+        isObservabilityEnabled = mockk()
+        observabilityWorkerManager = mockk(relaxUnitFun = true)
+        repository = mockk(relaxUnitFun = true)
+        sendObservabilityEvents = mockk(relaxUnitFun = true)
+    }
+
+
+    @Test
+    fun observabilityIsDisabled() {
+        coEvery { isObservabilityEnabled.invoke() } returns false
+
+        val result = makeAndRunWorker()
+        assertEquals(ListenableWorker.Result.success(), result)
+
+        coVerify(exactly = 0) { sendObservabilityEvents.invoke(any()) }
+        coVerify { observabilityWorkerManager.setLastSentNow() }
+        coVerify { repository.deleteAllEvents() }
+    }
+
+    @Test
+    fun noEvents() {
+        coEvery { isObservabilityEnabled.invoke() } returns true
+        coEvery { repository.getEvents(any()) } returns emptyList()
+
+        val result = makeAndRunWorker()
+        assertEquals(ListenableWorker.Result.success(), result)
+
+        coVerify(exactly = 0) { sendObservabilityEvents.invoke(any()) }
+    }
+
+    @Test
+    fun singleEvent() {
+        val events = listOf(mockk<ObservabilityEvent>())
+
+        coEvery { isObservabilityEnabled.invoke() } returns true
+        coEvery { repository.getEvents(any()) } returns events andThen emptyList()
+
+        val result = makeAndRunWorker()
+        assertEquals(ListenableWorker.Result.success(), result)
+
+        coVerify(exactly = 1) { sendObservabilityEvents.invoke(events) }
+        coVerify(exactly = 1) { repository.deleteEvents(events) }
+        coVerify(exactly = 1) { observabilityWorkerManager.setLastSentNow() }
+    }
+
+    @Test
+    fun batching() {
+        val events1 = listOf(mockk<ObservabilityEvent>())
+        val events2 = listOf(mockk<ObservabilityEvent>())
+
+        coEvery { isObservabilityEnabled.invoke() } returns true
+        coEvery { repository.getEvents(any()) }.returnsMany(events1, events2, emptyList())
+
+        val result = makeAndRunWorker()
+        assertEquals(ListenableWorker.Result.success(), result)
+
+        coVerify(exactly = 1) { sendObservabilityEvents.invoke(events1) }
+        coVerify(exactly = 1) { repository.deleteEvents(events1) }
+        coVerify(exactly = 1) { sendObservabilityEvents.invoke(events2) }
+        coVerify(exactly = 1) { repository.deleteEvents(events2) }
+
+        coVerify(exactly = 2) { sendObservabilityEvents.invoke(any()) }
+        coVerify(exactly = 2) { repository.deleteEvents(any()) }
+
+        coVerify(exactly = 1) { observabilityWorkerManager.setLastSentNow() }
+    }
+
+    @Test
+    fun retryableError() {
+        coEvery { isObservabilityEnabled.invoke() } returns true
+        coEvery { repository.getEvents(any()) } returns listOf(mockk())
+        coEvery { sendObservabilityEvents.invoke(any()) } throws
+            ApiException(ApiResult.Error.Http(HttpResponseCodes.HTTP_TOO_MANY_REQUESTS, "Error"))
+
+        val result = makeAndRunWorker()
+        assertEquals(ListenableWorker.Result.retry(), result)
+
+        coVerify(exactly = 0) { observabilityWorkerManager.setLastSentNow() }
+    }
+
+    @Test
+    fun unrecoverableHttpError() {
+        coEvery { isObservabilityEnabled.invoke() } returns true
+        coEvery { repository.getEvents(any()) } returns listOf(mockk())
+        coEvery { sendObservabilityEvents.invoke(any()) } throws
+            ApiException(ApiResult.Error.Http(HttpResponseCodes.HTTP_BAD_REQUEST, "Error"))
+
+        val result = makeAndRunWorker()
+        assertIs<ListenableWorker.Result.Failure>(result)
+
+        coVerify(exactly = 0) { observabilityWorkerManager.setLastSentNow() }
+    }
+
+    @Test
+    fun unrecoverableError() {
+        coEvery { isObservabilityEnabled.invoke() } returns true
+        coEvery { repository.getEvents(any()) } returns listOf(mockk())
+        coEvery { sendObservabilityEvents.invoke(any()) } throws Throwable("Unknown error")
+
+        val result = makeAndRunWorker()
+        assertIs<ListenableWorker.Result.Failure>(result)
+
+        coVerify(exactly = 0) { observabilityWorkerManager.setLastSentNow() }
+    }
+
+    private fun makeWorker(): ObservabilityWorker = TestListenableWorkerBuilder<ObservabilityWorker>(context)
+        .setWorkerFactory(hiltWorkerFactory)
+        .build()
+
+    private fun makeAndRunWorker(): ListenableWorker.Result = runBlocking { makeWorker().doWork() }
+}