feat: live relay activity UI for event sync screen
ViewModel now emits a LiveSyncActivity StateFlow alongside the existing
SyncState, carrying:
- activeRelays: which relays are currently being queried
- recentCompletions: last 100 completed relays with event counts
- outboxTargets / inboxTargets / dmTargets: destination relay sets
downloadFromPool gains onRelayStart/onRelayComplete(relay, eventsFound)
callbacks; downloadFromRelay returns Int (total events across all pages).
Tracking uses ConcurrentHashMap.newKeySet + synchronized ArrayDeque for
thread safety across 50 concurrent workers.
Screen adds three new live cards:
ActiveRelaysCard — FlowRow of chips with a shared pulsing dot
animation (one InfiniteTransition for all chips)
DestinationRelaysCard — Outbox / Inbox / DMs sections, color-coded
using primary / secondary / tertiary
ActivityLogCard — fixed-height (260dp) inner-scrollable list;
filled dot + count for active relays, muted for
empty/unreachable ones; K/M formatting for counts
String resources updated: event_sync_batch_of → event_sync_relays_progress,
paused_body copy updated to relay language; 7 new strings added.
https://claude.ai/code/session_01U8qF9mK4UBvXsXP1zNMXfX
This commit is contained in:
+362
-23
@@ -21,25 +21,40 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.ui.screen.loggedIn.relays.eventsync
|
||||
|
||||
import androidx.compose.animation.core.RepeatMode
|
||||
import androidx.compose.animation.core.animateFloat
|
||||
import androidx.compose.animation.core.infiniteRepeatable
|
||||
import androidx.compose.animation.core.rememberInfiniteTransition
|
||||
import androidx.compose.animation.core.tween
|
||||
import androidx.compose.foundation.background
|
||||
import androidx.compose.foundation.layout.Arrangement
|
||||
import androidx.compose.foundation.layout.Box
|
||||
import androidx.compose.foundation.layout.Column
|
||||
import androidx.compose.foundation.layout.ExperimentalLayoutApi
|
||||
import androidx.compose.foundation.layout.FlowRow
|
||||
import androidx.compose.foundation.layout.Row
|
||||
import androidx.compose.foundation.layout.Spacer
|
||||
import androidx.compose.foundation.layout.fillMaxSize
|
||||
import androidx.compose.foundation.layout.fillMaxWidth
|
||||
import androidx.compose.foundation.layout.height
|
||||
import androidx.compose.foundation.layout.padding
|
||||
import androidx.compose.foundation.layout.size
|
||||
import androidx.compose.foundation.layout.width
|
||||
import androidx.compose.foundation.rememberScrollState
|
||||
import androidx.compose.foundation.shape.CircleShape
|
||||
import androidx.compose.foundation.verticalScroll
|
||||
import androidx.compose.material3.AlertDialog
|
||||
import androidx.compose.material3.Button
|
||||
import androidx.compose.material3.ButtonDefaults
|
||||
import androidx.compose.material3.Card
|
||||
import androidx.compose.material3.CardDefaults
|
||||
import androidx.compose.material3.HorizontalDivider
|
||||
import androidx.compose.material3.LinearProgressIndicator
|
||||
import androidx.compose.material3.MaterialTheme
|
||||
import androidx.compose.material3.OutlinedButton
|
||||
import androidx.compose.material3.Scaffold
|
||||
import androidx.compose.material3.SuggestionChip
|
||||
import androidx.compose.material3.SuggestionChipDefaults
|
||||
import androidx.compose.material3.Text
|
||||
import androidx.compose.material3.TextButton
|
||||
import androidx.compose.runtime.Composable
|
||||
@@ -49,7 +64,10 @@ import androidx.compose.runtime.remember
|
||||
import androidx.compose.runtime.setValue
|
||||
import androidx.compose.ui.Alignment
|
||||
import androidx.compose.ui.Modifier
|
||||
import androidx.compose.ui.draw.alpha
|
||||
import androidx.compose.ui.draw.clip
|
||||
import androidx.compose.ui.text.font.FontWeight
|
||||
import androidx.compose.ui.text.style.TextOverflow
|
||||
import androidx.compose.ui.unit.dp
|
||||
import androidx.lifecycle.compose.collectAsStateWithLifecycle
|
||||
import com.vitorpamplona.amethyst.R
|
||||
@@ -57,6 +75,7 @@ import com.vitorpamplona.amethyst.ui.navigation.navs.INav
|
||||
import com.vitorpamplona.amethyst.ui.navigation.topbars.TopBarWithBackButton
|
||||
import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel
|
||||
import com.vitorpamplona.amethyst.ui.stringRes
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
|
||||
@Composable
|
||||
fun EventSyncScreen(
|
||||
@@ -66,6 +85,7 @@ fun EventSyncScreen(
|
||||
val syncViewModel = accountViewModel.eventSyncViewModel
|
||||
|
||||
val syncState by syncViewModel.syncState.collectAsStateWithLifecycle()
|
||||
val liveActivity by syncViewModel.liveActivity.collectAsStateWithLifecycle()
|
||||
val isMobileOrMetered by accountViewModel.settings.isMobileOrMeteredConnection.collectAsStateWithLifecycle()
|
||||
var showMobileDataDialog by remember { mutableStateOf(false) }
|
||||
|
||||
@@ -152,6 +172,27 @@ fun EventSyncScreen(
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Live relay activity (shown during and after sync) ----
|
||||
val isRunning = syncState is EventSyncViewModel.SyncState.Running
|
||||
|
||||
if (liveActivity.activeRelays.isNotEmpty()) {
|
||||
ActiveRelaysCard(
|
||||
activeRelays = liveActivity.activeRelays,
|
||||
isRunning = isRunning,
|
||||
)
|
||||
}
|
||||
|
||||
if (liveActivity.outboxTargets.isNotEmpty() ||
|
||||
liveActivity.inboxTargets.isNotEmpty() ||
|
||||
liveActivity.dmTargets.isNotEmpty()
|
||||
) {
|
||||
DestinationRelaysCard(activity = liveActivity)
|
||||
}
|
||||
|
||||
if (liveActivity.recentCompletions.isNotEmpty()) {
|
||||
ActivityLogCard(completions = liveActivity.recentCompletions)
|
||||
}
|
||||
|
||||
// ---- Action buttons ----
|
||||
when (val state = syncState) {
|
||||
is EventSyncViewModel.SyncState.Idle,
|
||||
@@ -237,27 +278,9 @@ fun EventSyncScreen(
|
||||
}
|
||||
}
|
||||
|
||||
@Composable
|
||||
private fun StepRow(
|
||||
number: String,
|
||||
text: String,
|
||||
) {
|
||||
Row(
|
||||
horizontalArrangement = Arrangement.spacedBy(8.dp),
|
||||
verticalAlignment = Alignment.Top,
|
||||
) {
|
||||
Text(
|
||||
text = "$number.",
|
||||
style = MaterialTheme.typography.bodyMedium,
|
||||
fontWeight = FontWeight.Bold,
|
||||
color = MaterialTheme.colorScheme.primary,
|
||||
)
|
||||
Text(
|
||||
text = text,
|
||||
style = MaterialTheme.typography.bodyMedium,
|
||||
)
|
||||
}
|
||||
}
|
||||
// -------------------------------------------------------------------------
|
||||
// Progress / status cards
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Composable
|
||||
private fun SyncProgressCard(state: EventSyncViewModel.SyncState.Running) {
|
||||
@@ -268,13 +291,19 @@ private fun SyncProgressCard(state: EventSyncViewModel.SyncState.Running) {
|
||||
) {
|
||||
Column(modifier = Modifier.padding(16.dp)) {
|
||||
Text(
|
||||
text = stringRes(R.string.event_sync_batch_of, state.relaysCompleted, state.totalRelays),
|
||||
text = stringRes(R.string.event_sync_relays_progress, state.relaysCompleted, state.totalRelays),
|
||||
style = MaterialTheme.typography.titleSmall,
|
||||
fontWeight = FontWeight.SemiBold,
|
||||
)
|
||||
Spacer(Modifier.height(8.dp))
|
||||
LinearProgressIndicator(
|
||||
progress = { state.relaysCompleted.toFloat() / state.totalRelays },
|
||||
progress = {
|
||||
if (state.totalRelays > 0) {
|
||||
state.relaysCompleted.toFloat() / state.totalRelays
|
||||
} else {
|
||||
0f
|
||||
}
|
||||
},
|
||||
modifier = Modifier.fillMaxWidth(),
|
||||
)
|
||||
Spacer(Modifier.height(8.dp))
|
||||
@@ -375,3 +404,313 @@ private fun ErrorCard(message: String) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Live activity cards
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Shows all relays currently being queried, each with a pulsing dot indicator.
|
||||
*/
|
||||
@OptIn(ExperimentalLayoutApi::class)
|
||||
@Composable
|
||||
private fun ActiveRelaysCard(
|
||||
activeRelays: Set<NormalizedRelayUrl>,
|
||||
isRunning: Boolean,
|
||||
) {
|
||||
val infiniteTransition = rememberInfiniteTransition(label = "pulse")
|
||||
val pulseAlpha by infiniteTransition.animateFloat(
|
||||
initialValue = 1f,
|
||||
targetValue = 0.2f,
|
||||
animationSpec =
|
||||
infiniteRepeatable(
|
||||
animation = tween(durationMillis = 800),
|
||||
repeatMode = RepeatMode.Reverse,
|
||||
),
|
||||
label = "pulseAlpha",
|
||||
)
|
||||
|
||||
Card(
|
||||
modifier = Modifier.fillMaxWidth(),
|
||||
colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surface),
|
||||
elevation = CardDefaults.cardElevation(defaultElevation = 1.dp),
|
||||
) {
|
||||
Column(modifier = Modifier.padding(16.dp)) {
|
||||
Text(
|
||||
text = stringRes(R.string.event_sync_reading_from, activeRelays.size),
|
||||
style = MaterialTheme.typography.titleSmall,
|
||||
fontWeight = FontWeight.SemiBold,
|
||||
)
|
||||
Spacer(Modifier.height(10.dp))
|
||||
FlowRow(
|
||||
horizontalArrangement = Arrangement.spacedBy(8.dp),
|
||||
verticalArrangement = Arrangement.spacedBy(6.dp),
|
||||
) {
|
||||
activeRelays.forEach { relay ->
|
||||
ActiveRelayChip(
|
||||
relay = relay,
|
||||
pulseAlpha = if (isRunning) pulseAlpha else 1f,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Composable
|
||||
private fun ActiveRelayChip(
|
||||
relay: NormalizedRelayUrl,
|
||||
pulseAlpha: Float,
|
||||
) {
|
||||
SuggestionChip(
|
||||
onClick = {},
|
||||
label = {
|
||||
Text(
|
||||
text = relay.displayHost(),
|
||||
style = MaterialTheme.typography.labelSmall,
|
||||
maxLines = 1,
|
||||
overflow = TextOverflow.Ellipsis,
|
||||
)
|
||||
},
|
||||
icon = {
|
||||
Box(
|
||||
modifier =
|
||||
Modifier
|
||||
.size(8.dp)
|
||||
.alpha(pulseAlpha)
|
||||
.clip(CircleShape)
|
||||
.background(MaterialTheme.colorScheme.primary),
|
||||
)
|
||||
},
|
||||
colors =
|
||||
SuggestionChipDefaults.suggestionChipColors(
|
||||
containerColor = MaterialTheme.colorScheme.primaryContainer.copy(alpha = 0.4f),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Shows where events are being sent: outbox, inbox, and DM relay lists.
|
||||
*/
|
||||
@Composable
|
||||
private fun DestinationRelaysCard(activity: EventSyncViewModel.LiveSyncActivity) {
|
||||
Card(
|
||||
modifier = Modifier.fillMaxWidth(),
|
||||
colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surface),
|
||||
elevation = CardDefaults.cardElevation(defaultElevation = 1.dp),
|
||||
) {
|
||||
Column(modifier = Modifier.padding(16.dp)) {
|
||||
Text(
|
||||
text = stringRes(R.string.event_sync_sending_to),
|
||||
style = MaterialTheme.typography.titleSmall,
|
||||
fontWeight = FontWeight.SemiBold,
|
||||
)
|
||||
|
||||
if (activity.outboxTargets.isNotEmpty()) {
|
||||
Spacer(Modifier.height(10.dp))
|
||||
DestinationSection(
|
||||
label = stringRes(R.string.event_sync_outbox_relays),
|
||||
relays = activity.outboxTargets,
|
||||
color = MaterialTheme.colorScheme.primary,
|
||||
)
|
||||
}
|
||||
|
||||
if (activity.inboxTargets.isNotEmpty()) {
|
||||
Spacer(Modifier.height(10.dp))
|
||||
HorizontalDivider(color = MaterialTheme.colorScheme.outlineVariant)
|
||||
Spacer(Modifier.height(10.dp))
|
||||
DestinationSection(
|
||||
label = stringRes(R.string.event_sync_inbox_relays),
|
||||
relays = activity.inboxTargets,
|
||||
color = MaterialTheme.colorScheme.secondary,
|
||||
)
|
||||
}
|
||||
|
||||
if (activity.dmTargets.isNotEmpty()) {
|
||||
Spacer(Modifier.height(10.dp))
|
||||
HorizontalDivider(color = MaterialTheme.colorScheme.outlineVariant)
|
||||
Spacer(Modifier.height(10.dp))
|
||||
DestinationSection(
|
||||
label = stringRes(R.string.event_sync_dm_relays),
|
||||
relays = activity.dmTargets,
|
||||
color = MaterialTheme.colorScheme.tertiary,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalLayoutApi::class)
|
||||
@Composable
|
||||
private fun DestinationSection(
|
||||
label: String,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
color: androidx.compose.ui.graphics.Color,
|
||||
) {
|
||||
Text(
|
||||
text = label,
|
||||
style = MaterialTheme.typography.labelMedium,
|
||||
fontWeight = FontWeight.Medium,
|
||||
color = MaterialTheme.colorScheme.onSurfaceVariant,
|
||||
)
|
||||
Spacer(Modifier.height(6.dp))
|
||||
FlowRow(
|
||||
horizontalArrangement = Arrangement.spacedBy(8.dp),
|
||||
verticalArrangement = Arrangement.spacedBy(6.dp),
|
||||
) {
|
||||
relays.forEach { relay ->
|
||||
SuggestionChip(
|
||||
onClick = {},
|
||||
label = {
|
||||
Text(
|
||||
text = relay.displayHost(),
|
||||
style = MaterialTheme.typography.labelSmall,
|
||||
maxLines = 1,
|
||||
overflow = TextOverflow.Ellipsis,
|
||||
)
|
||||
},
|
||||
colors =
|
||||
SuggestionChipDefaults.suggestionChipColors(
|
||||
containerColor = color.copy(alpha = 0.12f),
|
||||
labelColor = color,
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Scrollable log of recently completed relays, newest at the top.
|
||||
* Uses a fixed-height inner scroll area so it doesn't compete with the outer scroll.
|
||||
*/
|
||||
@Composable
|
||||
private fun ActivityLogCard(
|
||||
completions: List<EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo>,
|
||||
) {
|
||||
Card(
|
||||
modifier = Modifier.fillMaxWidth(),
|
||||
colors = CardDefaults.cardColors(containerColor = MaterialTheme.colorScheme.surface),
|
||||
elevation = CardDefaults.cardElevation(defaultElevation = 1.dp),
|
||||
) {
|
||||
Column(modifier = Modifier.padding(16.dp)) {
|
||||
Text(
|
||||
text = stringRes(R.string.event_sync_activity_log, completions.size),
|
||||
style = MaterialTheme.typography.titleSmall,
|
||||
fontWeight = FontWeight.SemiBold,
|
||||
)
|
||||
Spacer(Modifier.height(10.dp))
|
||||
|
||||
Box(
|
||||
modifier =
|
||||
Modifier
|
||||
.fillMaxWidth()
|
||||
.height(260.dp)
|
||||
.verticalScroll(rememberScrollState()),
|
||||
) {
|
||||
Column(verticalArrangement = Arrangement.spacedBy(2.dp)) {
|
||||
completions.forEachIndexed { index, info ->
|
||||
if (index > 0) {
|
||||
HorizontalDivider(
|
||||
color = MaterialTheme.colorScheme.outlineVariant.copy(alpha = 0.5f),
|
||||
)
|
||||
}
|
||||
ActivityLogRow(info = info)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Composable
|
||||
private fun ActivityLogRow(info: EventSyncViewModel.LiveSyncActivity.CompletedRelayInfo) {
|
||||
val hasEvents = info.eventsFound > 0
|
||||
val dotColor =
|
||||
if (hasEvents) {
|
||||
MaterialTheme.colorScheme.primary
|
||||
} else {
|
||||
MaterialTheme.colorScheme.onSurfaceVariant.copy(alpha = 0.4f)
|
||||
}
|
||||
val textColor =
|
||||
if (hasEvents) {
|
||||
MaterialTheme.colorScheme.onSurface
|
||||
} else {
|
||||
MaterialTheme.colorScheme.onSurfaceVariant
|
||||
}
|
||||
|
||||
Row(
|
||||
modifier =
|
||||
Modifier
|
||||
.fillMaxWidth()
|
||||
.padding(vertical = 6.dp),
|
||||
verticalAlignment = Alignment.CenterVertically,
|
||||
horizontalArrangement = Arrangement.spacedBy(10.dp),
|
||||
) {
|
||||
Box(
|
||||
modifier =
|
||||
Modifier
|
||||
.size(8.dp)
|
||||
.clip(CircleShape)
|
||||
.background(dotColor),
|
||||
)
|
||||
Text(
|
||||
text = info.relay.displayHost(),
|
||||
style = MaterialTheme.typography.bodySmall,
|
||||
color = textColor,
|
||||
maxLines = 1,
|
||||
overflow = TextOverflow.Ellipsis,
|
||||
modifier = Modifier.weight(1f),
|
||||
)
|
||||
Text(
|
||||
text =
|
||||
if (hasEvents) {
|
||||
stringRes(R.string.event_sync_events_found_log, formatCount(info.eventsFound))
|
||||
} else {
|
||||
stringRes(R.string.event_sync_no_events)
|
||||
},
|
||||
style = MaterialTheme.typography.bodySmall,
|
||||
color = textColor,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Composable
|
||||
private fun StepRow(
|
||||
number: String,
|
||||
text: String,
|
||||
) {
|
||||
Row(
|
||||
horizontalArrangement = Arrangement.spacedBy(8.dp),
|
||||
verticalAlignment = Alignment.Top,
|
||||
) {
|
||||
Text(
|
||||
text = "$number.",
|
||||
style = MaterialTheme.typography.bodyMedium,
|
||||
fontWeight = FontWeight.Bold,
|
||||
color = MaterialTheme.colorScheme.primary,
|
||||
)
|
||||
Text(
|
||||
text = text,
|
||||
style = MaterialTheme.typography.bodyMedium,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/** Strips the WebSocket scheme and trailing slash for compact display. */
|
||||
private fun NormalizedRelayUrl.displayHost(): String =
|
||||
url
|
||||
.removePrefix("wss://")
|
||||
.removePrefix("ws://")
|
||||
.trimEnd('/')
|
||||
|
||||
/** Formats a count with K/M suffix for large numbers. */
|
||||
private fun formatCount(n: Int): String =
|
||||
when {
|
||||
n >= 1_000_000 -> "${n / 1_000_000}M"
|
||||
n >= 1_000 -> "${n / 1_000}K"
|
||||
else -> n.toString()
|
||||
}
|
||||
|
||||
+106
-8
@@ -27,14 +27,13 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.IRequestListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.newSubId
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.ensureActive
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.supervisorScope
|
||||
@@ -57,6 +56,9 @@ import java.util.concurrent.atomic.AtomicLong
|
||||
* Each relay is paginated individually: after EOSE the oldest [Event.createdAt] seen on that
|
||||
* relay becomes the next `until` cursor, repeating until the relay returns no new events.
|
||||
*
|
||||
* Live activity is emitted via [liveActivity] so the UI can show which relays are currently
|
||||
* being read, where events are being sent, and a running log of completed relays.
|
||||
*
|
||||
* The sync is pausable: calling [cancel] transitions to [SyncState.Paused] so the user can
|
||||
* [resume] from the last completed relay index rather than starting over.
|
||||
*
|
||||
@@ -72,8 +74,15 @@ class EventSyncViewModel(
|
||||
|
||||
/** How long (ms) to wait for a single relay to reply per page before giving up. */
|
||||
const val RELAY_TIMEOUT_MS = 30_000L
|
||||
|
||||
/** Maximum number of completed-relay entries kept in the activity log. */
|
||||
const val MAX_ACTIVITY_LOG = 100
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Public state
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
sealed class SyncState {
|
||||
object Idle : SyncState()
|
||||
|
||||
@@ -100,13 +109,71 @@ class EventSyncViewModel(
|
||||
) : SyncState()
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-relay activity snapshot emitted continuously while the sync runs.
|
||||
*
|
||||
* @param activeRelays Relays currently being queried.
|
||||
* @param recentCompletions Last [MAX_ACTIVITY_LOG] relays that finished, newest first.
|
||||
* @param outboxTargets Relays receiving events authored by the user.
|
||||
* @param inboxTargets Relays receiving events that mention the user.
|
||||
* @param dmTargets Relays receiving DMs addressed to the user.
|
||||
*/
|
||||
data class LiveSyncActivity(
|
||||
val activeRelays: Set<NormalizedRelayUrl> = emptySet(),
|
||||
val recentCompletions: List<CompletedRelayInfo> = emptyList(),
|
||||
val outboxTargets: Set<NormalizedRelayUrl> = emptySet(),
|
||||
val inboxTargets: Set<NormalizedRelayUrl> = emptySet(),
|
||||
val dmTargets: Set<NormalizedRelayUrl> = emptySet(),
|
||||
) {
|
||||
data class CompletedRelayInfo(
|
||||
val relay: NormalizedRelayUrl,
|
||||
/** Total events received across all pages. 0 = relay was empty or unreachable. */
|
||||
val eventsFound: Int,
|
||||
)
|
||||
}
|
||||
|
||||
private val _syncState = MutableStateFlow<SyncState>(SyncState.Idle)
|
||||
val syncState: StateFlow<SyncState> = _syncState
|
||||
|
||||
private val _liveActivity = MutableStateFlow(LiveSyncActivity())
|
||||
val liveActivity: StateFlow<LiveSyncActivity> = _liveActivity
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Live activity tracking (written from worker threads)
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private val trackingActiveRelays = ConcurrentHashMap.newKeySet<NormalizedRelayUrl>()
|
||||
private val trackingCompletions = ArrayDeque<LiveSyncActivity.CompletedRelayInfo>()
|
||||
private val trackingLock = Any()
|
||||
|
||||
/** Destination relay sets, set at the start of each sync run. */
|
||||
@Volatile private var liveOutboxTargets: Set<NormalizedRelayUrl> = emptySet()
|
||||
@Volatile private var liveInboxTargets: Set<NormalizedRelayUrl> = emptySet()
|
||||
@Volatile private var liveDmTargets: Set<NormalizedRelayUrl> = emptySet()
|
||||
|
||||
private fun emitLiveSnapshot() {
|
||||
val completions = synchronized(trackingLock) { trackingCompletions.toList() }
|
||||
_liveActivity.value =
|
||||
LiveSyncActivity(
|
||||
activeRelays = trackingActiveRelays.toSet(),
|
||||
recentCompletions = completions,
|
||||
outboxTargets = liveOutboxTargets,
|
||||
inboxTargets = liveInboxTargets,
|
||||
dmTargets = liveDmTargets,
|
||||
)
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Control functions
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private var syncJob: Job? = null
|
||||
|
||||
fun start() {
|
||||
if (_syncState.value is SyncState.Running) return
|
||||
trackingActiveRelays.clear()
|
||||
synchronized(trackingLock) { trackingCompletions.clear() }
|
||||
_liveActivity.value = LiveSyncActivity()
|
||||
syncJob =
|
||||
scope.launch(Dispatchers.IO) {
|
||||
runSync(startRelayIndex = 0, initialEventsSent = 0)
|
||||
@@ -140,6 +207,10 @@ class EventSyncViewModel(
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Sync logic
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private suspend fun runSync(
|
||||
startRelayIndex: Int,
|
||||
initialEventsSent: Int,
|
||||
@@ -165,6 +236,12 @@ class EventSyncViewModel(
|
||||
val inboxTargets = account.nip65RelayList.inboxFlow.value
|
||||
val dmTargets = account.dmRelays.flow.value
|
||||
|
||||
// Publish destination relays so the UI can show them before events arrive.
|
||||
liveOutboxTargets = outboxTargets
|
||||
liveInboxTargets = inboxTargets
|
||||
liveDmTargets = dmTargets
|
||||
emitLiveSnapshot()
|
||||
|
||||
val baseFilters =
|
||||
buildList {
|
||||
if (outboxTargets.isNotEmpty()) add(Filter(authors = listOf(myPubKey)))
|
||||
@@ -223,8 +300,19 @@ class EventSyncViewModel(
|
||||
}
|
||||
}
|
||||
},
|
||||
onRelayComplete = {
|
||||
onRelayStart = { relay ->
|
||||
trackingActiveRelays.add(relay)
|
||||
emitLiveSnapshot()
|
||||
},
|
||||
onRelayComplete = { relay, eventsFound ->
|
||||
trackingActiveRelays.remove(relay)
|
||||
val info = LiveSyncActivity.CompletedRelayInfo(relay, eventsFound)
|
||||
synchronized(trackingLock) {
|
||||
trackingCompletions.addFirst(info)
|
||||
while (trackingCompletions.size > MAX_ACTIVITY_LOG) trackingCompletions.removeLast()
|
||||
}
|
||||
val completed = relaysCompleted.incrementAndGet()
|
||||
emitLiveSnapshot()
|
||||
_syncState.value =
|
||||
SyncState.Running(
|
||||
relaysCompleted = completed,
|
||||
@@ -255,7 +343,8 @@ class EventSyncViewModel(
|
||||
relays: List<NormalizedRelayUrl>,
|
||||
baseFilters: List<Filter>,
|
||||
onEvent: (Event) -> Unit,
|
||||
onRelayComplete: () -> Unit,
|
||||
onRelayStart: (NormalizedRelayUrl) -> Unit,
|
||||
onRelayComplete: (NormalizedRelayUrl, Int) -> Unit,
|
||||
) {
|
||||
val semaphore = Semaphore(MAX_CONCURRENT_RELAYS)
|
||||
supervisorScope {
|
||||
@@ -264,8 +353,9 @@ class EventSyncViewModel(
|
||||
semaphore.acquire()
|
||||
launch {
|
||||
try {
|
||||
downloadFromRelay(relay, baseFilters, onEvent)
|
||||
onRelayComplete()
|
||||
onRelayStart(relay)
|
||||
val eventsFound = downloadFromRelay(relay, baseFilters, onEvent)
|
||||
onRelayComplete(relay, eventsFound)
|
||||
} finally {
|
||||
semaphore.release()
|
||||
}
|
||||
@@ -281,13 +371,16 @@ class EventSyncViewModel(
|
||||
* [Event.createdAt] minus one becomes the next `until` and the query repeats.
|
||||
* Stops when EOSE arrives with no new events (relay exhausted) or the relay
|
||||
* cannot be reached.
|
||||
*
|
||||
* @return total number of events received across all pages.
|
||||
*/
|
||||
private suspend fun downloadFromRelay(
|
||||
relay: NormalizedRelayUrl,
|
||||
baseFilters: List<Filter>,
|
||||
onEvent: (Event) -> Unit,
|
||||
) {
|
||||
): Int {
|
||||
var until: Long? = null
|
||||
var totalEvents = 0
|
||||
|
||||
while (true) {
|
||||
coroutineContext.ensureActive()
|
||||
@@ -345,10 +438,15 @@ class EventSyncViewModel(
|
||||
account.client.close(subId)
|
||||
done.close()
|
||||
|
||||
if (pageCount.get() == 0) break // relay exhausted or unreachable
|
||||
val count = pageCount.get()
|
||||
if (count == 0) break // relay exhausted or unreachable
|
||||
|
||||
totalEvents += count
|
||||
|
||||
// Advance cursor: next page starts just before the oldest event seen.
|
||||
until = pageMinTs.get() - 1
|
||||
}
|
||||
|
||||
return totalEvents
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1767,11 +1767,19 @@
|
||||
<string name="event_sync_resume">Resume</string>
|
||||
<string name="event_sync_start_over">Start Over</string>
|
||||
<string name="event_sync_cancel">Cancel</string>
|
||||
<string name="event_sync_batch_of">Batch %1$d of %2$d</string>
|
||||
<string name="event_sync_relays_progress">Relays: %1$d / %2$d</string>
|
||||
<string name="event_sync_events_sent">Events redistributed: %1$d</string>
|
||||
<string name="event_sync_paused_title">Sync Paused</string>
|
||||
<string name="event_sync_paused_body">Completed batch %1$d of %2$d — %3$d events redistributed so far. Tap Resume to continue.</string>
|
||||
<string name="event_sync_paused_body">Completed %1$d of %2$d relays — %3$d events redistributed so far. Tap Resume to continue.</string>
|
||||
<string name="event_sync_done_title">Sync complete</string>
|
||||
<string name="event_sync_done_body">Redistributed %1$d events in %2$d seconds.</string>
|
||||
<string name="event_sync_error_title">Sync error</string>
|
||||
<string name="event_sync_reading_from">Reading (%1$d active)</string>
|
||||
<string name="event_sync_sending_to">Sending To</string>
|
||||
<string name="event_sync_outbox_relays">Outbox</string>
|
||||
<string name="event_sync_inbox_relays">Inbox</string>
|
||||
<string name="event_sync_dm_relays">DMs</string>
|
||||
<string name="event_sync_activity_log">Activity (%1$d relays)</string>
|
||||
<string name="event_sync_events_found_log">%1$s events</string>
|
||||
<string name="event_sync_no_events">no events</string>
|
||||
</resources>
|
||||
|
||||
Reference in New Issue
Block a user