diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/database/repository/DatabaseNostrRepository.kt index 38bdce6b..22e511b7 100644 --- 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) { diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/NostrPublishResult.kt index 0a55a35f..dc47f089 100644 --- 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, ) diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelayPool.kt index 3648a998..94577370 100644 --- 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 = emptySet()) { + suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set = emptySet()): Flow { 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, nostrEvent: NostrEvent) { + private suspend fun handlePublishEventToRelays(relayConnections: List, nostrEvent: NostrEvent): Flow { val responseFlow = MutableSharedFlow() 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 diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/network/relays/RelaysSocketManager.kt index 3cabc34a..a48b5f0c 100644 --- 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 { + 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) { - relayPool.publishEvent( + suspend fun publishEvent(nostrEvent: NostrEvent, relays: Set): Flow { + return relayPool.publishEvent( nostrEvent = nostrEvent, relays = relays ) - - // TODO: Close relays and remove them from relayPool -// customPool.closePool() } fun tryConnectingToAllRelays() { diff --git a/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt b/composeApp/src/commonMain/kotlin/ac/aux/compose/repository/NostrRepository.kt index f706608c..d1c0714d 100644 --- 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 = 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") } diff --git 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 index 38c5b8c7..6f976985 100644 --- 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" + ) + } + } + } + } } }