- Refactors the entire EOSE caching structure.

- Refactors all Datasource classes into a 3 layer design with composeSubscriptions being registered, passed to a group of EOSE managers that can change eose caching capabilities based on needs, and migrates the old subscription Orchestrator into a simpler controller
- Breaks down all Datasource into multiple filter functions
- Improves EOSE support for many of the base filters like on NIP-04 DMs, notifications, etc
- Moves EOSE cache for public, non-user-dependent items like channels, hashtags, etc to their own caching system.
- Improves the Relay speed logger to better track how many events are being received, and check for duplications

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

@@ -26,7 +26,6 @@ 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
@@ -35,9 +34,10 @@ 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(
+class SubscriptionController(
     val client: NostrClient,
-) {
+    val updateSubscriptions: () -> Unit,
+) : SubscriptionControllerService {
     private val subscriptions = SubscriptionSet()
     private var active: Boolean = false
     private val changingFilters = AtomicBoolean()
@@ -50,13 +50,13 @@ abstract class SubscriptionOrchestrator(
                 event: Event,
                 subscriptionId: String,
                 relay: Relay,
+                arrivalTime: Long,
                 afterEOSE: Boolean,
             ) {
                 if (subscriptions.contains(subscriptionId)) {
                     stats.add(subscriptionId, event.kind)
-
                     if (afterEOSE) {
-                        runAfterEOSE(subscriptionId, relay)
+                        runAfterEOSE(subscriptionId, relay, arrivalTime)
                     }
                 }
             }
@@ -64,18 +64,20 @@ abstract class SubscriptionOrchestrator(
             override fun onEOSE(
                 relay: Relay,
                 subscriptionId: String,
+                arrivalTime: Long,
             ) {
                 if (subscriptions.contains(subscriptionId)) {
-                    runAfterEOSE(subscriptionId, relay)
+                    runAfterEOSE(subscriptionId, relay, arrivalTime)
                 }
             }
         }
 
-    fun runAfterEOSE(
+    private fun runAfterEOSE(
         subscriptionId: String,
         relay: Relay,
+        arrivalTime: Long,
     ) {
-        subscriptions[subscriptionId]?.callEose(TimeUtils.oneMinuteAgo(), relay.url)
+        subscriptions[subscriptionId]?.callEose(arrivalTime, relay.url)
     }
 
     init {
@@ -83,7 +85,7 @@ abstract class SubscriptionOrchestrator(
         client.subscribe(clientListener)
     }
 
-    fun destroy() {
+    override fun destroy() {
         // makes sure to run
         Log.d("${this.javaClass.simpleName}", "Destroy, Unsubscribe")
         stop()
@@ -91,14 +93,14 @@ abstract class SubscriptionOrchestrator(
         bundler.cancel()
     }
 
-    open fun start() {
+    override fun start() {
         Log.d("${this.javaClass.simpleName}", "Start")
         active = true
         invalidateFilters()
     }
 
     @OptIn(DelicateCoroutinesApi::class)
-    open fun stop() {
+    override fun stop() {
         active = false
         Log.d("${this.javaClass.simpleName}", "Stop")
 
@@ -108,13 +110,13 @@ abstract class SubscriptionOrchestrator(
         }
     }
 
+    override fun printStats(tag: String) = stats.printCounter(tag)
+
     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(subId: String) = getSub(subId)?.let { dismissSubscription(it) }
 
     fun dismissSubscription(subscription: Subscription) {
         client.close(subscription.id)
@@ -127,7 +129,7 @@ abstract class SubscriptionOrchestrator(
     // Refreshes observers in batches.
     private val bundler = BundledUpdate(300, Dispatchers.Default)
 
-    fun invalidateFilters() {
+    override fun invalidateFilters() {
         bundler.invalidate {
             // println("DataSource: ${this.javaClass.simpleName} InvalidateFilters")
 
@@ -211,6 +213,4 @@ abstract class SubscriptionOrchestrator(
             }
         }
     }
-
-    abstract fun updateSubscriptions()
 }

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

@@ -0,0 +1,33 @@
+/**
+ * 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
+
+interface SubscriptionControllerService {
+    fun start()
+
+    fun stop()
+
+    fun invalidateFilters()
+
+    fun destroy()
+
+    fun printStats(tag: String)
+}