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

3rd large migration to outbox.

Browse Source
This commit is contained in:
Vitor Pamplona
2025-07-01 20:38:18 -04:00
parent e10332c957
commit 0cfcfaf899
624 changed files with 19405 additions and 11339 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/Constants.kt
-57
View File
@@ -1,57 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import com.vitorpamplona.quartz.nip65RelayList.RelayUrlFormatter
object Constants {
val activeTypesFollows = setOf(FeedType.FOLLOWS, FeedType.PRIVATE_DMS)
val activeTypesChats = setOf(FeedType.FOLLOWS, FeedType.PUBLIC_CHATS, FeedType.PRIVATE_DMS)
val activeTypesGlobalChats = setOf(FeedType.FOLLOWS, FeedType.PUBLIC_CHATS, FeedType.PRIVATE_DMS, FeedType.GLOBAL)
val activeTypesSearch = setOf(FeedType.SEARCH)
val defaultRelays =
arrayOf(
// Free relays for only DMs, Chats and Follows due to the amount of spam
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostr.bitcoiner.social"), read = true, write = true, feedTypes = activeTypesChats),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://relay.nostr.bg"), read = true, write = true, feedTypes = activeTypesChats),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostr.oxtr.dev"), read = true, write = true, feedTypes = activeTypesChats),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostr.fmt.wiz.biz"), read = true, write = false, feedTypes = activeTypesChats),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://relay.damus.io"), read = true, write = true, feedTypes = activeTypesFollows),
// Global
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostr.mom"), read = true, write = true, feedTypes = activeTypesGlobalChats),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nos.lol"), read = true, write = true, feedTypes = activeTypesGlobalChats),
// Paid relays
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostrelites.org"), read = true, write = false, feedTypes = activeTypesGlobalChats),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostr.wine"), read = true, write = false, feedTypes = activeTypesGlobalChats),
// Supporting NIP-50
RelaySetupInfo(RelayUrlFormatter.normalize("wss://relay.nostr.band"), read = true, write = false, feedTypes = activeTypesSearch),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://nostr.wine"), read = true, write = false, feedTypes = activeTypesSearch),
RelaySetupInfo(RelayUrlFormatter.normalize("wss://relay.noswhere.com"), read = true, write = false, feedTypes = activeTypesSearch),
)
val defaultSearchRelaySet =
setOf(
RelayUrlFormatter.normalize("wss://relay.nostr.band"),
RelayUrlFormatter.normalize("wss://nostr.wine"),
RelayUrlFormatter.normalize("wss://relay.noswhere.com"),
)
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/MutableSubscriptionCache.kt
-44
View File
@@ -1,44 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
class MutableSubscriptionCache : SubscriptionCache {
private var subscriptions = mapOf<String, List<TypedFilter>>()
fun add(
subscriptionId: String,
filters: List<TypedFilter> = listOf(),
) {
subscriptions = subscriptions + Pair(subscriptionId, filters)
}
fun remove(subscriptionId: String) {
subscriptions = subscriptions.minus(subscriptionId)
}
override fun isActive(subscriptionId: String): Boolean = subscriptions.contains(subscriptionId)
override fun allSubscriptions(): Map<String, List<TypedFilter>> = subscriptions
override fun getSubscriptionFilters(subId: String): List<TypedFilter> = subscriptions[subId] ?: emptyList()
override fun getSubscriptionFiltersOrNull(subId: String): List<TypedFilter>? = subscriptions[subId]
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/NostrClient.kt
-475
View File
@@ -1,475 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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.nip01Core.relay.sockets.WebsocketBuilderFactory
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.UUID
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
/**
* The Nostr Client manages a relay pool
*/
class NostrClient(
private val websocketBuilder: WebsocketBuilderFactory,
) : RelayPool.Listener {
private val relayPool: RelayPool = RelayPool()
private val activeSubscriptions: MutableSubscriptionCache = MutableSubscriptionCache()
private var listeners = setOf<Listener>()
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)
// Reconnects all relays that may have disconnected
fun reconnect() = relayPool.requestAndWatch()
@Synchronized
fun reconnect(
relays: Array<RelaySetupInfoToConnect>?,
onlyIfChanged: Boolean = false,
) {
Log.d("Relay", "Relay Pool Reconnecting to ${relays?.size} relays: \n${relays?.joinToString("\n") { it.url + " " + it.forceProxy + " " + it.read + " " + it.write + " " + it.feedTypes.joinToString(",") { it.name } }}")
checkNotInMainThread()
if (onlyIfChanged) {
if (!isSameRelaySetConfig(relays)) {
if (this.relayPool.availableRelays() > 0) {
relayPool.disconnect()
relayPool.unregister(this)
relayPool.unloadRelays()
}
if (relays != null) {
relayPool.register(this)
relayPool.loadRelays(relays.map(::buildRelay))
relayPool.requestAndWatch()
}
} else {
// Reconnects all relays that may have disconnected
relayPool.requestAndWatch()
}
} else {
if (this.relayPool.availableRelays() > 0) {
relayPool.disconnect()
relayPool.unregister(this)
relayPool.unloadRelays()
}
if (relays != null) {
relayPool.register(this)
relayPool.loadRelays(relays.map(::buildRelay))
relayPool.requestAndWatch()
}
}
}
fun isSameRelaySetConfig(newRelayConfig: Array<RelaySetupInfoToConnect>?): Boolean {
if (relayPool.availableRelays() != newRelayConfig?.size) return false
newRelayConfig.forEach {
val relay = relayPool.getRelay(it.url) ?: return false
if (!relay.isSameRelayConfig(it)) return false
}
return true
}
fun sendFilter(
subscriptionId: String = UUID.randomUUID().toString().substring(0..10),
filters: List<TypedFilter> = listOf(),
) {
checkNotInMainThread()
activeSubscriptions.add(subscriptionId, filters)
relayPool.sendFilter(subscriptionId, filters)
}
fun sendFilterAndStopOnFirstResponse(
subscriptionId: String = UUID.randomUUID().toString().substring(0..10),
filters: List<TypedFilter> = listOf(),
onResponse: (Event) -> Unit,
) {
checkNotInMainThread()
subscribe(
object : Listener {
override fun onEvent(
event: Event,
subId: String,
relay: Relay,
arrivalTime: Long,
afterEOSE: Boolean,
) {
if (subId == subscriptionId) {
unsubscribe(this)
close(subscriptionId)
onResponse(event)
}
}
},
)
activeSubscriptions.add(subscriptionId, filters)
relayPool.sendFilter(subscriptionId, filters)
}
@OptIn(DelicateCoroutinesApi::class)
suspend fun sendAndWaitForResponse(
signedEvent: Event,
relay: String? = null,
forceProxy: Boolean = false,
feedTypes: Set<FeedType>? = null,
relayList: List<RelaySetupInfo>? = null,
onDone: (() -> Unit)? = null,
additionalListener: Listener? = null,
timeoutInSeconds: Long = 15,
): Boolean {
checkNotInMainThread()
val size = if (relay != null) 1 else relayList?.size ?: relayPool.availableRelays()
val latch = CountDownLatch(size)
val relayErrors = mutableMapOf<String, String>()
var result = false
Log.d("sendAndWaitForResponse", "Waiting for $size responses")
val subscription =
object : Listener {
override fun onError(
error: Error,
subscriptionId: String,
relay: Relay,
) {
relayErrors[relay.url]?.let {
latch.countDown()
}
Log.d("sendAndWaitForResponse", "onError Error from relay ${relay.url} count: ${latch.count} error: $error")
}
override fun onEOSE(
relay: Relay,
subscriptionId: String,
arrivalTime: Long,
) {
latch.countDown()
Log.d("sendAndWaitForResponse", "onEOSE relay ${relay.url} count: ${latch.count}")
}
override fun onRelayStateChange(
type: RelayState,
relay: Relay,
) {
if (type == RelayState.DISCONNECTED) {
latch.countDown()
}
if (type == RelayState.CONNECTED) {
Log.d("sendAndWaitForResponse", "${type.name} Sending event to relay ${relay.url} count: ${latch.count}")
relay.send(signedEvent)
}
Log.d("sendAndWaitForResponse", "onRelayStateChange ${type.name} from relay ${relay.url} count: ${latch.count}")
}
override fun onSendResponse(
eventId: String,
success: Boolean,
message: String,
relay: Relay,
) {
if (eventId == signedEvent.id) {
if (success) {
result = true
}
latch.countDown()
Log.d("sendAndWaitForResponse", "onSendResponse Received response for $eventId from relay ${relay.url} count: ${latch.count} message $message success $success")
}
}
}
subscribe(subscription)
additionalListener?.let { subscribe(it) }
val job =
@OptIn(DelicateCoroutinesApi::class)
GlobalScope.launch(Dispatchers.IO) {
if (relayList != null) {
send(signedEvent, relayList)
} else if (relay == null) {
send(signedEvent)
} else {
sendSingle(signedEvent, RelaySetupInfoToConnect(relay, forceProxy, true, true, emptySet()), onDone ?: {})
}
}
job.join()
runBlocking {
latch.await(timeoutInSeconds, TimeUnit.SECONDS)
}
Log.d("sendAndWaitForResponse", "countdown finished")
unsubscribe(subscription)
additionalListener?.let { unsubscribe(it) }
return result
}
fun getAll(): List<Relay> = relayPool.getAll()
fun sendFilterOnlyIfDisconnected(
subscriptionId: String = UUID.randomUUID().toString().substring(0..10),
filters: List<TypedFilter> = listOf(),
) {
checkNotInMainThread()
activeSubscriptions.add(subscriptionId, filters)
relayPool.connectAndSendFiltersIfDisconnected()
}
fun sendIfExists(
signedEvent: Event,
connectedRelay: Relay,
) {
checkNotInMainThread()
relayPool.getRelays(connectedRelay.url).forEach {
it.send(signedEvent)
}
}
fun sendSingle(
signedEvent: Event,
relayTemplate: RelaySetupInfoToConnect,
onDone: (() -> Unit),
) {
checkNotInMainThread()
relayPool.runCreatingIfNeeded(buildRelay(relayTemplate), onDone = onDone) {
it.send(signedEvent)
}
}
fun send(signedEvent: Event) {
checkNotInMainThread()
relayPool.send(signedEvent)
}
fun send(
signedEvent: Event,
relayList: List<RelaySetupInfo>,
) {
checkNotInMainThread()
relayPool.sendToSelectedRelays(relayList, signedEvent)
}
fun sendPrivately(
signedEvent: Event,
relayList: List<RelaySetupInfoToConnect>,
) {
checkNotInMainThread()
relayList.forEach { relayTemplate ->
relayPool.runCreatingIfNeeded(buildRelay(relayTemplate)) {
it.sendOverride(signedEvent)
}
}
}
fun close(subscriptionId: String) {
relayPool.close(subscriptionId)
activeSubscriptions.remove(subscriptionId)
}
fun isActive(subscriptionId: String): Boolean = activeSubscriptions.isActive(subscriptionId)
@OptIn(DelicateCoroutinesApi::class)
override fun onEvent(
event: Event,
subscriptionId: String,
relay: Relay,
arrivalTime: Long,
afterEOSE: Boolean,
) {
// Releases the Web thread for the new payload.
// May need to add a processing queue if processing new events become too costly.
listeners.forEach { it.onEvent(event, subscriptionId, relay, arrivalTime, afterEOSE) }
}
override fun onEOSE(
relay: Relay,
subscriptionId: String,
arrivalTime: Long,
) {
listeners.forEach { it.onEOSE(relay, subscriptionId, arrivalTime) }
}
override fun onRelayStateChange(
type: RelayState,
relay: Relay,
) {
listeners.forEach { it.onRelayStateChange(type, relay) }
}
@OptIn(DelicateCoroutinesApi::class)
override fun onSendResponse(
eventId: String,
success: Boolean,
message: String,
relay: Relay,
) {
// Releases the Web thread for the new payload.
// May need to add a processing queue if processing new events become too costly.
listeners.forEach { it.onSendResponse(eventId, success, message, relay) }
}
@OptIn(DelicateCoroutinesApi::class)
override fun onAuth(
relay: Relay,
challenge: String,
) {
// Releases the Web thread for the new payload.
// May need to add a processing queue if processing new events become too costly.
listeners.forEach { it.onAuth(relay, challenge) }
}
@OptIn(DelicateCoroutinesApi::class)
override fun onNotify(
relay: Relay,
description: String,
) {
// Releases the Web thread for the new payload.
// May need to add a processing queue if processing new events become too costly.
listeners.forEach { it.onNotify(relay, description) }
}
@OptIn(DelicateCoroutinesApi::class)
override fun onSend(
relay: Relay,
msg: String,
success: Boolean,
) {
listeners.forEach { it.onSend(relay, msg, success) }
}
@OptIn(DelicateCoroutinesApi::class)
override fun onBeforeSend(
relay: Relay,
event: Event,
) {
listeners.forEach { it.onBeforeSend(relay, event) }
}
@OptIn(DelicateCoroutinesApi::class)
override fun onError(
error: Error,
subscriptionId: String,
relay: Relay,
) {
listeners.forEach { it.onError(error, subscriptionId, relay) }
}
fun subscribe(listener: Listener) {
listeners = listeners.plus(listener)
}
fun isSubscribed(listener: Listener): Boolean = listeners.contains(listener)
fun unsubscribe(listener: Listener) {
listeners = listeners.minus(listener)
}
fun allSubscriptions(): Map<String, List<TypedFilter>> = activeSubscriptions.allSubscriptions()
fun getSubscriptionFilters(subId: String): List<TypedFilter> = activeSubscriptions.getSubscriptionFilters(subId)
fun getSubscriptionFiltersOrNull(subId: String): List<TypedFilter>? = activeSubscriptions.getSubscriptionFiltersOrNull(subId)
fun connectedRelays() = relayPool.connectedRelays()
fun relayStatusFlow() = relayPool.statusFlow
interface Listener {
/** A new message was received */
open fun onEvent(
event: Event,
subscriptionId: String,
relay: Relay,
arrivalTime: Long,
afterEOSE: Boolean,
) = Unit
/** Connected to or disconnected from a relay */
open fun onEOSE(
relay: Relay,
subscriptionId: String,
arrivalTime: Long,
) = Unit
/** Connected to or disconnected from a relay */
open fun onRelayStateChange(
type: RelayState,
relay: Relay,
) = Unit
/** When an relay saves or rejects a new event. */
open fun onSendResponse(
eventId: String,
success: Boolean,
message: String,
relay: Relay,
) = Unit
open fun onAuth(
relay: Relay,
challenge: String,
) = Unit
open fun onNotify(
relay: Relay,
description: String,
) = Unit
open fun onSend(
relay: Relay,
msg: String,
success: Boolean,
) = Unit
open fun onBeforeSend(
relay: Relay,
event: Event,
) = Unit
open fun onError(
error: Error,
subscriptionId: String,
relay: Relay,
) = Unit
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/Relay.kt
-274
View File
@@ -1,274 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.RelayState
import com.vitorpamplona.quartz.nip01Core.relay.SimpleClientRelay
import com.vitorpamplona.quartz.nip01Core.relay.SubscriptionCollection
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilderFactory
import com.vitorpamplona.quartz.nip42RelayAuth.RelayAuthEvent
enum class FeedType {
FOLLOWS,
PUBLIC_CHATS,
PRIVATE_DMS,
GLOBAL,
SEARCH,
WALLET_CONNECT,
}
val ALL_FEED_TYPES =
setOf(FeedType.FOLLOWS, FeedType.PUBLIC_CHATS, FeedType.PRIVATE_DMS, FeedType.GLOBAL, FeedType.SEARCH)
val COMMON_FEED_TYPES =
setOf(FeedType.FOLLOWS, FeedType.PUBLIC_CHATS, FeedType.PRIVATE_DMS, FeedType.GLOBAL)
val EVENT_FINDER_TYPES =
setOf(FeedType.FOLLOWS, FeedType.PUBLIC_CHATS, FeedType.GLOBAL)
class RelaySubFilter(
val url: String,
val activeTypes: Set<FeedType>,
val subs: SubscriptionCache,
) : SubscriptionCollection {
fun isMatch(filter: TypedFilter) = activeTypes.any { it in filter.types } && filter.isValidFor(url)
fun match(filters: List<TypedFilter>): Boolean =
filters.any { filter ->
isMatch(filter)
}
override fun isActive(subscriptionId: String): Boolean = subs.isActive(subscriptionId) && match(subs.getSubscriptionFilters(subscriptionId))
override fun getFilters(subscriptionId: String) = filter(subs.getSubscriptionFilters(subscriptionId))
override fun allSubscriptions(): List<com.vitorpamplona.quartz.nip01Core.relay.Subscription> =
subs.allSubscriptions().mapNotNull { filter ->
val filters = filter(filter.value)
if (filters.isNotEmpty()) {
com.vitorpamplona.quartz.nip01Core.relay
.Subscription(filter.key, filters)
} else {
null
}
}
override fun match(
subscriptionId: String,
event: Event,
): Boolean = subs.getSubscriptionFilters(subscriptionId).any { it.filter.match(event, url) }
fun filter(filters: List<TypedFilter>): List<Filter> =
filters.mapNotNull { filter ->
if (isMatch(filter)) {
filter.filter.toRelay(url)
} else {
null
}
}
}
class Relay(
val url: String,
val read: Boolean = true,
val write: Boolean = true,
val forceProxy: Boolean = false,
val activeTypes: Set<FeedType>,
socketBuilderFactory: WebsocketBuilderFactory,
subs: SubscriptionCache,
) : SimpleClientRelay.Listener {
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 brief = RelayBriefInfoCache.get(url)
fun register(listener: Listener) {
listeners = listeners.plus(listener)
}
fun unregister(listener: Listener) {
listeners = listeners.minus(listener)
}
fun isConnected() = inner.isConnected()
fun connect() = inner.connect()
fun connectAndRunAfterSync(onConnected: () -> Unit) {
// BRB is crashing OkHttp Deflater object :(
if (url.contains("brb.io")) return
inner.connectAndRunAfterSync(onConnected)
}
fun sendOutbox() = inner.sendOutbox()
fun disconnect() = inner.disconnect()
fun sendFilter(
requestId: String,
filters: List<TypedFilter>,
) {
if (read) {
inner.sendRequest(requestId, relaySubFilter.filter(filters))
}
}
fun connectAndSendFiltersIfDisconnected() = inner.connectAndSendFiltersIfDisconnected()
fun renewFilters() = inner.renewSubscriptions()
fun sendOverride(signedEvent: Event) = inner.send(signedEvent)
fun send(signedEvent: Event) {
if (signedEvent is RelayAuthEvent || write) {
inner.send(signedEvent)
}
}
fun close(subscriptionId: String) = inner.close(subscriptionId)
fun isSameRelayConfig(other: RelaySetupInfoToConnect): Boolean =
url == other.url &&
forceProxy == other.forceProxy &&
write == other.write &&
read == other.read &&
activeTypes == other.feedTypes
override fun onEvent(
relay: SimpleClientRelay,
subscriptionId: String,
event: Event,
time: Long,
afterEOSE: Boolean,
) = listeners.forEach { it.onEvent(this, subscriptionId, event, time, afterEOSE) }
override fun onError(
relay: SimpleClientRelay,
subscriptionId: String,
error: Error,
) = listeners.forEach { it.onError(this, subscriptionId, error) }
override fun onEOSE(
relay: SimpleClientRelay,
subscriptionId: String,
time: Long,
) = listeners.forEach { it.onEOSE(this, subscriptionId, time) }
override fun onRelayStateChange(
relay: SimpleClientRelay,
type: RelayState,
) = listeners.forEach { it.onRelayStateChange(this, type) }
override fun onSendResponse(
relay: SimpleClientRelay,
eventId: String,
success: Boolean,
message: String,
) = listeners.forEach { it.onSendResponse(this, eventId, success, message) }
override fun onAuth(
relay: SimpleClientRelay,
challenge: String,
) = listeners.forEach { it.onAuth(this, challenge) }
override fun onNotify(
relay: SimpleClientRelay,
description: String,
) = listeners.forEach { it.onNotify(this, description) }
override fun onClosed(
relay: SimpleClientRelay,
subscriptionId: String,
message: String,
) { }
override fun onSend(
relay: SimpleClientRelay,
msg: String,
success: Boolean,
) = listeners.forEach { it.onSend(this, msg, success) }
override fun onBeforeSend(
relay: SimpleClientRelay,
event: Event,
) = listeners.forEach { it.onBeforeSend(this, event) }
interface Listener {
fun onEvent(
relay: Relay,
subscriptionId: String,
event: Event,
time: Long,
afterEOSE: Boolean,
)
fun onEOSE(
relay: Relay,
subscriptionId: String,
time: Long,
)
fun onError(
relay: Relay,
subscriptionId: String,
error: Error,
)
fun onSendResponse(
relay: Relay,
eventId: String,
success: Boolean,
message: String,
)
fun onAuth(
relay: Relay,
challenge: String,
)
fun onRelayStateChange(
relay: Relay,
type: RelayState,
)
/** Relay sent a notification */
fun onNotify(
relay: Relay,
description: String,
)
fun onBeforeSend(
relay: Relay,
event: Event,
)
fun onSend(
relay: Relay,
msg: String,
success: Boolean,
)
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/RelayBriefInfoCache.kt
-46
View File
@@ -1,46 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import android.util.LruCache
import androidx.compose.runtime.Immutable
import com.vitorpamplona.quartz.nip65RelayList.RelayUrlFormatter
object RelayBriefInfoCache {
val cache = LruCache<String, RelayBriefInfo?>(50)
@Immutable
class RelayBriefInfo(
val url: String,
) {
val displayUrl: String = RelayUrlFormatter.displayUrl(url).intern()
val favIcon: String = "https://$displayUrl/favicon.ico".intern()
}
fun get(url: String): RelayBriefInfo {
val info = cache[url]
if (info != null) return info
val newInfo = RelayBriefInfo(url)
cache.put(url, newInfo)
return newInfo
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/RelayPool.kt
-311
View File
@@ -1,311 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import androidx.compose.runtime.Immutable
import com.vitorpamplona.ammolite.service.checkNotInMainThread
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.RelayState
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.launch
/**
* RelayPool manages the connection to multiple Relays and lets consumers deal with simple events.
*/
class RelayPool : Relay.Listener {
private var relays = listOf<Relay>()
private var listeners = setOf<Listener>()
// Backing property to avoid flow emissions from other classes
private var lastStatus = RelayPoolStatus(0, 0)
private val _statusFlow =
MutableSharedFlow<RelayPoolStatus>(1, 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val statusFlow: SharedFlow<RelayPoolStatus> = _statusFlow.asSharedFlow()
fun availableRelays(): Int = relays.size
fun connectedRelays(): Int = relays.count { it.isConnected() }
fun getRelay(url: String): Relay? = relays.firstOrNull { it.url == url }
fun getRelays(url: String): List<Relay> = relays.filter { it.url == url }
fun getAll() = relays
fun runCreatingIfNeeded(
relay: Relay,
timeout: Long = 60000,
onDone: (() -> Unit)? = null,
whenConnected: (Relay) -> Unit,
) {
synchronized(this) {
val matching = getRelays(relay.url)
if (matching.isNotEmpty()) {
matching.forEach { whenConnected(it) }
} else {
addRelay(relay)
relay.connectAndRunAfterSync {
whenConnected(relay)
@OptIn(DelicateCoroutinesApi::class)
GlobalScope.launch(Dispatchers.IO) {
delay(timeout) // waits for a reply
relay.disconnect()
removeRelay(relay)
if (onDone != null) {
onDone()
}
}
}
}
}
}
fun loadRelays(relayList: List<Relay>) {
check(relayList.isNotEmpty()) { "Relay list should never be empty" }
relayList.forEach { addRelayInner(it) }
updateStatus()
}
fun unloadRelays() {
relays.forEach { it.unregister(this) }
relays = listOf()
}
fun requestAndWatch() {
checkNotInMainThread()
relays.forEach { it.connect() }
}
fun sendFilter(
subscriptionId: String,
filters: List<TypedFilter>,
) {
relays.forEach { relay ->
relay.sendFilter(subscriptionId, filters)
}
}
fun connectAndSendFiltersIfDisconnected() {
relays.forEach { it.connectAndSendFiltersIfDisconnected() }
}
fun sendToSelectedRelays(
list: List<RelaySetupInfo>,
signedEvent: Event,
) {
list.forEach { relay -> relays.filter { it.url == relay.url }.forEach { it.sendOverride(signedEvent) } }
}
fun send(signedEvent: Event) {
relays.forEach { it.send(signedEvent) }
}
fun sendOverride(signedEvent: Event) {
relays.forEach { it.sendOverride(signedEvent) }
}
fun close(subscriptionId: String) {
relays.forEach { it.close(subscriptionId) }
}
fun disconnect() {
relays.forEach { it.disconnect() }
}
fun addRelay(relay: Relay) {
addRelayInner(relay)
updateStatus()
}
private fun addRelayInner(relay: Relay) {
relay.register(this)
relays += relay
}
fun removeRelay(relay: Relay) {
relay.unregister(this)
relays = relays.minus(relay)
updateStatus()
}
fun register(listener: Listener) {
listeners = listeners.plus(listener)
}
fun unregister(listener: Listener) {
listeners = listeners.minus(listener)
}
interface Listener {
fun onEvent(
event: Event,
subscriptionId: String,
relay: Relay,
arrivalTime: Long,
afterEOSE: Boolean,
)
fun onEOSE(
relay: Relay,
subscriptionId: String,
arrivalTime: Long,
)
fun onRelayStateChange(
type: RelayState,
relay: Relay,
)
fun onSendResponse(
eventId: String,
success: Boolean,
message: String,
relay: Relay,
)
fun onAuth(
relay: Relay,
challenge: String,
)
fun onNotify(
relay: Relay,
description: String,
)
fun onSend(
relay: Relay,
msg: String,
success: Boolean,
)
fun onBeforeSend(
relay: Relay,
event: Event,
)
fun onError(
error: Error,
subscriptionId: String,
relay: Relay,
)
}
override fun onEvent(
relay: Relay,
subscriptionId: String,
event: Event,
arrivalTime: Long,
afterEOSE: Boolean,
) {
listeners.forEach { it.onEvent(event, subscriptionId, relay, arrivalTime, afterEOSE) }
}
override fun onError(
relay: Relay,
subscriptionId: String,
error: Error,
) {
listeners.forEach { it.onError(error, subscriptionId, relay) }
updateStatus()
}
override fun onEOSE(
relay: Relay,
subscriptionId: String,
arrivalTime: Long,
) {
listeners.forEach { it.onEOSE(relay, subscriptionId, arrivalTime) }
updateStatus()
}
override fun onRelayStateChange(
relay: Relay,
type: RelayState,
) {
listeners.forEach { it.onRelayStateChange(type, relay) }
}
override fun onSendResponse(
relay: Relay,
eventId: String,
success: Boolean,
message: String,
) {
listeners.forEach { it.onSendResponse(eventId, success, message, relay) }
}
override fun onAuth(
relay: Relay,
challenge: String,
) {
listeners.forEach { it.onAuth(relay, challenge) }
}
override fun onNotify(
relay: Relay,
description: String,
) {
listeners.forEach { it.onNotify(relay, description) }
}
override fun onSend(
relay: Relay,
msg: String,
success: Boolean,
) {
listeners.forEach { it.onSend(relay, msg, success) }
}
override fun onBeforeSend(
relay: Relay,
event: Event,
) {
listeners.forEach { it.onBeforeSend(relay, event) }
}
private fun updateStatus() {
val connected = connectedRelays()
val available = availableRelays()
if (lastStatus.connected != connected || lastStatus.available != available) {
lastStatus = RelayPoolStatus(connected, available)
_statusFlow.tryEmit(lastStatus)
}
}
}
@Immutable
data class RelayPoolStatus(
val connected: Int,
val available: Int,
val isConnected: Boolean = connected > 0,
)
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/RelaySetupInfo.kt
-40
View File
@@ -1,40 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import androidx.compose.runtime.Immutable
@Immutable
data class RelaySetupInfo(
val url: String,
val read: Boolean,
val write: Boolean,
val feedTypes: Set<FeedType>,
)
@Immutable
data class RelaySetupInfoToConnect(
val url: String,
val forceProxy: Boolean,
val read: Boolean,
val write: Boolean,
val feedTypes: Set<FeedType>,
)
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/RelayStats.kt
-71
View File
@@ -1,71 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import com.vitorpamplona.quartz.nip01Core.relay.RelayStat
object RelayStats {
private val innerCache = mutableMapOf<String, RelayStat>()
fun get(url: String): RelayStat = innerCache.getOrPut(url) { RelayStat() }
fun addBytesReceived(
url: String,
bytesUsedInMemory: Long,
) {
get(url).addBytesReceived(bytesUsedInMemory)
}
fun addBytesSent(
url: String,
bytesUsedInMemory: Long,
) {
get(url).addBytesSent(bytesUsedInMemory)
}
fun newError(
url: String,
error: String?,
) {
get(url).newError(error)
}
fun newNotice(
url: String,
notice: String?,
) {
get(url).newNotice(notice)
}
fun setPing(
url: String,
pingInMs: Long,
) {
get(url).pingInMs = pingInMs
}
fun newSpam(
url: String,
explanation: String,
) {
get(url).newSpam(explanation)
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/SubscriptionCache.kt
-31
View File
@@ -1,31 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
interface SubscriptionCache {
fun isActive(subscriptionId: String): Boolean
fun allSubscriptions(): Map<String, List<TypedFilter>>
fun getSubscriptionFilters(subId: String): List<TypedFilter>
fun getSubscriptionFiltersOrNull(subId: String): List<TypedFilter>?
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/TypedFilter.kt
-31
View File
@@ -1,31 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays
import com.vitorpamplona.ammolite.relays.filters.IPerRelayFilter
class TypedFilter(
val types: Set<FeedType>,
val filter: IPerRelayFilter,
) {
// This only exists because some relays confuse empty lists with null lists
fun isValidFor(url: String) = filter.isValidFor(url)
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/EventCollector.kt
-58
View File
@@ -1,58 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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,
arrivalTime: Long,
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
-276
View File
@@ -1,276 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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 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,
arrivalTime: Long,
afterEOSE: Boolean,
) {
if (subscriptions.contains(subscriptionId)) {
stats.add(subscriptionId, event.kind)
consume(event, relay)
if (afterEOSE) {
markAsEOSE(subscriptionId, relay, arrivalTime)
}
}
}
override fun onEOSE(
relay: Relay,
subscriptionId: String,
arrivalTime: Long,
) {
if (subscriptions.contains(subscriptionId)) {
markAsEOSE(subscriptionId, relay, arrivalTime)
}
}
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,
arrivalTime: Long,
) {
subscriptions[subscriptionId]?.callEose(
arrivalTime,
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
-51
View File
@@ -1,51 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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
-59
View File
@@ -1,59 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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/RelayLogger.kt
-67
View File
@@ -1,67 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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 onNotify messages from the relay
*/
class RelayLogger(
val client: NostrClient,
val notify: (message: String, relay: Relay) -> Unit,
) {
private val clientListener =
object : NostrClient.Listener {
/** A new message was received */
override fun onEvent(
event: Event,
subscriptionId: String,
relay: Relay,
arrivalTime: Long,
afterEOSE: Boolean,
) {
Log.d("Relay", "Relay onEVENT ${relay.url} ($subscriptionId - $afterEOSE) ${event.toJson()}")
}
override fun onSend(
relay: Relay,
msg: String,
success: Boolean,
) {
Log.d("Relay", "Relay send ${relay.url} (${msg.length} chars) $msg")
}
}
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
-58
View File
@@ -1,58 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.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,
) {
companion object {
val TAG = RelayNotifier::class.java.simpleName
}
private val clientListener =
object : NostrClient.Listener {
override fun onNotify(
relay: Relay,
message: String,
) {
notify(message, relay)
}
}
init {
Log.d(TAG, "Init, Subscribe")
client.subscribe(clientListener)
}
fun destroy() {
// makes sure to run
Log.d(TAG, "Destroy, Unsubscribe")
client.unsubscribe(clientListener)
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/Subscription.kt
+18 -88
View File
@@ -20,116 +20,46 @@
*/
package com.vitorpamplona.ammolite.relays.datasources
import com.vitorpamplona.ammolite.relays.TypedFilter
import com.vitorpamplona.ammolite.relays.filters.NormalFilter
import com.vitorpamplona.ammolite.relays.filters.SinceAuthorPerRelayFilter
import com.vitorpamplona.ammolite.relays.filters.SincePerRelayFilter
import java.util.UUID
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
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 kotlin.collections.forEachIndexed
data class Subscription(
val id: String = UUID.randomUUID().toString().substring(0, 4),
val onEose: ((time: Long, relayUrl: String) -> Unit)? = null,
val id: String = newSubId(),
val onEose: ((time: Long, relayUrl: NormalizedRelayUrl) -> Unit)? = null,
) {
var typedFilters: List<TypedFilter>? = null // Inactive when null
var relayBasedFilters: List<RelayBasedFilter>? = null // Inactive when null
fun reset() {
typedFilters = null
relayBasedFilters = null
}
fun callEose(
time: Long,
relay: String,
relay: NormalizedRelayUrl,
) {
onEose?.let { it(time, relay) }
}
fun hasChangedFiltersFrom(otherFilters: List<TypedFilter>?): Boolean {
if (typedFilters == null && otherFilters == null) return false
if (typedFilters?.size != otherFilters?.size) return true
fun hasChangedFiltersFrom(otherFilters: List<RelayBasedFilter>?): Boolean {
if (relayBasedFilters == null && otherFilters == null) return false
if (relayBasedFilters?.size != otherFilters?.size) return true
typedFilters?.forEachIndexed { index, typedFilter ->
relayBasedFilters?.forEachIndexed { index, relaySetFilter ->
val otherFilter = otherFilters?.getOrNull(index) ?: return true
if (typedFilter.filter is SincePerRelayFilter && otherFilter.filter is SincePerRelayFilter) {
return isDifferent(typedFilter.filter, otherFilter.filter)
}
if (relaySetFilter.relay != otherFilter.relay) return true
if (typedFilter.filter is SinceAuthorPerRelayFilter && otherFilter.filter is SinceAuthorPerRelayFilter) {
return isDifferent(typedFilter.filter, otherFilter.filter)
}
if (typedFilter.filter is NormalFilter && otherFilter.filter is NormalFilter) {
return isDifferent(typedFilter.filter, otherFilter.filter)
}
return true
return isDifferent(relaySetFilter.filter, otherFilter.filter)
}
return false
}
fun isDifferent(
filter1: SincePerRelayFilter,
filter2: SincePerRelayFilter,
): Boolean {
// Does not check SINCE on purpose. Avoids replacing the filter if SINCE was all that changed.
// fast check
if (filter1.authors?.size != filter2.authors?.size ||
filter1.ids?.size != filter2.ids?.size ||
filter1.tags?.size != filter2.tags?.size ||
filter1.kinds?.size != filter2.kinds?.size ||
filter1.limit != filter2.limit ||
filter1.search?.length != filter2.search?.length ||
filter1.until != filter2.until
) {
return true
}
// deep check
if (filter1.ids != filter2.ids ||
filter1.authors != filter2.authors ||
filter1.tags != filter2.tags ||
filter1.kinds != filter2.kinds ||
filter1.search != filter2.search
) {
return true
}
return false
}
fun isDifferent(
filter1: SinceAuthorPerRelayFilter,
filter2: SinceAuthorPerRelayFilter,
): Boolean {
// Does not check SINCE on purpose. Avoids replacing the filter if SINCE was all that changed.
// fast check
if (filter1.authors?.size != filter2.authors?.size ||
filter1.ids?.size != filter2.ids?.size ||
filter1.tags?.size != filter2.tags?.size ||
filter1.kinds?.size != filter2.kinds?.size ||
filter1.limit != filter2.limit ||
filter1.search?.length != filter2.search?.length ||
filter1.until != filter2.until
) {
return true
}
// deep check
if (filter1.ids != filter2.ids ||
filter1.authors != filter2.authors ||
filter1.tags != filter2.tags ||
filter1.kinds != filter2.kinds ||
filter1.search != filter2.search
) {
return true
}
return false
}
fun isDifferent(
filter1: NormalFilter,
filter2: NormalFilter,
filter1: Filter,
filter2: Filter,
): Boolean {
// Does not check SINCE on purpose. Avoids replacing the filter if SINCE was all that changed.
// fast check
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/SubscriptionController.kt
+19 -17
View File
@@ -21,10 +21,12 @@
package com.vitorpamplona.ammolite.relays.datasources
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.client.NostrClient
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.IRelayClientListener
import com.vitorpamplona.quartz.nip01Core.relay.client.pool.RelayBasedFilter
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.Dispatchers
import java.util.concurrent.atomic.AtomicBoolean
@@ -44,36 +46,36 @@ class SubscriptionController(
val stats = SubscriptionStats()
private val clientListener =
object : NostrClient.Listener {
object : IRelayClientListener {
override fun onEvent(
relay: IRelayClient,
subId: String,
event: Event,
subscriptionId: String,
relay: Relay,
arrivalTime: Long,
afterEOSE: Boolean,
) {
if (subscriptions.contains(subscriptionId)) {
stats.add(subscriptionId, event.kind)
if (subscriptions.contains(subId)) {
stats.add(subId, event.kind)
if (afterEOSE) {
runAfterEOSE(subscriptionId, relay, arrivalTime)
runAfterEOSE(subId, relay, arrivalTime)
}
}
}
override fun onEOSE(
relay: Relay,
subscriptionId: String,
relay: IRelayClient,
subId: String,
arrivalTime: Long,
) {
if (subscriptions.contains(subscriptionId)) {
runAfterEOSE(subscriptionId, relay, arrivalTime)
if (subscriptions.contains(subId)) {
runAfterEOSE(subId, relay, arrivalTime)
}
}
}
private fun runAfterEOSE(
subscriptionId: String,
relay: Relay,
relay: IRelayClient,
arrivalTime: Long,
) {
subscriptions[subscriptionId]?.callEose(arrivalTime, relay.url)
@@ -109,7 +111,7 @@ class SubscriptionController(
fun getSub(subId: String) = subscriptions.get(subId)
fun requestNewSubscription(onEOSE: ((Long, String) -> Unit)? = null): Subscription = subscriptions.newSub(onEOSE)
fun requestNewSubscription(onEOSE: ((Long, NormalizedRelayUrl) -> Unit)? = null): Subscription = subscriptions.newSub(onEOSE)
fun dismissSubscription(subId: String) = getSub(subId)?.let { dismissSubscription(it) }
@@ -163,9 +165,9 @@ class SubscriptionController(
fun updateRelaysIfNeeded(
updatedSubscription: Subscription,
currentFilters: List<TypedFilter>?,
currentFilters: List<RelayBasedFilter>?,
) {
val updatedSubscriptionNewFilters = updatedSubscription.typedFilters
val updatedSubscriptionNewFilters = updatedSubscription.relayBasedFilters
val isActive = client.isActive(updatedSubscription.id)
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/datasources/SubscriptionSet.kt
+4 -2
View File
@@ -20,6 +20,8 @@
*/
package com.vitorpamplona.ammolite.relays.datasources
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
class SubscriptionSet {
private var subscriptions = mapOf<String, Subscription>()
@@ -35,11 +37,11 @@ class SubscriptionSet {
fun remove(sub: Subscription) = remove(sub.id)
fun newSub(onEOSE: ((Long, String) -> Unit)? = null): Subscription = Subscription(onEose = onEOSE).also { add(it) }
fun newSub(onEOSE: ((Long, NormalizedRelayUrl) -> 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 }
fun actives() = subscriptions.values.filter { it.relayBasedFilters != null }
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/filters/IPerRelayFilter.kt
-40
View File
@@ -1,40 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays.filters
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
interface IPerRelayFilter {
fun toRelay(forRelay: String): Filter
fun toJson(forRelay: String): String
fun match(
event: Event,
forRelay: String,
): Boolean
fun toDebugJson(): String
// This only exists because some relays confuse empty lists with null lists
fun isValidFor(url: String): Boolean
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/filters/EOSETime.kt → ammolite/src/main/java/com/vitorpamplona/ammolite/relays/filters/MutableTime.kt
+15 -4
View File
@@ -23,14 +23,25 @@ package com.vitorpamplona.ammolite.relays.filters
/*
* Wrapper class to allow changing in EOSE without modifying the list it is included within
*/
class EOSETime(
var time: Long,
) {
class MutableTime(startTime: Long) {
var time: Long = startTime
private set
override fun toString(): String = time.toString()
fun update(newTime: Long) {
fun updateIfNewer(newTime: Long) {
if (newTime > time) {
time = newTime
}
}
fun updateIfOlder(newTime: Long) {
if (newTime < time) {
time = newTime
}
}
fun minus(delta: Int) = MutableTime(time - delta)
fun copy() = MutableTime(time)
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/filters/NormalFilter.kt
-57
View File
@@ -1,57 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays.filters
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.jackson.EventMapper
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterMatcher
import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterSerializer
/**
* This is a nostr filter with per-relay authors list and since parameters
*/
class NormalFilter(
val ids: List<String>? = null,
val authors: List<String>? = null,
val kinds: List<Int>? = null,
val tags: Map<String, List<String>>? = null,
val since: Long? = null,
val until: Long? = null,
val limit: Int? = null,
val search: String? = null,
) : IPerRelayFilter {
override fun isValidFor(url: String) = true
override fun toRelay(forRelay: String) = Filter(ids, authors, kinds, tags, since, until, limit, search)
override fun toJson(forRelay: String) = FilterSerializer.toJson(ids, authors, kinds, tags, since, until, limit, search)
override fun match(
event: Event,
forRelay: String,
) = FilterMatcher.match(event, ids, authors, kinds, tags, since, until)
override fun toDebugJson(): String {
val obj = FilterSerializer.toJsonObject(ids, authors, kinds, tags, since, until, limit, search)
return EventMapper.toJson(obj)
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/filters/SinceAuthorPerRelayFilter.kt
-89
View File
@@ -1,89 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays.filters
import com.fasterxml.jackson.databind.node.JsonNodeFactory
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.jackson.EventMapper
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterMatcher
import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterSerializer
/**
* This is a nostr filter with per-relay authors list and since parameters
*/
class SinceAuthorPerRelayFilter(
val ids: List<HexKey>? = null,
val authors: Map<String, List<HexKey>>? = null,
val kinds: List<Int>? = null,
val tags: Map<String, List<String>>? = null,
val since: Map<String, EOSETime>? = null,
val until: Long? = null,
val limit: Int? = null,
val search: String? = null,
) : IPerRelayFilter {
// This only exists because some relays consider empty arrays as null and return everything.
// So, if there is an author list, but no list for the specific relay or if the list is empty
// don't send it.
override fun isValidFor(forRelay: String) = authors == null || !authors[forRelay].isNullOrEmpty()
override fun toRelay(forRelay: String) = Filter(ids, authors?.get(forRelay), kinds, tags, since?.get(forRelay)?.time, until, limit, search)
override fun toJson(forRelay: String): String = FilterSerializer.toJson(ids, authors?.get(forRelay), kinds, tags, since?.get(forRelay)?.time, until, limit, search)
override fun match(
event: Event,
forRelay: String,
) = FilterMatcher.match(event, ids, authors?.get(forRelay), kinds, tags, since?.get(forRelay)?.time, until)
override fun toDebugJson(): String {
val factory = JsonNodeFactory.instance
val obj = FilterSerializer.toJsonObject(ids, null, kinds, tags, null, until, limit, search)
authors?.run {
if (isNotEmpty()) {
val jsonObjectPerRelayAuthors = factory.objectNode()
entries.forEach { relayAuthorPairs ->
jsonObjectPerRelayAuthors.replace(
relayAuthorPairs.key,
factory.arrayNode(relayAuthorPairs.value.size).apply {
relayAuthorPairs.value.forEach {
add(it)
}
},
)
}
obj.replace("authors", jsonObjectPerRelayAuthors)
}
}
since?.run {
if (isNotEmpty()) {
val jsonObjectSince = factory.objectNode()
entries.forEach { sincePairs ->
jsonObjectSince.put(sincePairs.key, "${sincePairs.value}")
}
obj.replace("since", jsonObjectSince)
}
}
return EventMapper.mapper.writeValueAsString(obj)
}
}
ammolite/src/main/java/com/vitorpamplona/ammolite/relays/filters/SincePerRelayFilter.kt
-69
View File
@@ -1,69 +0,0 @@
/**
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.ammolite.relays.filters
import com.fasterxml.jackson.databind.node.JsonNodeFactory
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.jackson.EventMapper
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterMatcher
import com.vitorpamplona.quartz.nip01Core.relay.filters.FilterSerializer
/**
* This is a nostr filter with per-relay authors list and since parameters
*/
class SincePerRelayFilter(
val ids: List<String>? = null,
val authors: List<String>? = null,
val kinds: List<Int>? = null,
val tags: Map<String, List<String>>? = null,
val since: Map<String, EOSETime>? = null,
val until: Long? = null,
val limit: Int? = null,
val search: String? = null,
) : IPerRelayFilter {
override fun isValidFor(url: String) = true
override fun toRelay(forRelay: String) = Filter(ids, authors, kinds, tags, since?.get(forRelay)?.time, until, limit, search)
override fun toJson(forRelay: String) = FilterSerializer.toJson(ids, authors, kinds, tags, since?.get(forRelay)?.time, until, limit, search)
override fun match(
event: Event,
forRelay: String,
) = FilterMatcher.match(event, ids, authors, kinds, tags, since?.get(forRelay)?.time, until)
override fun toDebugJson(): String {
val factory = JsonNodeFactory.instance
val obj = FilterSerializer.toJsonObject(ids, authors, kinds, tags, null, until, limit, search)
since?.run {
if (isNotEmpty()) {
val jsonObjectSince = factory.objectNode()
entries.forEach { sincePairs ->
jsonObjectSince.put(sincePairs.key, "${sincePairs.value}")
}
obj.replace("since", jsonObjectSince)
}
}
return EventMapper.toJson(obj)
}
}