optimize flow usage

shared/src/commonMain/kotlin/su/reya/coop/nostr/Messaging.kt

@@ -164,7 +164,7 @@ class MessageManager(private val nostr: Nostr) {
         try {
             val userPubkey =
                 signer.getPublicKeyAsync() ?: throw IllegalStateException("User not signed in")
-            
+
             val kind = Kind.fromStd(KindStandard.APPLICATION_SPECIFIC_DATA)
             val kTag = SingleLetterTag.lowercase(Alphabet.K)
 

shared/src/commonMain/kotlin/su/reya/coop/nostr/Nostr.kt

@@ -2,11 +2,13 @@ package su.reya.coop.nostr
 
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 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 rust.nostr.sdk.AsyncNostrSigner
@@ -30,6 +32,7 @@ import rust.nostr.sdk.Timestamp
 import rust.nostr.sdk.UnsignedEvent
 import rust.nostr.sdk.initLogger
 import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
 
 object NostrManager {
     val instance = Nostr()
@@ -59,17 +62,18 @@ class Nostr {
 
     private val isInitialized = MutableStateFlow(false)
 
-    private val _newEvents = MutableSharedFlow<UnsignedEvent>(extraBufferCapacity = 100)
+    private val _newEvents = MutableSharedFlow<UnsignedEvent>(
+        replay = 0,
+        extraBufferCapacity = 100,
+        onBufferOverflow = BufferOverflow.SUSPEND
+    )
     val newEvents = _newEvents.asSharedFlow()
 
-    suspend fun emitNewEvent(event: UnsignedEvent) {
-        _newEvents.emit(event)
+    fun emitNewEvent(event: UnsignedEvent) {
+        _newEvents.tryEmit(event)
     }
 
-    suspend fun init(
-        dbPath: String,
-        logLevel: LogLevel = LogLevel.WARN
-    ) {
+    suspend fun init(dbPath: String, logLevel: LogLevel = LogLevel.WARN) {
         try {
             if (isInitialized.value) return
 
@@ -105,7 +109,8 @@ class Nostr {
     }
 
     suspend fun waitUntilInitialized() {
-        isInitialized.first { it }
+        withTimeoutOrNull(30.seconds) { isInitialized.first { it } }
+            ?: throw IllegalStateException("Nostr initialization timed out")
     }
 
     suspend fun connectBootstrapRelays() {

shared/src/commonMain/kotlin/su/reya/coop/nostr/ProfileManager.kt

@@ -4,6 +4,7 @@ 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.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import rust.nostr.sdk.AckPolicy
@@ -33,10 +34,19 @@ class ProfileManager(private val nostr: Nostr) {
     private val signer: UniversalSigner get() = nostr.signer
 
     private val _metadataUpdates =
-        MutableSharedFlow<Pair<PublicKey, Metadata>>(extraBufferCapacity = 100)
+        MutableSharedFlow<Pair<PublicKey, Metadata>>(
+            replay = 0,
+            extraBufferCapacity = 100,
+            onBufferOverflow = BufferOverflow.SUSPEND
+        )
     val metadataUpdates = _metadataUpdates.asSharedFlow()
 
-    private val _contactListUpdates = MutableSharedFlow<List<PublicKey>>(extraBufferCapacity = 100)
+    private val _contactListUpdates =
+        MutableSharedFlow<List<PublicKey>>(
+            replay = 0,
+            extraBufferCapacity = 100,
+            onBufferOverflow = BufferOverflow.SUSPEND
+        )
     val contactListUpdates = _contactListUpdates.asSharedFlow()
 
     suspend fun emitMetadataUpdate(pubkey: PublicKey, metadata: Metadata) {
@@ -138,19 +148,24 @@ class ProfileManager(private val nostr: Nostr) {
         bio: String? = null,
         picture: String? = null
     ): Metadata {
-        val currentUser =
-            signer.getPublicKeyAsync() ?: throw IllegalStateException("User not signed in")
-
         try {
-            val record = getLatestMetadata(currentUser)?.asRecord() ?: MetadataRecord()
-            val newRecord = record.copy(
-                displayName = name ?: record.displayName,
-                about = bio ?: record.about,
-                picture = picture ?: record.picture
-            )
-            val newMetadata = Metadata.fromRecord(newRecord)
-            val event = EventBuilder.metadata(newMetadata).finalizeAsync(signer)
+            val currentUser =
+                signer.getPublicKeyAsync() ?: throw IllegalStateException("User not signed in")
 
+            // Get the latest metadata event
+            val record = getLatestMetadata(currentUser)?.asRecord() ?: MetadataRecord()
+
+            // Build a new metadata based on old records
+            val newMetadata = Metadata.fromRecord(
+                record.copy(
+                    displayName = name ?: record.displayName,
+                    about = bio ?: record.about,
+                    picture = picture ?: record.picture
+                )
+            )
+
+            // Send the new metadata event
+            val event = EventBuilder.metadata(newMetadata).finalizeAsync(signer)
             client?.sendEvent(
                 event = event,
                 target = SendEventTarget.broadcast(),

shared/src/commonMain/kotlin/su/reya/coop/nostr/RelayManager.kt

@@ -104,6 +104,32 @@ class RelayManager(private val nostr: Nostr) {
         return msgRelayList
     }
 
+    suspend fun getRelayList(publicKey: PublicKey): Map<RelayUrl, RelayMetadata?> {
+        try {
+            val kind = Kind.fromStd(KindStandard.RELAY_LIST)
+            val filter = Filter().kind(kind).author(publicKey).limit(1u)
+            val events = client?.database()?.query(filter)
+
+            return extractRelayList(events?.toVec()?.firstOrNull() ?: return emptyMap())
+        } catch (e: Exception) {
+            throw IllegalStateException("Failed to get relay list: ${e.message}", e)
+        }
+    }
+
+    suspend fun setRelaylist(relays: Map<RelayUrl, RelayMetadata?>) {
+        try {
+            val event = EventBuilder.relayList(relays).finalizeAsync(signer)
+
+            client?.sendEvent(
+                event = event,
+                target = SendEventTarget.broadcast(),
+                ackPolicy = AckPolicy.none(),
+            )
+        } catch (e: Exception) {
+            throw IllegalStateException("Failed to set msg relays: ${e.message}", e)
+        }
+    }
+    
     suspend fun setMsgRelays(urls: List<RelayUrl>) {
         try {
             val event = EventBuilder.nip17RelayList(urls).finalizeAsync(signer)
@@ -154,30 +180,4 @@ class RelayManager(private val nostr: Nostr) {
             throw IllegalStateException("Failed to fetch msg relays: ${e.message}", e)
         }
     }
-
-    suspend fun getRelayList(publicKey: PublicKey): Map<RelayUrl, RelayMetadata?> {
-        try {
-            val kind = Kind.fromStd(KindStandard.RELAY_LIST)
-            val filter = Filter().kind(kind).author(publicKey).limit(1u)
-            val events = client?.database()?.query(filter)
-
-            return extractRelayList(events?.toVec()?.firstOrNull() ?: return emptyMap())
-        } catch (e: Exception) {
-            throw IllegalStateException("Failed to get relay list: ${e.message}", e)
-        }
-    }
-
-    suspend fun setRelaylist(relays: Map<RelayUrl, RelayMetadata?>) {
-        try {
-            val event = EventBuilder.relayList(relays).finalizeAsync(signer)
-
-            client?.sendEvent(
-                event = event,
-                target = SendEventTarget.broadcast(),
-                ackPolicy = AckPolicy.none(),
-            )
-        } catch (e: Exception) {
-            throw IllegalStateException("Failed to set msg relays: ${e.message}", e)
-        }
-    }
 }

shared/src/commonMain/kotlin/su/reya/coop/viewmodel/BaseViewModel.kt

@@ -1,11 +1,16 @@
 package su.reya.coop.viewmodel
 
 import androidx.lifecycle.ViewModel
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.asSharedFlow
 
 abstract class BaseViewModel : ViewModel() {
-    private val _errorEvents = MutableSharedFlow<String>(extraBufferCapacity = 10)
+    private val _errorEvents = MutableSharedFlow<String>(
+        replay = 0,
+        extraBufferCapacity = 10,
+        onBufferOverflow = BufferOverflow.SUSPEND
+    )
     val errorEvents = _errorEvents.asSharedFlow()
 
     protected fun showError(message: String) {

shared/src/commonMain/kotlin/su/reya/coop/viewmodel/ChatViewModel.kt

@@ -2,11 +2,11 @@ package su.reya.coop.viewmodel
 
 import androidx.lifecycle.viewModelScope
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.asSharedFlow
-import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.flow.update
@@ -16,7 +16,6 @@ import rust.nostr.sdk.EventId
 import rust.nostr.sdk.Kind
 import rust.nostr.sdk.KindStandard
 import rust.nostr.sdk.PublicKey
-import rust.nostr.sdk.RelayUrl
 import rust.nostr.sdk.Tag
 import rust.nostr.sdk.UnsignedEvent
 import su.reya.coop.Room
@@ -26,7 +25,6 @@ import su.reya.coop.roomId
 
 data class ChatState(
     val rooms: Map<Long, Room> = emptyMap(),
-    val isSyncing: Boolean = false,
     val isPartialProcessedGiftWrap: Boolean = false,
 )
 
@@ -35,25 +33,23 @@ class ChatViewModel(
     private val mediaRepository: MediaRepository,
 ) : BaseViewModel() {
     private val _state = MutableStateFlow(ChatState())
-    val state = combine(
-        _state,
-        nostr.messages.messageSyncState
-    ) { local, state -> local.copy(isSyncing = state.isSyncing) }.stateIn(
+    val state = _state.stateIn(
         viewModelScope,
         SharingStarted.WhileSubscribed(5000),
         ChatState()
     )
 
-    private val _newEvents = MutableSharedFlow<UnsignedEvent>(extraBufferCapacity = 100)
+    private val _newEvents = MutableSharedFlow<UnsignedEvent>(
+        replay = 0,
+        extraBufferCapacity = 100,
+        onBufferOverflow = BufferOverflow.SUSPEND
+    )
     val newEvents = _newEvents.asSharedFlow()
 
-    private val _sentReports = MutableSharedFlow<Map<EventId, List<RelayUrl>>>()
-    val sentReport = _sentReports.asSharedFlow()
-
     val chatRooms = state.map { it.rooms.values.sortedByDescending { it.createdAt.asSecs() } }
         .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())
 
-    val isSyncing = state.map { it.isSyncing }
+    val isSyncing = nostr.messages.messageSyncState.map { it.isSyncing }
         .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), false)
 
     val isPartialProcessedGiftWrap = state.map { it.isPartialProcessedGiftWrap }
@@ -92,7 +88,7 @@ class ChatViewModel(
                         updateRoomList(roomId, event)
                     }
 
-                    _newEvents.emit(event)
+                    _newEvents.tryEmit(event)
                 }
             }
         }
@@ -156,6 +152,8 @@ class ChatViewModel(
     suspend fun getChatRoomMessages(roomId: Long): List<UnsignedEvent> {
         try {
             return nostr.messages.getChatRoomMessages(roomId)
+        } catch (e: CancellationException) {
+            throw e
         } catch (e: Exception) {
             showError("Error: ${e.message}")
         }
@@ -191,7 +189,7 @@ class ChatViewModel(
                     replies = replies,
                     onRumorCreated = { event ->
                         updateRoomList(roomId, event)
-                        viewModelScope.launch { _newEvents.emit(event) }
+                        viewModelScope.launch { _newEvents.tryEmit(event) }
                     },
                 )
             } catch (e: Exception) {