nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

update metadata loading

Browse Source
This commit is contained in:
nrobi144
2026-01-20 15:18:11 +02:00
parent 8f6de1d304
commit 8155cfee9d
77 changed files with 1490 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinator.kt
+221
View File
@@ -0,0 +1,221 @@
/**
* 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.amethyst.commons.relayClient.assemblers
import com.vitorpamplona.amethyst.commons.model.Note
import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataPreloader
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.PrioritizedSubscriptionQueue
import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.SubscriptionPriority
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
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 com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
import kotlinx.coroutines.CoroutineScope
/**
* Coordinates metadata and reactions loading for feed items.
* Ensures metadata (display names, avatars) loads before reactions.
*
* Priority order:
* 1. METADATA - Display names, avatars (highest priority)
* 2. REACTIONS - Likes, zaps, reposts (second priority)
*
* Usage:
* ```
* val coordinator = FeedMetadataCoordinator(client, scope, indexRelays, preloader)
* coordinator.start()
*
* // When feed loads new notes
* LaunchedEffect(notes) {
* coordinator.loadMetadataForNotes(notes)
* }
* ```
*/
class FeedMetadataCoordinator(
private val client: INostrClient,
private val scope: CoroutineScope,
private val indexRelays: Set<NormalizedRelayUrl>,
private val preloader: MetadataPreloader? = null,
private val onEvent: ((Event, NormalizedRelayUrl) -> Unit)? = null,
) {
private val priorityQueue = PrioritizedSubscriptionQueue(scope)
// Track what we've already queued to avoid duplicates
private val queuedPubkeys = mutableSetOf<HexKey>()
private val queuedNoteIds = mutableSetOf<HexKey>()
/**
* Start processing the subscription queue.
* Call once when coordinator is created.
*/
fun start() {
priorityQueue.start { filter ->
// Convert filter to relay-based map for all index relays
val filterMap = indexRelays.associateWith { listOf(filter) }
// Create listener to pass events to the callback
val listener =
if (onEvent != null) {
object : IRequestListener {
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
onEvent.invoke(event, relay)
}
}
} else {
null
}
client.openReqSubscription(
subId = newSubId(),
filters = filterMap,
listener = listener,
)
}
}
/**
* Load metadata and reactions for a list of notes.
* Metadata loads first (priority 1), then reactions (priority 2).
*
* @param notes The notes to load metadata/reactions for
*/
fun loadMetadataForNotes(notes: List<Note>) {
if (notes.isEmpty()) return
// Extract unique authors that we haven't already queued
val authors =
notes
.mapNotNull { it.author?.pubkeyHex }
.filter { it !in queuedPubkeys }
.distinct()
// Extract unique note IDs that we haven't already queued
val noteIds =
notes
.map { it.idHex }
.filter { it !in queuedNoteIds }
.distinct()
// Queue metadata first (highest priority)
if (authors.isNotEmpty()) {
queuedPubkeys.addAll(authors)
// Use preloader if available for rate-limited loading
if (preloader != null) {
notes.mapNotNull { it.author }.forEach { user ->
preloader.preloadForUser(user)
}
} else {
// Direct queue without rate limiting
val metadataFilter =
Filter(
kinds = listOf(MetadataEvent.KIND),
authors = authors,
limit = authors.size,
)
priorityQueue.enqueue(
SubscriptionPriority.METADATA,
metadataFilter,
tag = "feed-metadata",
)
}
}
// Queue reactions second (lower priority)
if (noteIds.isNotEmpty()) {
queuedNoteIds.addAll(noteIds)
val reactionsFilter =
Filter(
kinds = listOf(ReactionEvent.KIND),
tags = mapOf("e" to noteIds),
)
priorityQueue.enqueue(
SubscriptionPriority.REACTIONS,
reactionsFilter,
tag = "feed-reactions",
)
}
}
/**
* Load metadata for specific pubkeys.
* Useful for loading follower/following metadata.
*/
fun loadMetadataForPubkeys(pubkeys: List<HexKey>) {
val newPubkeys = pubkeys.filter { it !in queuedPubkeys }
if (newPubkeys.isEmpty()) return
queuedPubkeys.addAll(newPubkeys)
val filter =
Filter(
kinds = listOf(MetadataEvent.KIND),
authors = newPubkeys,
limit = newPubkeys.size,
)
priorityQueue.enqueue(
SubscriptionPriority.METADATA,
filter,
tag = "pubkey-metadata",
)
}
/**
* Load reactions for specific note IDs.
*/
fun loadReactionsForNotes(noteIds: List<HexKey>) {
val newNoteIds = noteIds.filter { it !in queuedNoteIds }
if (newNoteIds.isEmpty()) return
queuedNoteIds.addAll(newNoteIds)
val filter =
Filter(
kinds = listOf(ReactionEvent.KIND),
tags = mapOf("e" to newNoteIds),
)
priorityQueue.enqueue(
SubscriptionPriority.REACTIONS,
filter,
tag = "note-reactions",
)
}
/**
* Clear queued items. Call when switching feeds.
*/
fun clear() {
priorityQueue.clear()
queuedPubkeys.clear()
queuedNoteIds.clear()
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/MetadataFilterAssembler.kt
+93
View File
@@ -0,0 +1,93 @@
/**
* 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.amethyst.commons.relayClient.assemblers
import com.vitorpamplona.amethyst.commons.relayClient.eoseManagers.SingleSubEoseManager
import com.vitorpamplona.amethyst.commons.relays.SincePerRelayMap
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
/**
* Query state for metadata subscriptions.
* Groups pubkeys with their preferred index relays.
*/
data class MetadataQueryState(
val pubkeys: Set<HexKey>,
val indexRelays: Set<NormalizedRelayUrl>,
)
/**
* Subscribes to Kind 0 (user metadata) for a set of pubkeys.
* Used to load display names, avatars, and other profile information.
*
* This assembler:
* - Batches multiple pubkey requests into single subscription
* - Sends requests to index relays for efficient discovery
* - Caches EOSE to avoid re-fetching known metadata
*/
class MetadataFilterAssembler(
client: INostrClient,
allKeys: () -> Set<MetadataQueryState>,
) : SingleSubEoseManager<MetadataQueryState>(client, allKeys, invalidateAfterEose = true) {
override fun distinct(key: MetadataQueryState): Any = key.pubkeys.hashCode()
override fun updateFilter(
keys: List<MetadataQueryState>,
since: SincePerRelayMap?,
): List<RelayBasedFilter>? {
// Collect all pubkeys and relays from all query states
val allPubkeys = mutableSetOf<HexKey>()
val allRelays = mutableSetOf<NormalizedRelayUrl>()
keys.forEach { state ->
allPubkeys.addAll(state.pubkeys)
allRelays.addAll(state.indexRelays)
}
if (allPubkeys.isEmpty() || allRelays.isEmpty()) return null
val pubkeyList = allPubkeys.toList()
// Create filter for metadata (Kind 0)
val filter =
Filter(
kinds = listOf(MetadataEvent.KIND),
authors = pubkeyList,
limit = pubkeyList.size,
)
// Apply since times per relay
return allRelays.map { relay ->
val sinceTime = since?.get(relay)?.time
val filterWithSince =
if (sinceTime != null) {
filter.copy(since = sinceTime)
} else {
filter
}
RelayBasedFilter(relay, filterWithSince)
}
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/ReactionsFilterAssembler.kt
+92
View File
@@ -0,0 +1,92 @@
/**
* 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.amethyst.commons.relayClient.assemblers
import com.vitorpamplona.amethyst.commons.relayClient.eoseManagers.SingleSubEoseManager
import com.vitorpamplona.amethyst.commons.relays.SincePerRelayMap
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
/**
* Query state for reactions subscriptions.
* Groups note IDs with their preferred relays.
*/
data class ReactionsQueryState(
val noteIds: Set<HexKey>,
val relays: Set<NormalizedRelayUrl>,
)
/**
* Subscribes to Kind 7 (reactions) for a set of note IDs.
* Used to load like counts, zap counts, and other reactions.
*
* This assembler:
* - Batches multiple note ID requests into single subscription
* - Uses e-tags to filter reactions for specific notes
* - Caches EOSE to avoid re-fetching known reactions
*/
class ReactionsFilterAssembler(
client: INostrClient,
allKeys: () -> Set<ReactionsQueryState>,
) : SingleSubEoseManager<ReactionsQueryState>(client, allKeys, invalidateAfterEose = true) {
override fun distinct(key: ReactionsQueryState): Any = key.noteIds.hashCode()
override fun updateFilter(
keys: List<ReactionsQueryState>,
since: SincePerRelayMap?,
): List<RelayBasedFilter>? {
// Collect all note IDs and relays from all query states
val allNoteIds = mutableSetOf<HexKey>()
val allRelays = mutableSetOf<NormalizedRelayUrl>()
keys.forEach { state ->
allNoteIds.addAll(state.noteIds)
allRelays.addAll(state.relays)
}
if (allNoteIds.isEmpty() || allRelays.isEmpty()) return null
val noteIdList = allNoteIds.toList()
// Create filter for reactions (Kind 7) targeting these notes via e-tags
val filter =
Filter(
kinds = listOf(ReactionEvent.KIND),
tags = mapOf("e" to noteIdList),
)
// Apply since times per relay
return allRelays.map { relay ->
val sinceTime = since?.get(relay)?.time
val filterWithSince =
if (sinceTime != null) {
filter.copy(since = sinceTime)
} else {
filter
}
RelayBasedFilter(relay, filterWithSince)
}
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/ComposeSubscriptionManager.kt
+52
View File
@@ -0,0 +1,52 @@
/**
* 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.amethyst.commons.relayClient.composeSubscriptionManagers
import java.util.concurrent.ConcurrentHashMap
/**
* This allows composables to directly register their queries
* to relays. There may be multiple duplications in these
* subscriptions since we do not control when screens are removed.
*/
abstract class ComposeSubscriptionManager<T> : ComposeSubscriptionManagerControls {
private var composeSubscriptions: ConcurrentHashMap<T, T> = ConcurrentHashMap()
// This is called by main. Keep it really fast.
fun subscribe(query: T?) {
if (query == null) return
composeSubscriptions.put(query, query)
invalidateKeys()
}
// This is called by main. Keep it really fast.
fun unsubscribe(query: T?) {
if (query == null) return
composeSubscriptions.remove(query)
invalidateKeys()
}
fun allKeys() = composeSubscriptions.keys
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/ComposeSubscriptionManagerControls.kt
+29
View File
@@ -0,0 +1,29 @@
/**
* 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.amethyst.commons.relayClient.composeSubscriptionManagers
interface ComposeSubscriptionManagerControls {
fun invalidateKeys()
fun invalidateFilters()
fun destroy()
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/composeSubscriptionManagers/MutableComposeSubscriptionManager.kt
+78
View File
@@ -0,0 +1,78 @@
/**
* 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.amethyst.commons.relayClient.composeSubscriptionManagers
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap
/**
* This allows composables to directly register their queries
* to relays. There may be multiple duplications in these
* subscriptions since we do not control when screens are removed.
*
* This is similar to QueryBasedSubscriptionOrchestrator, but it
* also allows the subscription itself to change over time as a
* flow, which trigger an update on the relay subscriptions
*/
abstract class MutableComposeSubscriptionManager<T : MutableQueryState>(
val scope: CoroutineScope,
) : ComposeSubscriptionManagerControls {
private var composeSubscriptions: ConcurrentHashMap<T, Job?> = ConcurrentHashMap()
// This is called by main. Keep it really fast.
fun subscribe(query: T?) {
if (query == null) return
composeSubscriptions[query]?.cancel()
composeSubscriptions[query] =
scope.launch {
query.flow().collectLatest {
invalidateKeys()
}
}
invalidateKeys()
}
// This is called by main. Keep it really fast.
fun unsubscribe(query: T?) {
if (query == null) return
composeSubscriptions[query]?.cancel()
composeSubscriptions.remove(query)
invalidateKeys()
}
fun allKeys() = composeSubscriptions.keys
fun forEachSubscriber(action: (T) -> Unit) {
composeSubscriptions.keys.forEach(action)
}
}
interface MutableQueryState {
fun flow(): Flow<*>
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/BaseEoseManager.kt
+69
View File
@@ -0,0 +1,69 @@
/**
* 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.amethyst.commons.relayClient.eoseManagers
import com.vitorpamplona.amethyst.commons.service.BundledUpdate
import com.vitorpamplona.amethyst.commons.utils.isDebug
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.SubscriptionController
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.Dispatchers
abstract class BaseEoseManager<T>(
val client: INostrClient,
val allKeys: () -> Set<T>,
val sampleTime: Long = 500,
) : IEoseManager {
protected val logTag: String = this::class.simpleName ?: "BaseEoseManager"
private val orchestrator = SubscriptionController(client)
abstract fun updateSubscriptions(keys: Set<T>)
fun newSubscriptionId() = if (isDebug) logTag + newSubId() else newSubId()
fun getSubscription(subId: String) = orchestrator.getSub(subId)
fun requestNewSubscription(listener: IRequestListener) = orchestrator.requestNewSubscription(newSubscriptionId(), listener)
fun dismissSubscription(subId: String) = orchestrator.dismissSubscription(subId)
// Refreshes observers in batches.
private val bundler = BundledUpdate(sampleTime, Dispatchers.IO)
override fun invalidateFilters(ignoreIfDoing: Boolean) {
bundler.invalidate(ignoreIfDoing, ::forceInvalidate)
}
fun forceInvalidate() {
updateSubscriptions(allKeys())
orchestrator.updateRelays()
}
override fun destroy() {
bundler.cancel()
if (isDebug) {
Log.d(logTag, "Destroy, Unsubscribe")
}
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/IEoseManager.kt
+27
View File
@@ -0,0 +1,27 @@
/**
* 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.amethyst.commons.relayClient.eoseManagers
interface IEoseManager {
fun invalidateFilters(ignoreIfDoing: Boolean = false)
fun destroy()
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/PerKeyEoseManager.kt
+162
View File
@@ -0,0 +1,162 @@
/**
* 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.amethyst.commons.relayClient.eoseManagers
import com.vitorpamplona.amethyst.commons.relays.EOSECache
import com.vitorpamplona.amethyst.commons.relays.SincePerRelayMap
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.client.subscriptions.Subscription
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.TimeUtils
/**
* Generic per-key EOSE manager that creates a subscription for each unique key.
*
* This query type creates a new relay subscription for every distinct key K
* extracted from query state T. It is ideal for screens that need separate
* subscriptions per entity (user, thread, etc.).
*
* @param T The query state type (e.g., ThreadQueryState)
* @param K The key type used to deduplicate subscriptions (e.g., String noteId)
*
* This class keeps EOSEs for each key for as long as possible and can be
* shared among multiple query states that map to the same key.
*/
abstract class PerKeyEoseManager<T, K : Any>(
client: INostrClient,
allKeys: () -> Set<T>,
val invalidateAfterEose: Boolean = false,
cacheSize: Int = 200,
) : BaseEoseManager<T>(client, allKeys) {
// EOSE cache keyed by K
private val latestEOSEs = EOSECache<K>(cacheSize)
// Map from key K to subscription ID
private val keySubscriptionMap = mutableMapOf<K, String>()
/**
* Get the since map for a query state's key.
*/
fun since(queryState: T): SincePerRelayMap? = latestEOSEs.since(extractKey(queryState))
/**
* Record a new EOSE for a query state.
*/
open fun newEose(
queryState: T,
relayUrl: NormalizedRelayUrl,
time: Long,
filters: List<Filter>? = null,
) {
latestEOSEs.newEose(extractKey(queryState), relayUrl, time)
if (invalidateAfterEose) {
invalidateFilters()
}
}
/**
* Create a new subscription for a query state.
*/
open fun newSub(queryState: T): Subscription =
requestNewSubscription(
object : IRequestListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
newEose(queryState, relay, TimeUtils.now(), forFilters)
}
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
if (isLive) {
newEose(queryState, relay, TimeUtils.now(), forFilters)
}
}
},
)
/**
* End a subscription for a key.
*/
open fun endSub(
key: K,
subId: String,
) {
dismissSubscription(subId)
keySubscriptionMap.remove(key)
}
/**
* Find or create a subscription for a query state.
*/
fun findOrCreateSubFor(queryState: T): Subscription {
val key = extractKey(queryState)
val subId = keySubscriptionMap[key]
return if (subId == null) {
newSub(queryState).also { keySubscriptionMap[key] = it.id }
} else {
getSubscription(subId) ?: newSub(queryState).also { keySubscriptionMap[key] = it.id }
}
}
override fun updateSubscriptions(keys: Set<T>) {
val uniqueByKey = keys.distinctBy { extractKey(it) }
val updatedKeys = mutableSetOf<K>()
uniqueByKey.forEach { queryState ->
val key = extractKey(queryState)
val newFilters = updateFilter(queryState, since(queryState))?.ifEmpty { null }
findOrCreateSubFor(queryState).updateFilters(newFilters?.groupByRelay())
updatedKeys.add(key)
}
// Clean up subscriptions for keys no longer active
keySubscriptionMap.filter { it.key !in updatedKeys }.forEach {
endSub(it.key, it.value)
}
}
/**
* Build filters for a query state.
* @return List of relay-based filters, or null to clear the subscription
*/
abstract fun updateFilter(
queryState: T,
since: SincePerRelayMap?,
): List<RelayBasedFilter>?
/**
* Extract the deduplication key from a query state.
* Subscriptions are shared among query states with the same key.
*/
abstract fun extractKey(queryState: T): K
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/eoseManagers/SingleSubEoseManager.kt
+103
View File
@@ -0,0 +1,103 @@
/**
* 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.amethyst.commons.relayClient.eoseManagers
import com.vitorpamplona.amethyst.commons.relays.EOSERelayList
import com.vitorpamplona.amethyst.commons.relays.SincePerRelayMap
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.groupByRelay
import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import com.vitorpamplona.quartz.utils.TimeUtils
/**
* This query type creates only ONE relay subscription. It filters duplicates
* by DistinctById.uniqueId() but it doesn't create a new subscription for each.
*
* All filters are passed as a single sub.
*
* It is ideal for temporary filters, including event, user finding that must be
* disabled after the user is found.
*
* This class keeps EOSEs for as long as possible and
* shares all EOSEs among all users.
*/
abstract class SingleSubEoseManager<T>(
client: INostrClient,
allKeys: () -> Set<T>,
val invalidateAfterEose: Boolean = false,
) : BaseEoseManager<T>(client, allKeys) {
// long term EOSE cache
private val latestEOSEs = EOSERelayList()
fun since() = latestEOSEs.since()
open fun newEose(
relay: NormalizedRelayUrl,
time: Long,
filters: List<Filter>? = null,
) {
latestEOSEs.newEose(relay, time)
if (invalidateAfterEose) {
invalidateFilters()
}
}
val sub =
requestNewSubscription(
object : IRequestListener {
override fun onEose(
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
newEose(relay, TimeUtils.now(), forFilters)
}
override fun onEvent(
event: Event,
isLive: Boolean,
relay: NormalizedRelayUrl,
forFilters: List<Filter>?,
) {
if (isLive) {
newEose(relay, TimeUtils.now(), forFilters)
}
}
},
)
override fun updateSubscriptions(keys: Set<T>) {
val uniqueSubscribedAccounts = keys.distinctBy { distinct(it) }
val newFilters = updateFilter(uniqueSubscribedAccounts, since())?.ifEmpty { null }
sub.updateFilters(newFilters?.groupByRelay())
}
abstract fun updateFilter(
keys: List<T>,
since: SincePerRelayMap?,
): List<RelayBasedFilter>?
abstract fun distinct(key: T): Any
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/preload/MetadataPreloader.kt
+109
View File
@@ -0,0 +1,109 @@
/**
* 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.amethyst.commons.relayClient.preload
import com.vitorpamplona.amethyst.commons.model.User
/**
* Preloads user metadata and avatar images for feed items.
* Coordinates between metadata subscription (rate-limited) and image preloading.
*
* Priority order:
* 1. Metadata (display names, avatar URLs) - FIRST
* 2. Avatar images (prefetch when metadata arrives) - SECOND
*/
class MetadataPreloader(
private val rateLimiter: MetadataRateLimiter,
private val imagePrefetcher: ImagePrefetcher? = null,
) {
/**
* Queue users for metadata preloading.
* If user already has metadata, prefetch their avatar image.
* Otherwise, queue for metadata subscription.
*/
fun preloadForUsers(users: Collection<User>) {
users.forEach { user ->
val metadata = user.info
if (metadata != null) {
// Already have metadata, prefetch avatar
metadata.picture?.let { avatarUrl ->
imagePrefetcher?.prefetch(avatarUrl)
}
} else {
// Need to fetch metadata first
rateLimiter.enqueue(user.pubkeyHex)
}
}
}
/**
* Queue a single user for metadata preloading.
*/
fun preloadForUser(user: User) {
val metadata = user.info
if (metadata != null) {
metadata.picture?.let { avatarUrl ->
imagePrefetcher?.prefetch(avatarUrl)
}
} else {
rateLimiter.enqueue(user.pubkeyHex)
}
}
/**
* Called when metadata arrives for a user.
* Triggers avatar image prefetch.
*/
fun onMetadataReceived(user: User) {
user.info?.picture?.let { avatarUrl ->
imagePrefetcher?.prefetch(avatarUrl)
}
}
/**
* Prefetch avatar images for users that already have metadata.
*/
fun prefetchAvatars(users: Collection<User>) {
users.forEach { user ->
user.info?.picture?.let { avatarUrl ->
imagePrefetcher?.prefetch(avatarUrl)
}
}
}
}
/**
* Interface for image prefetching.
* Platform-specific implementations use Coil (Android) or similar (Desktop).
*/
interface ImagePrefetcher {
/**
* Prefetch an image URL into the cache.
*/
fun prefetch(url: String)
/**
* Prefetch multiple image URLs.
*/
fun prefetchAll(urls: Collection<String>) {
urls.forEach { prefetch(it) }
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/preload/MetadataRateLimiter.kt
+109
View File
@@ -0,0 +1,109 @@
/**
* 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.amethyst.commons.relayClient.preload
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.launch
/**
* Global rate limiter for metadata requests to prevent thundering herd on fast scroll.
* Batches pubkeys and processes at a controlled rate.
*
* @param maxRequestsPerSecond Maximum requests to process per second (default 20)
* @param scope CoroutineScope for processing
*/
class MetadataRateLimiter(
private val maxRequestsPerSecond: Int = 20,
private val scope: CoroutineScope,
) {
private val queue = Channel<String>(Channel.BUFFERED)
private val processed = mutableSetOf<String>()
/**
* Enqueue a pubkey for metadata fetching.
* Duplicates within the same batch are automatically filtered.
*/
fun enqueue(pubkey: String) {
if (pubkey !in processed) {
queue.trySend(pubkey)
}
}
/**
* Enqueue multiple pubkeys for metadata fetching.
*/
fun enqueueAll(pubkeys: Collection<String>) {
pubkeys.forEach { enqueue(it) }
}
/**
* Start processing the queue with rate limiting.
* @param onRequest Callback invoked for each pubkey to process
*/
fun start(onRequest: suspend (String) -> Unit) {
scope.launch {
val batch = mutableListOf<String>()
queue.consumeAsFlow().collect { pubkey ->
if (pubkey !in processed) {
batch.add(pubkey)
processed.add(pubkey)
}
// Process batch when we hit the limit or queue is empty
if (batch.size >= maxRequestsPerSecond) {
processBatch(batch, onRequest)
batch.clear()
delay(1000) // Wait 1 second before next batch
}
}
// Process remaining
if (batch.isNotEmpty()) {
processBatch(batch, onRequest)
}
}
}
private suspend fun processBatch(
batch: List<String>,
onRequest: suspend (String) -> Unit,
) {
batch.forEach { pubkey ->
onRequest(pubkey)
}
}
/**
* Clear the processed set to allow re-fetching.
* Call this when switching accounts or clearing cache.
*/
fun reset() {
processed.clear()
}
/**
* Check if a pubkey has already been processed.
*/
fun isProcessed(pubkey: String): Boolean = pubkey in processed
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/subscriptions/KeyDataSourceSubscription.kt
+49
View File
@@ -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.amethyst.commons.relayClient.subscriptions
import androidx.compose.runtime.Composable
import androidx.compose.runtime.DisposableEffect
import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.ComposeSubscriptionManager
import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.MutableComposeSubscriptionManager
import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.MutableQueryState
@Composable
fun <T> KeyDataSourceSubscription(
state: T,
dataSource: ComposeSubscriptionManager<T>,
) = DisposableEffect(state) {
dataSource.subscribe(state)
onDispose {
dataSource.unsubscribe(state)
}
}
@Composable
fun <T : MutableQueryState> KeyDataSourceSubscription(
state: T,
dataSource: MutableComposeSubscriptionManager<T>,
) = DisposableEffect(state) {
dataSource.subscribe(state)
onDispose {
dataSource.unsubscribe(state)
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/subscriptions/PrioritizedSubscriptionQueue.kt
+129
View File
@@ -0,0 +1,129 @@
/**
* 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.amethyst.commons.relayClient.subscriptions
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.launch
/**
* A subscription request with its priority.
*/
data class PrioritizedFilter(
val priority: SubscriptionPriority,
val filter: Filter,
val tag: String? = null,
)
/**
* Queue that processes subscription requests in priority order.
* Ensures metadata loads before reactions, reactions before replies, etc.
*
* Usage:
* ```
* val queue = PrioritizedSubscriptionQueue(scope)
* queue.start { filter -> relayClient.subscribe(filter) }
*
* // Add subscriptions - they'll be processed in priority order
* queue.enqueue(SubscriptionPriority.METADATA, metadataFilter)
* queue.enqueue(SubscriptionPriority.REACTIONS, reactionsFilter)
* ```
*/
class PrioritizedSubscriptionQueue(
private val scope: CoroutineScope,
) {
// Separate channels per priority for efficient processing
private val queues =
SubscriptionPriority.entries.associateWith {
Channel<PrioritizedFilter>(Channel.BUFFERED)
}
private var isRunning = false
/**
* Enqueue a filter with the given priority.
*/
fun enqueue(
priority: SubscriptionPriority,
filter: Filter,
tag: String? = null,
) {
queues[priority]?.trySend(PrioritizedFilter(priority, filter, tag))
}
/**
* Enqueue multiple filters with the same priority.
*/
fun enqueueAll(
priority: SubscriptionPriority,
filters: List<Filter>,
tag: String? = null,
) {
filters.forEach { enqueue(priority, it, tag) }
}
/**
* Start processing the queue.
* Processes higher priority items first within each batch.
*
* @param onSubscribe Callback to execute the subscription
*/
fun start(onSubscribe: suspend (Filter) -> Unit) {
if (isRunning) return
isRunning = true
// Process each priority queue in order
SubscriptionPriority.sortedByPriority().forEach { priority ->
scope.launch {
queues[priority]?.consumeAsFlow()?.collect { item ->
onSubscribe(item.filter)
}
}
}
}
/**
* Process all pending items immediately in priority order.
* Useful for batch operations.
*/
suspend fun flush(onSubscribe: suspend (Filter) -> Unit) {
SubscriptionPriority.sortedByPriority().forEach { priority ->
val queue = queues[priority] ?: return@forEach
while (true) {
val item = queue.tryReceive().getOrNull() ?: break
onSubscribe(item.filter)
}
}
}
/**
* Clear all pending items.
*/
fun clear() {
queues.values.forEach { channel ->
while (channel.tryReceive().isSuccess) {
// Drain the channel
}
}
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relayClient/subscriptions/SubscriptionPriority.kt
+55
View File
@@ -0,0 +1,55 @@
/**
* 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.amethyst.commons.relayClient.subscriptions
/**
* Priority levels for relay subscriptions.
* Lower order = higher priority = processed first.
*
* Priority order ensures progressive rendering:
* 1. METADATA - Display names, avatars load first
* 2. REACTIONS - Like/zap counts load second
* 3. REPLIES - Reply counts
* 4. CONTENT - Additional content
*/
enum class SubscriptionPriority(
val order: Int,
) {
/** User metadata (Kind 0) - display names, avatars. Highest priority. */
METADATA(1),
/** Reactions (Kind 7) - likes, zaps, reposts */
REACTIONS(2),
/** Replies - reply counts and threads */
REPLIES(3),
/** Additional content - lower priority items */
CONTENT(4),
;
companion object {
/**
* Get priorities sorted by order (highest priority first).
*/
fun sortedByPriority(): List<SubscriptionPriority> = entries.sortedBy { it.order }
}
}
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/EOSECache.kt
+144
View File
@@ -0,0 +1,144 @@
/**
* 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.amethyst.commons.relays
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
/**
* Generic EOSE cache keyed by any type K.
* KMP-compatible version (no LruCache dependency).
*
* For memory-constrained environments, consider using platform-specific
* LRU implementations in androidMain/jvmMain.
*/
open class EOSECache<K : Any>(
private val maxSize: Int = 200,
) {
private val cache = linkedMapOf<K, EOSERelayList>()
private val lock = Any()
fun addOrUpdate(
key: K,
relayUrl: NormalizedRelayUrl,
time: Long,
) {
synchronized(lock) {
val relayList = cache[key]
if (relayList == null) {
// Evict oldest if at capacity
if (cache.size >= maxSize) {
cache.remove(cache.keys.first())
}
val newList = EOSERelayList()
newList.addOrUpdate(relayUrl, time)
cache[key] = newList
} else {
relayList.addOrUpdate(relayUrl, time)
}
}
}
fun since(key: K): SincePerRelayMap? =
synchronized(lock) {
cache[key]?.relayList?.toMutableMap()
}
fun newEose(
key: K,
relayUrl: NormalizedRelayUrl,
time: Long,
) = addOrUpdate(key, relayUrl, time)
fun remove(key: K) {
synchronized(lock) {
cache.remove(key)
}
}
fun clear() {
synchronized(lock) {
cache.clear()
}
}
fun size(): Int = synchronized(lock) { cache.size }
}
/**
* Two-level EOSE cache: outer key -> inner key -> relay list.
* Useful for user + list combinations.
*/
open class EOSETwoLevelCache<K1 : Any, K2 : Any>(
private val outerMaxSize: Int = 20,
private val innerMaxSize: Int = 200,
) {
private val cache = linkedMapOf<K1, EOSECache<K2>>()
private val lock = Any()
fun addOrUpdate(
outerKey: K1,
innerKey: K2,
relayUrl: NormalizedRelayUrl,
time: Long,
) {
synchronized(lock) {
val innerCache = cache[outerKey]
if (innerCache == null) {
// Evict oldest if at capacity
if (cache.size >= outerMaxSize) {
cache.remove(cache.keys.first())
}
val newCache = EOSECache<K2>(innerMaxSize)
newCache.addOrUpdate(innerKey, relayUrl, time)
cache[outerKey] = newCache
} else {
innerCache.addOrUpdate(innerKey, relayUrl, time)
}
}
}
fun since(
outerKey: K1,
innerKey: K2,
): SincePerRelayMap? =
synchronized(lock) {
cache[outerKey]?.since(innerKey)
}
fun newEose(
outerKey: K1,
innerKey: K2,
relayUrl: NormalizedRelayUrl,
time: Long,
) = addOrUpdate(outerKey, innerKey, relayUrl, time)
fun removeOuter(key: K1) {
synchronized(lock) {
cache.remove(key)
}
}
fun clear() {
synchronized(lock) {
cache.clear()
}
}
}
commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/network/RelayConnectionManager.kt
+15 -12
View File
@@ -44,24 +44,27 @@ import kotlinx.coroutines.flow.asStateFlow
open class RelayConnectionManager(
websocketBuilder: WebsocketBuilder,
) : IRelayClientListener {
private val client = NostrClient(websocketBuilder)
private val _client = NostrClient(websocketBuilder)
/** Exposes the underlying INostrClient for subscription coordinators */
val client: com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient get() = _client
private val _relayStatuses = MutableStateFlow<Map<NormalizedRelayUrl, RelayStatus>>(emptyMap())
val relayStatuses: StateFlow<Map<NormalizedRelayUrl, RelayStatus>> = _relayStatuses.asStateFlow()
val connectedRelays: StateFlow<Set<NormalizedRelayUrl>> = client.connectedRelaysFlow()
val availableRelays: StateFlow<Set<NormalizedRelayUrl>> = client.availableRelaysFlow()
val connectedRelays: StateFlow<Set<NormalizedRelayUrl>> = _client.connectedRelaysFlow()
val availableRelays: StateFlow<Set<NormalizedRelayUrl>> = _client.availableRelaysFlow()
init {
client.subscribe(this)
_client.subscribe(this)
}
fun connect() {
client.connect()
_client.connect()
}
fun disconnect() {
client.disconnect()
_client.disconnect()
}
fun addRelay(url: String): NormalizedRelayUrl? {
@@ -85,18 +88,18 @@ open class RelayConnectionManager(
listener: IRequestListener? = null,
) {
val filterMap = relays.associateWith { filters }
client.openReqSubscription(subId, filterMap, listener)
_client.openReqSubscription(subId, filterMap, listener)
}
fun unsubscribe(subId: String) {
client.close(subId)
_client.close(subId)
}
fun send(
event: Event,
relays: Set<NormalizedRelayUrl> = connectedRelays.value,
) {
client.send(event, relays)
_client.send(event, relays)
}
/**
@@ -118,7 +121,7 @@ open class RelayConnectionManager(
if (relay !in availableRelays.value) {
updateRelayStatus(relay) { it.copy(connected = false, error = null) }
}
client.send(event, setOf(relay))
_client.send(event, setOf(relay))
}
/**
@@ -135,7 +138,7 @@ open class RelayConnectionManager(
updateRelayStatus(relay) { it.copy(connected = false, error = null) }
}
val filterMap = mapOf(relay to filters)
client.openReqSubscription(
_client.openReqSubscription(
subId = subId,
filters = filterMap,
listener =
@@ -159,7 +162,7 @@ open class RelayConnectionManager(
relay: NormalizedRelayUrl,
subId: String,
) {
client.close(subId)
_client.close(subId)
}
private fun updateRelayStatus(