Lock it up
This commit is contained in:
Kgothatso Ngako
@@ -6,6 +6,7 @@ import ac.cord.auxiliary.compose.database.model.traits.NostrEventEntity
|
||||
import ac.cord.auxiliary.compose.database.model.traits.SoftDeletableEntity
|
||||
import ac.cord.auxiliary.compose.database.model.traits.TimestampedEntity
|
||||
import ac.cord.auxiliary.compose.database.model.traits.UserViewableEntity
|
||||
import ac.cord.auxiliary.compose.repository.NostrRepository
|
||||
import ac.cord.auxiliary.compose.ui.composable.navigation.routes.NostrEventDetailRoute
|
||||
import ac.cord.auxiliary.compose.ui.composable.navigation.routes.Route
|
||||
import ac.cord.auxiliary.compose.ui.composable.widgets.profile.ProfileAvatar
|
||||
@@ -31,6 +32,8 @@ import androidx.room3.Entity
|
||||
import androidx.room3.ForeignKey
|
||||
import androidx.room3.Index
|
||||
import androidx.room3.PrimaryKey
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip51Lists.relayLists.RelayFeedsListEvent
|
||||
import kotlin.time.Clock
|
||||
import kotlin.time.Instant
|
||||
|
||||
@@ -100,6 +103,32 @@ data class Profile(
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun profilePublicRelays(
|
||||
nostrRepository: NostrRepository
|
||||
): List<NormalizedRelayUrl>? {
|
||||
val relays = nostrRepository.getNostrEvent(
|
||||
publicKey,
|
||||
kind = RelayFeedsListEvent.KIND
|
||||
)?.let { relayFeedNostrEvent ->
|
||||
val relayFeedEvent = RelayFeedsListEvent(
|
||||
id = relayFeedNostrEvent.id,
|
||||
pubKey = relayFeedNostrEvent.pubKey,
|
||||
content = relayFeedNostrEvent.content,
|
||||
tags = relayFeedNostrEvent.tags,
|
||||
sig = relayFeedNostrEvent.sig,
|
||||
createdAt = relayFeedNostrEvent.createdAt.epochSeconds
|
||||
)
|
||||
|
||||
relayFeedEvent.publicRelays()
|
||||
}
|
||||
|
||||
if (relays?.isNotEmpty() == true) {
|
||||
return relays
|
||||
}
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
@Composable
|
||||
fun RenderAsListItem(
|
||||
onNavigateToEvent: (Route) -> Unit
|
||||
|
||||
@@ -390,17 +390,14 @@ class DatabaseNostrRepository(
|
||||
broadcastNostrEventRequest: BroadcastNostrEventRequest,
|
||||
status: String
|
||||
) {
|
||||
logger.d("Update local reference: $broadcastNostrEventRequest")
|
||||
database.broadcastNostrEventRequestDao().upsert(
|
||||
broadcastNostrEventRequest.copy(
|
||||
status = status
|
||||
)
|
||||
)
|
||||
logger.d("$status: $broadcastNostrEventRequest")
|
||||
}
|
||||
|
||||
override suspend fun synchronizeNostrEventRequestProcessed(synchronizeNostrEventRequest: SynchronizeNostrEventRequest) {
|
||||
logger.i("Update local reference: $synchronizeNostrEventRequest")
|
||||
database.synchronizeNostrEventRequestDao().upsert(
|
||||
synchronizeNostrEventRequest.copy(
|
||||
status = "sent",
|
||||
|
||||
@@ -124,15 +124,11 @@ class RelayPool(
|
||||
|
||||
suspend fun addRelaysIfMissing(relays: Set<RelayDTO>) {
|
||||
relayMutex.withLock {
|
||||
logger.d("addRelaysIfMissing: ${relays.map { it.url }}")
|
||||
val existingRelayUrls = socketClients.map { NormalizedRelayUrl(it.socketUrl).displayUrl() }
|
||||
logger.d("Existing Relays: $existingRelayUrls")
|
||||
val newRelayUrls = relays.map { it.url }
|
||||
|
||||
val toAddRelayUrls = newRelayUrls.filter { NormalizedRelayUrl(it).displayUrl() !in existingRelayUrls }.toSet()
|
||||
logger.d("toAddRelayUrls: $toAddRelayUrls")
|
||||
val toAddSocketClients = relays.filter { it.url in toAddRelayUrls }.mapAsNostrSocketClient()
|
||||
logger.d("toAddSocketClients: ${toAddSocketClients.map { it.socketUrl }}" )
|
||||
if (toAddSocketClients.isNotEmpty()) {
|
||||
val newSocketClients = socketClients.toMutableSet().apply {
|
||||
addAll(toAddSocketClients)
|
||||
@@ -242,7 +238,6 @@ class RelayPool(
|
||||
)
|
||||
)
|
||||
|
||||
logger.d("socketClients: ${socketClients.map { it.socketUrl }}")
|
||||
val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() }
|
||||
|
||||
val closeSubscription = OptimizedJsonMapper.toJson(closeCmd)
|
||||
|
||||
@@ -79,7 +79,8 @@ fun TextNoteEventDetail(
|
||||
val inReplyToNostrEventFeedListViewModel: InReplyToViewModel = viewModel(
|
||||
key = "replies-${localNostrEvent.nostrEvent.id}",
|
||||
factory = InReplyToViewModel.factory(
|
||||
nostrEventId = localNostrEvent.nostrEvent.id,
|
||||
nostrEvent = localNostrEvent.nostrEvent,
|
||||
profile = localNostrEvent.profile,
|
||||
initialFeedListUIState = FeedListUIState.Loading,
|
||||
nostrRepository = nostrRepository,
|
||||
createdAt = localNostrEvent.nostrEvent.createdAt
|
||||
|
||||
@@ -53,7 +53,7 @@ class FeedListViewModel(
|
||||
|
||||
val isSynchronizationPending: MutableState<Boolean> = mutableStateOf(false)
|
||||
|
||||
val negentropySynchronizeRequests = Relays.negentropicRelaySet.shuffled().take(1).map { normalizedRelayUrl ->
|
||||
val negentropySynchronizeRequests = Relays.negentropicRelaySet.shuffled().map { normalizedRelayUrl ->
|
||||
NegentropySynchronizeRequest(
|
||||
id = NegentropySynchronizeRequest.computeId(
|
||||
normalizedRelayUrl.url,
|
||||
|
||||
@@ -76,7 +76,7 @@ class FollowersListViewModel(
|
||||
viewModelScope.launch(Dispatchers.IO) {
|
||||
// Sync Notifications... might want to also run this in the background
|
||||
nostrRepository.queueNegentropySynchronizeRequest(
|
||||
Relays.eventPublishRelaySet.take(1).map { normalizedRelayUrl ->
|
||||
Relays.negentropicRelaySet.shuffled().map { normalizedRelayUrl ->
|
||||
NegentropySynchronizeRequest(
|
||||
id = NegentropySynchronizeRequest.computeId(
|
||||
relayURL = normalizedRelayUrl.url,
|
||||
|
||||
@@ -75,7 +75,7 @@ class FollowingListViewModel(
|
||||
viewModelScope.launch(Dispatchers.IO) {
|
||||
// Sync Notifications... might want to also run this in the background
|
||||
nostrRepository.queueNegentropySynchronizeRequest(
|
||||
Relays.eventPublishRelaySet.take(1).map { normalizedRelayUrl ->
|
||||
Relays.negentropicRelaySet.shuffled().map { normalizedRelayUrl ->
|
||||
NegentropySynchronizeRequest(
|
||||
id = NegentropySynchronizeRequest.computeId(
|
||||
relayURL = normalizedRelayUrl.url,
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package ac.cord.auxiliary.compose.ui.view.model
|
||||
|
||||
import ac.cord.auxiliary.compose.database.model.NegentropySynchronizeRequest
|
||||
import ac.cord.auxiliary.compose.database.model.NostrEvent
|
||||
import ac.cord.auxiliary.compose.database.model.NostrEventRelay
|
||||
import ac.cord.auxiliary.compose.database.model.Profile
|
||||
import ac.cord.auxiliary.compose.database.model.SynchronizeNostrEventRequest
|
||||
import ac.cord.auxiliary.compose.database.model.types.SynchronizationFilter
|
||||
import ac.cord.auxiliary.compose.nostr.Relays
|
||||
@@ -37,6 +40,7 @@ import co.touchlab.kermit.Logger
|
||||
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
|
||||
import com.vitorpamplona.quartz.nip18Reposts.RepostEvent
|
||||
import com.vitorpamplona.quartz.nip25Reactions.ReactionEvent
|
||||
import com.vitorpamplona.quartz.nip51Lists.relayLists.RelayFeedsListEvent
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.IO
|
||||
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
@@ -44,7 +48,8 @@ import kotlinx.coroutines.launch
|
||||
import kotlin.time.Instant
|
||||
|
||||
class InReplyToViewModel(
|
||||
nostrEventId: String,
|
||||
val nostrEvent: NostrEvent,
|
||||
val profile: Profile?,
|
||||
initialFeedListUIState: FeedListUIState,
|
||||
val synchronizationFilter: SynchronizationFilter,
|
||||
val nostrRepository: NostrRepository
|
||||
@@ -79,8 +84,12 @@ class InReplyToViewModel(
|
||||
logger.d("scheduleSynchronization")
|
||||
viewModelScope.launch(Dispatchers.IO) {
|
||||
// Sync Notifications... might want to also run this in the background
|
||||
val publicRelays = profile?.profilePublicRelays(
|
||||
nostrRepository
|
||||
) ?: Relays.negentropicRelaySet
|
||||
|
||||
nostrRepository.queueNegentropySynchronizeRequest(
|
||||
Relays.eventPublishRelaySet.take(1).map { normalizedRelayUrl ->
|
||||
publicRelays.take(3).map { normalizedRelayUrl ->
|
||||
NegentropySynchronizeRequest(
|
||||
id = NegentropySynchronizeRequest.computeId(
|
||||
relayURL = normalizedRelayUrl.url,
|
||||
@@ -193,7 +202,8 @@ class InReplyToViewModel(
|
||||
const val TAG = "InReplyToViewModel"
|
||||
|
||||
fun factory(
|
||||
nostrEventId: String,
|
||||
nostrEvent: NostrEvent,
|
||||
profile: Profile?,
|
||||
createdAt: Instant,
|
||||
initialFeedListUIState: FeedListUIState = FeedListUIState.Loading,
|
||||
synchronizationFilter: SynchronizationFilter = SynchronizationFilter(
|
||||
@@ -204,7 +214,7 @@ class InReplyToViewModel(
|
||||
),
|
||||
tags = mapOf(
|
||||
"e" to listOf(
|
||||
nostrEventId
|
||||
nostrEvent.id
|
||||
)
|
||||
),
|
||||
since = createdAt
|
||||
@@ -214,7 +224,8 @@ class InReplyToViewModel(
|
||||
initializer {
|
||||
InReplyToViewModel(
|
||||
initialFeedListUIState = initialFeedListUIState,
|
||||
nostrEventId = nostrEventId,
|
||||
nostrEvent = nostrEvent,
|
||||
profile = profile,
|
||||
synchronizationFilter = synchronizationFilter,
|
||||
nostrRepository = nostrRepository
|
||||
)
|
||||
|
||||
@@ -43,6 +43,8 @@ import kotlinx.coroutines.flow.distinctUntilChanged
|
||||
import kotlinx.coroutines.flow.getAndUpdate
|
||||
import kotlinx.coroutines.flow.timeout
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
import kotlin.time.Instant
|
||||
|
||||
@@ -62,6 +64,8 @@ class NavigationViewModel(
|
||||
companion object {
|
||||
private const val TAG = "NavigationViewModel"
|
||||
|
||||
private val mutex = Mutex()
|
||||
|
||||
fun factory(
|
||||
initialNavigationUIState: NavigationUIState,
|
||||
nostrRepository: NostrRepository,
|
||||
@@ -221,168 +225,178 @@ class NavigationViewModel(
|
||||
scope.launch(Dispatchers.IO) {
|
||||
nostrRepository.observePendingNegentropySynchronizeRequests().distinctUntilChanged().collect { negentropySynchronizeRequestOrNull ->
|
||||
negentropySynchronizeRequestOrNull?.let { negentropySynchronizeRequest ->
|
||||
logger.i("negentropySynchronizeRequest: $negentropySynchronizeRequest")
|
||||
mutex.withLock {
|
||||
logger.i("negentropySynchronizeRequest: $negentropySynchronizeRequest")
|
||||
|
||||
val events = nostrRepository.getNostrFeedIds(
|
||||
arrayOf(
|
||||
negentropySynchronizeRequest.synchronizationFilter
|
||||
),
|
||||
applyLimits = false
|
||||
)
|
||||
logger.d("Events: ${events.size}")
|
||||
val storage = StorageVector().apply {
|
||||
events.forEach { event ->
|
||||
insert(
|
||||
event.createdAt.toEpochMilliseconds(),
|
||||
event.id
|
||||
)
|
||||
val events = nostrRepository.getNostrFeedIds(
|
||||
arrayOf(
|
||||
negentropySynchronizeRequest.synchronizationFilter
|
||||
),
|
||||
applyLimits = false
|
||||
)
|
||||
logger.d("Events: ${events.size}")
|
||||
val storage = StorageVector().apply {
|
||||
events.forEach { event ->
|
||||
insert(
|
||||
event.createdAt.toEpochMilliseconds(),
|
||||
event.id
|
||||
)
|
||||
}
|
||||
|
||||
seal()
|
||||
}
|
||||
val negentropy = Negentropy(
|
||||
storage,
|
||||
)
|
||||
|
||||
seal()
|
||||
}
|
||||
val negentropy = Negentropy(
|
||||
storage,
|
||||
)
|
||||
val negOpenCmd = NegOpenCmd(
|
||||
subId = negentropySynchronizeRequest.uuid,
|
||||
filter = Filter(
|
||||
ids = negentropySynchronizeRequest.synchronizationFilter.ids?.toList(),
|
||||
authors = negentropySynchronizeRequest.synchronizationFilter.authors?.toList(),
|
||||
kinds = negentropySynchronizeRequest.synchronizationFilter.kinds?.toList(),
|
||||
tags = negentropySynchronizeRequest.synchronizationFilter.tags,
|
||||
tagsAll = negentropySynchronizeRequest.synchronizationFilter.tagsAll,
|
||||
since = negentropySynchronizeRequest.synchronizationFilter.since?.epochSeconds,
|
||||
until = negentropySynchronizeRequest.synchronizationFilter.until?.epochSeconds,
|
||||
limit = null, // We don't do limits when performing negentropy...
|
||||
search = negentropySynchronizeRequest.synchronizationFilter.search
|
||||
),
|
||||
initialMessage = negentropy.initiate().toHexString()
|
||||
)
|
||||
logger.d("negOpenCmd: ${negOpenCmd.initialMessage}")
|
||||
|
||||
val negOpenCmd = NegOpenCmd(
|
||||
subId = negentropySynchronizeRequest.uuid,
|
||||
filter = Filter(
|
||||
ids = negentropySynchronizeRequest.synchronizationFilter.ids?.toList(),
|
||||
authors = negentropySynchronizeRequest.synchronizationFilter.authors?.toList(),
|
||||
kinds = negentropySynchronizeRequest.synchronizationFilter.kinds?.toList(),
|
||||
tags = negentropySynchronizeRequest.synchronizationFilter.tags,
|
||||
tagsAll = negentropySynchronizeRequest.synchronizationFilter.tagsAll,
|
||||
since = negentropySynchronizeRequest.synchronizationFilter.since?.epochSeconds,
|
||||
until = negentropySynchronizeRequest.synchronizationFilter.until?.epochSeconds,
|
||||
limit = null, // We don't do limits when performing negentropy...
|
||||
search = negentropySynchronizeRequest.synchronizationFilter.search
|
||||
),
|
||||
initialMessage = negentropy.initiate().toHexString()
|
||||
)
|
||||
logger.d("negOpenCmd: ${negOpenCmd.initialMessage}")
|
||||
nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest)
|
||||
|
||||
nostrRepository.negentropySynchronizeRequestProcessed(negentropySynchronizeRequest)
|
||||
scope.launch(Dispatchers.IO) {
|
||||
try {
|
||||
val negCloseCmd = NegCloseCmd(
|
||||
subId = negentropySynchronizeRequest.uuid,
|
||||
)
|
||||
|
||||
scope.launch(Dispatchers.IO) {
|
||||
try {
|
||||
val negCloseCmd = NegCloseCmd(
|
||||
subId = negentropySynchronizeRequest.uuid,
|
||||
)
|
||||
relaysSocketManager.negentropySync(
|
||||
negOpenCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
).collect { nostrIncomingMessage ->
|
||||
when (nostrIncomingMessage) {
|
||||
is NostrIncomingMessage.EventMessage -> {
|
||||
scope.launch(Dispatchers.IO) {
|
||||
logger.d("Import message: $nostrIncomingMessage")
|
||||
nostrIncomingMessage.nostrEvent?.let {
|
||||
nostrRepository.saveNostrEvent(
|
||||
nostrEvent = it,
|
||||
negentropySynchronizeRequest,
|
||||
synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
is NostrIncomingMessage.EventsMessage -> {
|
||||
logger.d("Import messages: $nostrIncomingMessage")
|
||||
|
||||
relaysSocketManager.negentropySync(
|
||||
negOpenCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
).collect { nostrIncomingMessage ->
|
||||
when (nostrIncomingMessage) {
|
||||
is NostrIncomingMessage.EventMessage -> {
|
||||
scope.launch(Dispatchers.IO) {
|
||||
logger.d("Import message: $nostrIncomingMessage")
|
||||
nostrIncomingMessage.nostrEvent?.let {
|
||||
nostrIncomingMessage.nostrEvents.forEach { nostrEvent ->
|
||||
nostrRepository.saveNostrEvent(
|
||||
nostrEvent = it,
|
||||
nostrEvent = nostrEvent,
|
||||
negentropySynchronizeRequest,
|
||||
synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
is NostrIncomingMessage.EventsMessage -> {
|
||||
logger.d("Import messages: $nostrIncomingMessage")
|
||||
is NostrIncomingMessage.EoseMessage -> {
|
||||
logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage")
|
||||
|
||||
nostrIncomingMessage.nostrEvents.forEach { nostrEvent ->
|
||||
nostrRepository.saveNostrEvent(
|
||||
nostrEvent = nostrEvent,
|
||||
negentropySynchronizeRequest,
|
||||
synchronizationRelayURLs = listOf(negentropySynchronizeRequest.relayURL) // TODO: + Relays.eventPublishRelaySet.map { normalizedRelayUrl -> normalizedRelayUrl.url }
|
||||
relaysSocketManager.closeNegentropySync(
|
||||
negCloseCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
|
||||
return@collect
|
||||
}
|
||||
}
|
||||
is NostrIncomingMessage.EoseMessage -> {
|
||||
logger.d("Sync request has been successfully processed (${negentropySynchronizeRequest.relayURL}): $nostrIncomingMessage")
|
||||
is NostrIncomingMessage.NegentropyError -> {
|
||||
logger.d("Negentropy Error (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}")
|
||||
|
||||
relaysSocketManager.closeNegentropySync(
|
||||
negCloseCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
}
|
||||
is NostrIncomingMessage.NegentropyError -> {
|
||||
logger.d("Negentropy Error (need to synchronize like normal): ${nostrIncomingMessage.negentropyReason}")
|
||||
// TODO: Don't schedule a sync when on mobile internet.
|
||||
val since = negentropySynchronizeRequest.synchronizationFilter.since // TODO: Update since -> until based on what we have in the local db...
|
||||
val until = negentropySynchronizeRequest.synchronizationFilter.until
|
||||
|
||||
// TODO: Don't schedule a sync when on mobile internet.
|
||||
val since = negentropySynchronizeRequest.synchronizationFilter.since // TODO: Update since -> until based on what we have in the local db...
|
||||
val until = negentropySynchronizeRequest.synchronizationFilter.until
|
||||
|
||||
val synchronizationFilter = negentropySynchronizeRequest.synchronizationFilter.copy(
|
||||
since = since,
|
||||
until = until
|
||||
)
|
||||
|
||||
nostrRepository.queueSynchronizeNostrEvent(
|
||||
listOf(
|
||||
SynchronizeNostrEventRequest(
|
||||
purpose = negentropySynchronizeRequest.purpose,
|
||||
synchronizationFilters = arrayOf(
|
||||
synchronizationFilter
|
||||
),
|
||||
relayURL = negentropySynchronizeRequest.relayURL,
|
||||
level = negentropySynchronizeRequest.level,
|
||||
)
|
||||
val synchronizationFilter = negentropySynchronizeRequest.synchronizationFilter.copy(
|
||||
since = since,
|
||||
until = until
|
||||
)
|
||||
)
|
||||
}
|
||||
is NostrIncomingMessage.NegentropyMessage -> {
|
||||
logger.d("NegentropyMessage: ${nostrIncomingMessage.negentropyMessage}")
|
||||
|
||||
val result = negentropy.reconcile(
|
||||
nostrIncomingMessage.negentropyMessage.hexToByteArray()
|
||||
)
|
||||
logger.d("NeedIds: ${result.needIds.map { it.toHexString() }}")
|
||||
logger.d("SendIds: ${result.sendIds.map { it.toHexString() }}")
|
||||
logger.d("EventsIds: ${events.map { it.id }}")
|
||||
logger.d("Timestamp: ${events.map { it.createdAt.epochSeconds }}")
|
||||
|
||||
if (result.needIds.isNotEmpty()) {
|
||||
// Schedule a sync from this relay...
|
||||
nostrRepository.queueSynchronizeNostrEvent(
|
||||
listOf(
|
||||
SynchronizeNostrEventRequest(
|
||||
purpose = negentropySynchronizeRequest.purpose,
|
||||
synchronizationFilters = arrayOf(
|
||||
SynchronizationFilter(
|
||||
ids = result.needIds.map { it.toHexString() }.toTypedArray()
|
||||
)
|
||||
synchronizationFilter
|
||||
),
|
||||
relayURL = negentropySynchronizeRequest.relayURL,
|
||||
level = negentropySynchronizeRequest.level,
|
||||
)
|
||||
)
|
||||
)
|
||||
return@collect
|
||||
}
|
||||
val broadcastNostrEventRequests = result.sendIds.map { nostrEventId ->
|
||||
BroadcastNostrEventRequest(
|
||||
nostrEventId = nostrEventId.toHexString(),
|
||||
relayURL = negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
}
|
||||
nostrRepository.scheduleBroadcastNostrEventRequests(
|
||||
broadcastNostrEventRequests
|
||||
)
|
||||
is NostrIncomingMessage.NegentropyMessage -> {
|
||||
logger.d("NegentropyMessage: ${nostrIncomingMessage.negentropyMessage}")
|
||||
|
||||
relaysSocketManager.closeNegentropySync(
|
||||
negCloseCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
}
|
||||
else -> {
|
||||
logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage")
|
||||
val result = negentropy.reconcile(
|
||||
nostrIncomingMessage.negentropyMessage.hexToByteArray()
|
||||
)
|
||||
logger.d("NeedIds: ${result.needIds.map { it.toHexString() }}")
|
||||
logger.d("SendIds: ${result.sendIds.map { it.toHexString() }}")
|
||||
logger.d("EventsIds: ${events.map { it.id }}")
|
||||
logger.d("Timestamp: ${events.map { it.createdAt.epochSeconds }}")
|
||||
|
||||
if (result.needIds.isNotEmpty()) {
|
||||
// Schedule a sync from this relay...
|
||||
nostrRepository.queueSynchronizeNostrEvent(
|
||||
listOf(
|
||||
SynchronizeNostrEventRequest(
|
||||
purpose = negentropySynchronizeRequest.purpose,
|
||||
synchronizationFilters = arrayOf(
|
||||
SynchronizationFilter(
|
||||
ids = result.needIds.map { it.toHexString() }.toTypedArray()
|
||||
)
|
||||
),
|
||||
relayURL = negentropySynchronizeRequest.relayURL,
|
||||
level = negentropySynchronizeRequest.level,
|
||||
)
|
||||
)
|
||||
)
|
||||
}
|
||||
val broadcastNostrEventRequests = result.sendIds.map { nostrEventId ->
|
||||
BroadcastNostrEventRequest(
|
||||
nostrEventId = nostrEventId.toHexString(),
|
||||
relayURL = negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
}
|
||||
nostrRepository.scheduleBroadcastNostrEventRequests(
|
||||
broadcastNostrEventRequests
|
||||
)
|
||||
|
||||
relaysSocketManager.closeNegentropySync(
|
||||
negCloseCmd,
|
||||
negentropySynchronizeRequest.relayURL
|
||||
)
|
||||
|
||||
return@collect
|
||||
}
|
||||
else -> {
|
||||
logger.d("Unhandled message ${negentropySynchronizeRequest.relayURL}: $nostrIncomingMessage")
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.d("Queried Sync")
|
||||
} catch (e: Throwable) {
|
||||
logger.e("Failed to sync", e)
|
||||
}
|
||||
|
||||
} catch (e: Throwable) {
|
||||
logger.e("Failed to sync", e)
|
||||
}
|
||||
|
||||
logger.i("After launch")
|
||||
}
|
||||
logger.i("Lock released")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user