feat: websocket events update payment db

2026-06-01 15:44:19 +02:00
parent 18c8697ce9
commit 385536db25
6 changed files with 277 additions and 68 deletions
@@ -260,11 +260,19 @@ data class WsPaymentMessage(

 data class WsPayment(
    @SerializedName("payment_hash") val paymentHash: String,
-    val amount: Long,
-    val status: String,
-    val memo: String?
+    @SerializedName("checking_id")  val checkingId: String,
+    @SerializedName("amount")       val amount: Long,
+    @SerializedName("fee")          val fee: Long,
+    @SerializedName("memo")         val memo: String?,
+    @SerializedName("time")         val time: String,
+    @SerializedName("status")       val status: String,
+    @SerializedName("bolt11")       val bolt11: String?,
+    @SerializedName("pending")      val pending: Boolean,
+    @SerializedName("preimage")     val preimage: String?,
+    @SerializedName("extra")        val extra: PaymentExtra?
 )

+
 // ── Fiat rate ────────────────────────────────────────────────────────────────
 data class FiatRateResponse(
    val rate: Double,   // sat to fiat rate
@@ -36,6 +36,40 @@ class PaymentCacheRepository(context: Context) {
        }
    }

+    /**
+     * Load a page of payments from the DB.
+     * Returns empty list if nothing cached yet.
+     */
+    suspend fun loadCachedPage(offset: Int, limit: Int): List<PaymentRecord> {
+        return dao.getPage(limit = limit, offset = offset)
+            .map { it.toDomain() }
+            .also {
+                if (it.isNotEmpty())
+                    Log.d(TAG, "PAYMENTS [CACHE HIT ] offset=$offset got ${it.size} rows from Room")
+                else
+                    Log.d(TAG, "PAYMENTS [CACHE MISS] offset=$offset — no rows in Room")
+            }
+    }
+
+    /** Total number of payment rows in the DB. */
+    suspend fun countCached(): Int = dao.count()
+
+    /**
+     * Merge a fetched page into the DB.
+     * Uses INSERT OR REPLACE — never deletes existing rows.
+     * Safe to call for any page, not just the first.
+     */
+    suspend fun mergePayments(payments: List<PaymentRecord>) {
+        if (payments.isEmpty()) return
+        val now = System.currentTimeMillis()
+        runCatching {
+            dao.insertAll(payments.map { it.toEntity(now) })
+            Log.d(TAG, "PAYMENTS [MERGE    ] upserted ${payments.size} payments (total: ${dao.count()})")
+        }.onFailure {
+            Log.w(TAG, "PAYMENTS [MERGE ERR] ${it.message}")
+        }
+    }
+
    /** Persist the first page of payments to the database. */
    suspend fun savePayments(payments: List<PaymentRecord>) {
        val now = System.currentTimeMillis()
@@ -62,6 +96,28 @@ class PaymentCacheRepository(context: Context) {
        Log.d(TAG, "PAYMENTS [DETAIL SET] $checkingId")
    }

+    /** Upsert PaymentRecord to database. */
+    suspend fun upsertPayment(payment: PaymentRecord) {
+        val entity = payment.toEntity(savedAt = System.currentTimeMillis())
+        dao.insert(entity)
+        Log.d(TAG, "PAYMENTS [UPSERT   ] ${payment.checkingId}")
+    }
+
+    /** Check the DB for a stored payment record and return it as a detail response.
+     *  Returns null if not found. Never hits the network. */
+    suspend fun getDetailFromDb(checkingId: String): PaymentDetailResponse? {
+        val entity = dao.getByCheckingId(checkingId) ?: return null
+        Log.d(TAG, "PAYMENTS [DETAIL DB ] $checkingId — found in Room")
+        return PaymentDetailResponse(
+            paid    = entity.status == "success",
+            details = entity.toDomain()
+        )
+    }
+
+    /** Returns true if this checkingId is already stored in the DB. */
+    suspend fun containsPayment(checkingId: String): Boolean =
+        dao.getByCheckingId(checkingId) != null
+
    /** Clear everything — useful for testing and logout. */
    suspend fun clearAll() {
        detailCache.clear()
@@ -11,14 +11,28 @@ interface PaymentDao {
    @Insert(onConflict = OnConflictStrategy.REPLACE)
    suspend fun insertAll(payments: List<PaymentRecordEntity>)

+    @Insert(onConflict = OnConflictStrategy.REPLACE)
+    suspend fun insert(payment: PaymentRecordEntity)
+
    /** Returns all rows ordered by time descending. */
    @Query("SELECT * FROM payment_records ORDER BY time DESC")
    suspend fun getAll(): List<PaymentRecordEntity>

+    /** A single page, ordered newest-first. */
+    @Query("SELECT * FROM payment_records ORDER BY time DESC LIMIT :limit OFFSET :offset")
+    suspend fun getPage(limit: Int, offset: Int): List<PaymentRecordEntity>
+
+    /** Total number of rows — used to know if we have a full page cached. */
+    @Query("SELECT COUNT(*) FROM payment_records")
+    suspend fun count(): Int
+
    /** Returns the savedAt timestamp of the most recently written row, or null if table is empty. */
    @Query("SELECT savedAt FROM payment_records ORDER BY savedAt DESC LIMIT 1")
    suspend fun getLatestSavedAt(): Long?

+    @Query("SELECT * FROM payment_records WHERE checkingId = :checkingId LIMIT 1")
+    suspend fun getByCheckingId(checkingId: String): PaymentRecordEntity?
+
    @Query("DELETE FROM payment_records")
    suspend fun deleteAll()
 }
@@ -9,12 +9,13 @@ import android.os.IBinder
 import android.util.Log
 import com.bitcointxoko.gudariwallet.api.GsonProvider
 import com.bitcointxoko.gudariwallet.api.LNbitsClient
+import com.bitcointxoko.gudariwallet.api.PaymentRecord
 import com.bitcointxoko.gudariwallet.api.WsPaymentMessage
+import com.bitcointxoko.gudariwallet.data.PaymentCacheRepository
 import com.bitcointxoko.gudariwallet.security.EncryptedSecretStore
 import com.bitcointxoko.gudariwallet.util.NotificationConstants
 import com.bitcointxoko.gudariwallet.util.NotificationHelper
 import com.bitcointxoko.gudariwallet.util.WalletConstants
-import com.google.gson.Gson
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -48,6 +49,8 @@ private const val TAG = "WalletNotifService"
 */
 class WalletNotificationService : Service() {

+    private val paymentCache by lazy { PaymentCacheRepository(applicationContext) }
+
    // ── Companion: shared state accessible by ViewModel ───────────────────────

    companion object {
@@ -231,6 +234,26 @@ class WalletNotificationService : Service() {
                )
            )
        }
+
+        // Persist the incoming payment directly to DB
+        // so the cache is fresh even if the History screen isn't open
+        serviceScope.launch {
+            paymentCache.upsertPayment(
+                PaymentRecord(
+                    checkingId = payment.checkingId,
+                    paymentHash = payment.paymentHash,
+                    amountMsat = payment.amount,
+                    feeMsat = 0L,           // WsPaymentMessage doesn't carry fee
+                    memo = payment.memo,
+                    time = payment.time,
+                    status = payment.status,
+                    bolt11 = payment.bolt11,
+                    pending = false,        // we only act on status == "success"
+                    preimage = payment.preimage,
+                    extra = payment.extra
+                )
+            )
+        }
    }

    private fun scheduleReconnect() {
@@ -8,21 +8,20 @@ import com.bitcointxoko.gudariwallet.data.NodeAliasRepository
 import com.bitcointxoko.gudariwallet.api.PaymentRecord
 import com.bitcointxoko.gudariwallet.data.PaymentCacheRepository
 import com.bitcointxoko.gudariwallet.data.WalletRepository
+import com.bitcointxoko.gudariwallet.service.WalletNotificationService
 import com.bitcointxoko.gudariwallet.util.WalletConstants
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext

 class HistoryViewModel(
    private val repo         : WalletRepository,
    private val aliasRepo    : NodeAliasRepository,
    private val paymentCache : PaymentCacheRepository,
-    val fiatCurrency           : StateFlow<String?>,   // passed from WalletViewModel.selectedCurrency
-    val fiatSatsPerUnit        : StateFlow<Double?>    // passed from WalletViewModel.fiatSatsPerUnit
+    val fiatCurrency         : StateFlow<String?>,
+    val fiatSatsPerUnit      : StateFlow<Double?>
 ) : ViewModel() {

    private val _state = MutableStateFlow<HistoryState>(HistoryState.Loading)
@@ -32,27 +31,126 @@ class HistoryViewModel(
    private var loadJob: Job? = null

    init {
-        // Cache read moved to IO dispatcher — never blocks the main thread,
-        // so navigation to HistoryScreen is instant on tap.
        viewModelScope.launch {
-            val cached = paymentCache.loadCachedPayments()
+            // Serve first page from DB immediately — no TTL, always valid
+            val cached = paymentCache.loadCachedPage(
+                offset = 0,
+                limit  = WalletConstants.PAYMENTS_PAGE_SIZE
+            )
            if (cached.isNotEmpty()) {
-                Log.d("HistoryViewModel", "PAYMENTS [INIT      ] Showing ${cached.size} cached payments immediately")
+                Log.d(TAG, "PAYMENTS [INIT      ] Showing ${cached.size} cached payments immediately")
+                currentOffset = cached.size
                _state.value = HistoryState.Success(
                    payments     = cached,
                    canLoadMore  = cached.size >= WalletConstants.PAYMENTS_PAGE_SIZE,
                    isRefreshing = true
                )
            } else {
-                Log.d("HistoryViewModel", "PAYMENTS [INIT      ] No cache — cold load")
+                Log.d(TAG, "PAYMENTS [INIT      ] No cache — cold load")
+            }
+            syncNewPayments()
+        }
+        // ── Live WebSocket updates ────────────────────────────────────────────────
+        // The payment is already written to the DB by WalletNotificationService
+        // before this event is emitted. syncNewPayments() will find it immediately
+        // via containsPayment() on the first page and stop after merging it.
+        viewModelScope.launch {
+            WalletNotificationService.paymentEvents.collect { event ->
+                Log.d(TAG, "PAYMENTS [WS EVENT  ] incoming event ${event.paymentHash} — refreshing list")
+                val current = _state.value
+                if (current is HistoryState.Success) {
+                    _state.value = current.copy(isRefreshing = true)
+                }
+                syncNewPayments()
            }
-            loadPage()
        }
-        // Note: no loadFiatRate() here — fiatSatsPerUnit is owned by WalletViewModel
-        // and already kept up to date. Zero extra network calls.
    }

-    // ── Payments list ─────────────────────────────────────────────────────────
+    // ── Sync: fetch pages from network until we hit a known checkingId ────────
+
+    /**
+     * Fetches pages from the server starting at offset 0, merging each into the
+     * DB, until a payment is found that already exists in the DB (meaning we've
+     * caught up), or the server returns a partial page (end of history).
+     *
+     * This fills any gap caused by payments that arrived while the app was
+     * offline, without re-fetching history we already have.
+     */
+    private suspend fun syncNewPayments(maxPayments: Int = WalletConstants.PAYMENTS_SYNC_CHECKPOINT) {
+        var syncOffset   = 0
+        var totalNew     = 0
+        var pagesChecked = 0
+
+        Log.d(TAG, "PAYMENTS [SYNC START] scanning for new payments (cap=$maxPayments)")
+
+        while (true) {
+            // ── Checkpoint guard ──────────────────────────────────────────────────
+            if (totalNew >= maxPayments) {
+                Log.d(TAG, "PAYMENTS [SYNC CAP  ] reached $maxPayments payment limit — stopping early")
+                break
+            }
+
+            val page = runCatching {
+                repo.getPayments(
+                    offset = syncOffset,
+                    limit  = WalletConstants.PAYMENTS_PAGE_SIZE
+                )
+            }.getOrElse { e ->
+                val current = _state.value
+                if (current is HistoryState.Success) {
+                    Log.w(TAG, "PAYMENTS [SYNC ERR  ] ${e.message} — keeping cached list")
+                    _state.value = current.copy(isRefreshing = false)
+                } else {
+                    Log.w(TAG, "PAYMENTS [SYNC ERR  ] ${e.message} — no cache to fall back on")
+                    _state.value = HistoryState.Error(e.message ?: "Failed to load payments")
+                }
+                return
+            }
+
+            pagesChecked++
+
+            val overlapIndex = page.indexOfFirst { paymentCache.containsPayment(it.checkingId) }
+
+            val newPayments = if (overlapIndex == -1) page
+            else page.subList(0, overlapIndex)
+
+            // Respect the cap — don't write more than allowed
+            val allowedNew  = maxPayments - totalNew
+            val toMerge     = if (newPayments.size > allowedNew) newPayments.subList(0, allowedNew)
+            else newPayments
+
+            if (toMerge.isNotEmpty()) {
+                paymentCache.mergePayments(toMerge)
+                totalNew += toMerge.size
+                Log.d(TAG, "PAYMENTS [SYNC PAGE ] page $pagesChecked — ${toMerge.size} new payments merged")
+            }
+
+            val reachedOverlap = overlapIndex != -1
+            val reachedEnd     = page.size < WalletConstants.PAYMENTS_PAGE_SIZE
+
+            if (reachedOverlap || reachedEnd) {
+                Log.d(TAG, "PAYMENTS [SYNC DONE ] $totalNew new payment(s) across $pagesChecked page(s) — " +
+                        if (reachedOverlap) "stopped at known record" else "reached end of history")
+                break
+            }
+
+            syncOffset += page.size
+        }
+
+        // Reload first page from DB to reflect any newly merged payments
+        val refreshed = paymentCache.loadCachedPage(
+            offset = 0,
+            limit  = WalletConstants.PAYMENTS_PAGE_SIZE
+        )
+        currentOffset = refreshed.size
+        _state.value = HistoryState.Success(
+            payments     = refreshed,
+            canLoadMore  = refreshed.size >= WalletConstants.PAYMENTS_PAGE_SIZE,
+            isRefreshing = false
+        )
+    }
+
+    // ── Pull-to-refresh ───────────────────────────────────────────────────────

    fun refresh() {
        loadJob?.cancel()
@@ -63,58 +161,50 @@ class HistoryViewModel(
        } else {
            _state.value = HistoryState.Loading
        }
-        loadPage()
+        viewModelScope.launch { syncNewPayments() }
    }

+    // ── Load more (pagination) ────────────────────────────────────────────────
+
    fun loadMore() {
        if (loadJob?.isActive == true) return
        val current = _state.value
-        if (current is HistoryState.Success && !current.canLoadMore) return
-        loadPage()
-    }
+        if (current !is HistoryState.Success || !current.canLoadMore) return

-    private fun loadPage() {
        loadJob = viewModelScope.launch {
-            runCatching {
-                repo.getPayments(
+            // Try DB first — if we have a full page cached, serve it instantly
+            val fromDb = paymentCache.loadCachedPage(
                offset = currentOffset,
                limit  = WalletConstants.PAYMENTS_PAGE_SIZE
            )
-            }.onSuccess { newPage ->
-                val canLoadMore  = newPage.size >= WalletConstants.PAYMENTS_PAGE_SIZE
-                val existing     = (_state.value as? HistoryState.Success)?.payments ?: emptyList()
-                val combined     = if (currentOffset == 0) newPage else existing + newPage
-                val deduplicated = combined.distinctBy { it.paymentHash }
-                currentOffset   += newPage.size
-
-                val previousCount = existing.size
-                val newCount      = deduplicated.size
-                if (currentOffset <= WalletConstants.PAYMENTS_PAGE_SIZE) {
-                    when {
-                        previousCount == 0      -> Log.d("HistoryViewModel", "PAYMENTS [NETWORK   ] Cold load — got ${newPage.size} payments")
-                        newCount > previousCount -> Log.d("HistoryViewModel", "PAYMENTS [NETWORK   ] ${newCount - previousCount} new payment(s) since cache")
-                        else                    -> Log.d("HistoryViewModel", "PAYMENTS [NETWORK   ] Cache was up to date — no new payments")
-                    }
-                }
-
+            if (fromDb.size == WalletConstants.PAYMENTS_PAGE_SIZE) {
+                Log.d(TAG, "PAYMENTS [MORE DB   ] offset=$currentOffset — served ${fromDb.size} rows from Room")
+                val deduplicated = (current.payments + fromDb).distinctBy { it.paymentHash }
+                currentOffset += fromDb.size
                _state.value = HistoryState.Success(
                    payments     = deduplicated,
-                    canLoadMore  = canLoadMore,
+                    canLoadMore  = true,
                    isRefreshing = false
                )
+                return@launch
+            }

-                if (currentOffset <= WalletConstants.PAYMENTS_PAGE_SIZE) {
-                    paymentCache.savePayments(deduplicated)
-                }
+            // Partial or empty DB page — fetch from network and merge
+            Log.d(TAG, "PAYMENTS [MORE NET  ] offset=$currentOffset — DB has ${fromDb.size}, fetching from network")
+            runCatching {
+                repo.getPayments(offset = currentOffset, limit = WalletConstants.PAYMENTS_PAGE_SIZE)
+            }.onSuccess { newPage ->
+                paymentCache.mergePayments(newPage)
+                val deduplicated = (current.payments + newPage).distinctBy { it.paymentHash }
+                currentOffset += newPage.size
+                _state.value = HistoryState.Success(
+                    payments     = deduplicated,
+                    canLoadMore  = newPage.size >= WalletConstants.PAYMENTS_PAGE_SIZE,
+                    isRefreshing = false
+                )
            }.onFailure { e ->
-                val current = _state.value
-                if (current is HistoryState.Success) {
-                    Log.w("HistoryViewModel", "PAYMENTS [NET ERROR ] Keeping cached list — ${e.message}")
+                Log.w(TAG, "PAYMENTS [MORE ERR  ] ${e.message}")
                _state.value = current.copy(isRefreshing = false)
-                } else {
-                    Log.w("HistoryViewModel", "PAYMENTS [NET ERROR ] No cache to fall back on — ${e.message}")
-                    _state.value = HistoryState.Error(e.message ?: "Failed to load payments")
-                }
            }
        }
    }
@@ -125,29 +215,41 @@ class HistoryViewModel(
    val detailState: StateFlow<DetailState> = _detailState

    fun loadDetail(checkingId: String, record: PaymentRecord? = null) {
+        // Tier 1: in-memory cache
        paymentCache.getCachedDetail(checkingId)?.let {
-            Log.d("HistoryViewModel", "PAYMENTS [DETAIL HIT] $checkingId — instant from cache")
+            Log.d(TAG, "PAYMENTS [DETAIL HIT] $checkingId — instant from cache")
            _detailState.value = DetailState.Success(it)
            return
        }

+        // Show partial UI immediately
        if (record != null) {
-            Log.d("HistoryViewModel", "PAYMENTS [DETAIL PAR] $checkingId — showing partial from PaymentRecord")
+            Log.d(TAG, "PAYMENTS [DETAIL PAR] $checkingId — showing partial from PaymentRecord")
            _detailState.value = DetailState.Partial(record)
        } else {
-            Log.d("HistoryViewModel", "PAYMENTS [DETAIL LOD] $checkingId — no record, showing spinner")
+            Log.d(TAG, "PAYMENTS [DETAIL LOD] $checkingId — no record, showing spinner")
            _detailState.value = DetailState.Loading
        }

        viewModelScope.launch {
+            // Tier 2: Room DB
+            val fromDb = paymentCache.getDetailFromDb(checkingId)
+            if (fromDb != null) {
+                paymentCache.saveDetail(checkingId, fromDb)
+                _detailState.value = DetailState.Success(fromDb)
+                Log.d(TAG, "PAYMENTS [DETAIL DB ] $checkingId — served from Room, skipping network")
+                return@launch
+            }
+
+            // Tier 3: network
            runCatching {
                repo.getPaymentDetail(checkingId)
            }.onSuccess { detail ->
-                Log.d("HistoryViewModel", "PAYMENTS [DETAIL NET] $checkingId — fetched from network")
+                Log.d(TAG, "PAYMENTS [DETAIL NET] $checkingId — fetched from network")
                paymentCache.saveDetail(checkingId, detail)
                _detailState.value = DetailState.Success(detail)
            }.onFailure { e ->
-                Log.w("HistoryViewModel", "PAYMENTS [DETAIL ERR] $checkingId — ${e.message}")
+                Log.w(TAG, "PAYMENTS [DETAIL ERR] $checkingId — ${e.message}")
                if (_detailState.value !is DetailState.Partial) {
                    _detailState.value = DetailState.Error(e.message ?: "Failed to load payment")
                }
@@ -155,11 +257,9 @@ class HistoryViewModel(
        }
    }

-    fun clearDetail() {
-        _detailState.value = DetailState.Idle
-    }
+    fun clearDetail() { _detailState.value = DetailState.Idle }

-    // ── Destination pubkey + alias resolution ─────────────────────────────────
+    // ── Destination resolution ────────────────────────────────────────────────

    val nodeAliases: StateFlow<Map<String, String>> = aliasRepo.aliases

@@ -176,8 +276,13 @@ class HistoryViewModel(
            if (pubkey != null) aliasRepo.resolve(pubkey)
        }
    }
+
+    companion object {
+        private const val TAG = "HistoryViewModel"
+    }
 }

+
 // ── Factory ───────────────────────────────────────────────────────────────────

 class HistoryViewModelFactory(
@@ -41,6 +41,9 @@ object WalletConstants {
 //    const val PAYMENTS_PREFS_KEY        = "payment_list"
 //    const val PAYMENTS_PREFS_SAVED_AT   = "payment_list_saved_at"
    const val PAYMENTS_CACHE_TTL_MS     = 300_000L      // 5 minutes — show stale, refresh in background
+    // ── Payments sync ─────────────────────────────────────────────────────────────
+    const val PAYMENTS_SYNC_CHECKPOINT  = 100   // max payments fetched on auto-sync at startup
+
    // ── Balance cache ─────────────────────────────────────────────────────────
    const val BALANCE_PREFS_NAME        = "balance_cache_prefs"
    const val BALANCE_PREFS_KEY         = "last_balance_sats"