diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt index 7e982ebe7..780efd58b 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncScreen.kt @@ -50,7 +50,6 @@ import androidx.compose.runtime.setValue import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier 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 @@ -134,14 +133,16 @@ fun EventSyncScreen( // ---- Progress / Status area ---- when (val state = syncState) { - is EventSyncViewModel.SyncState.Idle -> { - // No progress shown — just the start button below - } + is EventSyncViewModel.SyncState.Idle -> Unit is EventSyncViewModel.SyncState.Running -> { SyncProgressCard(state = state) } + is EventSyncViewModel.SyncState.Paused -> { + PausedCard(state = state) + } + is EventSyncViewModel.SyncState.Done -> { DoneCard(state = state) } @@ -152,7 +153,7 @@ fun EventSyncScreen( } // ---- Action buttons ---- - when (syncState) { + when (val state = syncState) { is EventSyncViewModel.SyncState.Idle, is EventSyncViewModel.SyncState.Done, is EventSyncViewModel.SyncState.Error, @@ -171,6 +172,27 @@ fun EventSyncScreen( } } + is EventSyncViewModel.SyncState.Paused -> { + Button( + onClick = { + if (isMobileOrMetered) { + showMobileDataDialog = true + } else { + syncViewModel.resume() + } + }, + modifier = Modifier.fillMaxWidth(), + ) { + Text(stringRes(R.string.event_sync_resume)) + } + OutlinedButton( + onClick = { syncViewModel.start() }, + modifier = Modifier.fillMaxWidth(), + ) { + Text(stringRes(R.string.event_sync_start_over)) + } + } + is EventSyncViewModel.SyncState.Running -> { OutlinedButton( onClick = { syncViewModel.cancel() }, @@ -180,12 +202,14 @@ fun EventSyncScreen( contentColor = MaterialTheme.colorScheme.error, ), ) { - Text(stringRes(R.string.event_sync_cancel)) + Text(stringRes(R.string.event_sync_pause)) } } } + // ---- Mobile-data confirmation dialog ---- if (showMobileDataDialog) { + val isPaused = syncState is EventSyncViewModel.SyncState.Paused AlertDialog( onDismissRequest = { showMobileDataDialog = false }, title = { Text(stringRes(R.string.event_sync_mobile_data_dialog_title)) }, @@ -194,7 +218,7 @@ fun EventSyncScreen( Button( onClick = { showMobileDataDialog = false - syncViewModel.start() + if (isPaused) syncViewModel.resume() else syncViewModel.start() }, ) { Text(stringRes(R.string.event_sync_start_anyway)) @@ -223,7 +247,7 @@ private fun StepRow( verticalAlignment = Alignment.Top, ) { Text( - text = number + ".", + text = "$number.", style = MaterialTheme.typography.bodyMedium, fontWeight = FontWeight.Bold, color = MaterialTheme.colorScheme.primary, @@ -244,34 +268,13 @@ private fun SyncProgressCard(state: EventSyncViewModel.SyncState.Running) { ) { Column(modifier = Modifier.padding(16.dp)) { Text( - text = phaseLabel(state.phase), + text = stringRes(R.string.event_sync_batch_of, state.chunkIndex, state.totalChunks), style = MaterialTheme.typography.titleSmall, fontWeight = FontWeight.SemiBold, ) - Spacer(Modifier.height(4.dp)) - Text( - text = stringRes(R.string.event_sync_relay_of, state.relayIndex, state.totalRelays), - style = MaterialTheme.typography.bodySmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - ) - Spacer(Modifier.height(4.dp)) - Text( - text = state.currentRelay, - style = MaterialTheme.typography.bodySmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - maxLines = 1, - overflow = TextOverflow.Ellipsis, - ) Spacer(Modifier.height(8.dp)) - - // Overall progress across all phases and relays - val overallProgress = - ( - (state.phase - 1).toFloat() / state.phaseTotal + - (state.relayIndex.toFloat() / state.totalRelays) / state.phaseTotal - ).coerceIn(0f, 1f) LinearProgressIndicator( - progress = { overallProgress }, + progress = { state.chunkIndex.toFloat() / state.totalChunks }, modifier = Modifier.fillMaxWidth(), ) Spacer(Modifier.height(8.dp)) @@ -284,6 +287,38 @@ private fun SyncProgressCard(state: EventSyncViewModel.SyncState.Running) { } } +@Composable +private fun PausedCard(state: EventSyncViewModel.SyncState.Paused) { + Card( + modifier = Modifier.fillMaxWidth(), + colors = + CardDefaults.cardColors( + containerColor = MaterialTheme.colorScheme.secondaryContainer, + ), + ) { + Column(modifier = Modifier.padding(16.dp)) { + Text( + text = stringRes(R.string.event_sync_paused_title), + style = MaterialTheme.typography.titleSmall, + fontWeight = FontWeight.Bold, + color = MaterialTheme.colorScheme.onSecondaryContainer, + ) + Spacer(Modifier.height(4.dp)) + Text( + text = + stringRes( + R.string.event_sync_paused_body, + state.nextChunkIndex, + state.totalChunks, + state.eventsSent, + ), + style = MaterialTheme.typography.bodyMedium, + color = MaterialTheme.colorScheme.onSecondaryContainer, + ) + } + } +} + @Composable private fun DoneCard(state: EventSyncViewModel.SyncState.Done) { Card( @@ -340,12 +375,3 @@ private fun ErrorCard(message: String) { } } } - -@Composable -private fun phaseLabel(phase: Int): String = - when (phase) { - 1 -> stringRes(R.string.event_sync_phase1) - 2 -> stringRes(R.string.event_sync_phase2) - 3 -> stringRes(R.string.event_sync_phase3) - else -> "" - } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt index 77ac8fe00..ee242acae 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/eventsync/EventSyncViewModel.kt @@ -36,32 +36,48 @@ import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull +import java.util.concurrent.atomic.AtomicInteger /** * Syncs the user's events across all known relays: * 1. Downloads all events authored by the user and sends them to their outbox relays. - * 2. Downloads all events that p-tag the user and sends them to their inbox relays. + * 2. Downloads all events that p-tag the user (non-DM) and sends them to their inbox relays. * 3. Downloads kind-4 and kind-1059 events that p-tag the user and sends them to DM relays. * - * The procedure iterates over every relay URL in LocalCache.relayHints.relayDB. - * Events are deduplicated per destination to avoid redundant sends. + * Relays are processed in batches of [CHUNK_SIZE]. All three filters are sent to every relay + * in a chunk simultaneously, and events are routed to the appropriate destination by their + * properties. This removes the need for three sequential passes over the full relay list. * - * Scoped to the AccountViewModel so the sync continues in the background while - * the user navigates to other screens or away from the app. + * The sync is pausable: calling [cancel] transitions to [SyncState.Paused] so the user can + * [resume] from the last completed chunk rather than starting over. + * + * Scoped to the AccountViewModel so the sync survives navigation within the same session. */ class EventSyncViewModel( val account: Account, private val scope: CoroutineScope, ) { + companion object { + /** Number of relays queried in each batch. */ + const val CHUNK_SIZE = 50 + + /** How long (ms) to wait for all relays in a chunk to reply before moving on. */ + const val CHUNK_TIMEOUT_MS = 120_000L + } + sealed class SyncState { object Idle : SyncState() data class Running( - val phase: Int, - val phaseTotal: Int, - val currentRelay: String, - val relayIndex: Int, - val totalRelays: Int, + val chunkIndex: Int, + val totalChunks: Int, + val eventsSent: Int, + ) : SyncState() + + /** Cancelled mid-run; can be resumed from [nextChunkIndex]. */ + data class Paused( + val nextChunkIndex: Int, + val totalChunks: Int, val eventsSent: Int, ) : SyncState() @@ -84,129 +100,135 @@ class EventSyncViewModel( if (_syncState.value is SyncState.Running) return syncJob = scope.launch(Dispatchers.IO) { - runSync() + runSync(startChunkIndex = 0, initialEventsSent = 0) + } + } + + fun resume() { + val paused = _syncState.value as? SyncState.Paused ?: return + if (_syncState.value is SyncState.Running) return + syncJob = + scope.launch(Dispatchers.IO) { + runSync( + startChunkIndex = paused.nextChunkIndex, + initialEventsSent = paused.eventsSent, + ) } } fun cancel() { syncJob?.cancel() - _syncState.value = SyncState.Idle + val current = _syncState.value + _syncState.value = + if (current is SyncState.Running) { + SyncState.Paused( + nextChunkIndex = current.chunkIndex - 1, // retry the in-progress chunk + totalChunks = current.totalChunks, + eventsSent = current.eventsSent, + ) + } else { + SyncState.Idle + } } - private suspend fun runSync() { + private suspend fun runSync( + startChunkIndex: Int, + initialEventsSent: Int, + ) { val startTime = System.currentTimeMillis() - val myPubKey = account.signer.pubKey + val allRelays = account.cache.relayHints.relayDB .keys() .toList() if (allRelays.isEmpty()) { - _syncState.value = SyncState.Error("No known relays found. Browse some content first to discover relays.") + _syncState.value = + SyncState.Error("No known relays found. Browse some content first to discover relays.") return } + val chunks = allRelays.chunked(CHUNK_SIZE) + val totalChunks = chunks.size + val outboxTargets = account.outboxRelays.flow.value val inboxTargets = account.nip65RelayList.inboxFlow.value val dmTargets = account.dmRelays.flow.value - // Deduplication sets — one per destination category + // Build the minimal set of filters needed across all three categories. + // Events are routed to their correct destination(s) by property after retrieval. + val filters = + buildList { + if (outboxTargets.isNotEmpty()) add(Filter(authors = listOf(myPubKey))) + if (inboxTargets.isNotEmpty() || dmTargets.isNotEmpty()) { + add(Filter(tags = mapOf("p" to listOf(myPubKey)))) + } + } + + if (filters.isEmpty()) { + _syncState.value = + SyncState.Error("No outbox, inbox, or DM relays configured.") + return + } + + // Deduplication sets — one per destination category. val outboxSent = HashSet(1024) val inboxSent = HashSet(1024) val dmSent = HashSet(1024) - var totalSent = 0 - val totalPhases = 3 + var totalSent = initialEventsSent try { - // ------------------------- - // Phase 1: Author's events → Outbox relays - // ------------------------- - if (outboxTargets.isNotEmpty()) { - allRelays.forEachIndexed { index, relay -> - if (!isActive) return + for (i in startChunkIndex until totalChunks) { + if (!isActive) return - _syncState.value = - SyncState.Running( - phase = 1, - phaseTotal = totalPhases, - currentRelay = relay.url, - relayIndex = index + 1, - totalRelays = allRelays.size, - eventsSent = totalSent, - ) + _syncState.value = + SyncState.Running( + chunkIndex = i + 1, + totalChunks = totalChunks, + eventsSent = totalSent, + ) - downloadFromRelay( - relay = relay, - filter = Filter(authors = listOf(myPubKey)), - ) { event -> + downloadFromChunk( + relays = chunks[i], + filters = filters, + ) { event -> + // Route to outbox if I authored it. + if (event.pubKey == myPubKey && outboxTargets.isNotEmpty()) { if (outboxSent.add(event.id)) { account.client.send(event, outboxTargets) totalSent++ } } - } - } - - // ------------------------- - // Phase 2: P-tagged (non-DM) events → Inbox relays - // ------------------------- - if (inboxTargets.isNotEmpty()) { - allRelays.forEachIndexed { index, relay -> - if (!isActive) return - - _syncState.value = - SyncState.Running( - phase = 2, - phaseTotal = totalPhases, - currentRelay = relay.url, - relayIndex = index + 1, - totalRelays = allRelays.size, - eventsSent = totalSent, - ) - - downloadFromRelay( - relay = relay, - filter = Filter(tags = mapOf("p" to listOf(myPubKey))), - ) { event -> - // Skip DM kinds — handled in phase 3 - if (event.kind != 4 && event.kind != 1059) { - if (inboxSent.add(event.id)) { + // Route p-tagged events to inbox or DM relays. + val pTagsMe = + event.tags.any { tag -> + tag.size >= 2 && tag[0] == "p" && tag[1] == myPubKey + } + if (pTagsMe) { + if (event.kind == 4 || event.kind == 1059) { + if (dmTargets.isNotEmpty() && dmSent.add(event.id)) { + account.client.send(event, dmTargets) + totalSent++ + } + } else { + if (inboxTargets.isNotEmpty() && inboxSent.add(event.id)) { account.client.send(event, inboxTargets) totalSent++ } } } } - } - - // ------------------------- - // Phase 3: DM events (kind 4 & 1059) → DM relays - // ------------------------- - if (dmTargets.isNotEmpty()) { - allRelays.forEachIndexed { index, relay -> - if (!isActive) return + // Update the running count after the chunk completes. + if (isActive) { _syncState.value = SyncState.Running( - phase = 3, - phaseTotal = totalPhases, - currentRelay = relay.url, - relayIndex = index + 1, - totalRelays = allRelays.size, + chunkIndex = i + 1, + totalChunks = totalChunks, eventsSent = totalSent, ) - - downloadFromRelay( - relay = relay, - filter = Filter(kinds = listOf(4, 1059), tags = mapOf("p" to listOf(myPubKey))), - ) { event -> - if (dmSent.add(event.id)) { - account.client.send(event, dmTargets) - totalSent++ - } - } } } @@ -223,14 +245,16 @@ class EventSyncViewModel( } /** - * Opens a REQ subscription on [relay] with [filter], delivers every event to [onEvent], - * and suspends until the relay sends EOSE (or 90 seconds pass, whichever comes first). + * Sends [filters] to every relay in [relays] simultaneously (one subscription covering + * the whole chunk). Suspends until every relay has replied with EOSE/CLOSED/error, or + * [CHUNK_TIMEOUT_MS] elapses, whichever comes first. */ - private suspend fun downloadFromRelay( - relay: NormalizedRelayUrl, - filter: Filter, + private suspend fun downloadFromChunk( + relays: List, + filters: List, onEvent: (Event) -> Unit, ) { + val remaining = AtomicInteger(relays.size) val done = Channel(Channel.CONFLATED) val subId = newSubId() @@ -249,7 +273,7 @@ class EventSyncViewModel( relay: NormalizedRelayUrl, forFilters: List?, ) { - done.trySend(Unit) + if (remaining.decrementAndGet() <= 0) done.trySend(Unit) } override fun onClosed( @@ -257,7 +281,7 @@ class EventSyncViewModel( relay: NormalizedRelayUrl, forFilters: List?, ) { - done.trySend(Unit) + if (remaining.decrementAndGet() <= 0) done.trySend(Unit) } override fun onCannotConnect( @@ -265,13 +289,14 @@ class EventSyncViewModel( message: String, forFilters: List?, ) { - done.trySend(Unit) + if (remaining.decrementAndGet() <= 0) done.trySend(Unit) } } - account.client.openReqSubscription(subId, mapOf(relay to listOf(filter)), listener) + val filtersMap = relays.associateWith { filters } + account.client.openReqSubscription(subId, filtersMap, listener) - withTimeoutOrNull(90_000L) { + withTimeoutOrNull(CHUNK_TIMEOUT_MS) { done.receive() } diff --git a/amethyst/src/main/res/values/strings.xml b/amethyst/src/main/res/values/strings.xml index c0295cf4d..d7c2063ee 100644 --- a/amethyst/src/main/res/values/strings.xml +++ b/amethyst/src/main/res/values/strings.xml @@ -1763,12 +1763,14 @@ Use Mobile Data? Start Sync Start Anyway (mobile data) + Pause + Resume + Start Over Cancel - Phase 1 of 3 — Syncing your posts to outbox relays - Phase 2 of 3 — Syncing mentions to inbox relays - Phase 3 of 3 — Syncing direct messages to DM relays - Relay %1$d of %2$d + Batch %1$d of %2$d Events redistributed: %1$d + Sync Paused + Completed batch %1$d of %2$d — %3$d events redistributed so far. Tap Resume to continue. Sync complete Redistributed %1$d events in %2$d seconds. Sync error