diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt index 27f2b47f8..7e2854a16 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSync.kt @@ -23,10 +23,9 @@ package com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync import com.vitorpamplona.amethyst.model.Account import com.vitorpamplona.quartz.nip01Core.core.Event import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.downloadFromRelay import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener -import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient -import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter @@ -37,15 +36,12 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job -import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.supervisorScope import kotlinx.coroutines.sync.Semaphore -import kotlinx.coroutines.withTimeoutOrNull import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicLong @@ -407,11 +403,7 @@ class EventSync( /** * Fetches all pages from a single [relay] using paginated `until` cursors. - * - * Sends [baseFilters] and waits for EOSE. If events were returned, the oldest - * [Event.createdAt] minus one becomes the next `until` and the query repeats. - * Stops when EOSE arrives with no new events (relay exhausted) or the relay - * cannot be reached. + * Delegates to the Quartz [downloadFromRelay] extension. * * @return total number of events received across all pages. */ @@ -419,75 +411,5 @@ class EventSync( relay: NormalizedRelayUrl, baseFilters: List, onEvent: (Event) -> Unit, - ): Int { - var until: Long? = null - var totalEvents = 0 - - while (true) { - coroutineContext.ensureActive() - val pageCount = AtomicInteger(0) - val pageMinTs = AtomicLong(Long.MAX_VALUE) - val done = Channel(Channel.CONFLATED) - val subId = newSubId() - - val filters = - if (until == null) { - baseFilters - } else { - baseFilters.map { it.copy(until = until) } - } - - val listener = - object : IRequestListener { - override fun onEvent( - event: Event, - isLive: Boolean, - relay: NormalizedRelayUrl, - forFilters: List?, - ) { - onEvent(event) - pageCount.incrementAndGet() - pageMinTs.updateAndGet { minOf(it, event.createdAt) } - } - - override fun onEose( - relay: NormalizedRelayUrl, - forFilters: List?, - ) { - done.trySend(Unit) - } - - override fun onClosed( - message: String, - relay: NormalizedRelayUrl, - forFilters: List?, - ) { - done.trySend(Unit) - } - - override fun onCannotConnect( - relay: NormalizedRelayUrl, - message: String, - forFilters: List?, - ) { - done.trySend(Unit) - } - } - - account.client.openReqSubscription(subId, mapOf(relay to filters), listener) - withTimeoutOrNull(RELAY_TIMEOUT_MS) { done.receive() } - account.client.close(subId) - done.close() - - val count = pageCount.get() - if (count == 0) break // relay exhausted or unreachable - - totalEvents += count - - // Advance cursor: next page starts just before the oldest event seen. - until = pageMinTs.get() - 1 - } - - return totalEvents - } + ): Int = account.client.downloadFromRelay(relay, baseFilters, RELAY_TIMEOUT_MS, onEvent) } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDownloadFromRelayExt.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDownloadFromRelayExt.kt new file mode 100644 index 000000000..ff099b595 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/NostrClientDownloadFromRelayExt.kt @@ -0,0 +1,143 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay.client.accessories + +import com.vitorpamplona.quartz.nip01Core.core.Event +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED +import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.withTimeoutOrNull +import kotlin.coroutines.coroutineContext + +/** + * Downloads all pages of events matching [baseFilters] from a single [relay] using + * paginated `until` cursors. + * + * After EOSE the oldest [Event.createdAt] seen in that page minus one becomes the + * next `until`, and the query repeats until the relay returns no new events. + * + * @param relay The relay to query. + * @param baseFilters Filters to apply on every page (the `until` field is overwritten per page). + * @param timeoutMs Maximum time to wait for a single page's EOSE before giving up. + * @param onEvent Called for every event received (in page order, after each EOSE). + * @return Total number of events received across all pages. + */ +suspend fun INostrClient.downloadFromRelay( + relay: NormalizedRelayUrl, + baseFilters: List, + timeoutMs: Long = 30_000L, + onEvent: (Event) -> Unit, +): Int { + var until: Long? = null + var totalEvents = 0 + + while (true) { + coroutineContext.ensureActive() + + val eventChannel = Channel(UNLIMITED) + val doneChannel = Channel(Channel.CONFLATED) + val subId = newSubId() + + val filters = + if (until == null) { + baseFilters + } else { + baseFilters.map { it.copy(until = until) } + } + + val listener = + object : IRequestListener { + override fun onEvent( + event: Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + eventChannel.trySend(event) + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + doneChannel.trySend(Unit) + } + + override fun onClosed( + message: String, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + doneChannel.trySend(Unit) + } + + override fun onCannotConnect( + relay: NormalizedRelayUrl, + message: String, + forFilters: List?, + ) { + doneChannel.trySend(Unit) + } + } + + openReqSubscription(subId, mapOf(relay to filters), listener) + withTimeoutOrNull(timeoutMs) { doneChannel.receive() } + close(subId) + eventChannel.close() + doneChannel.close() + + var pageCount = 0 + var pageMinTs = Long.MAX_VALUE + for (event in eventChannel) { + onEvent(event) + pageCount++ + if (event.createdAt < pageMinTs) pageMinTs = event.createdAt + } + + if (pageCount == 0) break + + totalEvents += pageCount + + // Advance cursor: next page starts just before the oldest event seen. + until = pageMinTs - 1 + } + + return totalEvents +} + +suspend fun INostrClient.downloadFromRelay( + relay: String, + baseFilters: List, + timeoutMs: Long = 30_000L, + onEvent: (Event) -> Unit, +): Int = + downloadFromRelay( + relay = RelayUrlNormalizer.normalize(relay), + baseFilters = baseFilters, + timeoutMs = timeoutMs, + onEvent = onEvent, + ) diff --git a/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientDownloadFromRelayTest.kt b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientDownloadFromRelayTest.kt new file mode 100644 index 000000000..c02d023da --- /dev/null +++ b/quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/NostrClientDownloadFromRelayTest.kt @@ -0,0 +1,68 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quartz.nip01Core.relay + +import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent +import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.downloadFromRelay +import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class NostrClientDownloadFromRelayTest : BaseNostrClientTest() { + @Test + fun testDownloadFromRelayReturnsMetadataEvents() = + runBlocking { + val appScope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val client = NostrClient(socketBuilder, appScope) + + val events = mutableListOf() + + val totalFound = + client.downloadFromRelay( + relay = "wss://nos.lol", + baseFilters = + listOf( + Filter( + kinds = listOf(MetadataEvent.KIND), + limit = 10, + ), + ), + ) { event -> + events.add(event) + } + + client.disconnect() + appScope.cancel() + + assertTrue(totalFound > 0, "Expected at least one event from wss://nos.lol") + assertTrue(events.isNotEmpty(), "Events list should not be empty") + events.forEach { event -> + assertEquals(MetadataEvent.KIND, event.kind, "All events should be kind ${MetadataEvent.KIND}") + } + } +}