diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt new file mode 100644 index 00000000..7f833932 --- /dev/null +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/AppLifecycle.kt @@ -0,0 +1,32 @@ +package press.mantra.compose + +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow + +/** + * Whether the app is in the foreground, for the parts of the app that hold network + * connections open. + * + * A singleton rather than something threaded through the composition, because the consumers + * are not composables: they are long-lived coroutines at application scope, started before + * any screen exists and outliving all of them. + * + * Defaults to foreground. Nothing here can observe a platform that has not wired + * [enteredForeground]/[enteredBackground] up, and on such a platform "always on" is the + * behaviour that predates this file — a subscription that never opens is a far worse failure + * than one that stays open too long. + */ +object AppLifecycle { + private val _isForeground = MutableStateFlow(true) + + val isForeground: StateFlow = _isForeground.asStateFlow() + + fun enteredForeground() { + _isForeground.value = true + } + + fun enteredBackground() { + _isForeground.value = false + } +} diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt index a37fd240..04145fc1 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/managers/LiveSubscriptionManager.kt @@ -19,6 +19,8 @@ import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.map @@ -26,7 +28,9 @@ import kotlinx.coroutines.flow.transformWhile import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.withContext +import press.mantra.compose.database.model.NegentropySynchronizeRequest import press.mantra.compose.database.model.NostrEvent +import press.mantra.compose.database.model.types.SynchronizationFilter import press.mantra.compose.network.relays.RelaysSocketManager import press.mantra.compose.network.relays.isRelayBackPressure import press.mantra.compose.network.sockets.NostrIncomingMessage @@ -62,6 +66,7 @@ class LiveSubscriptionManager( private val relaysSocketManager: RelaysSocketManager, private val nostrRepository: NostrRepository, private val chatRepository: ChatRepository, + private val isForeground: StateFlow, ) { private val logger = Logger.withTag(TAG) @@ -112,6 +117,17 @@ class LiveSubscriptionManager( private val INITIAL_REOPEN_DELAY = 5.seconds private val MAX_REOPEN_DELAY = 5.minutes + + /** + * Matches what `ChatRoomListViewModel` queued, deliberately: the id a negentropy + * request is stored under is a hash of its filter, so an identical filter shape + * collapses into the same row rather than queueing the same reconciliation twice + * while both callers exist. + * + * The value itself barely matters — the negentropy pump drops `limit` outright, and it + * only survives into the plain-REQ fallback. + */ + private const val CATCH_UP_LIMIT = 50 } /** @@ -129,9 +145,33 @@ class LiveSubscriptionManager( * active wallet, so a wallet switch tears every subscription down and the new wallet's * call builds its own. */ - suspend fun observe(keyPair: KeyPair): Unit = coroutineScope { + suspend fun observe(keyPair: KeyPair) { val publicKey = keyPair.pubKey.toHexKey() - logger.i("Opening live subscriptions for $publicKey") + + // collectLatest is the whole lifecycle mechanism: going to the background cancels the + // block below, and every subscription's `finally` sends its CLOSE on the way out. + isForeground.collectLatest { foreground -> + if (!foreground) { + logger.i("Backgrounded; live subscriptions closed") + return@collectLatest + } + + runWhileForeground(publicKey = publicKey, keyPair = keyPair) + } + } + + private suspend fun runWhileForeground(publicKey: HexKey, keyPair: KeyPair): Unit = coroutineScope { + logger.i("Foregrounded; opening live subscriptions for $publicKey") + + // Before anything is asked for. A socket that was open when the OS suspended us + // reports itself connected on the way back while being functionally dead, and the + // relay dropped our subscriptions long ago. + runCatching { relaysSocketManager.reconnectAll() } + .onFailure { logger.w(throwable = it) { "Reconnect on foreground failed" } } + + // Live subscriptions cover the window we are online for; this covers the gap we were + // not. Neither subsumes the other. + launch(Dispatchers.IO) { queueCatchUpSynchronization(publicKey) } Relays.DefaultDMRelayList.forEach { relay -> launch(Dispatchers.IO) { @@ -147,6 +187,62 @@ class LiveSubscriptionManager( launch(Dispatchers.IO) { followGroupMembership(publicKey = publicKey, keyPair = keyPair) } } + /** + * Reconciles what we hold against what the relays hold, for the time we were away. + * + * A live subscription answers "what is new since I connected"; negentropy answers "what do + * you have that I don't". Coming back from the background is exactly the question only the + * second one can answer, and `limit` on the re-opened subscriptions is a window, not a + * guarantee. + */ + private suspend fun queueCatchUpSynchronization(publicKey: HexKey) { + val groupIds = liveGroupIds(publicKey) + + val giftWrapFilter = SynchronizationFilter( + kinds = arrayOf(GiftWrapEvent.KIND), + tags = mapOf("p" to listOf(publicKey)), + limit = CATCH_UP_LIMIT, + ) + + val requests = mutableListOf() + + Relays.DefaultDMRelayList.forEach { relay -> + requests += NegentropySynchronizeRequest( + id = NegentropySynchronizeRequest.computeId( + relayURL = relay.url, + synchronizationFilter = giftWrapFilter, + ), + purpose = "chat", + synchronizationFilter = giftWrapFilter, + relayURL = relay.url, + level = 0, + ) + + if (groupIds.isEmpty()) return@forEach + + val groupSynchronizationFilter = SynchronizationFilter( + kinds = arrayOf(GroupEvent.KIND), + tags = mapOf("h" to groupIds), + ) + requests += NegentropySynchronizeRequest( + id = NegentropySynchronizeRequest.computeId( + relayURL = relay.url, + synchronizationFilter = groupSynchronizationFilter, + ), + purpose = "mlsMessages", + synchronizationFilter = groupSynchronizationFilter, + relayURL = relay.url, + level = 0, + ) + } + + logger.i("Queueing ${requests.size} catch-up reconciliation(s) over ${groupIds.size} group(s)") + nostrRepository.queueNegentropySynchronizeRequest(requests) + } + + private suspend fun liveGroupIds(publicKey: HexKey): List = + chatRepository.getChatRoomListByPublicKey(publicKey).toLiveGroupIds() + /** * Every gift wrap addressed to us: direct messages, and the Marmot Welcome events that * make us a member of a group. @@ -195,18 +291,7 @@ class LiveSubscriptionManager( val subscriptions = mutableMapOf() chatRepository.observeChatRoomListByPublicKey(publicKey) - .map { rooms -> - rooms - // An MLS group is a room with group state; a NIP-17 room has none and is - // served by the gift wrap subscription instead. A room we have left or - // deleted keeps its history locally but must stop pulling new messages. - .filter { it.chatRoom.mlsGroupState != null } - .filter { it.chatRoom.leftGroupAt == null && it.chatRoom.deletedAt == null } - .map { it.chatRoom.id } - .sorted() - } - // Sorted first so that the same membership in a different row order is the same - // value, and a re-emit that changes nothing costs nothing. + .map { rooms -> rooms.toLiveGroupIds() } .distinctUntilChanged() .debounce(GROUP_CHANGE_DEBOUNCE) .collect { groupIds -> @@ -284,6 +369,21 @@ class LiveSubscriptionManager( } } + /** + * An MLS group is a room with group state; a NIP-17 room has none and is served by the + * gift wrap subscription instead. A room we have left or deleted keeps its history + * locally but must stop pulling new messages. + * + * Sorted so the same membership in a different row order is the same value, and a + * re-emit that changes nothing costs nothing downstream. + */ + private fun List.toLiveGroupIds() = + this + .filter { it.chatRoom.mlsGroupState != null } + .filter { it.chatRoom.leftGroupAt == null && it.chatRoom.deletedAt == null } + .map { it.chatRoom.id } + .sorted() + private fun subscriptionKey(relayUrl: String, subId: String) = "$relayUrl|$subId" private fun subscriptionIndex(key: String) = diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt index 73a08734..d4a1be20 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt @@ -166,6 +166,27 @@ class RelayPool( fun hasRelays() = relays.isNotEmpty() + /** + * Tears every socket down and immediately builds it again. + * + * For coming back from the background, where trusting the connection is the mistake: a + * socket that was open when the OS suspended the process reports itself connected on the + * way back while being functionally dead, and the relay has long since dropped the + * subscriptions it was carrying. + * + * Re-opening here rather than leaving it to the next send is deliberate — it is what makes + * this a RE-connection, so the retained subscriptions are replayed and any collector still + * attached from before the gap starts receiving again. + */ + suspend fun reconnectAll() { + socketClients.forEach { client -> + updateRelayStatus(url = client.socketUrl, connected = false) + runCatching { client.close() } + runCatching { client.ensureSocketConnectionOrThrow() } + .onFailure { logger.w(throwable = it) { "Could not re-open ${client.socketUrl}" } } + } + } + suspend fun tryConnectingToRelay(url: String) { runCatching { socketClients.find { it.socketUrl == url }?.ensureSocketConnectionOrThrow() diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt index 256828de..8b18428f 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelaysSocketManager.kt @@ -126,6 +126,9 @@ class RelaysSocketManager( ) } + /** @see RelayPool.reconnectAll */ + suspend fun reconnectAll() = relayPool.reconnectAll() + fun tryConnectingToAllRelays() { relayPool.relays.forEach { scope.launch { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt index 9f9d612e..1a6cd763 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/composable/navigation/MantraNavHost.kt @@ -3,10 +3,14 @@ package press.mantra.compose.ui.composable.navigation import androidx.compose.foundation.layout.fillMaxSize import androidx.compose.material3.Surface import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.remember import androidx.compose.ui.Modifier import androidx.compose.ui.graphics.Color +import press.mantra.compose.AppLifecycle +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleEventObserver import androidx.lifecycle.compose.LocalLifecycleOwner import androidx.lifecycle.viewmodel.compose.viewModel import androidx.navigation.NavHostController @@ -219,6 +223,22 @@ fun MantraNavHost( // TODO: Produce a synchronization UI element... + // The one place in the app that knows whether it is on screen. Everything that holds a + // relay connection open reads AppLifecycle rather than a lifecycle owner, because those + // consumers are application-scoped coroutines that outlive any composition. + DisposableEffect(lifecycleOwner) { + val observer = LifecycleEventObserver { _, event -> + when (event) { + Lifecycle.Event.ON_START -> AppLifecycle.enteredForeground() + Lifecycle.Event.ON_STOP -> AppLifecycle.enteredBackground() + else -> Unit + } + } + + lifecycleOwner.lifecycle.addObserver(observer) + onDispose { lifecycleOwner.lifecycle.removeObserver(observer) } + } + LaunchedEffect(lifecycleOwner) { navigationViewModel.navigationUIState.collect { state -> when (state) { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt index 146fbc81..fab0fb71 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/ui/view/model/SynchronizationViewModel.kt @@ -8,6 +8,7 @@ import press.mantra.compose.database.model.BroadcastNostrEventRequest import press.mantra.compose.database.model.SynchronizeNostrEventRequest import press.mantra.compose.database.model.types.SynchronizationFilter import press.mantra.compose.network.dto.toRelayDTO +import press.mantra.compose.AppLifecycle import press.mantra.compose.managers.LiveSubscriptionManager import press.mantra.compose.network.relays.RelayPool import press.mantra.compose.network.relays.RelaysSocketManager @@ -81,6 +82,7 @@ class SynchronizationViewModel( relaysSocketManager = relaysSocketManager, nostrRepository = nostrRepository, chatRepository = chatRepository, + isForeground = AppLifecycle.isForeground, ) companion object {