Makes sure Filters are not updated if the since is the only change

2025-07-07 13:37:57 -04:00
parent 28d182cc55
commit cf48516602
44 changed files with 397 additions and 525 deletions
@@ -25,12 +25,12 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientLis
 import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayState
 import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolEventOutboxRepository
 import com.vitorpamplona.quartz.nip01Core.relay.client.pool.PoolSubscriptionRepository
-import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
 import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayPool
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
 import com.vitorpamplona.quartz.nip01Core.relay.client.stats.RelayStats
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
 import kotlinx.coroutines.CoroutineScope
@@ -44,14 +44,15 @@ import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.stateIn

 /**
- * The Nostr Client manages a relay pool, keeps active subscriptions and manages sending of events.
+ * The Nostr Client manages a relay pool, keeps active subscriptions and manages re-sending of events.
 */
 class NostrClient(
    private val websocketBuilder: WebsocketBuilder,
    private val scope: CoroutineScope,
 ) : IRelayClientListener {
    private val relayPool: RelayPool = RelayPool(this, ::buildRelay)
-    private val activeSubscriptions: PoolSubscriptionRepository = PoolSubscriptionRepository()
+    private val activeRequests: PoolSubscriptionRepository = PoolSubscriptionRepository()
+    private val activeCounts: PoolSubscriptionRepository = PoolSubscriptionRepository()
    private val eventOutbox: PoolEventOutboxRepository = PoolEventOutboxRepository()

    private var listeners = setOf<IRelayClientListener>()
@@ -62,19 +63,20 @@ class NostrClient(
     */
    private val allRelays =
        combine(
-            activeSubscriptions.relays,
+            activeRequests.relays,
+            activeCounts.relays,
            eventOutbox.relays,
-        ) { subs, outbox ->
-            subs + outbox
+        ) { reqs, counts, outbox ->
+            reqs + counts + outbox
        }.onStart {
-            activeSubscriptions.relays.value + eventOutbox.relays.value
+            activeRequests.relays.value + activeCounts.relays.value + eventOutbox.relays.value
        }.onEach {
            relayPool.updatePool(it)
        }.flowOn(Dispatchers.Default)
            .stateIn(
                scope,
                SharingStarted.Companion.Eagerly,
-                activeSubscriptions.relays.value + eventOutbox.relays.value,
+                activeRequests.relays.value + activeCounts.relays.value + eventOutbox.relays.value,
            )

    fun buildRelay(relay: NormalizedRelayUrl): IRelayClient =
@@ -84,7 +86,8 @@ class NostrClient(
            listener = relayPool,
            stats = RelayStats.get(relay),
        ) { liveRelay ->
-            activeSubscriptions.forEachSub(relay, liveRelay::sendRequest)
+            activeRequests.forEachSub(relay, liveRelay::sendRequest)
+            activeCounts.forEachSub(relay, liveRelay::sendCount)
            eventOutbox.forEachUnsentEvent(relay, liveRelay::send)
        }

@@ -109,20 +112,124 @@ class NostrClient(
        }
    }

-    fun sendFilter(
-        subscriptionId: String = newSubId(),
-        filters: List<RelayBasedFilter> = listOf(),
-    ) {
-        activeSubscriptions.addOrUpdate(subscriptionId, filters)
-        relayPool.sendRequest(subscriptionId, filters)
+    fun needsToResendRequest(
+        oldFilters: List<Filter>,
+        newFilters: List<Filter>,
+    ): Boolean {
+        if (oldFilters.size != newFilters.size) return true
+
+        oldFilters.forEachIndexed { index, oldFilter ->
+            val newFilter = newFilters.getOrNull(index) ?: return true
+
+            return needsToResendRequest(oldFilter, newFilter)
+        }
+        return false
    }

-    fun sendFilterOnlyIfDisconnected(
-        subscriptionId: String = newSubId(),
-        filters: List<RelayBasedFilter> = listOf(),
+    /**
+     * Checks if the filter has changed, with a special case for when the since changes due to new
+     * EOSE times.
+     */
+    fun needsToResendRequest(
+        oldFilter: Filter,
+        newFilter: Filter,
+    ): Boolean {
+        // Does not check SINCE on purpose. Avoids replacing the filter if SINCE was all that changed.
+        // fast check
+        if (oldFilter.authors?.size != newFilter.authors?.size ||
+            oldFilter.ids?.size != newFilter.ids?.size ||
+            oldFilter.tags?.size != newFilter.tags?.size ||
+            oldFilter.kinds?.size != newFilter.kinds?.size ||
+            oldFilter.limit != newFilter.limit ||
+            oldFilter.search?.length != newFilter.search?.length ||
+            oldFilter.until != newFilter.until
+        ) {
+            return true
+        }
+
+        // deep check
+        if (oldFilter.ids != newFilter.ids ||
+            oldFilter.authors != newFilter.authors ||
+            oldFilter.tags != newFilter.tags ||
+            oldFilter.kinds != newFilter.kinds ||
+            oldFilter.search != newFilter.search
+        ) {
+            return true
+        }
+
+        if (oldFilter.since != null) {
+            if (newFilter.since == null) {
+                // went was checking the future only and now wants everything
+                return true
+            } else if (oldFilter.since > newFilter.since) {
+                // went backwards in time, forces update
+                return true
+            }
+        }
+
+        return false
+    }
+
+    fun sendRequest(
+        subId: String = newSubId(),
+        filters: Map<NormalizedRelayUrl, List<Filter>>,
    ) {
-        activeSubscriptions.addOrUpdate(subscriptionId, filters)
-        relayPool.connectIfDisconnected()
+        val oldFilters = activeRequests.getSubscriptionFiltersOrNull(subId) ?: emptyMap()
+        activeRequests.addOrUpdate(subId, filters)
+
+        val allRelays = filters.keys + oldFilters.keys
+
+        allRelays.forEach { relay ->
+            val oldFilters = oldFilters[relay]
+            val newFilters = filters[relay]
+
+            if (newFilters.isNullOrEmpty()) {
+                // some relays are not in this sub anymore. Stop their subscriptions
+                relayPool.close(relay, subId)
+            } else if (oldFilters.isNullOrEmpty()) {
+                // new relays were added. Start a new sub in them
+                relayPool.sendRequest(relay, subId, newFilters)
+            } else if (needsToResendRequest(oldFilters, newFilters)) {
+                // filters were changed enough (not only an update in since) to warn a new update
+                relayPool.sendRequest(relay, subId, newFilters)
+            } else {
+                // makes sure the relay wakes up if it was disconnected by the server
+                // upon connection, the relay will run the default Sync and update all
+                // filters, including this one.
+                relayPool.connectIfDisconnected(relay)
+            }
+        }
+    }
+
+    fun sendCount(
+        subId: String = newSubId(),
+        filters: Map<NormalizedRelayUrl, List<Filter>>,
+    ) {
+        val oldFilters = activeCounts.getSubscriptionFiltersOrNull(subId) ?: emptyMap()
+        activeCounts.addOrUpdate(subId, filters)
+
+        val allRelays = filters.keys + oldFilters.keys
+
+        allRelays.forEach { relay ->
+            val oldFilters = oldFilters[relay]
+            val newFilters = filters[relay]
+
+            if (newFilters.isNullOrEmpty()) {
+                // some relays are not in this sub anymore. Stop their subscriptions
+                relayPool.close(relay, subId)
+            } else if (oldFilters.isNullOrEmpty()) {
+                // new relays were added. Start a new sub in them
+                relayPool.sendCount(relay, subId, newFilters)
+            } else if (needsToResendRequest(oldFilters, newFilters)) {
+                // filters were changed enough (not only an update in since) to warn a new update
+                relayPool.sendCount(relay, subId, newFilters)
+            } else {
+                // makes sure the relay wakes up if it was disconnected by the server
+                // upon connection, the relay will run the default Sync and update all
+                // filters, including this one.
+                relayPool.connectIfDisconnected(relay)
+            }
+        }
    }

    fun sendIfExists(
@@ -142,11 +249,10 @@ class NostrClient(

    fun close(subscriptionId: String) {
        relayPool.close(subscriptionId)
-        activeSubscriptions.remove(subscriptionId)
+        activeRequests.remove(subscriptionId)
+        activeCounts.remove(subscriptionId)
    }

-    fun isActive(subscriptionId: String): Boolean = activeSubscriptions.isActive(subscriptionId)
-
    override fun onEvent(
        relay: IRelayClient,
        subId: String,
@@ -236,7 +342,7 @@ class NostrClient(
        listeners = listeners.minus(listener)
    }

-    fun getSubscriptionFiltersOrNull(subId: String): List<RelayBasedFilter>? = activeSubscriptions.getSubscriptionFiltersOrNull(subId)
+    fun getSubscriptionFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = activeRequests.getSubscriptionFiltersOrNull(subId)

    fun relayStatusFlow() = relayPool.statusFlow
 }
@@ -23,9 +23,10 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.acessories
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
-import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
 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 kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.delay
@@ -33,7 +34,7 @@ import kotlinx.coroutines.launch

 fun NostrClient.downloadFirstEvent(
    subscriptionId: String = newSubId(),
-    filters: List<RelayBasedFilter> = listOf(),
+    filters: Map<NormalizedRelayUrl, List<Filter>>,
    onResponse: (Event) -> Unit,
 ) {
    val listener =
@@ -56,7 +57,7 @@ fun NostrClient.downloadFirstEvent(

    subscribe(listener)

-    sendFilter(subscriptionId, filters)
+    sendRequest(subscriptionId, filters)

    GlobalScope.launch(Dispatchers.IO) {
        delay(30000)
@@ -1,29 +0,0 @@
-/**
- * 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.pool
-
-import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
-
-class PoolSubscription(
-    var filters: List<RelayBasedFilter> = emptyList(),
-) {
-    fun toFilter(relay: NormalizedRelayUrl) = filters.mapNotNull { it.toFilter(relay) }
-}
@@ -22,20 +22,17 @@ package com.vitorpamplona.quartz.nip01Core.relay.client.pool

 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.utils.LargeCache
 import kotlinx.coroutines.flow.MutableStateFlow

 class PoolSubscriptionRepository {
-    private var subscriptions = mapOf<String, PoolSubscription>()
+    private var subscriptions = LargeCache<String, Map<NormalizedRelayUrl, List<Filter>>>()
    val relays = MutableStateFlow(setOf<NormalizedRelayUrl>())

    fun updateRelays() {
        val myRelays = mutableSetOf<NormalizedRelayUrl>()
-        subscriptions.values.forEach {
-            it.filters.forEach {
-                if (!myRelays.contains(it.relay)) {
-                    myRelays.add(it.relay)
-                }
-            }
+        subscriptions.forEach { sub, perRelayFilters ->
+            myRelays.addAll(perRelayFilters.keys)
        }

        if (relays.value != myRelays) {
@@ -45,20 +42,15 @@ class PoolSubscriptionRepository {

    fun addOrUpdate(
        subscriptionId: String,
-        filters: List<RelayBasedFilter> = listOf(),
+        filters: Map<NormalizedRelayUrl, List<Filter>>,
    ) {
-        val currentFilter = subscriptions[subscriptionId]
-        if (currentFilter == null) {
-            subscriptions = subscriptions + Pair(subscriptionId, PoolSubscription(filters))
-        } else {
-            currentFilter.filters = filters
-        }
+        subscriptions.put(subscriptionId, filters)
        updateRelays()
    }

    fun remove(subscriptionId: String) {
-        if (subscriptions.contains(subscriptionId)) {
-            subscriptions = subscriptions.minus(subscriptionId)
+        if (subscriptions.containsKey(subscriptionId)) {
+            subscriptions.remove(subscriptionId)
            updateRelays()
        }
    }
@@ -67,9 +59,9 @@ class PoolSubscriptionRepository {
        relay: NormalizedRelayUrl,
        run: (String, List<Filter>) -> Unit,
    ) {
-        subscriptions.forEach { (subId, filters) ->
-            val filters = filters.toFilter(relay)
-            if (filters.isNotEmpty()) {
+        subscriptions.forEach { subId, filters ->
+            val filters = filters[relay]
+            if (!filters.isNullOrEmpty()) {
                run(subId, filters)
            } else {
                null
@@ -77,11 +69,5 @@ class PoolSubscriptionRepository {
        }
    }

-    fun isActive(subscriptionId: String): Boolean = subscriptions.contains(subscriptionId)
-
-    fun allSubscriptions(): Map<String, PoolSubscription> = subscriptions
-
-    fun getSubscriptionFilters(subId: String): List<RelayBasedFilter> = subscriptions[subId]?.filters ?: emptyList()
-
-    fun getSubscriptionFiltersOrNull(subId: String): List<RelayBasedFilter>? = subscriptions[subId]?.filters
+    fun getSubscriptionFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = subscriptions.get(subId)
 }
@@ -40,3 +40,11 @@ class RelayBasedFilter(
            null
        }
 }
+
+fun List<RelayBasedFilter>.groupByRelay(): Map<NormalizedRelayUrl, List<Filter>> {
+    val result = mutableMapOf<NormalizedRelayUrl, MutableList<Filter>>()
+    for (relayBasedFilter in this) {
+        result.getOrPut(relayBasedFilter.relay) { mutableListOf() }.add(relayBasedFilter.filter)
+    }
+    return result
+}
@@ -26,6 +26,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.EmptyClientList
 import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
 import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayState
 import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.utils.LargeCache
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -33,7 +34,7 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlin.collections.forEach
 import kotlin.collections.isNotEmpty
-import kotlin.collections.mapNotNull
+import kotlin.collections.isNullOrEmpty

 val UnsupportedRelayCreation: (url: NormalizedRelayUrl) -> IRelayClient = {
    throw UnsupportedOperationException("Cannot create new relays")
@@ -77,30 +78,48 @@ class RelayPool(
            relay.connectAndSyncFiltersIfDisconnected()
        }

+    fun connectIfDisconnected(relay: NormalizedRelayUrl) = relays.get(relay)?.connectAndSyncFiltersIfDisconnected()
+
    fun disconnect() =
        relays.forEach { url, relay ->
            relay.disconnect()
        }

    fun sendRequest(
+        relay: NormalizedRelayUrl,
        subId: String,
-        filters: List<RelayBasedFilter>,
+        filters: List<Filter>,
+    ) {
+        relays.get(relay)?.sendRequest(subId, filters)
+    }
+
+    fun sendRequest(
+        subId: String,
+        filters: Map<NormalizedRelayUrl, List<Filter>>,
    ) {
        relays.forEach { url, relay ->
-            val filters = filters.mapNotNull { it.toFilter(url) }
-            if (filters.isNotEmpty()) {
+            val filters = filters[relay.url]
+            if (!filters.isNullOrEmpty()) {
                relay.sendRequest(subId, filters)
            }
        }
    }

-    fun sendCounter(
+    fun sendCount(
+        relay: NormalizedRelayUrl,
        subId: String,
-        filters: List<RelayBasedFilter>,
+        filters: List<Filter>,
+    ) {
+        relays.get(relay)?.sendRequest(subId, filters)
+    }
+
+    fun sendCount(
+        subId: String,
+        filters: Map<NormalizedRelayUrl, List<Filter>>,
    ) {
        relays.forEach { url, relay ->
-            val filters = filters.mapNotNull { it.toFilter(url) }
-            if (filters.isNotEmpty()) {
+            val filters = filters[relay.url]
+            if (!filters.isNullOrEmpty()) {
                relay.sendCount(subId, filters)
            }
        }
@@ -111,6 +130,11 @@ class RelayPool(
            relay.close(subscriptionId)
        }

+    fun close(
+        relay: NormalizedRelayUrl,
+        subscriptionId: String,
+    ) = relays.get(relay)?.close(subscriptionId)
+
    fun send(
        signedEvent: Event,
        list: Set<NormalizedRelayUrl>,
@@ -293,7 +293,7 @@ open class BasicRelayClient(
    }

    private fun processClosed(msg: ClosedMessage) {
-        // Log.w(logTag, "Relay Closed Subscription $newMessage")
+        Log.w(logTag, "Relay Closed Subscription ${msg.subscriptionId} ${msg.message}")
        afterEOSEPerSubscription[msg.subscriptionId] = false
        listener.onClosed(this@BasicRelayClient, msg.subscriptionId, msg.message)
    }
@@ -20,8 +20,10 @@
 */
 package com.vitorpamplona.quartz.utils

+import android.R.attr.value
 import java.util.concurrent.ConcurrentSkipListMap
 import java.util.function.BiConsumer
+import kotlin.collections.LinkedHashMap

 class LargeCache<K, V> {
    private val cache = ConcurrentSkipListMap<K, V>()
@@ -170,14 +172,32 @@ class LargeCache<K, V> {
        return runner.count
    }

-    private fun innerForEach(runner: BiConsumer<K, V>) {
-        // val (value, elapsed) =
-        //    measureTimedValue {
-        cache.forEach(runner)
-        //    }
-        // println("LargeCache full loop $elapsed \t for $runner")
+    public fun <T, U> associate(transform: (K, V) -> Pair<T, U>): Map<T, U> {
+        val runner = BiAssociateCollector(size(), transform)
+        innerForEach(runner)
+        return runner.results
+    }

-        listOf(1, 2, 3).joinToString()
+    public fun <T, U> associateNotNull(transform: (K, V) -> Pair<T, U>?): Map<T, U> {
+        val runner = BiAssociateNotNullCollector(size(), transform)
+        innerForEach(runner)
+        return runner.results
+    }
+
+    public fun <U> associateWith(transform: (K, V) -> U?): Map<K, U?> {
+        val runner = BiAssociateWithCollector(size(), transform)
+        innerForEach(runner)
+        return runner.results
+    }
+
+    public fun <U> associateNotNullWith(transform: (K, V) -> U): Map<K, U> {
+        val runner = BiAssociateNotNullWithCollector(size(), transform)
+        innerForEach(runner)
+        return runner.results
+    }
+
+    private fun innerForEach(runner: BiConsumer<K, V>) {
+        cache.forEach(runner)
    }

    fun joinToString(
@@ -255,6 +275,13 @@ fun interface BiMapper<K, V, R> {
    ): R?
 }

+fun interface BiMapperNotNull<K, V, R> {
+    fun map(
+        k: K,
+        v: V,
+    ): R
+}
+
 class BiMapCollector<K, V, R>(
    val mapper: BiMapper<K, V, R?>,
 ) : BiConsumer<K, V> {
@@ -271,6 +298,69 @@ class BiMapCollector<K, V, R>(
    }
 }

+class BiAssociateCollector<K, V, T, U>(
+    val size: Int,
+    val mapper: BiMapperNotNull<K, V, Pair<T, U>>,
+) : BiConsumer<K, V> {
+    var results: LinkedHashMap<T, U> = LinkedHashMap(size)
+
+    override fun accept(
+        k: K,
+        v: V,
+    ) {
+        val pair = mapper.map(k, v)
+        results.put(pair.first, pair.second)
+    }
+}
+
+class BiAssociateNotNullCollector<K, V, T, U>(
+    val size: Int,
+    val mapper: BiMapper<K, V, Pair<T, U>?>,
+) : BiConsumer<K, V> {
+    var results: LinkedHashMap<T, U> = LinkedHashMap(size)
+
+    override fun accept(
+        k: K,
+        v: V,
+    ) {
+        val pair = mapper.map(k, v)
+        if (pair != null) {
+            results.put(pair.first, pair.second)
+        }
+    }
+}
+
+class BiAssociateWithCollector<K, V, U>(
+    val size: Int,
+    val mapper: BiMapper<K, V, U?>,
+) : BiConsumer<K, V> {
+    var results: LinkedHashMap<K, U?> = LinkedHashMap(size)
+
+    override fun accept(
+        k: K,
+        v: V,
+    ) {
+        results.put(k, mapper.map(k, v))
+    }
+}
+
+class BiAssociateNotNullWithCollector<K, V, U>(
+    val size: Int,
+    val mapper: BiMapper<K, V, U>,
+) : BiConsumer<K, V> {
+    var results: LinkedHashMap<K, U> = LinkedHashMap(size)
+
+    override fun accept(
+        k: K,
+        v: V,
+    ) {
+        val newValue = mapper.map(k, v)
+        if (newValue != null) {
+            results.put(k, newValue)
+        }
+    }
+}
+
 class BiMapUniqueCollector<K, V, R>(
    val mapper: BiMapper<K, V, R?>,
 ) : BiConsumer<K, V> {