From 44b25b7c7cb9006fe839e72644b542e00a5d5061 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Wed, 29 Nov 2023 10:06:28 -0500 Subject: [PATCH] Refines the EOSE strategy to make sure relays that have not replied have a chance to --- .../service/NostrSingleEventDataSource.kt | 99 ++++++++++----- .../service/NostrSingleUserDataSource.kt | 114 +++++++----------- 2 files changed, 112 insertions(+), 101 deletions(-) diff --git a/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleEventDataSource.kt b/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleEventDataSource.kt index 545af39de..5baf073a4 100644 --- a/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleEventDataSource.kt +++ b/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleEventDataSource.kt @@ -2,6 +2,7 @@ package com.vitorpamplona.amethyst.service import com.vitorpamplona.amethyst.model.AddressableNote 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.EOSETime import com.vitorpamplona.amethyst.service.relays.JsonFilter @@ -21,34 +22,32 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") { private var addressesToWatch = setOf() private fun createReactionsToWatchInAddressFilter(): List? { - val addressesToWatch = eventsToWatch.filter { it.address() != null } + addressesToWatch + val addressesToWatch = eventsToWatch.filter { it.address() != null } + addressesToWatch.filter { it.address() != null } if (addressesToWatch.isEmpty()) { return null } - return addressesToWatch.mapNotNull { - it.address()?.let { aTag -> - TypedFilter( - types = COMMON_FEED_TYPES, - filter = JsonFilter( - kinds = listOf( - TextNoteEvent.kind, - ReactionEvent.kind, - RepostEvent.kind, - GenericRepostEvent.kind, - ReportEvent.kind, - LnZapEvent.kind, - PollNoteEvent.kind, - CommunityPostApprovalEvent.kind, - LiveActivitiesChatMessageEvent.kind - ), - tags = mapOf("a" to listOf(aTag.toTag())), - since = it.lastReactionsDownloadTime, - limit = 1000 // Max amount of "replies" to download on a specific event. - ) + return groupByEOSEPresence(eventsToWatch).mapNotNull { + TypedFilter( + types = COMMON_FEED_TYPES, + filter = JsonFilter( + kinds = listOf( + TextNoteEvent.kind, + ReactionEvent.kind, + RepostEvent.kind, + GenericRepostEvent.kind, + ReportEvent.kind, + LnZapEvent.kind, + PollNoteEvent.kind, + CommunityPostApprovalEvent.kind, + LiveActivitiesChatMessageEvent.kind + ), + tags = mapOf("a" to it.mapNotNull { it.address()?.toTag() }), + since = findMinimumEOSEs(it), + limit = 1000 // Max amount of "replies" to download on a specific event. ) - } + ) } } @@ -67,7 +66,7 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") { filter = JsonFilter( kinds = listOf(aTag.kind), authors = listOf(aTag.pubKeyHex), - limit = 1000 // Max amount of "replies" to download on a specific event. + limit = 5 ) ) } else { @@ -77,7 +76,7 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") { kinds = listOf(aTag.kind), tags = mapOf("d" to listOf(aTag.dTag)), 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? { - val reactionsToWatch = eventsToWatch - - if (reactionsToWatch.isEmpty()) { + if (eventsToWatch.isEmpty()) { return null } - return reactionsToWatch.map { + return groupByEOSEPresence(eventsToWatch).map { TypedFilter( types = COMMON_FEED_TYPES, filter = JsonFilter( @@ -105,8 +102,8 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") { LnZapEvent.kind, PollNoteEvent.kind ), - tags = mapOf("e" to listOf(it.idHex)), - since = it.lastReactionsDownloadTime, + tags = mapOf("e" to it.map { it.idHex }), + since = findMinimumEOSEs(it), limit = 1000 // Max amount of "replies" to download on a specific event. ) ) @@ -207,3 +204,45 @@ object NostrSingleEventDataSource : NostrDataSource("SingleEventFeed") { } } } + +fun groupByEOSEPresence(notes: Set): Collection> { + return notes.groupBy { it.lastReactionsDownloadTime.keys.sorted().joinToString(",") }.values +} + +fun groupByEOSEPresence(users: Iterable): Collection> { + return users.groupBy { it.latestEOSEs.keys.sorted().joinToString(",") }.values +} + +fun findMinimumEOSEs(notes: List): Map { + val minLatestEOSEs = mutableMapOf() + + 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): Map { + val minLatestEOSEs = mutableMapOf() + + 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 +} diff --git a/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleUserDataSource.kt b/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleUserDataSource.kt index a4fc612d0..152350619 100644 --- a/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleUserDataSource.kt +++ b/app/src/main/java/com/vitorpamplona/amethyst/service/NostrSingleUserDataSource.kt @@ -17,99 +17,71 @@ object NostrSingleUserDataSource : NostrDataSource("SingleUserFeed") { val firstTimers = usersToWatch.filter { it.info?.latestMetadata == null }.map { it.pubkeyHex } + if (firstTimers.isEmpty()) return null + return listOf( TypedFilter( types = COMMON_FEED_TYPES, filter = JsonFilter( - kinds = listOf(MetadataEvent.kind, StatusEvent.kind), - authors = firstTimers, - limit = 10 * firstTimers.size + kinds = listOf(MetadataEvent.kind), + authors = firstTimers ) ) ) } - fun createUserMetadataFilter(minLatestEOSEs: Map): TypedFilter? { + fun createUserMetadataStatusReportFilter(): List? { if (usersToWatch.isEmpty()) return null - return TypedFilter( - types = COMMON_FEED_TYPES, - filter = JsonFilter( - kinds = listOf(StatusEvent.kind), - authors = usersToWatch.map { it.pubkeyHex }, - since = minLatestEOSEs + 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( + types = COMMON_FEED_TYPES, + filter = JsonFilter( + kinds = listOf(MetadataEvent.kind, StatusEvent.kind), + authors = groupIds, + since = minEOSEs + ) + ), + TypedFilter( + types = COMMON_FEED_TYPES, + filter = JsonFilter( + kinds = listOf(ReportEvent.kind), + tags = mapOf("p" to groupIds), + since = minEOSEs + ) + ) ) - ) - } - - fun createUserStatusFilter(minLatestEOSEs: Map): 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): TypedFilter? { - if (usersToWatch.isEmpty()) return null - - return TypedFilter( - types = COMMON_FEED_TYPES, - filter = JsonFilter( - kinds = listOf(ReportEvent.kind), - tags = mapOf("p" to usersToWatch.map { it.pubkeyHex }), - since = minLatestEOSEs - ) - ) + }.flatten() } val userChannel = requestNewChannel() { time, relayUrl -> - usersToWatch.forEach { - val eose = it.latestEOSEs[relayUrl] - if (eose == null) { - it.latestEOSEs = it.latestEOSEs + Pair(relayUrl, EOSETime(time)) - } else { - eose.time = time - } - } - } + checkNotInMainThread() - 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() { - val minLatestEOSEs = mutableMapOf() - val neverGottenAnEOSE = mutableSetOf() 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 - } + if (it.info?.latestMetadata != null) { + val eose = it.latestEOSEs[relayUrl] + if (eose == null) { + it.latestEOSEs = it.latestEOSEs + Pair(relayUrl, EOSETime(time)) + } else { + eose.time = time } } } + } + + override fun updateChannelFilters() { + checkNotInMainThread() userChannel.typedFilters = listOfNotNull( - createUserMetadataFilter(minLatestEOSEs), - createUserStatusFilter(minLatestEOSEs), - createUserReportFilter(minLatestEOSEs) - ).ifEmpty { null } - userChannelFirstTimers.typedFilters = listOfNotNull( - createUserMetadataFilter() + createUserMetadataFilter(), + createUserMetadataStatusReportFilter() ).flatten().ifEmpty { null } }