Refines the EOSE strategy to make sure relays that have not replied have a chance to

This commit is contained in:
Vitor Pamplona
2023-11-29 10:06:28 -05:00
parent 0dabf3f2d3
commit 44b25b7c7c
2 changed files with 112 additions and 101 deletions
@@ -2,6 +2,7 @@ package com.vitorpamplona.amethyst.service
import com.vitorpamplona.amethyst.model.AddressableNote import com.vitorpamplona.amethyst.model.AddressableNote
import com.vitorpamplona.amethyst.model.Note import com.vitorpamplona.amethyst.model.Note
import com.vitorpamplona.amethyst.model.User
import com.vitorpamplona.amethyst.service.relays.COMMON_FEED_TYPES import com.vitorpamplona.amethyst.service.relays.COMMON_FEED_TYPES
import com.vitorpamplona.amethyst.service.relays.EOSETime import com.vitorpamplona.amethyst.service.relays.EOSETime
import com.vitorpamplona.amethyst.service.relays.JsonFilter import com.vitorpamplona.amethyst.service.relays.JsonFilter
@@ -21,14 +22,13 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
private var addressesToWatch = setOf<Note>() private var addressesToWatch = setOf<Note>()
private fun createReactionsToWatchInAddressFilter(): List<TypedFilter>? { private fun createReactionsToWatchInAddressFilter(): List<TypedFilter>? {
val addressesToWatch = eventsToWatch.filter { it.address() != null } + addressesToWatch val addressesToWatch = eventsToWatch.filter { it.address() != null } + addressesToWatch.filter { it.address() != null }
if (addressesToWatch.isEmpty()) { if (addressesToWatch.isEmpty()) {
return null return null
} }
return addressesToWatch.mapNotNull { return groupByEOSEPresence(eventsToWatch).mapNotNull {
it.address()?.let { aTag ->
TypedFilter( TypedFilter(
types = COMMON_FEED_TYPES, types = COMMON_FEED_TYPES,
filter = JsonFilter( filter = JsonFilter(
@@ -43,14 +43,13 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
CommunityPostApprovalEvent.kind, CommunityPostApprovalEvent.kind,
LiveActivitiesChatMessageEvent.kind LiveActivitiesChatMessageEvent.kind
), ),
tags = mapOf("a" to listOf(aTag.toTag())), tags = mapOf("a" to it.mapNotNull { it.address()?.toTag() }),
since = it.lastReactionsDownloadTime, since = findMinimumEOSEs(it),
limit = 1000 // Max amount of "replies" to download on a specific event. limit = 1000 // Max amount of "replies" to download on a specific event.
) )
) )
} }
} }
}
private fun createAddressFilter(): List<TypedFilter>? { private fun createAddressFilter(): List<TypedFilter>? {
val addressesToWatch = addressesToWatch.filter { it.event == null } val addressesToWatch = addressesToWatch.filter { it.event == null }
@@ -67,7 +66,7 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
filter = JsonFilter( filter = JsonFilter(
kinds = listOf(aTag.kind), kinds = listOf(aTag.kind),
authors = listOf(aTag.pubKeyHex), authors = listOf(aTag.pubKeyHex),
limit = 1000 // Max amount of "replies" to download on a specific event. limit = 5
) )
) )
} else { } else {
@@ -77,7 +76,7 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
kinds = listOf(aTag.kind), kinds = listOf(aTag.kind),
tags = mapOf("d" to listOf(aTag.dTag)), tags = mapOf("d" to listOf(aTag.dTag)),
authors = listOf(aTag.pubKeyHex), authors = listOf(aTag.pubKeyHex),
limit = 1000 // Max amount of "replies" to download on a specific event. limit = 5
) )
) )
} }
@@ -86,13 +85,11 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
} }
private fun createRepliesAndReactionsFilter(): List<TypedFilter>? { private fun createRepliesAndReactionsFilter(): List<TypedFilter>? {
val reactionsToWatch = eventsToWatch if (eventsToWatch.isEmpty()) {
if (reactionsToWatch.isEmpty()) {
return null return null
} }
return reactionsToWatch.map { return groupByEOSEPresence(eventsToWatch).map {
TypedFilter( TypedFilter(
types = COMMON_FEED_TYPES, types = COMMON_FEED_TYPES,
filter = JsonFilter( filter = JsonFilter(
@@ -105,8 +102,8 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
LnZapEvent.kind, LnZapEvent.kind,
PollNoteEvent.kind PollNoteEvent.kind
), ),
tags = mapOf("e" to listOf(it.idHex)), tags = mapOf("e" to it.map { it.idHex }),
since = it.lastReactionsDownloadTime, since = findMinimumEOSEs(it),
limit = 1000 // Max amount of "replies" to download on a specific event. limit = 1000 // Max amount of "replies" to download on a specific event.
) )
) )
@@ -207,3 +204,45 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") {
} }
} }
} }
fun groupByEOSEPresence(notes: Set<Note>): Collection<List<Note>> {
return notes.groupBy { it.lastReactionsDownloadTime.keys.sorted().joinToString(",") }.values
}
fun groupByEOSEPresence(users: Iterable<User>): Collection<List<User>> {
return users.groupBy { it.latestEOSEs.keys.sorted().joinToString(",") }.values
}
fun findMinimumEOSEs(notes: List<Note>): Map<String, EOSETime> {
val minLatestEOSEs = mutableMapOf<String, EOSETime>()
notes.forEach {
it.lastReactionsDownloadTime.forEach {
val minEose = minLatestEOSEs[it.key]
if (minEose == null) {
minLatestEOSEs.put(it.key, EOSETime(it.value.time))
} else if (it.value.time < minEose.time) {
minEose.time = it.value.time
}
}
}
return minLatestEOSEs
}
fun findMinimumEOSEsForUsers(users: List<User>): Map<String, EOSETime> {
val minLatestEOSEs = mutableMapOf<String, EOSETime>()
users.forEach {
it.latestEOSEs.forEach {
val minEose = minLatestEOSEs[it.key]
if (minEose == null) {
minLatestEOSEs.put(it.key, EOSETime(it.value.time))
} else if (it.value.time < minEose.time) {
minEose.time = it.value.time
}
}
}
return minLatestEOSEs
}
@@ -17,59 +17,55 @@ object NostrSingleUserDataSource : NostrDataSource("SingleUserFeed") {
val firstTimers = usersToWatch.filter { it.info?.latestMetadata == null }.map { it.pubkeyHex } val firstTimers = usersToWatch.filter { it.info?.latestMetadata == null }.map { it.pubkeyHex }
if (firstTimers.isEmpty()) return null
return listOf( return listOf(
TypedFilter(
types = COMMON_FEED_TYPES,
filter = JsonFilter(
kinds = listOf(MetadataEvent.kind),
authors = firstTimers
)
)
)
}
fun createUserMetadataStatusReportFilter(): List<TypedFilter>? {
if (usersToWatch.isEmpty()) return null
val secondTimers = usersToWatch.filter { it.info?.latestMetadata != null }
if (secondTimers.isEmpty()) return null
return groupByEOSEPresence(secondTimers).map { group ->
val groupIds = group.map { it.pubkeyHex }
val minEOSEs = findMinimumEOSEsForUsers(group)
listOf(
TypedFilter( TypedFilter(
types = COMMON_FEED_TYPES, types = COMMON_FEED_TYPES,
filter = JsonFilter( filter = JsonFilter(
kinds = listOf(MetadataEvent.kind, StatusEvent.kind), kinds = listOf(MetadataEvent.kind, StatusEvent.kind),
authors = firstTimers, authors = groupIds,
limit = 10 * firstTimers.size since = minEOSEs
) )
) ),
) TypedFilter(
}
fun createUserMetadataFilter(minLatestEOSEs: Map<String, EOSETime>): TypedFilter? {
if (usersToWatch.isEmpty()) return null
return TypedFilter(
types = COMMON_FEED_TYPES,
filter = JsonFilter(
kinds = listOf(StatusEvent.kind),
authors = usersToWatch.map { it.pubkeyHex },
since = minLatestEOSEs
)
)
}
fun createUserStatusFilter(minLatestEOSEs: Map<String, EOSETime>): TypedFilter? {
if (usersToWatch.isEmpty()) return null
return TypedFilter(
types = COMMON_FEED_TYPES,
filter = JsonFilter(
kinds = listOf(StatusEvent.kind),
authors = usersToWatch.map { it.pubkeyHex },
since = minLatestEOSEs
)
)
}
fun createUserReportFilter(minLatestEOSEs: Map<String, EOSETime>): TypedFilter? {
if (usersToWatch.isEmpty()) return null
return TypedFilter(
types = COMMON_FEED_TYPES, types = COMMON_FEED_TYPES,
filter = JsonFilter( filter = JsonFilter(
kinds = listOf(ReportEvent.kind), kinds = listOf(ReportEvent.kind),
tags = mapOf("p" to usersToWatch.map { it.pubkeyHex }), tags = mapOf("p" to groupIds),
since = minLatestEOSEs since = minEOSEs
) )
) )
)
}.flatten()
} }
val userChannel = requestNewChannel() { time, relayUrl -> val userChannel = requestNewChannel() { time, relayUrl ->
checkNotInMainThread()
usersToWatch.forEach { usersToWatch.forEach {
if (it.info?.latestMetadata != null) {
val eose = it.latestEOSEs[relayUrl] val eose = it.latestEOSEs[relayUrl]
if (eose == null) { if (eose == null) {
it.latestEOSEs = it.latestEOSEs + Pair(relayUrl, EOSETime(time)) it.latestEOSEs = it.latestEOSEs + Pair(relayUrl, EOSETime(time))
@@ -78,38 +74,14 @@ object NostrSingleUserDataSource : NostrDataSource("SingleUserFeed") {
} }
} }
} }
val userChannelFirstTimers = requestNewChannel() { time, relayUrl ->
// Many relays operate with limits in the amount of filters.
// As information comes, the filters will be rotated to get more data.
invalidateFilters()
} }
override fun updateChannelFilters() { override fun updateChannelFilters() {
val minLatestEOSEs = mutableMapOf<String, EOSETime>() checkNotInMainThread()
val neverGottenAnEOSE = mutableSetOf<String>()
usersToWatch.forEach {
if (it.latestEOSEs.isEmpty()) { // first time
neverGottenAnEOSE.add(it.pubkeyHex)
} else {
it.latestEOSEs.forEach {
val minEose = minLatestEOSEs[it.key]
if (minEose == null) {
minLatestEOSEs.put(it.key, EOSETime(it.value.time))
} else if (it.value.time < minEose.time) {
minEose.time = it.value.time
}
}
}
}
userChannel.typedFilters = listOfNotNull( userChannel.typedFilters = listOfNotNull(
createUserMetadataFilter(minLatestEOSEs), createUserMetadataFilter(),
createUserStatusFilter(minLatestEOSEs), createUserMetadataStatusReportFilter()
createUserReportFilter(minLatestEOSEs)
).ifEmpty { null }
userChannelFirstTimers.typedFilters = listOfNotNull(
createUserMetadataFilter()
).flatten().ifEmpty { null } ).flatten().ifEmpty { null }
} }