feat: move downloadFromRelay pagination logic to Quartz library

Extract the paginated relay download into a reusable INostrClient extension (NostrClientDownloadFromRelayExt.kt) in the Quartz commonMain accessories package. Uses Channel-based event collection for KMP compatibility (no JVM-only atomics). EventSyncViewModel.downloadFromRelay now delegates to the Quartz extension, keeping the ViewModel thin. Add NostrClientDownloadFromRelayTest integration test against wss://nos.lol with Filter(kinds = [MetadataEvent.KIND], limit = 10) to verify paginated downloads return correctly-typed events. https://claude.ai/code/session_01U8qF9mK4UBvXsXP1zNMXfX
2026-03-17 15:32:42 +00:00
parent e0dc4edf7a
commit 034cc0ab6d
3 changed files with 214 additions and 81 deletions
@@ -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<Filter>,
        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<Unit>(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<Filter>?,
-                    ) {
-                        onEvent(event)
-                        pageCount.incrementAndGet()
-                        pageMinTs.updateAndGet { minOf(it, event.createdAt) }
-                    }
-
-                    override fun onEose(
-                        relay: NormalizedRelayUrl,
-                        forFilters: List<Filter>?,
-                    ) {
-                        done.trySend(Unit)
-                    }
-
-                    override fun onClosed(
-                        message: String,
-                        relay: NormalizedRelayUrl,
-                        forFilters: List<Filter>?,
-                    ) {
-                        done.trySend(Unit)
-                    }
-
-                    override fun onCannotConnect(
-                        relay: NormalizedRelayUrl,
-                        message: String,
-                        forFilters: List<Filter>?,
-                    ) {
-                        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)
 }
@@ -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<Filter>,
+    timeoutMs: Long = 30_000L,
+    onEvent: (Event) -> Unit,
+): Int {
+    var until: Long? = null
+    var totalEvents = 0
+
+    while (true) {
+        coroutineContext.ensureActive()
+
+        val eventChannel = Channel<Event>(UNLIMITED)
+        val doneChannel = Channel<Unit>(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<Filter>?,
+                ) {
+                    eventChannel.trySend(event)
+                }
+
+                override fun onEose(
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) {
+                    doneChannel.trySend(Unit)
+                }
+
+                override fun onClosed(
+                    message: String,
+                    relay: NormalizedRelayUrl,
+                    forFilters: List<Filter>?,
+                ) {
+                    doneChannel.trySend(Unit)
+                }
+
+                override fun onCannotConnect(
+                    relay: NormalizedRelayUrl,
+                    message: String,
+                    forFilters: List<Filter>?,
+                ) {
+                    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<Filter>,
+    timeoutMs: Long = 30_000L,
+    onEvent: (Event) -> Unit,
+): Int =
+    downloadFromRelay(
+        relay = RelayUrlNormalizer.normalize(relay),
+        baseFilters = baseFilters,
+        timeoutMs = timeoutMs,
+        onEvent = onEvent,
+    )
@@ -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<com.vitorpamplona.quartz.nip01Core.core.Event>()
+
+            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}")
+            }
+        }
+}