diff --git a/android/app/build.gradle.kts b/android/app/build.gradle.kts index 4e45501..d856236 100644 --- a/android/app/build.gradle.kts +++ b/android/app/build.gradle.kts @@ -103,6 +103,7 @@ dependencies { implementation(libs.ktor.client.content.negotiation) implementation(libs.ktor.serialization.kotlinx.json) implementation(libs.ktor.client.logging) + implementation(libs.ktor.client.websockets) // Serialization implementation(libs.kotlinx.serialization.json) diff --git a/android/app/src/main/java/com/ethosprotocol/api/ApiClient.kt b/android/app/src/main/java/com/ethosprotocol/api/ApiClient.kt index 81a510f..d616d87 100644 --- a/android/app/src/main/java/com/ethosprotocol/api/ApiClient.kt +++ b/android/app/src/main/java/com/ethosprotocol/api/ApiClient.kt @@ -11,9 +11,11 @@ import io.ktor.client.* import io.ktor.client.call.* import io.ktor.client.engine.HttpClientEngine import io.ktor.client.engine.android.* +import io.ktor.client.plugins.HttpRequestTimeoutException import io.ktor.client.plugins.HttpTimeout import io.ktor.client.plugins.contentnegotiation.* import io.ktor.client.plugins.logging.* +import io.ktor.client.plugins.websocket.WebSockets import io.ktor.client.request.* import io.ktor.http.* import io.ktor.serialization.kotlinx.json.* @@ -33,7 +35,7 @@ sealed class ApiResult { } @Singleton -class ApiClient @Inject constructor( +class ApiClient( private val tokenProvider: TokenProvider, private val networkMonitor: NetworkMonitor, private val offlineCache: OfflineCache, @@ -152,7 +154,9 @@ class ApiClient @Inject constructor( else ApiResult.NetworkUnavailable } return runCatching { - val response = client.get("$baseUrl$path") { bearerAuth() } + val response = withRetry(retryPolicy, ::isRetryableNetworkError) { + client.get("$baseUrl$path") { bearerAuth() } + } when (response.status.value) { in 200..299 -> { val body: T = response.body() @@ -220,6 +224,15 @@ class ApiClient @Inject constructor( if (result is ApiResult.Success) tokenProvider.setSession(result.data) } + // GET is the only idempotent verb this client issues — retrying POST/DELETE + // automatically could double-submit a mutation (check-in, withdrawal, 2FA + // disable, ...), so only get() calls withRetry with this predicate. + // HttpRequestTimeoutException is checked explicitly because it subclasses + // CancellationException (so HttpTimeout can cooperate with coroutine + // cancellation) rather than IOException. + private fun isRetryableNetworkError(e: Throwable): Boolean = + e is HttpRequestTimeoutException || e is IOException + private fun HttpRequestBuilder.bearerAuth() { tokenProvider.token?.let { header(HttpHeaders.Authorization, "Bearer $it") } } diff --git a/android/app/src/main/java/com/ethosprotocol/api/RetryPolicy.kt b/android/app/src/main/java/com/ethosprotocol/api/RetryPolicy.kt new file mode 100644 index 0000000..de562aa --- /dev/null +++ b/android/app/src/main/java/com/ethosprotocol/api/RetryPolicy.kt @@ -0,0 +1,37 @@ +package com.ethosprotocol.api + +import kotlinx.coroutines.delay + +// Retry configuration for idempotent network calls. ApiClient applies this only +// to GET requests — POST/DELETE must never be retried automatically, since a +// retried mutation (check-in, withdrawal, 2FA disable, ...) could double-submit. +data class RetryPolicy( + val maxAttempts: Int, + val baseDelayMillis: Long, + val sleep: suspend (Long) -> Unit = { delay(it) } +) { + companion object { + val networkDefault = RetryPolicy(maxAttempts = 3, baseDelayMillis = 500) + } +} + +// Retries [operation] with exponential backoff (`baseDelayMillis * 2^attempt`) up to +// `policy.maxAttempts` total attempts, but only for errors [isRetryable] accepts. +// Any other error — or exhausting the attempt budget — is rethrown immediately. +suspend fun withRetry( + policy: RetryPolicy, + isRetryable: (Throwable) -> Boolean, + operation: suspend () -> T +): T { + var attempt = 0 + while (true) { + try { + return operation() + } catch (e: Throwable) { + attempt++ + if (attempt >= policy.maxAttempts || !isRetryable(e)) throw e + val delayMillis = policy.baseDelayMillis * (1L shl (attempt - 1)) + policy.sleep(delayMillis) + } + } +} diff --git a/android/app/src/main/java/com/ethosprotocol/models/Models.kt b/android/app/src/main/java/com/ethosprotocol/models/Models.kt index b290954..a7eebf5 100644 --- a/android/app/src/main/java/com/ethosprotocol/models/Models.kt +++ b/android/app/src/main/java/com/ethosprotocol/models/Models.kt @@ -18,6 +18,25 @@ data class Vault( val formattedBalance: String get() = "%.7f XLM".format(balance / 10_000_000.0) } +// A single real-time event delivered over the `wss://.../ws?vault_id={id}` socket +// (see shared/api-contract.md). `vault` carries the full updated Vault so consumers +// can update state in place without an extra round trip. +@Serializable +data class VaultEvent( + val type: String, + val vault: Vault? = null +) + +// A single page of GET /vaults, requested via `offset`/`limit` query params (see +// shared/api-contract.md). `nextOffset` is the offset to pass for the following +// page; null once `hasMore` is false. +@Serializable +data class VaultPage( + val vaults: List, + @SerialName("next_offset") val nextOffset: Int? = null, + @SerialName("has_more") val hasMore: Boolean = false +) + @Serializable enum class VaultStatus { active, expired, released, paused } diff --git a/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt b/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt new file mode 100644 index 0000000..de34f43 --- /dev/null +++ b/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt @@ -0,0 +1,90 @@ +package com.ethosprotocol.services + +import com.ethosprotocol.api.ApiClient +import com.ethosprotocol.api.TokenProvider +import com.ethosprotocol.models.VaultEvent +import io.ktor.client.plugins.websocket.webSocketSession +import io.ktor.client.request.header +import io.ktor.client.request.url +import io.ktor.http.HttpHeaders +import io.ktor.websocket.Frame +import io.ktor.websocket.WebSocketSession +import io.ktor.websocket.readText +import javax.inject.Inject +import javax.inject.Singleton +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.isActive +import kotlinx.serialization.json.Json + +// Reconnect/backoff schedule for VaultEventSocket. Unlike RetryPolicy (bounded +// attempts for a single request), a dropped socket should keep retrying +// indefinitely with the delay capped so a long outage doesn't leave the client +// waiting minutes between attempts once it reconnects. +data class ReconnectBackoff( + val baseDelayMillis: Long, + val maxDelayMillis: Long, + val sleep: suspend (Long) -> Unit = { delay(it) } +) { + fun delayForAttempt(attempt: Int): Long { + val shift = attempt.coerceIn(0, 20) + return (baseDelayMillis * (1L shl shift)).coerceAtMost(maxDelayMillis) + } + + companion object { + val default = ReconnectBackoff(baseDelayMillis = 1_000, maxDelayMillis = 30_000) + } +} + +// Client for the `wss://.../ws?vault_id={id}` endpoint (shared/api-contract.md). +// Reuses ApiClient's HttpClient/WebSockets plugin rather than a second client. +// A dropped or failed connection reconnects with exponential backoff +// ([ReconnectBackoff]) until the collecting coroutine is cancelled; the backoff +// resets once a new connection is established. +@Singleton +class VaultEventSocket( + private val apiClient: ApiClient, + private val tokenProvider: TokenProvider, + private val backoff: ReconnectBackoff = ReconnectBackoff.default +) { + // Distinct @Inject constructor for the same reason as ApiClient's: Dagger's + // codegen calls the full-arg constructor directly and ignores Kotlin default + // values, so Hilt needs an explicit constructor matching only its bindings. + @Inject constructor(apiClient: ApiClient, tokenProvider: TokenProvider) : + this(apiClient, tokenProvider, ReconnectBackoff.default) + + // internal (not private): tests replace this to simulate connect failures/frames + // without a real server, matching how ApiClient exposes its engine to tests. + internal var openSession: suspend (String) -> WebSocketSession = { vaultId -> + apiClient.client.webSocketSession { + url(apiClient.webSocketUrl(vaultId)) + tokenProvider.token?.let { header(HttpHeaders.Authorization, "Bearer $it") } + } + } + + fun events(vaultId: String): Flow = flow { + var attempt = 0 + while (currentCoroutineContext().isActive) { + try { + val session = openSession(vaultId) + attempt = 0 + for (frame in session.incoming) { + if (frame is Frame.Text) { + runCatching { Json.decodeFromString(frame.readText()) } + .onSuccess { emit(it) } + } + } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + // Connection failed or dropped — fall through to backoff and reconnect. + } + if (!currentCoroutineContext().isActive) break + backoff.sleep(backoff.delayForAttempt(attempt)) + attempt++ + } + } +} diff --git a/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt b/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt index 6f31719..86a80f3 100644 --- a/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt +++ b/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt @@ -252,6 +252,7 @@ class TwoFactorViewModel @Inject constructor( data class VaultUiState( val vaults: List = emptyList(), val isLoading: Boolean = false, + val isLoadingMore: Boolean = false, val error: String? = null, val isOffline: Boolean = false, val beneficiaryUpdated: Boolean = false @@ -268,11 +269,23 @@ class VaultViewModel @Inject constructor( private val _state = MutableStateFlow(VaultUiState()) val state = _state.asStateFlow() + private var nextOffset = 0 + private val eventJobs = mutableMapOf() + fun load() = viewModelScope.launch { _state.update { it.copy(isLoading = true, error = null) } - when (val result = apiClient.listVaults()) { + when (val result = apiClient.listVaults(offset = 0, limit = PAGE_SIZE)) { is ApiResult.Success -> { - _state.update { it.copy(vaults = result.data, isLoading = false, isOffline = false) } + nextOffset = result.data.nextOffset ?: result.data.vaults.size + _state.update { + it.copy( + vaults = result.data.vaults, + isLoading = false, + isOffline = false, + hasMore = result.data.hasMore + ) + } + subscribeToEvents(result.data.vaults.map { it.id }) } ApiResult.NetworkUnavailable -> { _state.update { it.copy(isLoading = false, isOffline = true) } diff --git a/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt b/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt index 33f0511..8300d10 100644 --- a/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt +++ b/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt @@ -324,6 +324,15 @@ fun VaultListScreen( ) } } + if (state.hasMore) item { + Box(Modifier.fillMaxWidth().padding(16.dp), contentAlignment = Alignment.Center) { + if (state.isLoadingMore) { + CircularProgressIndicator(Modifier.size(24.dp), strokeWidth = 2.dp) + } else { + OutlinedButton(onClick = { vm.loadMore() }) { Text("Load more") } + } + } + } } } } diff --git a/android/app/src/main/java/com/ethosprotocol/widget/VaultStatusWidget.kt b/android/app/src/main/java/com/ethosprotocol/widget/VaultStatusWidget.kt index 36378c0..01de06b 100644 --- a/android/app/src/main/java/com/ethosprotocol/widget/VaultStatusWidget.kt +++ b/android/app/src/main/java/com/ethosprotocol/widget/VaultStatusWidget.kt @@ -104,7 +104,7 @@ class VaultWidgetUpdateWorker @AssistedInject constructor( ) : CoroutineWorker(context, params) { override suspend fun doWork(): Result { - val result = apiClient.listVaults() + val result = apiClient.listVaults(limit = 1) if (result is ApiResult.Success) { // Pick the active vault with the lowest ttlRemaining — the same // "most urgent vault" selection that iOS TTLTimelineProvider uses diff --git a/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt b/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt new file mode 100644 index 0000000..6401319 --- /dev/null +++ b/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt @@ -0,0 +1,96 @@ +package com.ethosprotocol + +import com.ethosprotocol.api.ApiClient +import com.ethosprotocol.api.TokenProvider +import com.ethosprotocol.models.VaultEvent +import com.ethosprotocol.services.ReconnectBackoff +import com.ethosprotocol.services.VaultEventSocket +import io.ktor.websocket.Frame +import io.ktor.websocket.WebSocketSession +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import java.io.IOException +import java.util.concurrent.atomic.AtomicInteger +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.test.runTest +import kotlinx.serialization.json.Json +import org.junit.Assert.* +import org.junit.Test + +class VaultEventSocketTest { + + private val apiClient: ApiClient = mockk(relaxed = true) + private val tokenProvider: TokenProvider = mockk { every { token } returns "test-token" } + + @Test + fun `delayForAttempt doubles each attempt and caps at maxDelayMillis`() { + val backoff = ReconnectBackoff(baseDelayMillis = 1_000, maxDelayMillis = 30_000) + + assertEquals(1_000L, backoff.delayForAttempt(0)) + assertEquals(2_000L, backoff.delayForAttempt(1)) + assertEquals(4_000L, backoff.delayForAttempt(2)) + assertEquals(8_000L, backoff.delayForAttempt(3)) + assertEquals(30_000L, backoff.delayForAttempt(10)) + } + + @Test + fun `events reconnects with backoff after a connection failure and resumes emitting`() = runTest { + val delays = mutableListOf() + val backoff = ReconnectBackoff(baseDelayMillis = 1_000, maxDelayMillis = 30_000, sleep = { delays.add(it) }) + val socket = VaultEventSocket(apiClient, tokenProvider, backoff) + + val openAttempts = AtomicInteger(0) + val channel = Channel(capacity = 1) + val event = VaultEvent(type = "check_in", vault = null) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) + + socket.openSession = { + if (openAttempts.getAndIncrement() == 0) throw IOException("connection refused") + mockk { every { incoming } returns channel } + } + + val received = socket.events("vault-1").take(1).toList() + + assertEquals(listOf(event), received) + assertEquals(2, openAttempts.get()) + assertEquals(listOf(1_000L), delays) + } + + @Test + fun `events resets the backoff attempt counter after a successful connection`() = runTest { + val delays = mutableListOf() + val backoff = ReconnectBackoff(baseDelayMillis = 1_000, maxDelayMillis = 30_000, sleep = { delays.add(it) }) + val socket = VaultEventSocket(apiClient, tokenProvider, backoff) + + // Sequence: fail, succeed-then-immediately-close (no frames), fail, succeed-with-a-frame. + val openAttempts = AtomicInteger(0) + val event = VaultEvent(type = "check_in", vault = null) + socket.openSession = { + when (openAttempts.getAndIncrement()) { + 0 -> throw IOException("connection refused") + 1 -> mockk { every { incoming } returns Channel().apply { close() } } + 2 -> throw IOException("connection refused") + else -> mockk { + every { incoming } returns Channel(capacity = 1).apply { + trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) + } + } + } + } + + val received = socket.events("vault-1").take(1).toList() + + assertEquals(listOf(event), received) + assertEquals(4, openAttempts.get()) + // open#0 fails at attempt=0 -> delay 1000ms, attempt->1. + // open#1 connects (then closes with no frames) -> attempt reset to 0 -> delay + // 1000ms again (not 2000ms), proving the reset, attempt->1. + // open#2 fails at attempt=1 -> delay 2000ms, attempt->2. + // open#3 connects and delivers the frame; take(1) ends collection before any + // further delay is recorded. + assertEquals(listOf(1_000L, 1_000L, 2_000L), delays) + } +} diff --git a/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt b/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt index 1c0b177..e2a81d6 100644 --- a/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt +++ b/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt @@ -2,6 +2,8 @@ package com.ethosprotocol import com.ethosprotocol.api.ApiResult import com.ethosprotocol.models.Vault +import com.ethosprotocol.models.VaultEvent +import com.ethosprotocol.models.VaultPage import com.ethosprotocol.models.VaultStatus import com.ethosprotocol.ui.VaultUiState import com.ethosprotocol.ui.VaultViewModel @@ -16,6 +18,8 @@ import io.mockk.* import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.test.* import org.junit.After import org.junit.Assert.* @@ -49,18 +53,20 @@ class VaultViewModelTest { @Test fun `load success updates vaults`() = runTest { val vaults = listOf(makeVault("v1"), makeVault("v2")) - coEvery { apiClient.listVaults() } returns ApiResult.Success(vaults) + coEvery { apiClient.listVaults(offset = 0, limit = 20) } returns + ApiResult.Success(VaultPage(vaults, nextOffset = null, hasMore = false)) vm.load() assertEquals(vaults, vm.state.value.vaults) assertFalse(vm.state.value.isLoading) + assertFalse(vm.state.value.hasMore) assertNull(vm.state.value.error) } @Test fun `load network unavailable sets offline flag`() = runTest { - coEvery { apiClient.listVaults() } returns ApiResult.NetworkUnavailable + coEvery { apiClient.listVaults(offset = 0, limit = 20) } returns ApiResult.NetworkUnavailable vm.load() @@ -83,7 +89,7 @@ class VaultViewModelTest { @Test fun `load error sets error message`() = runTest { - coEvery { apiClient.listVaults() } returns ApiResult.Error("Server error", 500) + coEvery { apiClient.listVaults(offset = 0, limit = 20) } returns ApiResult.Error("Server error", 500) vm.load()