- Restructures the old static datasource model into dynamic filter assemblers.

- Moves filter assemblers, viewModels and DAL classes to their own packages.
- Creates Composable observers for Users and Notes
- Deletes most of the secondary LiveData objects in the move to Flow
- Manipulates nostr filters depending on the account of the current screen, not a global account.
- Unifies all FilterAssembly lifecycle watchers to a few classes.
- Prepares to separate The Nostr Client as an Engine of Amethyst.
- Moves the pre-caching processor of new events from the datasource to the accountViewModel
- Reorganizes search to be per-screen basis
- Moves authentication to a fixed Coordinator class for all accounts in all relays.
- Moves NOTIFY command to its own coordinator class for all accounts
- Moves the connection between filters and cache to its own class.
- Significantly reduces the dependency on a single ServiceManager class.

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/NostrClient.kt

@@ -20,16 +20,17 @@
  */
 package com.vitorpamplona.ammolite.relays
 
-import android.system.Os.close
 import android.util.Log
-import androidx.core.app.PendingIntentCompat.send
 import com.vitorpamplona.ammolite.service.checkNotInMainThread
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.RelayState
 import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilderFactory
+import kotlinx.coroutines.CoroutineExceptionHandler
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import java.util.UUID
@@ -37,26 +38,28 @@ import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 
 /**
- * The Nostr Client manages multiple personae the user may switch between. Events are received and
- * published through multiple relays. Events are stored with their respective persona.
+ * The Nostr Client manages a relay pool
  */
 class NostrClient(
     private val websocketBuilder: WebsocketBuilderFactory,
+    private val listenerScope: CoroutineScope =
+        CoroutineScope(
+            Dispatchers.Default + SupervisorJob() +
+                CoroutineExceptionHandler { _, throwable ->
+                    Log.e("NostrClient", "Caught exception: ${throwable.message}", throwable)
+                },
+        ),
 ) : RelayPool.Listener {
     private val relayPool: RelayPool = RelayPool()
-    private val subscriptions: MutableSubscriptionManager = MutableSubscriptionManager()
-
+    private val activeSubscriptions: MutableSubscriptionManager = MutableSubscriptionManager()
     private var listeners = setOf<Listener>()
-    private var relays = emptyArray<Relay>()
 
-    fun buildRelay(it: RelaySetupInfoToConnect): Relay = Relay(it.url, it.read, it.write, it.forceProxy, it.feedTypes, websocketBuilder, subscriptions)
+    fun buildRelay(it: RelaySetupInfoToConnect): Relay = Relay(it.url, it.read, it.write, it.forceProxy, it.feedTypes, websocketBuilder, activeSubscriptions)
 
     fun getRelay(url: String): Relay? = relayPool.getRelay(url)
 
-    fun reconnect() {
-        // Reconnects all relays that may have disconnected
-        relayPool.requestAndWatch()
-    }
+    // Reconnects all relays that may have disconnected
+    fun reconnect() = relayPool.requestAndWatch()
 
     @Synchronized
     fun reconnect(
@@ -68,47 +71,42 @@ class NostrClient(
 
         if (onlyIfChanged) {
             if (!isSameRelaySetConfig(relays)) {
-                if (this.relays.isNotEmpty()) {
+                if (this.relayPool.availableRelays() > 0) {
                     relayPool.disconnect()
                     relayPool.unregister(this)
                     relayPool.unloadRelays()
                 }
 
                 if (relays != null) {
-                    val newRelays = relays.map(::buildRelay)
                     relayPool.register(this)
-                    relayPool.loadRelays(newRelays)
+                    relayPool.loadRelays(relays.map(::buildRelay))
                     relayPool.requestAndWatch()
-                    this.relays = newRelays.toTypedArray()
                 }
             } else {
                 // Reconnects all relays that may have disconnected
                 relayPool.requestAndWatch()
             }
         } else {
-            if (this.relays.isNotEmpty()) {
+            if (this.relayPool.availableRelays() > 0) {
                 relayPool.disconnect()
                 relayPool.unregister(this)
                 relayPool.unloadRelays()
             }
 
             if (relays != null) {
-                val newRelays = relays.map(::buildRelay)
                 relayPool.register(this)
-                relayPool.loadRelays(newRelays)
+                relayPool.loadRelays(relays.map(::buildRelay))
                 relayPool.requestAndWatch()
-                this.relays = newRelays.toTypedArray()
             }
         }
     }
 
     fun isSameRelaySetConfig(newRelayConfig: Array<RelaySetupInfoToConnect>?): Boolean {
-        if (relays.size != newRelayConfig?.size) return false
+        if (relayPool.availableRelays() != newRelayConfig?.size) return false
 
-        relays.forEach { oldRelayInfo ->
-            val newRelayInfo = newRelayConfig.find { it.url == oldRelayInfo.url } ?: return false
-
-            if (!oldRelayInfo.isSameRelayConfig(newRelayInfo)) return false
+        newRelayConfig.forEach {
+            val relay = relayPool.getRelay(it.url) ?: return false
+            if (!relay.isSameRelayConfig(it)) return false
         }
 
         return true
@@ -120,7 +118,7 @@ class NostrClient(
     ) {
         checkNotInMainThread()
 
-        subscriptions.add(subscriptionId, filters)
+        activeSubscriptions.add(subscriptionId, filters)
         relayPool.sendFilter(subscriptionId, filters)
     }
 
@@ -148,7 +146,7 @@ class NostrClient(
             },
         )
 
-        subscriptions.add(subscriptionId, filters)
+        activeSubscriptions.add(subscriptionId, filters)
         relayPool.sendFilter(subscriptionId, filters)
     }
 
@@ -255,7 +253,7 @@ class NostrClient(
     ) {
         checkNotInMainThread()
 
-        subscriptions.add(subscriptionId, filters)
+        activeSubscriptions.add(subscriptionId, filters)
         relayPool.connectAndSendFiltersIfDisconnected()
     }
 
@@ -311,10 +309,10 @@ class NostrClient(
 
     fun close(subscriptionId: String) {
         relayPool.close(subscriptionId)
-        subscriptions.remove(subscriptionId)
+        activeSubscriptions.remove(subscriptionId)
     }
 
-    fun isActive(subscriptionId: String): Boolean = subscriptions.isActive(subscriptionId)
+    fun isActive(subscriptionId: String): Boolean = activeSubscriptions.isActive(subscriptionId)
 
     @OptIn(DelicateCoroutinesApi::class)
     override fun onEvent(
@@ -325,7 +323,7 @@ class NostrClient(
     ) {
         // Releases the Web thread for the new payload.
         // May need to add a processing queue if processing new events become too costly.
-        GlobalScope.launch(Dispatchers.Default) {
+        listenerScope.launch {
             listeners.forEach { it.onEvent(event, subscriptionId, relay, afterEOSE) }
         }
     }
@@ -334,14 +332,18 @@ class NostrClient(
         relay: Relay,
         subscriptionId: String,
     ) {
-        listeners.forEach { it.onEOSE(relay, subscriptionId) }
+        listenerScope.launch {
+            listeners.forEach { it.onEOSE(relay, subscriptionId) }
+        }
     }
 
     override fun onRelayStateChange(
         type: RelayState,
         relay: Relay,
     ) {
-        listeners.forEach { it.onRelayStateChange(type, relay) }
+        listenerScope.launch {
+            listeners.forEach { it.onRelayStateChange(type, relay) }
+        }
     }
 
     @OptIn(DelicateCoroutinesApi::class)
@@ -353,7 +355,7 @@ class NostrClient(
     ) {
         // Releases the Web thread for the new payload.
         // May need to add a processing queue if processing new events become too costly.
-        GlobalScope.launch(Dispatchers.Default) {
+        listenerScope.launch {
             listeners.forEach { it.onSendResponse(eventId, success, message, relay) }
         }
     }
@@ -365,7 +367,7 @@ class NostrClient(
     ) {
         // Releases the Web thread for the new payload.
         // May need to add a processing queue if processing new events become too costly.
-        GlobalScope.launch(Dispatchers.Default) { listeners.forEach { it.onAuth(relay, challenge) } }
+        listenerScope.launch { listeners.forEach { it.onAuth(relay, challenge) } }
     }
 
     @OptIn(DelicateCoroutinesApi::class)
@@ -375,7 +377,7 @@ class NostrClient(
     ) {
         // Releases the Web thread for the new payload.
         // May need to add a processing queue if processing new events become too costly.
-        GlobalScope.launch(Dispatchers.Default) {
+        listenerScope.launch {
             listeners.forEach { it.onNotify(relay, description) }
         }
     }
@@ -386,7 +388,7 @@ class NostrClient(
         msg: String,
         success: Boolean,
     ) {
-        GlobalScope.launch(Dispatchers.Default) {
+        listenerScope.launch {
             listeners.forEach { it.onSend(relay, msg, success) }
         }
     }
@@ -396,7 +398,7 @@ class NostrClient(
         relay: Relay,
         event: Event,
     ) {
-        GlobalScope.launch(Dispatchers.Default) {
+        listenerScope.launch {
             listeners.forEach { it.onBeforeSend(relay, event) }
         }
     }
@@ -407,7 +409,7 @@ class NostrClient(
         subscriptionId: String,
         relay: Relay,
     ) {
-        GlobalScope.launch(Dispatchers.Default) {
+        listenerScope.launch {
             listeners.forEach { it.onError(error, subscriptionId, relay) }
         }
     }
@@ -422,9 +424,9 @@ class NostrClient(
         listeners = listeners.minus(listener)
     }
 
-    fun allSubscriptions(): Map<String, List<TypedFilter>> = subscriptions.allSubscriptions()
+    fun allSubscriptions(): Map<String, List<TypedFilter>> = activeSubscriptions.allSubscriptions()
 
-    fun getSubscriptionFilters(subId: String): List<TypedFilter> = subscriptions.getSubscriptionFilters(subId)
+    fun getSubscriptionFilters(subId: String): List<TypedFilter> = activeSubscriptions.getSubscriptionFilters(subId)
 
     fun connectedRelays() = relayPool.connectedRelays()
 

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/NostrDataSource.kt

@@ -1,314 +0,0 @@
-/**
- * Copyright (c) 2024 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.ammolite.relays
-
-import android.util.Log
-import com.vitorpamplona.ammolite.service.checkNotInMainThread
-import com.vitorpamplona.quartz.nip01Core.core.Event
-import com.vitorpamplona.quartz.nip01Core.relay.RelayState
-import com.vitorpamplona.quartz.utils.TimeUtils
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.DelicateCoroutinesApi
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
-import kotlinx.coroutines.launch
-import java.util.UUID
-import java.util.concurrent.atomic.AtomicBoolean
-
-abstract class NostrDataSource(
-    val client: NostrClient,
-    val debugName: String,
-) {
-    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
-
-    private var subscriptions = mapOf<String, Subscription>()
-
-    data class Counter(
-        val subscriptionId: String,
-        val eventKind: Int,
-        var counter: Int,
-    )
-
-    private var eventCounter = mapOf<Int, Counter>()
-    var changingFilters = AtomicBoolean()
-
-    private var active: Boolean = false
-
-    fun printCounter() {
-        eventCounter.forEach {
-            Log.d(
-                "STATE DUMP ${this.javaClass.simpleName}",
-                "Received Events $debugName ${it.value.subscriptionId} ${it.value.eventKind}: ${it.value.counter}",
-            )
-        }
-    }
-
-    fun hashCodeFields(
-        str1: String,
-        str2: Int,
-    ): Int = 31 * str1.hashCode() + str2.hashCode()
-
-    private val clientListener =
-        object : NostrClient.Listener {
-            override fun onEvent(
-                event: Event,
-                subscriptionId: String,
-                relay: Relay,
-                afterEOSE: Boolean,
-            ) {
-                if (subscriptions.containsKey(subscriptionId)) {
-                    val key = hashCodeFields(subscriptionId, event.kind)
-                    val keyValue = eventCounter[key]
-                    if (keyValue != null) {
-                        keyValue.counter++
-                    } else {
-                        eventCounter = eventCounter + Pair(key, Counter(subscriptionId, event.kind, 1))
-                    }
-
-                    // Log.d(this@NostrDataSource.javaClass.simpleName, "Relay ${relay.url}: $subscriptionId ${event.kind} ")
-
-                    consume(event, relay)
-                    if (afterEOSE) {
-                        markAsEOSE(subscriptionId, relay)
-                    }
-                }
-            }
-
-            override fun onEOSE(
-                relay: Relay,
-                subscriptionId: String,
-            ) {
-                if (subscriptions.containsKey(subscriptionId)) {
-                    markAsEOSE(subscriptionId, relay)
-                }
-            }
-
-            override fun onRelayStateChange(
-                type: RelayState,
-                relay: Relay,
-            ) {}
-
-            override fun onSendResponse(
-                eventId: String,
-                success: Boolean,
-                message: String,
-                relay: Relay,
-            ) {
-                if (success) {
-                    markAsSeenOnRelay(eventId, relay)
-                }
-            }
-
-            override fun onAuth(
-                relay: Relay,
-                challenge: String,
-            ) {
-                auth(relay, challenge)
-            }
-
-            override fun onNotify(
-                relay: Relay,
-                description: String,
-            ) {
-                notify(relay, description)
-            }
-        }
-
-    init {
-        Log.d("DataSource", "${this.javaClass.simpleName} Subscribe")
-        client.subscribe(clientListener)
-    }
-
-    fun destroy() {
-        // makes sure to run
-        Log.d("DataSource", "${this.javaClass.simpleName} Unsubscribe")
-        stop()
-        client.unsubscribe(clientListener)
-        scope.cancel()
-        bundler.cancel()
-    }
-
-    open fun start() {
-        Log.d("DataSource", "${this.javaClass.simpleName} Start")
-        active = true
-        resetFilters()
-    }
-
-    open fun startSync() {
-        Log.d("DataSource", "${this.javaClass.simpleName} Start")
-        active = true
-        resetFiltersSuspend()
-    }
-
-    @OptIn(DelicateCoroutinesApi::class)
-    open fun stop() {
-        active = false
-        Log.d("DataSource", "${this.javaClass.simpleName} Stop")
-
-        GlobalScope.launch(Dispatchers.IO) {
-            subscriptions.values.forEach { subscription ->
-                client.close(subscription.id)
-                subscription.typedFilters = null
-            }
-        }
-    }
-
-    open fun stopSync() {
-        active = false
-        Log.d("DataSource", "${this.javaClass.simpleName} Stop")
-
-        subscriptions.values.forEach { subscription ->
-            client.close(subscription.id)
-            subscription.typedFilters = null
-        }
-    }
-
-    fun requestNewChannel(onEOSE: ((Long, String) -> Unit)? = null): Subscription {
-        val newSubscription = Subscription(UUID.randomUUID().toString().substring(0, 4), onEOSE)
-        subscriptions = subscriptions + Pair(newSubscription.id, newSubscription)
-        return newSubscription
-    }
-
-    fun dismissChannel(subscription: Subscription) {
-        client.close(subscription.id)
-        subscriptions = subscriptions.minus(subscription.id)
-    }
-
-    // Refreshes observers in batches.
-    private val bundler = BundledUpdate(300, Dispatchers.Default)
-
-    fun invalidateFilters() {
-        bundler.invalidate {
-            // println("DataSource: ${this.javaClass.simpleName} InvalidateFilters")
-
-            // adds the time to perform the refresh into this delay
-            // holding off new updates in case of heavy refresh routines.
-            resetFiltersSuspend()
-        }
-    }
-
-    fun resetFilters() {
-        scope.launch(Dispatchers.IO) { resetFiltersSuspend() }
-    }
-
-    private fun resetFiltersSuspend() {
-        Log.d("DataSource", "${this.javaClass.simpleName} resetFiltersSuspend $active")
-        checkNotInMainThread()
-
-        // saves the channels that are currently active
-        val activeSubscriptions = subscriptions.values.filter { it.typedFilters != null }
-        // saves the current content to only update if it changes
-        val currentFilters = activeSubscriptions.associate { it.id to it.typedFilters }
-
-        changingFilters.getAndSet(true)
-
-        updateChannelFilters()
-
-        // Makes sure to only send an updated filter when it actually changes.
-        subscriptions.values.forEach { updatedSubscription ->
-            val updatedSubscriptionNewFilters = updatedSubscription.typedFilters
-
-            val isActive = client.isActive(updatedSubscription.id)
-
-            if (!isActive && updatedSubscriptionNewFilters != null) {
-                // Filter was removed from the active list
-                if (active) {
-                    client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
-                }
-            } else {
-                if (currentFilters.containsKey(updatedSubscription.id)) {
-                    if (updatedSubscriptionNewFilters == null) {
-                        // was active and is not active anymore, just close.
-                        client.close(updatedSubscription.id)
-                    } else {
-                        // was active and is still active, check if it has changed.
-                        if (updatedSubscription.hasChangedFiltersFrom(currentFilters[updatedSubscription.id])) {
-                            client.close(updatedSubscription.id)
-                            if (active) {
-                                client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
-                            }
-                        } else {
-                            // hasn't changed, does nothing.
-                            if (active) {
-                                client.sendFilterOnlyIfDisconnected(
-                                    updatedSubscription.id,
-                                    updatedSubscriptionNewFilters,
-                                )
-                            }
-                        }
-                    }
-                } else {
-                    if (updatedSubscriptionNewFilters == null) {
-                        // was not active and is still not active, does nothing
-                    } else {
-                        // was not active and becomes active, sends the filter.
-                        if (updatedSubscription.hasChangedFiltersFrom(currentFilters[updatedSubscription.id])) {
-                            if (active) {
-                                Log.d(
-                                    this@NostrDataSource.javaClass.simpleName,
-                                    "Update Filter 3 ${updatedSubscription.id} ${client.isSubscribed(clientListener)}",
-                                )
-                                client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
-                            }
-                        }
-                    }
-                }
-            }
-        }
-
-        changingFilters.getAndSet(false)
-    }
-
-    open fun consume(
-        event: Event,
-        relay: Relay,
-    ) = Unit
-
-    open fun markAsSeenOnRelay(
-        eventId: String,
-        relay: Relay,
-    ) = Unit
-
-    open fun markAsEOSE(
-        subscriptionId: String,
-        relay: Relay,
-    ) {
-        subscriptions[subscriptionId]?.updateEOSE(
-            // in case people's clock is slighly off.
-            TimeUtils.oneMinuteAgo(),
-            relay.url,
-        )
-    }
-
-    abstract fun updateChannelFilters()
-
-    open fun auth(
-        relay: Relay,
-        challenge: String,
-    ) = Unit
-
-    open fun notify(
-        relay: Relay,
-        description: String,
-    ) = Unit
-}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/Relay.kt

@@ -100,8 +100,7 @@ class Relay(
     private var listeners = setOf<Listener>()
 
     val relaySubFilter = RelaySubFilter(url, activeTypes, subs)
-    val inner =
-        SimpleClientRelay(url, socketBuilderFactory.build(url, forceProxy), relaySubFilter, this@Relay, RelayStats.get(url))
+    val inner = SimpleClientRelay(url, socketBuilderFactory.build(url, forceProxy), relaySubFilter, this@Relay, RelayStats.get(url))
 
     val brief = RelayBriefInfoCache.get(url)
 

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/EventCollector.kt

@@ -0,0 +1,57 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+import com.vitorpamplona.ammolite.relays.NostrClient
+import com.vitorpamplona.ammolite.relays.Relay
+import com.vitorpamplona.quartz.nip01Core.core.Event
+
+/**
+ * Listens to NostrClient's onEvent messages for caching purposes.
+ */
+class EventCollector(
+    val client: NostrClient,
+    val onEvent: (event: Event, relay: Relay) -> Unit,
+) {
+    private val clientListener =
+        object : NostrClient.Listener {
+            override fun onEvent(
+                event: Event,
+                subscriptionId: String,
+                relay: Relay,
+                afterEOSE: Boolean,
+            ) {
+                onEvent(event, relay)
+            }
+        }
+
+    init {
+        Log.d("${this.javaClass.simpleName}", "Init, Subscribe")
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        // makes sure to run
+        Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
+        client.unsubscribe(clientListener)
+    }
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/NostrDataSource.kt

@@ -0,0 +1,275 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+import com.vitorpamplona.ammolite.relays.BundledUpdate
+import com.vitorpamplona.ammolite.relays.NostrClient
+import com.vitorpamplona.ammolite.relays.Relay
+import com.vitorpamplona.ammolite.relays.TypedFilter
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.RelayState
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * Semantically groups Nostr filters and subscriptions in data source objects that
+ * maintain the desired active filter with the relay.
+ *
+ * This model also allows each datasource to observe any new events from the network,
+ * regardless of the subscription
+ */
+abstract class NostrDataSource(
+    val client: NostrClient,
+) {
+    private var subscriptions = SubscriptionSet()
+    val stats = SubscriptionStats()
+
+    var changingFilters = AtomicBoolean()
+
+    private var active: Boolean = false
+
+    private val clientListener =
+        object : NostrClient.Listener {
+            override fun onEvent(
+                event: Event,
+                subscriptionId: String,
+                relay: Relay,
+                afterEOSE: Boolean,
+            ) {
+                if (subscriptions.contains(subscriptionId)) {
+                    stats.add(subscriptionId, event.kind)
+
+                    consume(event, relay)
+
+                    if (afterEOSE) {
+                        markAsEOSE(subscriptionId, relay)
+                    }
+                }
+            }
+
+            override fun onEOSE(
+                relay: Relay,
+                subscriptionId: String,
+            ) {
+                if (subscriptions.contains(subscriptionId)) {
+                    markAsEOSE(subscriptionId, relay)
+                }
+            }
+
+            override fun onRelayStateChange(
+                type: RelayState,
+                relay: Relay,
+            ) {}
+
+            override fun onSendResponse(
+                eventId: String,
+                success: Boolean,
+                message: String,
+                relay: Relay,
+            ) {
+                if (success) {
+                    markAsSeenOnRelay(eventId, relay)
+                }
+            }
+
+            override fun onAuth(
+                relay: Relay,
+                challenge: String,
+            ) {
+                auth(relay, challenge)
+            }
+
+            override fun onNotify(
+                relay: Relay,
+                description: String,
+            ) {
+                notify(relay, description)
+            }
+        }
+
+    init {
+        Log.d("${this.javaClass.simpleName}", "Init, Subscribe")
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        // makes sure to run
+        Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
+        stop()
+        client.unsubscribe(clientListener)
+        bundler.cancel()
+    }
+
+    open fun start() {
+        Log.d("${this.javaClass.simpleName}", "Start")
+        active = true
+        invalidateFilters()
+    }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    open fun stop() {
+        active = false
+        Log.d("${this.javaClass.simpleName}", "Stop")
+
+        subscriptions.forEach { subscription ->
+            client.close(subscription.id)
+            subscription.reset()
+        }
+    }
+
+    fun getSub(subId: String) = subscriptions.get(subId)
+
+    fun requestNewSubscription(onEOSE: ((Long, String) -> Unit)? = null): Subscription = subscriptions.newSub(onEOSE)
+
+    fun dismissSubscription(subId: String) {
+        getSub(subId)?.let { dismissSubscription(it) }
+    }
+
+    fun dismissSubscription(subscription: Subscription) {
+        client.close(subscription.id)
+        subscription.reset()
+        subscriptions.remove(subscription)
+    }
+
+    // Refreshes observers in batches.
+    private val bundler = BundledUpdate(300, Dispatchers.Default)
+
+    fun invalidateFilters() {
+        bundler.invalidate {
+            // println("DataSource: ${this.javaClass.simpleName} InvalidateFilters")
+
+            // adds the time to perform the refresh into this delay
+            // holding off new updates in case of heavy refresh routines.
+            resetFiltersSuspend()
+        }
+    }
+
+    suspend fun resetFiltersSuspend() {
+        // only runs one at a time. Ignores the others
+        if (changingFilters.compareAndSet(false, true)) {
+            Log.d("${this.javaClass.simpleName}", "resetFiltersSuspend $active")
+            try {
+                resetFiltersSuspendInner()
+            } finally {
+                changingFilters.getAndSet(false)
+            }
+        }
+    }
+
+    private fun resetFiltersSuspendInner() {
+        // saves the channels that are currently active
+        val activeSubscriptions = subscriptions.actives()
+        // saves the current content to only update if it changes
+        val currentFilters = activeSubscriptions.associate { it.id to it.typedFilters }
+
+        // updates all filters
+        updateSubscriptions()
+
+        // Makes sure to only send an updated filter when it actually changes.
+        subscriptions.forEach { newSubscriptionFilters ->
+            val currentFilters = currentFilters[newSubscriptionFilters.id]
+            updateRelaysIfNeeded(newSubscriptionFilters, currentFilters)
+        }
+    }
+
+    fun updateRelaysIfNeeded(
+        updatedSubscription: Subscription,
+        currentFilters: List<TypedFilter>?,
+    ) {
+        val updatedSubscriptionNewFilters = updatedSubscription.typedFilters
+
+        val isActive = client.isActive(updatedSubscription.id)
+
+        if (!isActive && updatedSubscriptionNewFilters != null) {
+            // Filter was removed from the active list
+            // but it is supposed to be there. Send again.
+            if (active) {
+                client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
+            }
+        } else {
+            if (currentFilters != null) {
+                if (updatedSubscriptionNewFilters == null) {
+                    // was active and is not active anymore, just close.
+                    client.close(updatedSubscription.id)
+                } else {
+                    // was active and is still active, check if it has changed.
+                    if (updatedSubscription.hasChangedFiltersFrom(currentFilters)) {
+                        client.close(updatedSubscription.id)
+                        if (active) {
+                            client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
+                        }
+                    } else {
+                        // hasn't changed, does nothing.
+                        // unless the relay has disconnected, then reconnect.
+                        if (active) {
+                            client.sendFilterOnlyIfDisconnected(updatedSubscription.id, updatedSubscriptionNewFilters)
+                        }
+                    }
+                }
+            } else {
+                if (updatedSubscriptionNewFilters == null) {
+                    // was not active and is still not active, does nothing
+                } else {
+                    // was not active and becomes active, sends the filter.
+                    if (active) {
+                        client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
+                    }
+                }
+            }
+        }
+    }
+
+    open fun consume(
+        event: Event,
+        relay: Relay,
+    ) = Unit
+
+    open fun markAsSeenOnRelay(
+        eventId: String,
+        relay: Relay,
+    ) = Unit
+
+    open fun markAsEOSE(
+        subscriptionId: String,
+        relay: Relay,
+    ) {
+        subscriptions[subscriptionId]?.callEose(
+            // in case people's clock is slighly off.
+            TimeUtils.oneMinuteAgo(),
+            relay.url,
+        )
+    }
+
+    open fun auth(
+        relay: Relay,
+        challenge: String,
+    ) = Unit
+
+    open fun notify(
+        relay: Relay,
+        description: String,
+    ) = Unit
+
+    abstract fun updateSubscriptions()
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/RelayAuthenticator.kt

@@ -0,0 +1,51 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+import com.vitorpamplona.ammolite.relays.NostrClient
+import com.vitorpamplona.ammolite.relays.Relay
+
+class RelayAuthenticator(
+    val client: NostrClient,
+    val authenticate: (challenge: String, relay: Relay) -> Unit,
+) {
+    private val clientListener =
+        object : NostrClient.Listener {
+            override fun onAuth(
+                relay: Relay,
+                challenge: String,
+            ) {
+                authenticate(challenge, relay)
+            }
+        }
+
+    init {
+        Log.d("${this.javaClass.simpleName}", "Init, Subscribe")
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        // makes sure to run
+        Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
+        client.unsubscribe(clientListener)
+    }
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/RelayInsertConfirmationCollector.kt

@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+import com.vitorpamplona.ammolite.relays.NostrClient
+import com.vitorpamplona.ammolite.relays.Relay
+import com.vitorpamplona.quartz.nip01Core.core.HexKey
+
+/**
+ * Listens to NostrClient's onEvent messages for caching purposes.
+ */
+class RelayInsertConfirmationCollector(
+    val client: NostrClient,
+    val onRelayReceived: (eventId: HexKey, relay: Relay) -> Unit,
+) {
+    private val clientListener =
+        object : NostrClient.Listener {
+            override fun onSendResponse(
+                eventId: String,
+                success: Boolean,
+                message: String,
+                relay: Relay,
+            ) {
+                if (success) {
+                    onRelayReceived(eventId, relay)
+                }
+            }
+        }
+
+    init {
+        Log.d("${this.javaClass.simpleName}", "Init, Subscribe")
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        // makes sure to run
+        Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
+        client.unsubscribe(clientListener)
+    }
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/RelayNotifier.kt

@@ -0,0 +1,54 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+import com.vitorpamplona.ammolite.relays.NostrClient
+import com.vitorpamplona.ammolite.relays.Relay
+
+/**
+ * Listens to NostrClient's onNotify messages from the relay
+ */
+class RelayNotifier(
+    val client: NostrClient,
+    val notify: (message: String, relay: Relay) -> Unit,
+) {
+    private val clientListener =
+        object : NostrClient.Listener {
+            override fun onNotify(
+                relay: Relay,
+                message: String,
+            ) {
+                notify(message, relay)
+            }
+        }
+
+    init {
+        Log.d("${this.javaClass.simpleName}", "Init, Subscribe")
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        // makes sure to run
+        Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
+        client.unsubscribe(clientListener)
+    }
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/Subscription.kt

@@ -18,23 +18,28 @@
  * 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.ammolite.relays
+package com.vitorpamplona.ammolite.relays.datasources
 
+import com.vitorpamplona.ammolite.relays.TypedFilter
 import com.vitorpamplona.ammolite.relays.filters.SinceAuthorPerRelayFilter
 import com.vitorpamplona.ammolite.relays.filters.SincePerRelayFilter
 import java.util.UUID
 
 data class Subscription(
     val id: String = UUID.randomUUID().toString().substring(0, 4),
-    val onEOSE: ((Long, String) -> Unit)? = null,
+    val onEose: ((time: Long, relayUrl: String) -> Unit)? = null,
 ) {
     var typedFilters: List<TypedFilter>? = null // Inactive when null
 
-    fun updateEOSE(
+    fun reset() {
+        typedFilters = null
+    }
+
+    fun callEose(
         time: Long,
         relay: String,
     ) {
-        onEOSE?.let { it(time, relay) }
+        onEose?.let { it(time, relay) }
     }
 
     fun hasChangedFiltersFrom(otherFilters: List<TypedFilter>?): Boolean {

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/SubscriptionOrchestrator.kt

@@ -0,0 +1,216 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+import com.vitorpamplona.ammolite.relays.BundledUpdate
+import com.vitorpamplona.ammolite.relays.NostrClient
+import com.vitorpamplona.ammolite.relays.Relay
+import com.vitorpamplona.ammolite.relays.TypedFilter
+import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.utils.TimeUtils
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * Semantically groups Nostr filters and subscriptions in data source objects that
+ * maintain the desired active filter with the relay.
+ */
+abstract class SubscriptionOrchestrator(
+    val client: NostrClient,
+) {
+    private val subscriptions = SubscriptionSet()
+    private var active: Boolean = false
+    private val changingFilters = AtomicBoolean()
+
+    val stats = SubscriptionStats()
+
+    private val clientListener =
+        object : NostrClient.Listener {
+            override fun onEvent(
+                event: Event,
+                subscriptionId: String,
+                relay: Relay,
+                afterEOSE: Boolean,
+            ) {
+                if (subscriptions.contains(subscriptionId)) {
+                    stats.add(subscriptionId, event.kind)
+
+                    if (afterEOSE) {
+                        runAfterEOSE(subscriptionId, relay)
+                    }
+                }
+            }
+
+            override fun onEOSE(
+                relay: Relay,
+                subscriptionId: String,
+            ) {
+                if (subscriptions.contains(subscriptionId)) {
+                    runAfterEOSE(subscriptionId, relay)
+                }
+            }
+        }
+
+    fun runAfterEOSE(
+        subscriptionId: String,
+        relay: Relay,
+    ) {
+        subscriptions[subscriptionId]?.callEose(TimeUtils.oneMinuteAgo(), relay.url)
+    }
+
+    init {
+        Log.d("${this.javaClass.simpleName}", "Init, Subscribe")
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        // makes sure to run
+        Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
+        stop()
+        client.unsubscribe(clientListener)
+        bundler.cancel()
+    }
+
+    open fun start() {
+        Log.d("${this.javaClass.simpleName}", "Start")
+        active = true
+        invalidateFilters()
+    }
+
+    @OptIn(DelicateCoroutinesApi::class)
+    open fun stop() {
+        active = false
+        Log.d("${this.javaClass.simpleName}", "Stop")
+
+        subscriptions.forEach { subscription ->
+            client.close(subscription.id)
+            subscription.reset()
+        }
+    }
+
+    fun getSub(subId: String) = subscriptions.get(subId)
+
+    fun requestNewSubscription(onEOSE: ((Long, String) -> Unit)? = null): Subscription = subscriptions.newSub(onEOSE)
+
+    fun dismissSubscription(subId: String) {
+        getSub(subId)?.let { dismissSubscription(it) }
+    }
+
+    fun dismissSubscription(subscription: Subscription) {
+        client.close(subscription.id)
+        subscription.reset()
+        subscriptions.remove(subscription)
+    }
+
+    fun isUpdatingFilters() = changingFilters.get()
+
+    // Refreshes observers in batches.
+    private val bundler = BundledUpdate(300, Dispatchers.Default)
+
+    fun invalidateFilters() {
+        bundler.invalidate {
+            // println("DataSource: ${this.javaClass.simpleName} InvalidateFilters")
+
+            // adds the time to perform the refresh into this delay
+            // holding off new updates in case of heavy refresh routines.
+            resetFiltersSuspend()
+        }
+    }
+
+    private fun resetFiltersSuspend() {
+        // only runs one at a time. Ignores the others
+        if (changingFilters.compareAndSet(false, true)) {
+            Log.d("${this.javaClass.simpleName}", "resetFiltersSuspend $active")
+            try {
+                resetFiltersSuspendInner()
+            } finally {
+                changingFilters.getAndSet(false)
+            }
+        }
+    }
+
+    private fun resetFiltersSuspendInner() {
+        // saves the channels that are currently active
+        val activeSubscriptions = subscriptions.actives()
+        // saves the current content to only update if it changes
+        val currentFilters = activeSubscriptions.associate { it.id to it.typedFilters }
+
+        // updates all filters
+        updateSubscriptions()
+
+        // Makes sure to only send an updated filter when it actually changes.
+        subscriptions.forEach { newSubscriptionFilters ->
+            val currentFilters = currentFilters[newSubscriptionFilters.id]
+            updateRelaysIfNeeded(newSubscriptionFilters, currentFilters)
+        }
+    }
+
+    fun updateRelaysIfNeeded(
+        updatedSubscription: Subscription,
+        currentFilters: List<TypedFilter>?,
+    ) {
+        val updatedSubscriptionNewFilters = updatedSubscription.typedFilters
+
+        val isActive = client.isActive(updatedSubscription.id)
+
+        if (!isActive && updatedSubscriptionNewFilters != null) {
+            // Filter was removed from the active list
+            // but it is supposed to be there. Send again.
+            if (active) {
+                client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
+            }
+        } else {
+            if (currentFilters != null) {
+                if (updatedSubscriptionNewFilters == null) {
+                    // was active and is not active anymore, just close.
+                    client.close(updatedSubscription.id)
+                } else {
+                    // was active and is still active, check if it has changed.
+                    if (updatedSubscription.hasChangedFiltersFrom(currentFilters)) {
+                        client.close(updatedSubscription.id)
+                        if (active) {
+                            client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
+                        }
+                    } else {
+                        // hasn't changed, does nothing.
+                        // unless the relay has disconnected, then reconnect.
+                        if (active) {
+                            client.sendFilterOnlyIfDisconnected(updatedSubscription.id, updatedSubscriptionNewFilters)
+                        }
+                    }
+                }
+            } else {
+                if (updatedSubscriptionNewFilters == null) {
+                    // was not active and is still not active, does nothing
+                } else {
+                    // was not active and becomes active, sends the filter.
+                    if (active) {
+                        client.sendFilter(updatedSubscription.id, updatedSubscriptionNewFilters)
+                    }
+                }
+            }
+        }
+    }
+
+    abstract fun updateSubscriptions()
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/SubscriptionSet.kt

@@ -0,0 +1,45 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+class SubscriptionSet {
+    private var subscriptions = mapOf<String, Subscription>()
+
+    fun contains(subId: String) = subscriptions.containsKey(subId)
+
+    fun add(sub: Subscription) {
+        subscriptions = subscriptions + Pair(sub.id, sub)
+    }
+
+    fun remove(subId: String) {
+        subscriptions = subscriptions.minus(subId)
+    }
+
+    fun remove(sub: Subscription) = remove(sub.id)
+
+    fun newSub(onEOSE: ((Long, String) -> Unit)? = null): Subscription = Subscription(onEose = onEOSE).also { add(it) }
+
+    fun forEach(action: (Subscription) -> Unit) = subscriptions.values.forEach(action)
+
+    operator fun get(subId: String) = subscriptions[subId]
+
+    fun actives() = subscriptions.values.filter { it.typedFilters != null }
+}

ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/SubscriptionStats.kt

@@ -0,0 +1,60 @@
+/**
+ * Copyright (c) 2024 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.ammolite.relays.datasources
+
+import android.util.Log
+
+class SubscriptionStats {
+    data class Counter(
+        val subscriptionId: String,
+        val eventKind: Int,
+        var counter: Int,
+    )
+
+    private var eventCounter = mapOf<Int, Counter>()
+
+    fun eventCounterIndex(
+        str1: String,
+        str2: Int,
+    ): Int = 31 * str1.hashCode() + str2.hashCode()
+
+    fun add(
+        subscriptionId: String,
+        eventKind: Int,
+    ) {
+        val key = eventCounterIndex(subscriptionId, eventKind)
+        val keyValue = eventCounter[key]
+        if (keyValue != null) {
+            keyValue.counter++
+        } else {
+            eventCounter = eventCounter + Pair(key, Counter(subscriptionId, eventKind, 1))
+        }
+    }
+
+    fun printCounter() {
+        eventCounter.forEach {
+            Log.d(
+                "STATE DUMP ${this.javaClass.simpleName}",
+                "Received Events ${it.value.subscriptionId} ${it.value.eventKind}: ${it.value.counter}",
+            )
+        }
+    }
+}