Finalizes the RelaySync utility

2026-03-18 17:35:05 -04:00
parent e57023cad9
commit 3e3adaff10
13 changed files with 959 additions and 570 deletions
@@ -31,7 +31,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow

-interface INostrClient {
+interface INostrClient : AutoCloseable {
    fun connectedRelaysFlow(): StateFlow<Set<NormalizedRelayUrl>>

    fun availableRelaysFlow(): StateFlow<Set<NormalizedRelayUrl>>
@@ -136,4 +136,6 @@ object EmptyNostrClient : INostrClient {
    override fun activeCounts(url: NormalizedRelayUrl): Map<String, List<Filter>> = emptyMap()

    override fun activeOutboxCache(url: NormalizedRelayUrl): Set<HexKey> = emptySet()
+
+    override fun close() {}
 }
@@ -79,7 +79,8 @@ class NostrClient(
    private val websocketBuilder: WebsocketBuilder,
    private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()),
 ) : INostrClient,
-    IRelayClientListener {
+    IRelayClientListener,
+    AutoCloseable {
    private val relayPool: RelayPool = RelayPool(websocketBuilder, this)

    private val activeRequests: PoolRequests = PoolRequests()
@@ -326,4 +327,8 @@ class NostrClient(
    override fun connectedRelaysFlow() = relayPool.connectedRelays

    override fun availableRelaysFlow() = relayPool.availableRelays
+
+    override fun close() {
+        disconnect()
+    }
 }
@@ -28,7 +28,6 @@ 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
@@ -63,6 +62,8 @@ suspend fun INostrClient.reqBypassingRelayLimits(
    // Track how many matching events each filter has received so far.
    val matchCountPerFilter = IntArray(filters.size)

+    val subId = newSubId()
+
    while (true) {
        coroutineContext.ensureActive()

@@ -75,9 +76,7 @@ suspend fun INostrClient.reqBypassingRelayLimits(

        if (remainingFilters.isEmpty()) break

-        val eventChannel = Channel<Event>(UNLIMITED)
        val doneChannel = Channel<Unit>(Channel.CONFLATED)
-        val subId = newSubId()

        val activeFilters =
            if (until == null) {
@@ -86,19 +85,37 @@ suspend fun INostrClient.reqBypassingRelayLimits(
                remainingFilters.map { it.copy(until = until) }
            }

+        var pageCount = 0
+        var pageMinTs = Long.MAX_VALUE
+
        val listener =
            object : IRequestListener {
                override fun onEvent(
                    event: Event,
                    isLive: Boolean,
-                    relay: NormalizedRelayUrl,
+                    relayInner: NormalizedRelayUrl,
                    forFilters: List<Filter>?,
                ) {
-                    eventChannel.trySend(event)
+                    onEvent(event)
+                    pageCount++
+                    if (event.createdAt < pageMinTs) pageMinTs = event.createdAt
+
+                    // Count this event against every base filter it matches.
+                    if (matchCountPerFilter.size == 1) {
+                        // no need to run the match.
+                        matchCountPerFilter[0]++
+                    } else {
+                        for (i in filters.indices) {
+                            val limit = filters[i].limit
+                            if ((limit == null || matchCountPerFilter[i] < limit) && filters[i].match(event)) {
+                                matchCountPerFilter[i]++
+                            }
+                        }
+                    }
                }

                override fun onEose(
-                    relay: NormalizedRelayUrl,
+                    relayInner: NormalizedRelayUrl,
                    forFilters: List<Filter>?,
                ) {
                    doneChannel.trySend(Unit)
@@ -106,49 +123,30 @@ suspend fun INostrClient.reqBypassingRelayLimits(

                override fun onClosed(
                    message: String,
-                    relay: NormalizedRelayUrl,
+                    relayInner: NormalizedRelayUrl,
                    forFilters: List<Filter>?,
                ) {
                    doneChannel.trySend(Unit)
                }

                override fun onCannotConnect(
-                    relay: NormalizedRelayUrl,
+                    relayInner: NormalizedRelayUrl,
                    message: String,
                    forFilters: List<Filter>?,
                ) {
-                    println("AABBCC $message")
                    doneChannel.trySend(Unit)
                }
            }

        openReqSubscription(subId, mapOf(relay to activeFilters), 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
-
-            // Count this event against every base filter it matches.
-            if (matchCountPerFilter.size == 1) {
-                // no need to run the match.
-                matchCountPerFilter[0]++
-            } else {
-                for (i in filters.indices) {
-                    val limit = filters[i].limit
-                    if ((limit == null || matchCountPerFilter[i] < limit) && filters[i].match(event)) {
-                        matchCountPerFilter[i]++
-                    }
-                }
-            }
+        withTimeoutOrNull(timeoutMs) {
+            doneChannel.receive()
        }

+        close(subId)
+        doneChannel.close()
+
        if (pageCount == 0) break

        totalEvents += pageCount
@@ -67,7 +67,7 @@ suspend fun INostrClient.downloadFirstEvent(
    subscriptionId: String = newSubId(),
    filters: Map<NormalizedRelayUrl, List<Filter>>,
 ): Event? {
-    val resultChannel = Channel<Event>(UNLIMITED)
+    val resultChannel = Channel<Event?>(UNLIMITED)

    val listener =
        object : IRequestListener {
@@ -79,6 +79,29 @@ suspend fun INostrClient.downloadFirstEvent(
            ) {
                resultChannel.trySend(event)
            }
+
+            override fun onCannotConnect(
+                relay: NormalizedRelayUrl,
+                message: String,
+                forFilters: List<Filter>?,
+            ) {
+                resultChannel.trySend(null)
+            }
+
+            override fun onClosed(
+                message: String,
+                relay: NormalizedRelayUrl,
+                forFilters: List<Filter>?,
+            ) {
+                resultChannel.trySend(null)
+            }
+
+            override fun onEose(
+                relay: NormalizedRelayUrl,
+                forFilters: List<Filter>?,
+            ) {
+                resultChannel.trySend(null)
+            }
        }

    openReqSubscription(subscriptionId, filters, listener)
@@ -28,26 +28,63 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command
 import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.EventCmd
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.update
+import kotlin.compareTo

 class PoolEventOutbox {
    private var eventOutbox = mapOf<HexKey, PoolEventOutboxState>()
    val relays = MutableStateFlow(setOf<NormalizedRelayUrl>())

-    fun updateRelays() {
-        val myRelays = mutableSetOf<NormalizedRelayUrl>()
-        eventOutbox.values.forEach {
-            myRelays.addAll(it.relaysLeft())
+    fun needsToUpdateRelays(): Boolean {
+        val currentRelays = relays.value
+
+        var relaysToRemoveCounter = 0
+
+        currentRelays.forEach { currentRelay ->
+            if (eventOutbox.values.none { currentRelay in it.relaysRemaining }) {
+                relaysToRemoveCounter++
+            }
        }

-        if (relays.value != myRelays) {
-            relays.tryEmit(myRelays)
+        var relaysToAddCounter = 0
+        eventOutbox.values.forEach { outboxState ->
+            if (outboxState.relaysRemaining.any { it !in currentRelays }) {
+                relaysToAddCounter++
+            }
+        }
+
+        return relaysToRemoveCounter > 0 || relaysToAddCounter > 0
+    }
+
+    fun updateRelays() {
+        if (needsToUpdateRelays()) {
+            relays.update { currentRelays ->
+                val relaysToRemove = mutableSetOf<NormalizedRelayUrl>()
+
+                currentRelays.forEach { currentRelay ->
+                    if (eventOutbox.values.none { currentRelay in it.relaysRemaining }) {
+                        relaysToRemove.add(currentRelay)
+                    }
+                }
+
+                val relaysToAdd = mutableSetOf<NormalizedRelayUrl>()
+                eventOutbox.values.forEach { outboxState ->
+                    outboxState.relaysRemaining.forEach { relay ->
+                        if (relay !in relaysToAdd && relay !in currentRelays) {
+                            relaysToAdd.add(relay)
+                        }
+                    }
+                }
+
+                (currentRelays - relaysToRemove) + relaysToAdd
+            }
        }
    }

    fun activeOutboxCacheFor(url: NormalizedRelayUrl): Set<HexKey> {
        val myEvents = mutableSetOf<HexKey>()
        eventOutbox.forEach { (eventId, outboxCache) ->
-            if (url in outboxCache.relays) {
+            if (url in outboxCache.relaysRemaining) {
                myEvents.add(eventId)
            }
        }
@@ -72,7 +109,12 @@ class PoolEventOutbox {
        id: HexKey,
        url: NormalizedRelayUrl,
    ) {
-        eventOutbox[id]?.newTry(url)
+        val waiting = eventOutbox[id]
+        waiting?.newTry(url)
+        if (waiting?.isDone() == true) {
+            eventOutbox = eventOutbox - waiting.event.id
+            updateRelays()
+        }
    }

    fun newResponse(
@@ -84,15 +126,13 @@ class PoolEventOutbox {
        val waiting = eventOutbox[id]
        if (waiting != null) {
            waiting.newResponse(url, success, message)
-            clear()
+            if (waiting.isDone()) {
+                eventOutbox = eventOutbox - waiting.event.id
+                updateRelays()
+            }
        }
    }

-    fun clear() {
-        eventOutbox = eventOutbox.filter { !it.value.isDone() }
-        updateRelays()
-    }
-
    // --------------------------
    // State management functions
    // --------------------------
@@ -143,7 +183,7 @@ class PoolEventOutbox {
        errorMessage: String,
    ) {
        eventOutbox.forEach {
-            if (relay in it.value.relays) {
+            if (relay in it.value.relaysRemaining) {
                newResponse(it.key, relay, false, errorMessage)
            }
        }
@@ -26,35 +26,37 @@ import com.vitorpamplona.quartz.utils.TimeUtils

 class PoolEventOutboxState(
    val event: Event,
-    var relays: Set<NormalizedRelayUrl>,
+    var relaysRemaining: Set<NormalizedRelayUrl>,
 ) {
-    private var tries = mapOf<NormalizedRelayUrl, Tries>()
+    private var failures = mapOf<NormalizedRelayUrl, Tries>()

    fun updateRelays(newRelays: Set<NormalizedRelayUrl>) {
-        relays = newRelays
+        relaysRemaining = newRelays
    }

-    fun isDone(url: NormalizedRelayUrl) = tries[url]?.isDone() ?: false
+    fun isDone() = relaysRemaining.isEmpty()

-    fun isDone() = relays.all { isDone(it) }
+    fun relaysLeft(): Set<NormalizedRelayUrl> = relaysRemaining

-    fun relaysLeft(): Set<NormalizedRelayUrl> = relays.filterTo(mutableSetOf()) { !isDone(it) }
-
-    fun isSupposedToGo(url: NormalizedRelayUrl) = url in relays && !isDone(url)
+    fun isSupposedToGo(url: NormalizedRelayUrl) = url in relaysRemaining

    fun forEachUnsentEvent(
        url: NormalizedRelayUrl,
        run: (url: Event) -> Unit,
    ) = if (isSupposedToGo(url)) run(event) else null

-    fun remainingRelays() = relays.filterTo(mutableSetOf(), ::isSupposedToGo)
+    fun remainingRelays() = relaysRemaining

    fun newTry(url: NormalizedRelayUrl) {
-        val currentTries = tries[url]
+        val currentTries = failures[url]
        if (currentTries != null) {
            currentTries.addTriedTime(TimeUtils.now())
+            if (currentTries.isDone()) {
+                relaysRemaining = relaysRemaining - url
+                failures = failures - url
+            }
        } else {
-            tries = tries + (url to Tries(listOf(TimeUtils.now())))
+            failures = failures + (url to Tries(listOf(TimeUtils.now())))
        }
    }

@@ -63,38 +65,44 @@ class PoolEventOutboxState(
        success: Boolean,
        message: String,
    ) {
-        val currentTries = tries[url]
-        if (currentTries != null) {
-            currentTries.addResponse(Response(success, message))
+        val currentTries = failures[url]
+        if (success || message.shouldDiscard()) {
+            relaysRemaining = relaysRemaining - url
+            failures = failures - url
        } else {
-            tries = tries + (
-                url to
-                    Tries(
-                        listOf(TimeUtils.now() - 1),
-                        listOf(Response(success, message)),
-                    )
-            )
+            if (currentTries != null) {
+                currentTries.addResponse(message)
+            } else {
+                failures = failures + (
+                    url to
+                        Tries(
+                            listOf(TimeUtils.now() - 1),
+                            listOf(message),
+                        )
+                )
+            }
        }
    }

+    fun String.shouldDiscard() =
+        this.startsWith("replaced:") ||
+            this.startsWith("pow:") ||
+            this.startsWith("deleted:") ||
+            this.startsWith("invalid:")
+
    // Tries 3 times
    class Tries(
        var tries: List<Long> = listOf(),
-        var responses: List<Response> = listOf(),
+        var responses: List<String> = listOf(),
    ) {
-        fun isDone() = responses.any { it.success } || responses.size > 2 || tries.size > 3
+        fun isDone() = responses.size > 2 || tries.size > 3

-        fun addResponse(r: Response) {
-            responses += r
+        fun addResponse(msg: String) {
+            responses += msg
        }

        fun addTriedTime(tried: Long) {
            tries += tried
        }
    }
-
-    class Response(
-        val success: Boolean,
-        val message: String,
-    )
 }
@@ -42,6 +42,8 @@ class RelayStats(
            override fun create(key: NormalizedRelayUrl): RelayStat = RelayStat()
        }

+    fun snapshot(): Map<NormalizedRelayUrl, RelayStat> = innerCache.snapshot()
+
    fun get(url: NormalizedRelayUrl): RelayStat = innerCache[url] ?: throw IllegalArgumentException("Should never happen")

    private val clientListener =
@@ -101,6 +101,8 @@ private class TrackingNostrClient : INostrClient {
    override fun activeCounts(url: NormalizedRelayUrl): Map<String, List<Filter>> = emptyMap()

    override fun activeOutboxCache(url: NormalizedRelayUrl): Set<String> = emptySet()
+
+    override fun close() {}
 }

 /**