Add network constraints for fetching notifications with WorkManager

features/networkmonitor/api/src/main/kotlin/io/element/android/features/networkmonitor/api/NetworkMonitor.kt

@@ -24,5 +24,11 @@ interface NetworkMonitor {
     /**
      * Checks if the active network is being blocked by Doze, even if it's available.
      */
-    fun isNetworkBlocked(): Boolean
+    val isNetworkBlocked: StateFlow<Boolean>
+
+    /**
+    * A flow indicating whether the app is running in an air-gapped environment.
+     * An air-gapped environment is an environment that is not connected to the internet, and where the app can only communicate with a limited set of servers.
+    */
+    val isInAirGappedEnvironment: StateFlow<Boolean>
 }

features/networkmonitor/impl/build.gradle.kts

@@ -1,4 +1,5 @@
 import extension.setupDependencyInjection
+import extension.testCommonDependencies
 
 /*
  * Copyright (c) 2025 Element Creations Ltd.
@@ -23,4 +24,8 @@ dependencies {
     implementation(projects.libraries.core)
     implementation(projects.libraries.di)
     api(projects.features.networkmonitor.api)
+
+    testCommonDependencies(libs)
+    testImplementation(projects.libraries.matrix.test)
+    testImplementation(projects.features.networkmonitor.test)
 }

features/networkmonitor/impl/src/main/kotlin/io/element/android/features/networkmonitor/impl/DefaultNetworkMonitor.kt

@@ -13,18 +13,21 @@ package io.element.android.features.networkmonitor.impl
 import android.content.Context
 import android.net.ConnectivityManager
 import android.net.Network
+import android.net.NetworkCapabilities
 import android.net.NetworkRequest
 import dev.zacsweers.metro.AppScope
 import dev.zacsweers.metro.ContributesBinding
 import dev.zacsweers.metro.SingleIn
 import io.element.android.features.networkmonitor.api.NetworkMonitor
 import io.element.android.features.networkmonitor.api.NetworkStatus
+import io.element.android.libraries.core.meta.BuildMeta
 import io.element.android.libraries.di.annotations.AppCoroutineScope
 import io.element.android.libraries.di.annotations.ApplicationContext
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.channels.trySendBlocking
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.callbackFlow
@@ -39,13 +42,13 @@ import java.util.concurrent.atomic.AtomicInteger
 @SingleIn(AppScope::class)
 class DefaultNetworkMonitor(
     @ApplicationContext context: Context,
-    @AppCoroutineScope
-    appCoroutineScope: CoroutineScope,
+    @AppCoroutineScope appCoroutineScope: CoroutineScope,
+    private val buildMeta: BuildMeta,
 ) : NetworkMonitor {
     private val connectivityManager: ConnectivityManager = context.getSystemService(ConnectivityManager::class.java)
-    private val blockedNetworkBlockedChecker = NetworkBlockedChecker(connectivityManager)
 
-    override fun isNetworkBlocked(): Boolean = blockedNetworkBlockedChecker.isNetworkBlocked()
+    override val isNetworkBlocked = MutableStateFlow(false)
+    override val isInAirGappedEnvironment = MutableStateFlow(false)
 
     override val connectivity: StateFlow<NetworkStatus> = callbackFlow {
 
@@ -63,6 +66,27 @@ class DefaultNetworkMonitor(
                 }
             }
 
+            override fun onBlockedStatusChanged(network: Network, blocked: Boolean) {
+                Timber.d("Network ${network.networkHandle} blocked status changed: $blocked.")
+                if (network.networkHandle == connectivityManager.activeNetwork?.networkHandle) {
+                    // If the network is blocked, it means that Doze is preventing the app from using the network, even if it's available.
+                    isNetworkBlocked.value = blocked
+                }
+            }
+
+            override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) {
+                if (!buildMeta.isEnterpriseBuild) {
+                    // The air-gapped environment detection is only relevant for the enterprise build.
+                    return
+                }
+
+                if (network.networkHandle == connectivityManager.activeNetwork?.networkHandle) {
+                    // If the network doesn't have the NET_CAPABILITY_VALIDATED capability, it means that the network is not able to reach the internet
+                    // (according to Google), which is a common case in air-gapped environments.
+                    isInAirGappedEnvironment.value = !networkCapabilities.capabilities.contains(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
+                }
+            }
+
             override fun onAvailable(network: Network) {
                 if (activeNetworksCount.incrementAndGet() > 0) {
                     trySendBlocking(NetworkStatus.Connected)

features/networkmonitor/test/src/main/kotlin/io/element/android/features/networkmonitor/test/FakeNetworkMonitor.kt

@@ -14,8 +14,16 @@ import kotlinx.coroutines.flow.MutableStateFlow
 
 class FakeNetworkMonitor(
     initialStatus: NetworkStatus = NetworkStatus.Connected,
-    private val isNetworkBlockedLambda: () -> Boolean = { false },
 ) : NetworkMonitor {
     override val connectivity = MutableStateFlow(initialStatus)
-    override fun isNetworkBlocked(): Boolean = isNetworkBlockedLambda()
+    override val isNetworkBlocked = MutableStateFlow(false)
+    override val isInAirGappedEnvironment = MutableStateFlow(false)
+
+    fun givenNetworkBlocked(isBlocked: Boolean) {
+        isNetworkBlocked.value = isBlocked
+    }
+
+    fun givenIsInAirGappedEnvironment(isInAirGapped: Boolean) {
+        isInAirGappedEnvironment.value = isInAirGapped
+    }
 }

libraries/featureflag/api/src/main/kotlin/io/element/android/libraries/featureflag/api/FeatureFlags.kt

@@ -147,4 +147,12 @@ enum class FeatureFlags(
         defaultValue = { false },
         isFinished = false,
     ),
+    ValidateNetworkWhenSchedulingNotificationFetching(
+        key = "feature.validate_network_when_scheduling_notification_fetching",
+        title = "validate internet connectivity when scheduling notification fetching",
+        description = "Only fetch events for push notifications when the device has internet connectivity. " +
+            "Enabling this can be problematic in air-gapped environments.",
+        defaultValue = { true },
+        isFinished = false,
+    ),
 }

libraries/push/impl/src/main/kotlin/io/element/android/libraries/push/impl/push/DefaultPushHandler.kt

@@ -30,7 +30,6 @@ import io.element.android.libraries.workmanager.api.WorkManagerRequestType
 import io.element.android.libraries.workmanager.api.WorkManagerScheduler
 import io.element.android.services.analytics.api.AnalyticsLongRunningTransaction
 import io.element.android.services.analytics.api.AnalyticsService
-import io.element.android.services.toolbox.api.sdk.BuildVersionSdkIntProvider
 import io.element.android.services.toolbox.api.systemclock.SystemClock
 import kotlinx.coroutines.flow.first
 import timber.log.Timber
@@ -49,7 +48,7 @@ class DefaultPushHandler(
     private val analyticsService: AnalyticsService,
     private val systemClock: SystemClock,
     private val workManagerScheduler: WorkManagerScheduler,
-    private val buildVersionSdkIntProvider: BuildVersionSdkIntProvider,
+    private val syncPendingNotificationsRequestFactory: SyncPendingNotificationsRequestBuilder.Factory,
     resultProcessor: NotificationResultProcessor,
 ) : PushHandler {
     init {
@@ -134,12 +133,7 @@ class DefaultPushHandler(
 
             if (!workManagerScheduler.hasPendingWork(userId, WorkManagerRequestType.NOTIFICATION_SYNC)) {
                 Timber.d("No pending worker for push notifications found")
-                workManagerScheduler.submit(
-                    SyncPendingNotificationsRequestBuilder(
-                        sessionId = userId,
-                        buildVersionSdkIntProvider = buildVersionSdkIntProvider,
-                    )
-                )
+                workManagerScheduler.submit(syncPendingNotificationsRequestFactory.create(userId))
             }
         } catch (e: Exception) {
             Timber.tag(loggerTag.value).e(e, "## handleInternal() failed")

libraries/push/impl/src/main/kotlin/io/element/android/libraries/push/impl/workmanager/FetchPendingNotificationsWorker.kt

@@ -160,7 +160,8 @@ class FetchPendingNotificationsWorker(
         networkTimeoutSpans.finish()
 
         // If there is a problem with the updated network values, report it and retry if needed
-        if (reportConnectivityError(requests = requests, hasNetwork = hasNetwork, isNetworkBlocked = networkMonitor.isNetworkBlocked())) {
+        val isNetworkBlocked = networkMonitor.isNetworkBlocked.first()
+        if (reportConnectivityError(requests = requests, hasNetwork = hasNetwork, isNetworkBlocked = isNetworkBlocked)) {
             pushHistoryService.insertOrUpdatePushRequests(requests.map { request ->
                 request.copy(retries = request.retries + 1)
             })

libraries/push/impl/src/main/kotlin/io/element/android/libraries/push/impl/workmanager/SyncPendingNotificationsRequestBuilder.kt

@@ -8,32 +8,87 @@
 
 package io.element.android.libraries.push.impl.workmanager
 
+import android.net.NetworkCapabilities
+import android.net.NetworkRequest
 import android.os.Build
+import androidx.work.Constraints
 import androidx.work.ExistingWorkPolicy
+import androidx.work.NetworkType
 import androidx.work.OneTimeWorkRequestBuilder
 import androidx.work.OutOfQuotaPolicy
 import androidx.work.workDataOf
+import dev.zacsweers.metro.AppScope
+import dev.zacsweers.metro.Assisted
+import dev.zacsweers.metro.AssistedFactory
+import dev.zacsweers.metro.AssistedInject
+import dev.zacsweers.metro.ContributesBinding
+import io.element.android.features.networkmonitor.api.NetworkMonitor
+import io.element.android.libraries.featureflag.api.FeatureFlagService
+import io.element.android.libraries.featureflag.api.FeatureFlags
 import io.element.android.libraries.matrix.api.core.SessionId
+import io.element.android.libraries.push.impl.workmanager.SyncPendingNotificationsRequestBuilder.Companion.SESSION_ID
 import io.element.android.libraries.workmanager.api.WorkManagerRequestBuilder
 import io.element.android.libraries.workmanager.api.WorkManagerRequestType
 import io.element.android.libraries.workmanager.api.WorkManagerRequestWrapper
 import io.element.android.libraries.workmanager.api.WorkManagerWorkerType
 import io.element.android.libraries.workmanager.api.workManagerTag
 import io.element.android.services.toolbox.api.sdk.BuildVersionSdkIntProvider
+import kotlinx.coroutines.flow.first
+import timber.log.Timber
+
+interface SyncPendingNotificationsRequestBuilder : WorkManagerRequestBuilder {
+    interface Factory {
+        fun create(sessionId: SessionId): SyncPendingNotificationsRequestBuilder
+    }
 
-class SyncPendingNotificationsRequestBuilder(
-    private val sessionId: SessionId,
-    private val buildVersionSdkIntProvider: BuildVersionSdkIntProvider,
-) : WorkManagerRequestBuilder {
     companion object {
         const val SESSION_ID = "session_id"
     }
+}
+
+@AssistedInject
+class DefaultSyncPendingNotificationsRequestBuilder(
+    @Assisted private val sessionId: SessionId,
+    private val buildVersionSdkIntProvider: BuildVersionSdkIntProvider,
+    private val networkMonitor: NetworkMonitor,
+    private val featureFlagService: FeatureFlagService,
+) : SyncPendingNotificationsRequestBuilder {
+    @AssistedFactory
+    @ContributesBinding(AppScope::class)
+    interface Factory : SyncPendingNotificationsRequestBuilder.Factory {
+        override fun create(sessionId: SessionId): DefaultSyncPendingNotificationsRequestBuilder
+    }
 
     override suspend fun build(): Result<List<WorkManagerRequestWrapper>> {
         val type = WorkManagerWorkerType.Unique(
             name = workManagerTag(sessionId = sessionId, requestType = WorkManagerRequestType.NOTIFICATION_SYNC),
             policy = ExistingWorkPolicy.APPEND_OR_REPLACE,
         )
+
+        val networkRequestBuilder = NetworkRequest.Builder()
+            // Allow any kind of network that can have internet connectivity.
+            .addTransportType(NetworkCapabilities.TRANSPORT_CELLULAR)
+            .addTransportType(NetworkCapabilities.TRANSPORT_WIFI)
+            .addTransportType(NetworkCapabilities.TRANSPORT_VPN)
+            .addTransportType(NetworkCapabilities.TRANSPORT_ETHERNET)
+            // By default, the network request will require the device to not be in VPN, but since some customers use a VPN to connect to their homeserver,
+            // we need to allow VPN networks.
+            .removeCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
+
+        // If we're in an air-gapped environment, we shouldn't validate internet connectivity, as the checker will fail and the worker won't run at all.
+        // Note this will always be false for FOSS, since the feature is only enabled in Element Pro.
+        if (networkMonitor.isInAirGappedEnvironment.first()) {
+            Timber.d("In an air-gapped environment, not adding NET_CAPABILITY_VALIDATED to the network request")
+            networkRequestBuilder.removeCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
+        } else if (featureFlagService.isFeatureEnabled(FeatureFlags.ValidateNetworkWhenSchedulingNotificationFetching)) {
+            Timber.d("Not in an air-gapped environment, adding NET_CAPABILITY_VALIDATED to the network request")
+            networkRequestBuilder.addCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
+        }
+
+        val networkConstraints = Constraints.Builder()
+            .setRequiredNetworkRequest(networkRequestBuilder.build(), NetworkType.NOT_REQUIRED)
+            .build()
+
         val request = OneTimeWorkRequestBuilder<FetchPendingNotificationsWorker>()
             .setInputData(workDataOf(SESSION_ID to sessionId.value))
             .apply {
@@ -44,8 +99,10 @@ class SyncPendingNotificationsRequestBuilder(
                     setExpedited(OutOfQuotaPolicy.RUN_AS_NON_EXPEDITED_WORK_REQUEST)
                 }
             }
+            .setConstraints(networkConstraints)
             .setTraceTag(workManagerTag(sessionId, WorkManagerRequestType.NOTIFICATION_SYNC))
             .build()
+
         return Result.success(listOf(WorkManagerRequestWrapper(request, type)))
     }
 }

libraries/push/impl/src/test/kotlin/io/element/android/libraries/push/impl/push/DefaultPushHandlerTest.kt

@@ -26,15 +26,16 @@ import io.element.android.libraries.push.impl.history.PushHistoryService
 import io.element.android.libraries.push.impl.notifications.FakeNotificationResultProcessor
 import io.element.android.libraries.push.impl.test.DefaultTestPush
 import io.element.android.libraries.push.impl.troubleshoot.DiagnosticPushHandler
+import io.element.android.libraries.push.impl.workmanager.SyncPendingNotificationsRequestBuilder
 import io.element.android.libraries.pushproviders.api.PushData
 import io.element.android.libraries.pushstore.api.clientsecret.PushClientSecret
 import io.element.android.libraries.pushstore.test.userpushstore.FakeUserPushStore
 import io.element.android.libraries.pushstore.test.userpushstore.FakeUserPushStoreFactory
 import io.element.android.libraries.pushstore.test.userpushstore.clientsecret.FakePushClientSecret
 import io.element.android.libraries.workmanager.api.WorkManagerRequestBuilder
+import io.element.android.libraries.workmanager.api.WorkManagerRequestWrapper
 import io.element.android.libraries.workmanager.test.FakeWorkManagerScheduler
 import io.element.android.services.analytics.test.FakeAnalyticsService
-import io.element.android.services.toolbox.test.sdk.FakeBuildVersionSdkIntProvider
 import io.element.android.services.toolbox.test.systemclock.FakeSystemClock
 import io.element.android.tests.testutils.lambda.lambdaError
 import io.element.android.tests.testutils.lambda.lambdaRecorder
@@ -216,7 +217,6 @@ class DefaultPushHandlerTest {
         workManagerScheduler: FakeWorkManagerScheduler = FakeWorkManagerScheduler(),
         analyticsService: FakeAnalyticsService = FakeAnalyticsService(),
         systemClock: FakeSystemClock = FakeSystemClock(),
-        buildVersionSdkIntProvider: FakeBuildVersionSdkIntProvider = FakeBuildVersionSdkIntProvider(33),
         resultProcessor: FakeNotificationResultProcessor = FakeNotificationResultProcessor(
             emit = { Result.success(Unit) },
             start = {},
@@ -238,8 +238,16 @@ class DefaultPushHandlerTest {
             analyticsService = analyticsService,
             systemClock = systemClock,
             workManagerScheduler = workManagerScheduler,
-            buildVersionSdkIntProvider = buildVersionSdkIntProvider,
             resultProcessor = resultProcessor,
+            syncPendingNotificationsRequestFactory = object : SyncPendingNotificationsRequestBuilder.Factory {
+                override fun create(sessionId: SessionId): SyncPendingNotificationsRequestBuilder {
+                    return object : SyncPendingNotificationsRequestBuilder {
+                        override suspend fun build(): Result<List<WorkManagerRequestWrapper>> {
+                            return Result.success(emptyList())
+                        }
+                    }
+                }
+            }
         )
     }
 }