Replay pending actions when connectivity is restored

A raw network drop and recovery never flipped the offline flag, so the
queue only drained on the next sink op, an offline-mode toggle or a
restart. Observe the default network's availability and use it, together
with the persisted offline preference, to trigger a replay. This signal
only drains pending actions; it never changes offline mode.

app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt

@@ -52,3 +52,29 @@ fun networkMode(context: Context): Flow<NetworkMode> {
         awaitClose { if (registered) runCatching { connectivity?.unregisterNetworkCallback(callback) } }
     }.distinctUntilChanged()
 }
+
+fun networkAvailable(context: Context): Flow<Boolean> {
+    val connectivity = context.applicationContext.getSystemService(ConnectivityManager::class.java)
+    return callbackFlow {
+        fun push() {
+            val capabilities =
+                connectivity?.activeNetwork?.let { connectivity.getNetworkCapabilities(it) }
+            // INTERNET, not VALIDATED: a LAN-only server is reachable without internet validation.
+            trySend(capabilities?.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) == true)
+        }
+        val callback =
+            object : ConnectivityManager.NetworkCallback() {
+                override fun onAvailable(network: Network) = push()
+
+                override fun onLost(network: Network) = push()
+
+                override fun onCapabilitiesChanged(
+                    network: Network,
+                    capabilities: NetworkCapabilities,
+                ) = push()
+            }
+        val registered = runCatching { connectivity?.registerDefaultNetworkCallback(callback) }.isSuccess
+        push()
+        awaitClose { if (registered) runCatching { connectivity?.unregisterNetworkCallback(callback) } }
+    }.distinctUntilChanged()
+}

app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt

@@ -10,6 +10,8 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
@@ -21,20 +23,23 @@ class PendingServerActionSink(
     private val activeSourceId: suspend () -> Long?,
     private val offline: StateFlow<Boolean>,
     offlinePreferences: Flow<Boolean>,
+    networkAvailable: Flow<Boolean>,
     scope: CoroutineScope,
 ) : ServerActionSink {
     private val lock = Mutex()
 
     init {
         scope.launch {
-            offlinePreferences.collect { isOffline ->
+            combine(offlinePreferences, networkAvailable) { isOffline, online -> !isOffline && online }
-                // The persisted value is authoritative, unlike the process flag's default at cold
+                .distinctUntilChanged()
-                // start, so wait for that flag to agree before replaying.
+                .collect { ready ->
-                if (!isOffline) {
+                    // The persisted flag and connectivity are authoritative, unlike the process
-                    offline.first { !it }
+                    // flag's default at cold start, so wait for that flag to agree before replaying.
-                    flush()
+                    if (ready) {
+                        offline.first { !it }
+                        flush()
+                    }
                 }
-            }
         }
     }
 

app/src/main/kotlin/com/subtracks/di/AppModule.kt

@@ -14,6 +14,7 @@ import com.subtracks.data.download.DownloadEngine
 import com.subtracks.data.download.DownloadNotifier
 import com.subtracks.data.download.OkHttpArtworkFetcher
 import com.subtracks.data.download.SystemDownloadEngine
+import com.subtracks.data.net.networkAvailable
 import com.subtracks.data.net.networkMode
 import com.subtracks.data.prefs.UserPreferences
 import com.subtracks.data.prefs.createUserPreferences
@@ -85,6 +86,7 @@ fun appModule(
             activeSourceId = get<SourceRepository>()::activeSourceIdOnce,
             offline = get<SourceRepository>().offline,
             offlinePreferences = get<UserPreferences>().offlineMode(),
+            networkAvailable = networkAvailable(context.applicationContext),
             scope = get(),
         )
     }

app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt

@@ -15,9 +15,12 @@ import com.subtracks.data.source.StarType
 import com.subtracks.data.source.subsonic.SubsonicException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import org.junit.After
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertThrows
@@ -73,6 +76,7 @@ class PendingServerActionSinkTest {
     private lateinit var sink: PendingServerActionSink
     private val offline = MutableStateFlow(true)
     private val offlinePreferences = MutableStateFlow(true)
+    private val networkAvailable = MutableStateFlow(true)
     private var sourceId = 0L
 
     @Before
@@ -93,6 +97,7 @@ class PendingServerActionSinkTest {
                 activeSourceId = { sourceId },
                 offline = offline,
                 offlinePreferences = offlinePreferences,
+                networkAvailable = networkAvailable,
                 scope = scope,
             )
     }
@@ -131,6 +136,40 @@ class PendingServerActionSinkTest {
             assertEquals(0, db.pendingActionDao().pending(sourceId).size)
         }
 
+    @OptIn(ExperimentalCoroutinesApi::class)
+    @Test
+    fun aRestoredConnectionReplaysThroughTheCollector() =
+        runBlocking {
+            offline.value = false
+            networkAvailable.value = false
+            val reconnectScope = CoroutineScope(SupervisorJob() + UnconfinedTestDispatcher())
+            try {
+                val reconnectable =
+                    PendingServerActionSink(
+                        delegate = delegate,
+                        db = db,
+                        activeSourceId = { sourceId },
+                        offline = offline,
+                        offlinePreferences = flowOf(false),
+                        networkAvailable = networkAvailable,
+                        scope = reconnectScope,
+                    )
+                delegate.scrobbleFailure = IOException("down")
+
+                reconnectable.scrobble("s1", 1_000)
+                assertEquals(1, db.pendingActionDao().pending(sourceId).size)
+
+                delegate.scrobbleFailure = null
+                networkAvailable.value = true
+
+                awaitUntil("the queued scrobble to replay") { delegate.calls.isNotEmpty() }
+                assertEquals(listOf(Call.Scrobble("s1", 1_000)), delegate.calls)
+                assertEquals(0, db.pendingActionDao().pending(sourceId).size)
+            } finally {
+                cancelAndJoinBlocking(reconnectScope)
+            }
+        }
+
     @Test
     fun anOfflineUnstarCoalescesAwayAnEarlierStar() =
         runBlocking {