Moves subscription management from Ammolite to Quatrz

2025-07-11 08:51:02 -04:00
parent a94c8673d5
commit 2ecfeb6e7f
36 changed files with 128 additions and 48 deletions
@@ -46,7 +46,29 @@ import kotlinx.coroutines.flow.sample
 import kotlinx.coroutines.flow.stateIn

 /**
- * The Nostr Client manages a relay pool, keeps active subscriptions and manages re-sending of events.
+ * The NostrClient manages Nostr relay operations, subscriptions, and event delivery. It maintains:
+ * - A RelayPool for managing connections to a collection of Nostr relays
+ * - Active subscriptions tracking through PoolSubscriptionRepository
+ * - An event outbox for managing unsent events and retry logic
+ * - Automatic relay pool reconciliation based on subscription and event needs
+ *
+ * Core responsibilities include:
+ * - Initializing and managing relay connections using WebSocket builders
+ * - Coordinating subscription state across multiple relays
+ * - Handling event and filter resending when relays disconnect and reconnect
+ * - Aggregating relay status from the pool's state flow
+ * - Maintaining listeners for propagating relay events and state changes
+ *
+ * Features:
+ * - Reactive updating of relay sets based on subscription and outbox activity
+ * - Relayer reconnection strategy that processes pending requests
+ * - Filter comparison logic to detect significant subscription changes and avoid redundant requests
+ * - Integration with RelayStats for tracking relay performance
+ * - Thread-safe reconnection methods using coroutine flows
+ *
+ * The class combines flows from active subscriptions and the event outbox to ensure relays
+ * are connected to all necessary endpoints. It also listens to relay state changes to update
+ * subscriptions and message retries when connections are re-established.
 */
 class NostrClient(
    private val websocketBuilder: WebsocketBuilder,
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.core.Event
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.core.Event
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.core.Event
@@ -18,7 +18,7 @@
 * 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.acessories
+package com.vitorpamplona.quartz.nip01Core.relay.client.accessories

 import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.relay.client.NostrClient
@@ -41,7 +41,25 @@ val UnsupportedRelayCreation: (url: NormalizedRelayUrl) -> IRelayClient = {
 }

 /**
- * RelayPool manages the connection to multiple Relays and lets consumers deal with simple events.
+ * RelayPool manages a collection of Nostr relays, abstracting individual connections and providing
+ * unified methods for sending events, managing subscriptions, and tracking relay states.
+ *
+ * Key features:
+ * - Maintains a cache of relays using LargeCache for efficient relay management
+ * - Propagates event notifications to a shared listener across all relays
+ * - Provides state tracking through RelayPoolStatus (connected/available relays)
+ * - Supports relay lifecycle operations like connect/disconnect/reconnect
+ * - Maintains an immutable createNewRelay function for custom relay creation
+ *
+ * Listens to relay events and:
+ * 1. Forwards callbacks to parent listener
+ * 2. Updates statusFlow when relay connectivity or events change
+ * 3. Automatically reconnects relays that need reconnection
+ *
+ * Common use cases:
+ * - Sending events to multiple relays simultaneously (send method)
+ * - Managing subscriptions across the relay pool (sendRequest/closed methods)
+ * - Maintaining optimal relay connections (updatePool/addRelay/removeRelay methods)
 */
 class RelayPool(
    val listener: IRelayClientListener = EmptyClientListener,
@@ -52,6 +52,24 @@ import com.vitorpamplona.quartz.utils.bytesUsedInMemory
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.coroutines.cancellation.CancellationException

+/**
+ * A base implementation for a relay client that establishes and manages a WebSocket connection to a Nostr relay.
+ * This class provides fundamental connection handling, message parsing, reconnection logic, and event dispatching.
+ *
+ * @property url The relay's normalized URL.
+ * @property socketBuilder Provides the WebSocket instance for connection.
+ * @property listener Interface to notify the application of relay events and errors.
+ * @property stats Tracks operational statistics of the relay connection.
+ * @property defaultOnConnect Callback executed after a successful connection, allowing subclasses to add initialization logic.
+ *
+ * Reconnection Strategy:
+ * - Uses exponential backoff to retry connections, starting with [DELAY_TO_RECONNECT_IN_MSECS] (500ms).
+ * - Doubles the delay between reconnection attempts in case of failure.
+ *
+ * Message Handling:
+ * - Processes relay messages (e.g., `EVENT`, `EOSE`, `OK`, `AUTH`) and delegates to appropriate callbacks.
+ * - Dispatches received events, notices, and subscription closures via the [listener].
+ */
 open class BasicRelayClient(
    override val url: NormalizedRelayUrl,
    val socketBuilder: WebsocketBuilder,
@@ -0,0 +1,49 @@
+/**
+ * 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.subscriptions
+
+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
+
+data class Subscription(
+    val id: String = newSubId(),
+    val onEose: ((time: Long, relayUrl: NormalizedRelayUrl) -> Unit)? = null,
+) {
+    private var filters: Map<NormalizedRelayUrl, List<Filter>>? = null // Inactive when null
+
+    fun reset() {
+        filters = null
+    }
+
+    fun updateFilters(newFilters: Map<NormalizedRelayUrl, List<Filter>>?) {
+        filters = newFilters
+    }
+
+    fun filters() = filters
+
+    fun callEose(
+        time: Long,
+        relay: NormalizedRelayUrl,
+    ) {
+        onEose?.let { it(time, relay) }
+    }
+}
@@ -0,0 +1,140 @@
+/**
+ * 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.subscriptions
+
+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.single.IRelayClient
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.utils.LargeCache
+
+/**
+ * Manages Nostr subscriptions using a [NostrClient], allowing subscriptions to be created, modified,
+ * and synchronized with relay filters. Subscriptions are stored in a cache and processed through
+ * [updateRelays] to update relay filters dynamically. Also tracks event statistics and EOSE (End of
+ * Stored Events) events, and provides utility methods to interact with subscriptions like dismissal.
+ *
+ * Key responsibilities:
+ * 1. Maintain a cache of active [Subscription] instances.
+ * 2. Handle client events (onEvent, onEOSE) using [clientListener].
+ * 3. Synchronize relay filters via [updateRelays] when subscriptions change.
+ * 4. Provide methods to create, dismiss, and inspect subscriptions.
+ *
+ * Usage:
+ * - Use [requestNewSubscription] to create subscriptions.
+ * - Modify filters on [Subscription] at will and call [updateRelays] to apply changes.
+ * - Update filters based on EOSE callbacks on each subscription
+ * - Dismiss subscriptions with [dismissSubscription].
+ */
+class SubscriptionController(
+    val client: NostrClient,
+) {
+    private val subscriptions = LargeCache<String, Subscription>()
+    private val stats = SubscriptionStats()
+
+    private val clientListener =
+        object : IRelayClientListener {
+            override fun onEvent(
+                relay: IRelayClient,
+                subId: String,
+                event: Event,
+                arrivalTime: Long,
+                afterEOSE: Boolean,
+            ) {
+                if (subscriptions.containsKey(subId)) {
+                    stats.add(subId, event.kind)
+                    if (afterEOSE) {
+                        subscriptions.get(subId)?.callEose(arrivalTime, relay.url)
+                    }
+                }
+            }
+
+            override fun onEOSE(
+                relay: IRelayClient,
+                subId: String,
+                arrivalTime: Long,
+            ) {
+                if (subscriptions.containsKey(subId)) {
+                    subscriptions.get(subId)?.callEose(arrivalTime, relay.url)
+                }
+            }
+        }
+
+    init {
+        client.subscribe(clientListener)
+    }
+
+    fun destroy() {
+        client.unsubscribe(clientListener)
+    }
+
+    fun printStats(tag: String) = stats.printCounter(tag)
+
+    fun getSub(subId: String) = subscriptions.get(subId)
+
+    fun requestNewSubscription(
+        subId: String,
+        onEOSE: ((Long, NormalizedRelayUrl) -> Unit)? = null,
+    ): Subscription = Subscription(subId, onEose = onEOSE).also { subscriptions.put(it.id, it) }
+
+    fun dismissSubscription(subId: String) = getSub(subId)?.let { dismissSubscription(it) }
+
+    fun dismissSubscription(subscription: Subscription) {
+        client.close(subscription.id)
+        subscription.reset()
+        subscriptions.remove(subscription.id)
+    }
+
+    fun updateRelays() {
+        val currentFilters =
+            subscriptions.associateWith { id, sub ->
+                client.getSubscriptionFiltersOrNull(id)
+            }
+
+        subscriptions.forEach { id, sub ->
+            updateRelaysIfNeeded(id, sub.filters(), currentFilters[id])
+        }
+    }
+
+    fun updateRelaysIfNeeded(
+        subId: String,
+        updatedFilters: Map<NormalizedRelayUrl, List<Filter>>?,
+        currentFilters: Map<NormalizedRelayUrl, List<Filter>>?,
+    ) {
+        if (currentFilters != null) {
+            if (updatedFilters == null) {
+                // was active and is not active anymore, just close.
+                client.close(subId)
+            } else {
+                client.sendRequest(subId, updatedFilters)
+            }
+        } else {
+            if (updatedFilters == null) {
+                // was not active and is still not active, does nothing
+            } else {
+                // was not active and becomes active, sends the entire filter.
+                client.sendRequest(subId, updatedFilters)
+            }
+        }
+    }
+}
@@ -0,0 +1,58 @@
+/**
+ * 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.subscriptions
+
+import android.util.Log
+import com.vitorpamplona.quartz.utils.LargeCache
+
+class SubscriptionStats {
+    data class Counter(
+        val subscriptionId: String,
+        val eventKind: Int,
+    ) {
+        var counter: Int = 0
+    }
+
+    private var eventCounter = LargeCache<Int, Counter>()
+
+    private fun eventCounterIndex(
+        str1: String,
+        str2: Int,
+    ): Int = 31 * str1.hashCode() + str2.hashCode()
+
+    fun add(
+        subscriptionId: String,
+        eventKind: Int,
+    ) {
+        val key = eventCounterIndex(subscriptionId, eventKind)
+        val stats = eventCounter.getOrCreate(key) { Counter(subscriptionId, eventKind) }
+        stats.counter++
+    }
+
+    fun printCounter(tag: String) {
+        eventCounter.forEach { _, stats ->
+            Log.d(
+                tag,
+                "Received Events ${stats.subscriptionId} ${stats.eventKind}: ${stats.counter}",
+            )
+        }
+    }
+}
@@ -24,6 +24,23 @@ import android.util.Log
 import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.tags.addressables.Address

+/**
+ * A filter for Nostr events used in relay subscriptions. Supports various criteria
+ * to match events based on IDs, authors, kinds, tags, time ranges, search terms, and limits.
+ *
+ * Parameters:
+ * - ids: Optional list of event IDs to match (must be 64 characters).
+ * - authors: Optional list of author public keys (must be 64 characters).
+ * - kinds: Optional list of event kinds to include.
+ * - tags: Optional map of tag names to values arrays (common tags like 'p', 'e', 'a' are validated).
+ * - since: Optional timestamp for filtering events with publication time ≥ this value.
+ * - until: Optional timestamp for filtering events with publication time ≤ this value.
+ * - limit: Optional maximum number of events to request.
+ * - search: Optional string to search within event content.
+ *
+ * This class performs validation on construction to ensure all string-based identifiers
+ * follow Nostr requirements (64-char hex, onion addresses) and logs errors for invalid inputs.
+ */
 class Filter(
    val ids: List<String>? = null,
    val authors: List<String>? = null,
@@ -49,6 +66,9 @@ class Filter(
        search: String? = this.search,
    ) = Filter(ids, authors, kinds, tags, since, until, limit, search)

+    /**
+     * Returns true if this filter contains any non-null and non-empty criteria.
+     */
    fun isFilledFilter() =
        (ids != null && ids.isNotEmpty()) ||
            (authors != null && authors.isNotEmpty()) ||