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 7b38202..4220715 100644 --- a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/HomeScreen.kt +++ b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/HomeScreen.kt @@ -695,8 +695,7 @@ fun NewRequests(requests: List) { @Composable fun ChatRoom(room: Room, onClick: () -> Unit) { val profileCache = LocalProfileCache.current - val roomState by room.uiStateFlow(profileCache) - .collectAsStateWithLifecycle(RoomUiState()) + val roomState by room.uiStateFlow(profileCache).collectAsStateWithLifecycle(RoomUiState()) ListItem( modifier = Modifier.clickable(onClick = onClick), diff --git a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/NewChatScreen.kt b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/NewChatScreen.kt index a8d1076..c28de53 100644 --- a/composeApp/src/androidMain/kotlin/su/reya/coop/screens/NewChatScreen.kt +++ b/composeApp/src/androidMain/kotlin/su/reya/coop/screens/NewChatScreen.kt @@ -397,7 +397,7 @@ fun ContactListItem( supportingContent = { Text(text = pubkey.short()) }, content = { Text( - text = profile?.name ?: "", + text = profile?.name ?: "Unknown", style = MaterialTheme.typography.titleMediumEmphasized, ) } diff --git a/shared/src/commonMain/kotlin/su/reya/coop/Room.kt b/shared/src/commonMain/kotlin/su/reya/coop/Room.kt index 2a01f29..d761aad 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/Room.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/Room.kt @@ -112,13 +112,8 @@ fun UnsignedEvent.roomId(): Long { val pubkeys: MutableList = mutableListOf() pubkeys.add(this.author()) pubkeys.addAll(this.tags().publicKeys()) - - // Sort and hash the list of public keys - val sortedUniqueKeys = pubkeys - .distinctBy { it.toBech32() } - .sortedBy { it.toBech32() } - - return sortedUniqueKeys.hashCode().toLong() + + return pubkeys.map { it.toBech32() }.distinct().sorted().hashCode().toLong() } fun Timestamp.formatAsTime(): String { diff --git a/shared/src/commonMain/kotlin/su/reya/coop/nostr/Messaging.kt b/shared/src/commonMain/kotlin/su/reya/coop/nostr/Messaging.kt index 1b86a7b..3e3a002 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/nostr/Messaging.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/nostr/Messaging.kt @@ -39,12 +39,11 @@ class MessageManager(private val nostr: Nostr) { private val client: Client? get() = nostr.client private val signer: UniversalSigner get() = nostr.signer - val sentEvents: MutableMap> = mutableMapOf() - val rumorMap: MutableMap = mutableMapOf() - private val _messageSyncState = MutableStateFlow(MessageSyncState()) val messageSyncState = _messageSyncState.asStateFlow() + val rumorMap: MutableMap = mutableMapOf() + fun updateSyncState(update: (MessageSyncState) -> MessageSyncState) { _messageSyncState.update(update) } @@ -73,6 +72,8 @@ class MessageManager(private val nostr: Nostr) { target = ReqTarget.manual(target), id = "gift-wraps" ) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to fetch user messages: ${e.message}", e) } @@ -225,6 +226,8 @@ class MessageManager(private val nostr: Nostr) { // Filter out events without public keys (receivers) ?.filter { it.tags().publicKeys().isNotEmpty() } ?.sortedByDescending { it.createdAt().asSecs() } ?: emptyList() + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to get chat room messages: ${e.message}", e) } @@ -248,6 +251,8 @@ class MessageManager(private val nostr: Nostr) { connectMsgRelays(event) } } + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to fetch relays: ${e.message}", e) } @@ -327,9 +332,6 @@ class MessageManager(private val nostr: Nostr) { ) if (output != null) { - // Keep track of sent events - sentEvents[output.id] = emptyList() - // Keep track of rumor IDs val id = rumor.id() ?: throw IllegalStateException("Rumor ID is null") rumorMap[id] = output.id @@ -340,6 +342,8 @@ class MessageManager(private val nostr: Nostr) { } } } + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to send message: ${e.message}", e) } diff --git a/shared/src/commonMain/kotlin/su/reya/coop/nostr/Nostr.kt b/shared/src/commonMain/kotlin/su/reya/coop/nostr/Nostr.kt index e7617d4..2580334 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/nostr/Nostr.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/nostr/Nostr.kt @@ -9,9 +9,9 @@ import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.first -import kotlinx.coroutines.withTimeoutOrNull import kotlinx.coroutines.launch import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.withTimeoutOrNull import rust.nostr.sdk.AsyncNostrSigner import rust.nostr.sdk.Client import rust.nostr.sdk.ClientBuilder @@ -108,14 +108,6 @@ class Nostr( relays.connectBootstrapRelays() } - suspend fun reconnect() { - relays.reconnect() - } - - suspend fun disconnect() { - relays.disconnect() - } - suspend fun prune() { try { client?.database()?.wipe() @@ -182,8 +174,6 @@ class Nostr( when (notification) { is ClientNotification.Message -> { - val relayUrl = notification.relayUrl - when (val message = notification.message.asEnum()) { is RelayMessageEnum.EventMsg -> { val event = message.event @@ -237,14 +227,6 @@ class Nostr( } } - is RelayMessageEnum.Ok -> { - if (messages.sentEvents.containsKey(message.eventId)) { - val currentRelays = - messages.sentEvents[message.eventId] ?: emptyList() - messages.sentEvents[message.eventId] = currentRelays + relayUrl - } - } - else -> { /* Ignore other message types */ } diff --git a/shared/src/commonMain/kotlin/su/reya/coop/nostr/ProfileManager.kt b/shared/src/commonMain/kotlin/su/reya/coop/nostr/ProfileManager.kt index f228757..81439d6 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/nostr/ProfileManager.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/nostr/ProfileManager.kt @@ -27,6 +27,7 @@ import rust.nostr.sdk.ReqTarget import rust.nostr.sdk.SendEventTarget import rust.nostr.sdk.SubscribeAutoCloseOptions import rust.nostr.sdk.Timestamp +import kotlin.coroutines.cancellation.CancellationException import kotlin.time.Duration class ProfileManager(private val nostr: Nostr) { @@ -80,6 +81,8 @@ class ProfileManager(private val nostr: Nostr) { val opts = SubscribeAutoCloseOptions().exitPolicy(ReqExitPolicy.ExitOnEose) client?.subscribe(target = target, id = "user-metadata", closeOn = opts) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to fetch user metadata: ${e.message}", e) } @@ -92,6 +95,8 @@ class ProfileManager(private val nostr: Nostr) { val relays = RelayManager.BOOTSTRAP_RELAYS.map { RelayUrl.parse(it) } client?.sync(filter, relays) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { println("Failed to sync mutual contacts: ${e.message}") } @@ -174,6 +179,8 @@ class ProfileManager(private val nostr: Nostr) { ) return newMetadata + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to update identity: ${e.message}", e) } @@ -186,6 +193,8 @@ class ProfileManager(private val nostr: Nostr) { val event = client?.database()?.query(filter)?.first() ?: return null Metadata.fromJson(event.content()) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { println("Failed to get latest metadata: ${e.message}") null @@ -208,6 +217,8 @@ class ProfileManager(private val nostr: Nostr) { } return results + } catch (e: CancellationException) { + throw e } catch (e: Exception) { println("Failed to get all cache metadata: ${e.message}") return emptyMap() @@ -232,6 +243,8 @@ class ProfileManager(private val nostr: Nostr) { } client?.subscribe(target = ReqTarget.manual(target), closeOn = opts) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to fetch metadata batch: ${e.message}", e) } @@ -247,6 +260,8 @@ class ProfileManager(private val nostr: Nostr) { target = SendEventTarget.broadcast(), ackPolicy = AckPolicy.none(), ) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to set contact list: ${e.message}", e) } @@ -258,6 +273,8 @@ class ProfileManager(private val nostr: Nostr) { val bodyString: String = response.body() return Nip05Profile.fromJson(address, bodyString) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to fetch profile from address: ${e.message}", e) } @@ -269,6 +286,8 @@ class ProfileManager(private val nostr: Nostr) { val profile = profileFromAddress(httpClient, address) return profile.publicKey() + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to search address: ${e.message}", e) } @@ -304,6 +323,8 @@ class ProfileManager(private val nostr: Nostr) { } return results + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to search nostr: ${e.message}", e) } @@ -323,6 +344,8 @@ class ProfileManager(private val nostr: Nostr) { ) return events?.first()?.createdAt() + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to get latest activity: ${e.message}", e) } @@ -340,6 +363,8 @@ class ProfileManager(private val nostr: Nostr) { val pubkeys = events?.first()?.tags()?.publicKeys() ?: listOf() return pubkeys.contains(pubkey) + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to get mutual contacts: ${e.message}", e) } @@ -361,6 +386,8 @@ class ProfileManager(private val nostr: Nostr) { } return contacts.toSet() + } catch (e: CancellationException) { + throw e } catch (e: Exception) { throw IllegalStateException("Failed to get mutual contacts: ${e.message}", e) } diff --git a/shared/src/commonMain/kotlin/su/reya/coop/nostr/RelayManager.kt b/shared/src/commonMain/kotlin/su/reya/coop/nostr/RelayManager.kt index 12482bd..5242dad 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/nostr/RelayManager.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/nostr/RelayManager.kt @@ -130,7 +130,7 @@ class RelayManager(private val nostr: Nostr) { } } - suspend fun setRelaylist(relays: Map) { + suspend fun setRelayList(relays: Map) { try { val event = EventBuilder.relayList(relays).finalizeAsync(signer) @@ -143,7 +143,7 @@ class RelayManager(private val nostr: Nostr) { throw IllegalStateException("Failed to set msg relays: ${e.message}", e) } } - + suspend fun setMsgRelays(urls: List) { try { val event = EventBuilder.nip17RelayList(urls).finalizeAsync(signer) diff --git a/shared/src/commonMain/kotlin/su/reya/coop/repository/AccountRepository.kt b/shared/src/commonMain/kotlin/su/reya/coop/repository/AccountRepository.kt index e7562c7..f92956f 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/repository/AccountRepository.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/repository/AccountRepository.kt @@ -511,7 +511,7 @@ class AccountRepository( val relays = currentUserRelayListInternal().toMutableMap() relays[relayUrl] = RelayMetadata.WRITE - nostr.relays.setRelaylist(relays) + nostr.relays.setRelayList(relays) } catch (e: Exception) { showError("Error: ${e.message}") } @@ -525,7 +525,7 @@ class AccountRepository( val relays = currentUserRelayListInternal().toMutableMap() relays[relayUrl] = RelayMetadata.READ - nostr.relays.setRelaylist(relays) + nostr.relays.setRelayList(relays) } catch (e: Exception) { showError("Error: ${e.message}") } @@ -539,7 +539,7 @@ class AccountRepository( val relays = currentUserRelayListInternal().toMutableMap() relays.remove(relayUrl) - nostr.relays.setRelaylist(relays) + nostr.relays.setRelayList(relays) } catch (e: Exception) { showError("Error: ${e.message}") } diff --git a/shared/src/commonMain/kotlin/su/reya/coop/repository/ChatRepository.kt b/shared/src/commonMain/kotlin/su/reya/coop/repository/ChatRepository.kt index 6dbe60a..f26410b 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/repository/ChatRepository.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/repository/ChatRepository.kt @@ -51,13 +51,16 @@ class ChatRepository( ) val newEvents = _newEvents.asSharedFlow() - val chatRooms = state.map { it.rooms.values.sortedByDescending { it.createdAt.asSecs() } } + val chatRooms = state + .map { it.rooms.values.sortedByDescending { it.createdAt.asSecs() } } .stateIn(scope, SharingStarted.WhileSubscribed(5000), emptyList()) - val isSyncing = nostr.messages.messageSyncState.map { it.isSyncing } + val isSyncing = nostr.messages.messageSyncState + .map { it.isSyncing } .stateIn(scope, SharingStarted.WhileSubscribed(5000), false) - val isPartialProcessedGiftWrap = state.map { it.isPartialProcessedGiftWrap } + val isPartialProcessedGiftWrap = state + .map { it.isPartialProcessedGiftWrap } .stateIn(scope, SharingStarted.WhileSubscribed(5000), false) init { @@ -82,18 +85,7 @@ class ChatRepository( // Observe new messages launch { nostr.newEvents.collect { event -> - val roomId = event.roomId() - val existingRoom = _state.value.rooms[roomId] - - if (existingRoom == null) { - val currentUser = nostr.signer.getPublicKeyAsync() ?: return@collect - val newRoom = Room.new(event, currentUser) - _state.update { it.copy(rooms = it.rooms + (newRoom.id to newRoom)) } - } else { - updateRoomList(roomId, event) - } - - _newEvents.tryEmit(event) + updateRoomState(event) } } } @@ -194,9 +186,10 @@ class ChatRepository( content = message, subject = room.subject, replies = replies, - onRumorCreated = { event -> - updateRoomList(roomId, event) - scope.launch(defaultDispatcher) { _newEvents.tryEmit(event) } + onRumorCreated = { + scope.launch(defaultDispatcher) { + updateRoomState(it, roomId) + } }, ) } catch (e: Exception) { @@ -223,26 +216,32 @@ class ChatRepository( } } - fun isMessageSent(id: EventId): Boolean { - val giftWrapId = nostr.messages.rumorMap[id] + private suspend fun updateRoomState(event: UnsignedEvent, roomId: Long = event.roomId()) { + val currentUser = nostr.signer.getPublicKeyAsync() ?: return - if (giftWrapId != null) { - val isSent = nostr.messages.sentEvents[giftWrapId]?.isNotEmpty() ?: false - return isSent - } else { - return false - } - } - - private fun updateRoomList(roomId: Long, newMessage: UnsignedEvent) { _state.update { currentState -> - val room = currentState.rooms[roomId] ?: return@update currentState - val updatedRoom = room.copy( - lastMessage = newMessage.content(), - createdAt = newMessage.createdAt() - ) - currentState.copy(rooms = currentState.rooms + (roomId to updatedRoom)) + val rooms = currentState.rooms.toMutableMap() + val existingRoom = rooms[roomId] + + if (existingRoom == null) { + // New room discovery + val newRoom = Room.new(event, currentUser, roomId) + rooms[newRoom.id] = newRoom + } else if (event.createdAt().asSecs() >= existingRoom.createdAt.asSecs()) { + // Only update preview if message is newer (handles sync/late arrivals) + rooms[roomId] = existingRoom.copy( + lastMessage = event.content(), + createdAt = event.createdAt() + ) + } else { + // Don't update the room list state for older messages + return@update currentState + } + currentState.copy(rooms = rooms) } + + // Notify subscribers about the new event (for the active chat screen) + _newEvents.tryEmit(event) } fun resetInternalState() { diff --git a/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/AccountViewModel.kt b/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/AccountViewModel.kt index e25a53e..8030dab 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/AccountViewModel.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/AccountViewModel.kt @@ -13,7 +13,6 @@ import su.reya.coop.repository.AccountState class AccountViewModel( private val repository: AccountRepository, ) : ViewModel(), ErrorHost by repository { - val state: StateFlow = repository.state val isUpdatingProfile: StateFlow = repository.isUpdatingProfile val currentUserProfile: StateFlow = repository.currentUserProfile diff --git a/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/ProfileCache.kt b/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/ProfileCache.kt index 8e6a638..420043a 100644 --- a/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/ProfileCache.kt +++ b/shared/src/commonMain/kotlin/su/reya/coop/viewmodel/ProfileCache.kt @@ -1,16 +1,15 @@ package su.reya.coop.viewmodel +import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withTimeoutOrNull import rust.nostr.sdk.PublicKey import su.reya.coop.Profile @@ -19,27 +18,41 @@ import kotlin.time.Duration.Companion.milliseconds class ProfileCache( private val nostr: Nostr, + defaultDispatcher: CoroutineDispatcher = Dispatchers.Default, ) : ErrorHost by createErrorHost() { - private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - private val profilesMutex = Mutex() - private val profiles = mutableMapOf>() + private val scope = CoroutineScope(SupervisorJob() + defaultDispatcher) + private val profiles = MutableStateFlow>>(emptyMap()) + private val seenPublicKeys = MutableStateFlow>(emptySet()) private val metadataRequestChannel = Channel(Channel.UNLIMITED) - private val seenPublicKeys = mutableSetOf() init { - scope.launch { runObserver() } - scope.launch { runMetadataBatching() } scope.launch { - nostr.waitUntilInitialized() - loadCacheMetadata() + try { + runObserver() + } catch (e: Exception) { + showError("Metadata observer failed: ${e.message}") + } + } + scope.launch { + try { + runMetadataBatching() + } catch (e: Exception) { + showError("Metadata batching failed: ${e.message}") + } + } + scope.launch { + try { + nostr.waitUntilInitialized() + getCachedMetadata() + } catch (e: Exception) { + showError("Failed to load initial cache: ${e.message}") + } } } - private suspend fun runObserver() = coroutineScope { - launch { - nostr.profiles.metadataUpdates.collect { (pubkey, metadata) -> - updateMetadata(pubkey, Profile(pubkey, metadata)) - } + private suspend fun runObserver() { + nostr.profiles.metadataUpdates.collect { (pubkey, metadata) -> + updateMetadata(pubkey, Profile(pubkey, metadata)) } } @@ -57,37 +70,69 @@ class ProfileCache( batch.add(nextKey) } - nostr.profiles.fetchMetadataBatch(batch.toList()) - } - } - - private suspend fun loadCacheMetadata() { - val cache = nostr.profiles.getAllCacheMetadata() - - profilesMutex.withLock { - cache.forEach { (pubkey, metadata) -> - val profile = Profile(pubkey, metadata) - profiles.getOrPut(pubkey) { MutableStateFlow(null) }.value = profile - seenPublicKeys.add(pubkey) + try { + nostr.profiles.fetchMetadataBatch(batch.toList()) + } catch (e: Exception) { + // Allow these keys to be requested again since the fetch failed + seenPublicKeys.update { it - batch } + println("Failed to fetch metadata batch: ${e.message}") } } } - private fun requestMetadata(pubkey: PublicKey) { - if (seenPublicKeys.add(pubkey)) { - metadataRequestChannel.trySend(pubkey) + private suspend fun getCachedMetadata() { + val cache = nostr.profiles.getAllCacheMetadata() + cache.forEach { (pubkey, metadata) -> + updateMetadata(pubkey, Profile(pubkey, metadata)) } } - private suspend fun updateMetadata(pubkey: PublicKey, profile: Profile) { - profilesMutex.withLock { - profiles.getOrPut(pubkey) { MutableStateFlow(null) }.value = profile + private fun requestMetadata(pubkey: PublicKey) { + var added = false + seenPublicKeys.update { current -> + if (current.contains(pubkey)) { + added = false + current + } else { + added = true + current + pubkey + } } + if (added) metadataRequestChannel.trySend(pubkey) + } + + private fun updateMetadata(pubkey: PublicKey, profile: Profile) { + profiles.update { current -> + val flow = current[pubkey] ?: MutableStateFlow(null) + flow.value = profile + if (current.containsKey(pubkey)) current else current + (pubkey to flow) + } + seenPublicKeys.update { it + pubkey } } fun getMetadata(pubkey: PublicKey): StateFlow { - val flow = profiles.getOrPut(pubkey) { MutableStateFlow(null) } - if (flow.value == null) requestMetadata(pubkey) - return flow.asStateFlow() + val currentMap = profiles.value + val existingFlow = currentMap[pubkey] + + if (existingFlow != null) { + if (existingFlow.value == null) requestMetadata(pubkey) + return existingFlow.asStateFlow() + } + + val newFlow = MutableStateFlow(null) + var resultFlow = newFlow + + profiles.update { prev -> + if (prev.containsKey(pubkey)) { + resultFlow = prev[pubkey]!! + prev + } else { + prev + (pubkey to newFlow) + } + } + + if (resultFlow.value == null) requestMetadata(pubkey) + + return resultFlow.asStateFlow() } }