diff --git a/composeApp/src/androidMain/kotlin/su/reya/coop/NostrForegroundService.kt b/composeApp/src/androidMain/kotlin/su/reya/coop/NostrForegroundService.kt index abb13ad..82bcf57 100644 --- a/composeApp/src/androidMain/kotlin/su/reya/coop/NostrForegroundService.kt +++ b/composeApp/src/androidMain/kotlin/su/reya/coop/NostrForegroundService.kt @@ -65,8 +65,10 @@ class NostrForegroundService : Service() { } catch (e: Exception) { throw IllegalStateException("Failed to initialize Nostr Client", e) } + // Connect to bootstrap relays nostr.connectBootstrapRelays() + // Handle notifications nostr.handleNotifications( onMetadataUpdate = { pubkey, metadata -> @@ -75,9 +77,6 @@ class NostrForegroundService : Service() { onContactListUpdate = { contacts -> serviceScope.launch { nostr.emitContactListUpdate(contacts) } }, - onSubscriptionClose = { - serviceScope.launch { nostr.emitSubscriptionClosed() } - }, onNewMessage = { event -> serviceScope.launch { if (!isUserInApp()) { diff --git a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/ChatScreen.kt b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/ChatScreen.kt index fde4acf..2907e2d 100644 --- a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/ChatScreen.kt +++ b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/ChatScreen.kt @@ -4,15 +4,11 @@ import androidx.compose.foundation.clickable import androidx.compose.foundation.layout.Arrangement import androidx.compose.foundation.layout.Box import androidx.compose.foundation.layout.Column -import androidx.compose.foundation.layout.IntrinsicSize import androidx.compose.foundation.layout.PaddingValues import androidx.compose.foundation.layout.Row import androidx.compose.foundation.layout.Spacer -import androidx.compose.foundation.layout.aspectRatio -import androidx.compose.foundation.layout.fillMaxHeight import androidx.compose.foundation.layout.fillMaxSize import androidx.compose.foundation.layout.fillMaxWidth -import androidx.compose.foundation.layout.height import androidx.compose.foundation.layout.padding import androidx.compose.foundation.layout.size import androidx.compose.foundation.layout.widthIn @@ -526,9 +522,7 @@ fun ChatInput( ) { Surface(modifier = Modifier.fillMaxWidth()) { Row( - modifier = Modifier - .padding(horizontal = 16.dp, vertical = 8.dp) - .height(IntrinsicSize.Min), + modifier = Modifier.padding(horizontal = 16.dp, vertical = 8.dp), verticalAlignment = Alignment.Bottom ) { TextField( @@ -545,9 +539,7 @@ fun ChatInput( Spacer(modifier = Modifier.size(8.dp)) FilledTonalIconButton( onClick = onSend, - modifier = Modifier - .fillMaxHeight() - .aspectRatio(1f), + modifier = Modifier.size(56.dp), colors = IconButtonDefaults.filledTonalIconButtonColors( containerColor = MaterialTheme.colorScheme.surfaceVariant, contentColor = MaterialTheme.colorScheme.onSurfaceVariant diff --git a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/HomeScreen.kt b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/HomeScreen.kt index 0eeb362..b1dbf27 100644 --- a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/HomeScreen.kt +++ b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/HomeScreen.kt @@ -127,7 +127,8 @@ fun HomeScreen() { val userProfile by currentUserProfile.collectAsStateWithLifecycle() val chatRooms by viewModel.chatRooms.collectAsStateWithLifecycle() val isRelayListEmpty by viewModel.isRelayListEmpty.collectAsStateWithLifecycle() - val isPartialProcessedGiftWrap by viewModel.isPartialProcessedGiftWrap.collectAsState(initial = false) + val isSyncing by viewModel.isSyncing.collectAsStateWithLifecycle() + val isPartialProcessedGiftWrap by viewModel.isPartialProcessedGiftWrap.collectAsStateWithLifecycle() val isBannerDismissed by viewModel.isNotificationBannerDismissed.collectAsState() val expandedFab by remember { derivedStateOf { listState.firstVisibleItemIndex == 0 } } @@ -185,10 +186,18 @@ fun HomeScreen() { containerColor = MaterialTheme.colorScheme.surfaceContainer, ), title = { - Text( - text = "Coop", - style = MaterialTheme.typography.titleMediumEmphasized - ) + Row( + verticalAlignment = Alignment.CenterVertically, + horizontalArrangement = Arrangement.spacedBy(8.dp) + ) { + Text( + text = "Coop", + style = MaterialTheme.typography.titleMediumEmphasized + ) + if (isSyncing) { + LoadingIndicator(modifier = Modifier.size(24.dp)) + } + } }, actions = { // QR Scanner diff --git a/shared/src/commonMain/kotlin/su/reya/coop/Nostr.kt b/shared/src/commonMain/kotlin/su/reya/coop/Nostr.kt index ca2ce5f..ba0fcce 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/Nostr.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/Nostr.kt @@ -4,12 +4,15 @@ import io.ktor.client.HttpClient import io.ktor.client.call.body import io.ktor.client.request.get import io.ktor.client.statement.HttpResponse -import kotlinx.coroutines.Job -import kotlinx.coroutines.delay +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch import kotlinx.coroutines.supervisorScope import rust.nostr.sdk.AckPolicy @@ -56,7 +59,6 @@ import rust.nostr.sdk.nip17ExtractRelayList import rust.nostr.sdk.nip59MakeGiftWrapAsync import kotlin.coroutines.cancellation.CancellationException import kotlin.time.Duration -import kotlin.time.Duration.Companion.milliseconds object NostrManager { val instance = Nostr() @@ -95,18 +97,20 @@ class Nostr { private val _contactListUpdates = MutableSharedFlow>(extraBufferCapacity = 100) val contactListUpdates = _contactListUpdates.asSharedFlow() - private val _subscriptionClosed = MutableSharedFlow(extraBufferCapacity = 10) - val subscriptionClosed = _subscriptionClosed.asSharedFlow() + private val _messageSyncState = MutableStateFlow(MessageSyncState()) + val messageSyncState = _messageSyncState.asStateFlow() - suspend fun emitNewEvent(event: UnsignedEvent) = _newEvents.emit(event) + suspend fun emitNewEvent(event: UnsignedEvent) { + _newEvents.emit(event) + } - suspend fun emitSubscriptionClosed() = _subscriptionClosed.emit(Unit) - - suspend fun emitMetadataUpdate(pubkey: PublicKey, metadata: Metadata) = + suspend fun emitMetadataUpdate(pubkey: PublicKey, metadata: Metadata) { _metadataUpdates.emit(pubkey to metadata) + } - suspend fun emitContactListUpdate(contacts: List) = + suspend fun emitContactListUpdate(contacts: List) { _contactListUpdates.emit(contacts) + } suspend fun init( dbPath: String, @@ -266,17 +270,39 @@ class Nostr { } } + @OptIn(ExperimentalCoroutinesApi::class) suspend fun handleNotifications( onMetadataUpdate: (PublicKey, Metadata) -> Unit, onContactListUpdate: (List) -> Unit, onNewMessage: (UnsignedEvent) -> Unit, - onSubscriptionClose: () -> Unit, ) = supervisorScope { val now = Timestamp.now() val processedEvent = mutableSetOf() val notifications = client?.notifications() ?: return@supervisorScope - var eoseTrackerJob: Job? = null + val giftWrapQueue = Channel(Channel.UNLIMITED) + var processedCount = 0 + var eoseReceived = false + + launch(Dispatchers.Default) { + for (event in giftWrapQueue) { + val rumor = extractRumor(event) ?: continue + processedCount++ + + // Trigger new message notification + if (rumor.createdAt().asSecs() >= now.asSecs()) { + onNewMessage(rumor) + } + + // Update sync state + _messageSyncState.update { + it.copy( + processedCount = processedCount, + isSyncing = !eoseReceived || !giftWrapQueue.isEmpty + ) + } + } + } while (true) { val notification = notifications.next() ?: continue @@ -293,59 +319,47 @@ class Nostr { if (processedEvent.contains(event.id())) continue processedEvent.add(event.id()) - if (event.kind().asStd()?.equals(KindStandard.METADATA) == true) { - try { - val metadata = Metadata.fromJson(event.content()) - onMetadataUpdate(event.author(), metadata) - } catch (e: Exception) { - println("Failed to parse metadata: $e") - } - } - - if (event.kind().asStd()?.equals(KindStandard.CONTACT_LIST) == true) { - if (isSignedByUser(event = event)) { - val pubkeys = event.tags().publicKeys() - // Get mutual contacts - getMutualContacts(pubkeys) - // Emit contact list update - onContactListUpdate(pubkeys) - } - } - - if (event.kind().asStd()?.equals(KindStandard.INBOX_RELAYS) == true) { - // Get all gift wrap events for the current user - if (isSignedByUser(event = event)) { - getUserMessages(msgRelayList = event) - } - } - - if (event.kind().asStd()?.equals(KindStandard.GIFT_WRAP) == true) { - val rumor = extractRumor(event) - - // Logic to notify UI after processing - // Cancel previous tracker if it exists - eoseTrackerJob?.cancel() - - // Start a new tracker - eoseTrackerJob = launch { - delay(10000.milliseconds) // Wait for 10 seconds - onSubscriptionClose() - } - - // Handle new message - rumor?.createdAt()?.asSecs()?.let { - if (it >= now.asSecs()) { - onNewMessage(rumor) + when (event.kind().asStd()) { + KindStandard.METADATA -> { + try { + val metadata = Metadata.fromJson(event.content()) + onMetadataUpdate(event.author(), metadata) + } catch (e: Exception) { + println("Failed to parse metadata: $e") } } + + KindStandard.CONTACT_LIST -> { + if (isSignedByUser(event = event)) { + val pubkeys = event.tags().publicKeys() + // Get mutual contacts + getMutualContacts(pubkeys) + // Emit contact list update + onContactListUpdate(pubkeys) + } + } + + KindStandard.INBOX_RELAYS -> { + // Get all gift wrap events for the current user + if (isSignedByUser(event = event)) { + getUserMessages(msgRelayList = event) + } + } + + KindStandard.GIFT_WRAP -> { + giftWrapQueue.send(event) + } + + else -> {} } } is RelayMessageEnum.EndOfStoredEvents -> { - val subscriptionId = message.subscriptionId - - if (subscriptionId == "gift-wraps") { - onSubscriptionClose() + if (message.subscriptionId == "gift-wraps") { + eoseReceived = true + if (giftWrapQueue.isEmpty) { + _messageSyncState.update { it.copy(isSyncing = false) } + } } } @@ -1036,3 +1050,8 @@ class Nostr { } } } + +data class MessageSyncState( + val processedCount: Int = 0, + val isSyncing: Boolean = false +) diff --git a/shared/src/commonMain/kotlin/su/reya/coop/NostrViewModel.kt b/shared/src/commonMain/kotlin/su/reya/coop/NostrViewModel.kt index 0e7e915..ff6e71c 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/NostrViewModel.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/NostrViewModel.kt @@ -13,10 +13,13 @@ import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch import kotlinx.coroutines.withContext @@ -79,9 +82,14 @@ class NostrViewModel( val errorEvents = _errorEvents.receiveAsFlow() private val _metadataStore = mutableMapOf>() + private val metadataRequestChannel = Channel(Channel.UNLIMITED) + private val seenPublicKeys = mutableSetOf() + val isSyncing = nostr.messageSyncState.map { it.isSyncing } + .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), false) + init { // Skip the splash screen if a user is already logged in if (nostr.signer.currentUser != null) { @@ -108,7 +116,6 @@ class NostrViewModel( viewModelScope.launch { lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) { coroutineScope { - launch { refreshChatRooms() } launch { runObserver() } launch { runMetadataBatching() } } @@ -148,6 +155,21 @@ class NostrViewModel( } private suspend fun runObserver() = coroutineScope { + // Observe message sync progress + launch { + nostr.messageSyncState.collect { state -> + // When at least some messages are processed, allow UI to show the list + if (state.processedCount > 0) { + _isPartialProcessedGiftWrap.value = true + } + + // Refresh UI every 10 messages OR when sync is fully done + if (state.processedCount % 10 == 0 || !state.isSyncing) { + refreshChatRooms() + } + } + } + // Observe new messages launch { nostr.newEvents.collect { event -> @@ -181,14 +203,6 @@ class NostrViewModel( updateMetadata(pubkey, metadata) } } - - // Observes subscription close - launch { - nostr.subscriptionClosed.collect { - getChatRooms() - _isPartialProcessedGiftWrap.value = true - } - } } private suspend fun runMetadataBatching() = coroutineScope {