Handle published note events

2026-04-19 21:53:50 +02:00
parent 500f8db340
commit af3669af47
6 changed files with 108 additions and 71 deletions
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt
@@ -250,14 +250,17 @@ class DatabaseNostrRepository(
        )
    }

-    override suspend fun broadcastProcessed(broadcastNostrEventRequest: BroadcastNostrEventRequest) {
+    override suspend fun broadcastProcessed(
+        broadcastNostrEventRequest: BroadcastNostrEventRequest,
+        status: String
+    ) {
        logger.d("Update local reference: $broadcastNostrEventRequest")
        database.broadcastNostrEventRequestDao().upsert(
            broadcastNostrEventRequest.copy(
-                status = "sent"
+                status = status
            )
        )
-        logger.d("Processed: $broadcastNostrEventRequest")
+        logger.d("$status: $broadcastNostrEventRequest")
    }

    override suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) {
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt
@@ -5,4 +5,5 @@ import ac.aux.compose.network.sockets.NostrIncomingMessage
 data class NostrPublishResult(
    val result: NostrIncomingMessage? = null,
    val error: Throwable? = null,
+    val relayUrl: String,
 )
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt
@@ -161,10 +161,10 @@ class RelayPool(
        }

    @Throws(NostrPublishException::class)
-    suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set<RelayDTO> = emptySet()) {
+    suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set<RelayDTO> = emptySet()): Flow<NostrPublishResult> {
        val relayUrls = relays.map { it.url }

-        if (relayUrls.isEmpty()) {
+        return if (relayUrls.isEmpty()) {
            handlePublishEventToRelays(socketClients, nostrEvent)
        } else {
            addRelaysIfMissing(relays)
@@ -235,7 +235,7 @@ class RelayPool(
    }

    @OptIn(FlowPreview::class)
-    private suspend fun handlePublishEventToRelays(relayConnections: List<NostrSocketClient>, nostrEvent: NostrEvent) {
+    private suspend fun handlePublishEventToRelays(relayConnections: List<NostrSocketClient>, nostrEvent: NostrEvent): Flow<NostrPublishResult> {
        val responseFlow = MutableSharedFlow<NostrPublishResult>()
        relayConnections.forEach { nostrSocketClient ->
            scope.launch {
@@ -246,27 +246,41 @@ class RelayPool(
                        collectPublishResponse(eventId = nostrEvent.id)
                    }
                    sendEventResult.getOrNull()?.let {
-                        responseFlow.emit(NostrPublishResult(result = it))
+                        responseFlow.emit(
+                            NostrPublishResult(
+                                result = it,
+                                relayUrl = nostrSocketClient.socketUrl
+                            )
+                        )
                    }
                    sendEventResult.exceptionOrNull()?.let {
                        logger.w(throwable = it) { "sendEVENT failed to $socketUrl" }
-                        responseFlow.emit(NostrPublishResult(error = it))
+                        responseFlow.emit(
+                            NostrPublishResult(
+                                error = it,
+                                relayUrl = nostrSocketClient.socketUrl
+                            )
+                        )
                    }
                }
            }
        }

-        var responseCount = 0
-        responseFlow.timeout(PUBLISH_TIMEOUT.milliseconds)
-            .catch { throw NostrPublishException(cause = it) }
-            .transform {
-                emit(it)
-                responseCount++
-                if (relayConnections.size == responseCount && !it.isSuccessful()) {
-                    throw NostrPublishException(cause = null)
-                }
-            }
-            .first { it.isSuccessful() }
+//        var responseCount = 0
+//        responseFlow.timeout(PUBLISH_TIMEOUT.milliseconds)
+//            .catch {
+//                logger.e("Caught a nostr publish exception: $it")
+//                throw NostrPublishException(cause = it)
+//            }
+//            .transform {
+//                emit(it)
+//                responseCount++
+//                if (relayConnections.size == responseCount && !it.isSuccessful()) {
+//                    throw NostrPublishException(cause = null)
+//                }
+//            }
+//            .first { it.isSuccessful() }
+        return responseFlow
    }

    @FlowPreview
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt
@@ -106,8 +106,8 @@ class RelaysSocketManager constructor(
        }

    @Throws(NostrPublishException::class)
-    suspend fun publishEvent(nostrEvent: NostrEvent) {
-        if (userRelays.isNotEmpty()) {
+    suspend fun publishEvent(nostrEvent: NostrEvent): Flow<NostrPublishResult> {
+        return if (userRelays.isNotEmpty()) {
            relayPool.publishEvent(
                nostrEvent = nostrEvent,
                relays = userRelays
@@ -121,14 +121,11 @@ class RelaysSocketManager constructor(
    }

    @Throws(NostrPublishException::class)
-    suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set<RelayDTO>) {
-        relayPool.publishEvent(
+    suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set<RelayDTO>): Flow<NostrPublishResult> {
+        return relayPool.publishEvent(
            nostrEvent = nostrEvent,
            relays = relays
        )
-
-        // TODO: Close relays and remove them from relayPool
-//        customPool.closePool()
    }

    fun tryConnectingToAllRelays() {
--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt
@@ -52,7 +52,7 @@ interface NostrRepository {
        relayURLs: List<String> = emptyList()
    )

-    suspend fun broadcastProcessed(broadcastNostrEventRequest: BroadcastNostrEventRequest)
+    suspend fun broadcastProcessed(broadcastNostrEventRequest: BroadcastNostrEventRequest, status: String = "processing")

    suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest)

@@ -136,7 +136,10 @@ interface NostrRepository {

            }

-            override suspend fun broadcastProcessed(broadcastNostrEventRequest: BroadcastNostrEventRequest) {
+            override suspend fun broadcastProcessed(
+                broadcastNostrEventRequest: BroadcastNostrEventRequest,
+                status: String
+            ) {
                TODO("Not yet implemented")
            }

--- a/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt
+++ b/composeApp/src/commonMain/kotlin/ac/aux/compose/ui/view/model/NavigationViewModel.kt
@@ -2,8 +2,9 @@ package ac.aux.compose.ui.view.model

 import ac.aux.compose.database.model.NostrEvent
 import ac.aux.compose.managers.SeedManager
-import ac.aux.compose.network.NostrEventBroadcaster
 import ac.aux.compose.network.dto.RelayDTO
+import ac.aux.compose.network.relays.RelayPool
+import ac.aux.compose.network.relays.RelayPool.Companion.PUBLISH_TIMEOUT
 import ac.aux.compose.network.relays.RelaysSocketManager
 import ac.aux.compose.network.sockets.NostrIncomingMessage
 import ac.aux.compose.network.sockets.NostrSocketClientFactory
@@ -18,37 +19,25 @@ import androidx.lifecycle.viewmodel.initializer
 import androidx.lifecycle.viewmodel.viewModelFactory
 import co.touchlab.kermit.Logger
 import com.vitorpamplona.quartz.nip01Core.core.Event
-import com.vitorpamplona.quartz.nip01Core.core.OptimizedJsonMapper
 import com.vitorpamplona.quartz.nip01Core.core.toHexKey
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.CloseCmd
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.ReqCmd
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
-import com.vitorpamplona.quartz.nip01Core.tags.events.firstTaggedEvent
-import com.vitorpamplona.quartz.nip01Core.tags.people.firstTaggedUser
 import com.vitorpamplona.quartz.nip51Lists.encryption.PrivateTagsInContent
-import com.vitorpamplona.quartz.utils.text
-import io.ktor.client.HttpClient
-import io.ktor.client.plugins.websocket.WebSockets
-import io.ktor.client.plugins.websocket.webSocketSession
-import io.ktor.serialization.kotlinx.KotlinxWebsocketSerializationConverter
-import io.ktor.websocket.Frame
-import io.ktor.websocket.close
-import io.ktor.websocket.readText
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.IO
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.getAndUpdate
+import kotlinx.coroutines.flow.timeout
 import kotlinx.coroutines.launch
-import kotlinx.serialization.json.Json
-import kotlinx.serialization.json.JsonArray
-import kotlinx.serialization.json.jsonObject
-import kotlin.collections.set
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
 import kotlin.time.Instant

 class NavigationViewModel(
@@ -86,16 +75,6 @@ class NavigationViewModel(

    private val logger = Logger.withTag(TAG)

-    val httpClient = HttpClient() {
-        install(WebSockets) {
-            contentConverter = KotlinxWebsocketSerializationConverter(Json {
-                isLenient = true
-                ignoreUnknownKeys = true
-            })
-            pingIntervalMillis = 20_000
-        }
-    }
-
    private val _navigationUIState = MutableStateFlow(
        initialNavigationUIState
    )
@@ -231,6 +210,7 @@ class NavigationViewModel(
        }
    }

+    @OptIn(FlowPreview::class)
    private fun observePendingBroadcastNostrEventRequests() {
        logger.i { "observePendingBroadcastNostrEventRequests" }

@@ -238,26 +218,65 @@ class NavigationViewModel(
            nostrRepository.observePendingBroadcastNostrEventRequests().collect { localBroadcastNostrEventRequests ->
                localBroadcastNostrEventRequests.forEach { localBroadcastNostrEventRequest ->

+                    nostrRepository.broadcastProcessed(
+                        localBroadcastNostrEventRequest.broadcastNostrEventRequest
+                    )
+
                    if (localBroadcastNostrEventRequest.nostrEvent.broadcastedAt == null) {
-                        relaysSocketManager.publishEvent(localBroadcastNostrEventRequest.nostrEvent)
+                        scope.launch(Dispatchers.IO) {
+                            relaysSocketManager.publishEvent(
+                                localBroadcastNostrEventRequest.nostrEvent
+                            ).timeout(PUBLISH_TIMEOUT.milliseconds).catch {
+                                // Timeout...
+                            }.collect { nostrPublishResult ->
+                                if (nostrPublishResult.error != null) {
+                                    logger.e("Error publishing note: $nostrPublishResult")
+                                    nostrRepository.broadcastProcessed(
+                                        localBroadcastNostrEventRequest.broadcastNostrEventRequest,
+                                        "failed"
+                                    )
+                                } else {
+                                    logger.d("Nostr Publish Result: $nostrPublishResult")
+                                    nostrRepository.broadcastProcessed(
+                                        localBroadcastNostrEventRequest.broadcastNostrEventRequest,
+                                        "published"
+                                    )
+                                }
+                            }
+                        }
+
                    }

                    scope.launch(Dispatchers.IO) {
-                        nostrRepository.broadcastProcessed(
-                            localBroadcastNostrEventRequest.broadcastNostrEventRequest
-                        )
-                    }
-                    // Broadcast to the intended relay...
-                    relaysSocketManager.publishEvent(
-                        localBroadcastNostrEventRequest.nostrEvent,
-                        setOf(
-                            RelayDTO(
-                                localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL,
-                                write = true,
-                                read = true
+                        // Broadcast to the intended relay...
+                        relaysSocketManager.publishEvent(
+                            localBroadcastNostrEventRequest.nostrEvent,
+                            setOf(
+                                RelayDTO(
+                                    localBroadcastNostrEventRequest.broadcastNostrEventRequest.relayURL,
+                                    write = true,
+                                    read = true
+                                )
                            )
-                        )
-                    )
+                        ).timeout(PUBLISH_TIMEOUT.milliseconds).catch {
+                            // Timeout...
+                        }.collect { nostrPublishResult ->
+                            if (nostrPublishResult.error != null) {
+                                logger.e("Error publishing note: $nostrPublishResult")
+                                nostrRepository.broadcastProcessed(
+                                    localBroadcastNostrEventRequest.broadcastNostrEventRequest,
+                                    "failed"
+                                )
+                            } else {
+                                logger.d("Nostr Publish Result: $nostrPublishResult")
+                                nostrRepository.broadcastProcessed(
+                                    localBroadcastNostrEventRequest.broadcastNostrEventRequest,
+                                    "published"
+                                )
+                            }
+                        }
+                    }
+
                }
            }
        }