From daa83959b618f873b2eabec2ae4e6e79e1e28480 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 23 May 2026 16:13:19 +0000 Subject: [PATCH] refactor(onchain-zaps): extract verification coordinator from LocalCache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Moves the asynchronous chain-verification side of NIP-BC onchain zaps out of LocalCache into a dedicated OnchainZapResolver class living alongside other NIP-specific subpackages under model/nipBCOnchainZaps/. LocalCache shrinks by ~240 lines and now owns only the synchronous event-dispatch responsibility: loading the event, attaching the optimistic UNVERIFIED entry for the sender's own zap, and delegating the verifier launch to the resolver. The resolver owns: - launchVerification(event, source, repliesTo) — async fire-and-forget - reverifyOnchainZapsForNote(note) — used by the gallery's screen-driven loop - onchainTipHeightFlow — shared chain-tip poller, lazy + WhileSubscribed - verifyingEventIds / reverifyingNoteIds — in-flight de-duplication - reverifySemaphore — parallelism cap OnchainZapGallery now calls LocalCache.onchainZapResolver.{reverifyOnchainZaps ForNote, onchainTipHeightFlow} directly. consume(OnchainZapEvent) passes the already-computed repliesTo into launchVerification so the new-event path doesn't recompute it on the verifier side. No behavior change — all 22 onchain-zap tests still pass. --- .../amethyst/model/LocalCache.kt | 267 ++-------------- .../nipBCOnchainZaps/OnchainZapResolver.kt | 295 ++++++++++++++++++ .../amethyst/ui/note/OnchainZapGallery.kt | 8 +- 3 files changed, 320 insertions(+), 250 deletions(-) create mode 100644 amethyst/src/main/java/com/vitorpamplona/amethyst/model/nipBCOnchainZaps/OnchainZapResolver.kt diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt index 1163b08b42..e3ffb07729 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt @@ -42,6 +42,7 @@ import com.vitorpamplona.amethyst.commons.services.nwc.NwcPaymentTracker import com.vitorpamplona.amethyst.isDebug import com.vitorpamplona.amethyst.model.LocalCache.observeEvents import com.vitorpamplona.amethyst.model.nip51Lists.HiddenUsersState +import com.vitorpamplona.amethyst.model.nipBCOnchainZaps.OnchainZapResolver import com.vitorpamplona.amethyst.service.BundledInsert import com.vitorpamplona.amethyst.service.checkNotInMainThread import com.vitorpamplona.amethyst.ui.note.dateFormatter @@ -251,8 +252,6 @@ import com.vitorpamplona.quartz.nipACWebRtcCalls.events.CallRenegotiateEvent import com.vitorpamplona.quartz.nipB0WebBookmarks.WebBookmarkEvent import com.vitorpamplona.quartz.nipB7Blossom.BlossomServersEvent import com.vitorpamplona.quartz.nipBCOnchainZaps.chain.OnchainBackend -import com.vitorpamplona.quartz.nipBCOnchainZaps.verify.OnchainZapVerifier -import com.vitorpamplona.quartz.nipBCOnchainZaps.verify.VerifiedOnchainZap import com.vitorpamplona.quartz.nipBCOnchainZaps.zap.OnchainZapEvent import com.vitorpamplona.quartz.nipC0CodeSnippets.CodeSnippetEvent import com.vitorpamplona.quartz.nipC7Chats.ChatEvent @@ -263,32 +262,19 @@ import com.vitorpamplona.quartz.utils.TimeUtils import com.vitorpamplona.quartz.utils.cache.LargeCache import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.async -import kotlinx.coroutines.awaitAll import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.awaitClose -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.SharingStarted -import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asSharedFlow -import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.buffer import kotlinx.coroutines.flow.callbackFlow -import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.launch -import kotlinx.coroutines.supervisorScope -import kotlinx.coroutines.sync.Semaphore -import kotlinx.coroutines.sync.withPermit import java.io.File import java.io.FileOutputStream import java.io.IOException import java.util.SortedSet -import java.util.concurrent.ConcurrentHashMap interface ILocalCache { fun markAsSeen( @@ -299,13 +285,6 @@ interface ILocalCache { } } -/** - * Polling interval for the shared onchain chain-tip flow. Aligned with - * `CachingOnchainBackend.tipHeightTtlSeconds` (60s) — polling faster would - * just hit the cache and do no useful work. - */ -private const val ONCHAIN_TIP_POLL_INTERVAL_MS = 60_000L - object LocalCache : ILocalCache, ICacheProvider { val antiSpam = AntiSpamFilter() @@ -329,6 +308,15 @@ object LocalCache : ILocalCache, ICacheProvider { @Volatile var onchainBackend: OnchainBackend? = null + /** + * NIP-BC on-chain zap verification coordinator. Owns the chain-tip poller flow, + * the in-flight de-duplication of verifier calls across consume/reverify paths, + * and the parallelism cap. `consume(OnchainZapEvent)` delegates the async + * verification side here; the gallery's reverification driver also calls in here + * directly. + */ + val onchainZapResolver = OnchainZapResolver(this) + /** * Resolver for LNURL provider metadata used by [consume]`(LnZapEvent)` to * validate NIP-57 Appendix F. `null` skips the receipt-signer check (the @@ -1877,6 +1865,7 @@ object LocalCache : ILocalCache, ICacheProvider { // attachment; for incoming zaps from others, the sender-claimed `amount` tag // is untrusted and could mislead the viewer until the chain verifier responds. val isOwnEvent = relay == null + var repliesTo: List? = null if (!alreadyLoaded) { if (!(wasVerified || justVerify(event))) return false @@ -1886,8 +1875,9 @@ object LocalCache : ILocalCache, ICacheProvider { if (event.pubKey.equals(recipient, ignoreCase = true)) return false val author = getOrCreateUser(event.pubKey) - val repliesTo = computeReplyTo(event) - note.loadEvent(event, author, repliesTo) + val resolvedRepliesTo = computeReplyTo(event) + repliesTo = resolvedRepliesTo + note.loadEvent(event, author, resolvedRepliesTo) if (isOwnEvent) { // Optimistic attachment for the sender's own zap: surface it on the @@ -1902,7 +1892,7 @@ object LocalCache : ILocalCache, ICacheProvider { // via toLongOrNull() with no sign check, so a malicious sender // could otherwise put "-1" in the gallery as a negative-sats badge. val claimedSats = (event.claimedAmountInSats() ?: 0L).coerceAtLeast(0L) - repliesTo.forEach { + resolvedRepliesTo.forEach { it.addOnchainZap(note, txid, claimedSats, verifiedSats = 0L, OnchainZapStatus.UNVERIFIED) } } @@ -1911,232 +1901,15 @@ object LocalCache : ILocalCache, ICacheProvider { refreshNewNoteObservers(note) } - // Verification needs a chain backend. Without one (e.g. before AppModules - // wires its EsploraBackend) the event stays cached so subscriptions and - // profile zap views see it, but it can't contribute to Note totals. - val backend = onchainBackend ?: return !alreadyLoaded - - // Skip the verifier launch entirely once the chain has spoken definitively - // (Confirmed, or hard-rejected with a non-transient reason). The resolved - // flag is per-source-event and travels with the Note, so it survives across - // relay echoes and even covers profile-only zaps (which would otherwise - // bypass any per-target CONFIRMED check because they have no replyTo notes). - if (note.onchainZapResolved) return !alreadyLoaded - - // De-duplicate concurrent launches. `verifyingEventIds.add` returns false if - // another coroutine has already started verifying this exact event id — - // covers (a) two relays delivering the same event simultaneously, - // (b) a relay echo arriving while the prior verifier is still in flight, - // (c) `reverifyOnchainZapsForNote` racing with `consume`. - if (!verifyingEventIds.add(event.id)) return !alreadyLoaded - - val verifier = OnchainZapVerifier(backend) - val repliesTo = computeReplyTo(event) - - Amethyst.instance.applicationIOScope.launch { - try { - verifyAndUpgradeOnchainZap(event, note, repliesTo, verifier) - } finally { - verifyingEventIds.remove(event.id) - } - } + // Async chain verification is delegated to OnchainZapResolver, which owns the + // in-flight gates (so two relay echoes don't double-fetch) and the chain-tip + // polling flow used by the gallery driver. Reusing the repliesTo already + // computed above avoids the second computeReplyTo pass on the new-event path. + onchainZapResolver.launchVerification(event, note, repliesTo ?: computeReplyTo(event)) return !alreadyLoaded } - /** - * Run the chain verifier for a single onchain zap event and apply the result to every - * target note. Designed to be safe to call repeatedly — the [Note] entries upgrade - * monotonically (UNVERIFIED → PENDING → CONFIRMED) and won't move backwards, and - * source-scoped removal prevents one event's rejection from erasing another sender's - * legitimate entry that happens to share a txid. - * - * Callers must wrap the call site with a `verifyingEventIds`/`reverifyingNoteIds` - * gate; this function does not itself protect against duplicate concurrent runs. - */ - private suspend fun verifyAndUpgradeOnchainZap( - event: OnchainZapEvent, - source: Note, - repliesTo: List, - verifier: OnchainZapVerifier, - ) { - // Clamp claimedSats to non-negative; see consume() for rationale. - val claimedSats = (event.claimedAmountInSats() ?: 0L).coerceAtLeast(0L) - try { - when (val result = verifier.verify(event)) { - is VerifiedOnchainZap.Confirmed -> { - repliesTo.forEach { - it.addOnchainZap(source, result.txid, claimedSats, result.verifiedSats, OnchainZapStatus.CONFIRMED) - } - // Terminal — the chain has confirmed. Future relay echoes of this - // event id skip the verifier entirely via the `onchainZapResolved` - // gate in `consume()`. - source.onchainZapResolved = true - } - - is VerifiedOnchainZap.Pending -> { - repliesTo.forEach { - it.addOnchainZap(source, result.txid, claimedSats, result.verifiedSats, OnchainZapStatus.PENDING) - } - // Not terminal — the tx is in the mempool. A future tip-poll or - // reverify call can upgrade it to CONFIRMED, so leave the - // resolved flag false. - } - - is VerifiedOnchainZap.Rejected -> { - if (result.reason == VerifiedOnchainZap.Rejected.Reason.TX_NOT_FOUND) { - // Transient — the tx may not have propagated to the backend's - // indexer yet. Leave the entry as UNVERIFIED so a later - // reverifyOnchainZapsForNote() call (e.g. on chain tip change) - // can promote it. - Log.d("OnchainZap") { "tx not yet indexed for ${event.id} (${result.txid}); will retry" } - } else if (result.txid.isNotEmpty()) { - // Hard rejection — e.g. ZERO_VERIFIED_AMOUNT means this sender's - // tx did not pay the recipient. Drop only entries whose source - // matches THIS event AND that aren't CONFIRMED (the per-target - // CONFIRMED check inside `removeOnchainZapForSource` prevents a - // transient backend hiccup from wiping a previously-verified - // entry on a sibling target). - Log.d("OnchainZap") { "rejected ${result.txid}: ${result.reason}" } - repliesTo.forEach { it.removeOnchainZapForSource(result.txid, event.pubKey) } - // Terminal — don't re-verify on future relay echoes of this - // event id. (Different event ids with the same txid still go - // through their own verifier pass.) - source.onchainZapResolved = true - } else { - // MISSING_TXID etc. — log so we have observability when a - // malformed event reaches the backend. Mark terminal so we - // don't re-verify the same broken event on every echo. - Log.d("OnchainZap") { "rejected ${event.id}: ${result.reason} (no txid)" } - source.onchainZapResolved = true - } - } - } - } catch (t: Throwable) { - // Never swallow cancellation — it must propagate so screen-scoped callers - // (the gallery's reverification driver) can tear down cleanly. - if (t is CancellationException) throw t - Log.w("OnchainZap", "verification failed for ${event.id}", t) - } - } - - /** - * Re-run onchain-zap verification for every non-CONFIRMED entry attached to [note]. - * Safe to call from a screen-visibility hook or a chain-tip change observer; each - * verifier call is bounded by the chain backend's cache TTLs. - * - * - Per-note in-flight gate (`reverifyingNoteIds`) — if another caller is already - * reverifying this note (e.g. multiple visible galleries fired in the same tip - * tick) we skip the dup. - * - Per-event in-flight gate (`verifyingEventIds`) — coordinates with `consume()` - * so the same event isn't verified twice concurrently. - * - `supervisorScope` — a single verifier failure won't cancel sibling verifiers - * for other entries on the same note. - * - Bounded parallelism via [reverifySemaphore] so a thread with many pending - * entries doesn't blast Esplora. - */ - suspend fun reverifyOnchainZapsForNote(note: Note) { - val backend = onchainBackend ?: return - if (!reverifyingNoteIds.add(note.idHex)) return - try { - val pendingEntries = - note.onchainZaps.values.filter { it.status != OnchainZapStatus.CONFIRMED } - if (pendingEntries.isEmpty()) return - - val verifier = OnchainZapVerifier(backend) - supervisorScope { - pendingEntries - .mapNotNull { entry -> - val sourceEvent = entry.source.event as? OnchainZapEvent ?: return@mapNotNull null - val source = entry.source - if (source.onchainZapResolved) return@mapNotNull null - if (!verifyingEventIds.add(sourceEvent.id)) return@mapNotNull null - async { - try { - reverifySemaphore.withPermit { - val repliesTo = computeReplyTo(sourceEvent) - verifyAndUpgradeOnchainZap(sourceEvent, source, repliesTo, verifier) - } - } finally { - verifyingEventIds.remove(sourceEvent.id) - } - } - }.awaitAll() - } - } finally { - reverifyingNoteIds.remove(note.idHex) - } - } - - /** - * In-flight set of event ids currently being verified. Prevents two consume() - * calls (or a consume + a reverify) from issuing parallel Esplora fetches for - * the same event. `ConcurrentHashMap.newKeySet` gives lock-free atomic `add` - * returning `true` only for the inserting caller. - */ - private val verifyingEventIds: MutableSet = ConcurrentHashMap.newKeySet() - - /** - * In-flight set of note id strings currently being reverified. Lets the - * onchain-zap gallery driver be called from many visible composables without - * the same note's reverify pass running concurrently. The per-event gate is - * the backstop; this one short-circuits earlier and avoids creating async - * coroutines that would just no-op. - */ - private val reverifyingNoteIds: MutableSet = ConcurrentHashMap.newKeySet() - - /** - * Caps the parallelism of [reverifyOnchainZapsForNote] so a thread with many - * pending entries doesn't blast public Esplora endpoints with a burst of - * simultaneous requests. Tuned a little higher than the previous 4 so that - * concurrent galleries each running a small reverify aren't fully serialized - * behind a single big one. - */ - private val reverifySemaphore = Semaphore(permits = 8) - - /** - * Shared poller for the current bitcoin chain tip height. Each gallery that - * holds non-CONFIRMED onchain zaps subscribes to this flow and re-verifies its - * entries whenever the tip advances. Lazy + `WhileSubscribed` so the HTTP call - * only fires when at least one UI surface needs it. - * - * Lazy initialization is required because [Amethyst.instance] may not exist - * when the `LocalCache` singleton is class-loaded. Falls back to a constant - * null-emitting StateFlow if the application scope isn't available yet - * (e.g. unit tests, ContentProvider invocations), so the lazy field doesn't - * permanently fail with `UninitializedPropertyAccessException`. - */ - val onchainTipHeightFlow: StateFlow by lazy { - val scope = - runCatching { Amethyst.instance.applicationIOScope }.getOrNull() - ?: return@lazy MutableStateFlow(null).asStateFlow() - - flow { - while (true) { - val tip = - onchainBackend?.let { backend -> - // Explicit try/catch instead of `runCatching` because the latter - // would swallow CancellationException too — when the upstream - // scope cancels, we must let it propagate so the flow tears down - // promptly instead of looping through one more delay(). - try { - backend.tipHeight() - } catch (e: CancellationException) { - throw e - } catch (t: Throwable) { - null - } - } - emit(tip) - delay(ONCHAIN_TIP_POLL_INTERVAL_MS) - } - }.stateIn( - scope, - SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000L), - initialValue = null, - ) - } - private fun attachZapToLiveActivityChannel( event: LnZapEvent, note: Note, diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nipBCOnchainZaps/OnchainZapResolver.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nipBCOnchainZaps/OnchainZapResolver.kt new file mode 100644 index 0000000000..b22c441921 --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/nipBCOnchainZaps/OnchainZapResolver.kt @@ -0,0 +1,295 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.model.nipBCOnchainZaps + +import com.vitorpamplona.amethyst.Amethyst +import com.vitorpamplona.amethyst.commons.model.Note +import com.vitorpamplona.amethyst.commons.model.OnchainZapStatus +import com.vitorpamplona.amethyst.model.LocalCache +import com.vitorpamplona.quartz.nip01Core.core.HexKey +import com.vitorpamplona.quartz.nipBCOnchainZaps.verify.OnchainZapVerifier +import com.vitorpamplona.quartz.nipBCOnchainZaps.verify.VerifiedOnchainZap +import com.vitorpamplona.quartz.nipBCOnchainZaps.zap.OnchainZapEvent +import com.vitorpamplona.quartz.utils.Log +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharingStarted +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.stateIn +import kotlinx.coroutines.launch +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit +import java.util.concurrent.ConcurrentHashMap + +/** + * Coordinates NIP-BC on-chain zap verification on top of the chain backend. + * + * `LocalCache.consume(OnchainZapEvent)` owns the event dispatch and optimistic + * gallery attachment; this class owns the asynchronous chain-verification side: + * + * - Launching a verifier coroutine when a new event arrives (with per-event + * in-flight de-duplication so simultaneous relay echoes don't double-fetch). + * - Re-running verification for visible notes when the chain tip advances or + * when a screen first comes into view. + * - Owning the shared chain-tip poller flow that UI subscribes to. + * + * Stateless from the caller's perspective — the per-event resolution state lives + * on the source [Note] itself (`onchainZapResolved`), so it travels with the Note + * across cache eviction and doesn't accumulate here. + */ +class OnchainZapResolver( + private val cache: LocalCache, +) { + /** + * Caps the parallelism of [reverifyOnchainZapsForNote] so a thread with many + * pending entries doesn't blast public Esplora endpoints with a burst of + * simultaneous requests. + */ + private val reverifySemaphore = Semaphore(permits = 8) + + /** + * In-flight set of event ids currently being verified. Prevents two `consume()` + * calls (or a `consume` plus a reverify) from issuing parallel Esplora fetches + * for the same event. `ConcurrentHashMap.newKeySet` gives lock-free atomic `add` + * returning `true` only for the inserting caller. + */ + private val verifyingEventIds: MutableSet = ConcurrentHashMap.newKeySet() + + /** + * In-flight set of note id strings currently being reverified. Lets the on-chain + * zap gallery driver be called from many visible composables without the same + * note's reverify pass running concurrently. The per-event gate above is the + * backstop; this one short-circuits earlier and avoids creating async coroutines + * that would just no-op. + */ + private val reverifyingNoteIds: MutableSet = ConcurrentHashMap.newKeySet() + + /** + * Shared poller for the current bitcoin chain tip height. Each gallery that + * holds non-CONFIRMED on-chain zaps subscribes to this flow and re-verifies its + * entries whenever the tip advances. Lazy + `WhileSubscribed` so the HTTP call + * only fires when at least one UI surface needs it. + * + * Lazy initialization is required because [Amethyst.instance] may not exist + * when [OnchainZapResolver] is first constructed. Falls back to a constant + * null-emitting [StateFlow] if the application scope isn't available yet + * (e.g. unit tests, ContentProvider invocations), so the lazy field doesn't + * permanently fail with `UninitializedPropertyAccessException`. + */ + val onchainTipHeightFlow: StateFlow by lazy { + val scope = + runCatching { Amethyst.instance.applicationIOScope }.getOrNull() + ?: return@lazy MutableStateFlow(null).asStateFlow() + + flow { + while (true) { + val tip = + cache.onchainBackend?.let { backend -> + // Explicit try/catch instead of `runCatching` because the latter + // would swallow CancellationException too — when the upstream + // scope cancels, we must let it propagate so the flow tears down + // promptly instead of looping through one more delay(). + try { + backend.tipHeight() + } catch (e: CancellationException) { + throw e + } catch (t: Throwable) { + null + } + } + emit(tip) + delay(TIP_POLL_INTERVAL_MS) + } + }.stateIn( + scope, + SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000L), + initialValue = null, + ) + } + + /** + * Launches an asynchronous chain verification for [event] unless one is already + * in flight or [source] has already reached a terminal verdict. Returns + * immediately. Apply-to-Note updates happen on the application IO scope. + * + * [repliesTo] is the pre-computed list of notes the event references so consume() + * doesn't pay the cost of resolving it twice. + */ + fun launchVerification( + event: OnchainZapEvent, + source: Note, + repliesTo: List, + ) { + val backend = cache.onchainBackend ?: return + if (source.onchainZapResolved) return + if (!verifyingEventIds.add(event.id)) return + + val verifier = OnchainZapVerifier(backend) + + Amethyst.instance.applicationIOScope.launch { + try { + verifyAndUpgradeOnchainZap(event, source, repliesTo, verifier) + } finally { + verifyingEventIds.remove(event.id) + } + } + } + + /** + * Re-run on-chain zap verification for every non-CONFIRMED entry attached to + * [note]. Safe to call from a screen-visibility hook or a chain-tip change + * observer; each verifier call is bounded by the chain backend's cache TTLs. + * + * - Per-note in-flight gate ([reverifyingNoteIds]) — if another caller is + * already reverifying this note (e.g. multiple visible galleries fired in + * the same tip tick) we skip the dup. + * - Per-event in-flight gate ([verifyingEventIds]) — coordinates with + * [launchVerification] so the same event isn't verified twice concurrently. + * - [supervisorScope] — a single verifier failure won't cancel sibling + * verifiers for other entries on the same note. + * - Bounded parallelism via [reverifySemaphore] so a thread with many pending + * entries doesn't blast Esplora. + */ + suspend fun reverifyOnchainZapsForNote(note: Note) { + val backend = cache.onchainBackend ?: return + if (!reverifyingNoteIds.add(note.idHex)) return + try { + val pendingEntries = + note.onchainZaps.values.filter { it.status != OnchainZapStatus.CONFIRMED } + if (pendingEntries.isEmpty()) return + + val verifier = OnchainZapVerifier(backend) + supervisorScope { + pendingEntries + .mapNotNull { entry -> + val sourceEvent = entry.source.event as? OnchainZapEvent ?: return@mapNotNull null + val source = entry.source + if (source.onchainZapResolved) return@mapNotNull null + if (!verifyingEventIds.add(sourceEvent.id)) return@mapNotNull null + async { + try { + reverifySemaphore.withPermit { + val repliesTo = cache.computeReplyTo(sourceEvent) + verifyAndUpgradeOnchainZap(sourceEvent, source, repliesTo, verifier) + } + } finally { + verifyingEventIds.remove(sourceEvent.id) + } + } + }.awaitAll() + } + } finally { + reverifyingNoteIds.remove(note.idHex) + } + } + + /** + * Run the chain verifier for a single on-chain zap event and apply the result to + * every target note. Designed to be safe to call repeatedly — the [Note] entries + * upgrade monotonically (UNVERIFIED → PENDING → CONFIRMED) and won't move + * backwards, and source-scoped removal prevents one event's rejection from + * erasing another sender's legitimate entry that happens to share a txid. + * + * Callers must wrap the call site with the [verifyingEventIds] gate; this + * function does not itself protect against duplicate concurrent runs. + */ + private suspend fun verifyAndUpgradeOnchainZap( + event: OnchainZapEvent, + source: Note, + repliesTo: List, + verifier: OnchainZapVerifier, + ) { + // Clamp claimedSats to non-negative — `amount` tag parses via toLongOrNull() + // with no sign check, so a malicious sender could otherwise put "-1" in the + // gallery as a negative-sats badge. + val claimedSats = (event.claimedAmountInSats() ?: 0L).coerceAtLeast(0L) + try { + when (val result = verifier.verify(event)) { + is VerifiedOnchainZap.Confirmed -> { + repliesTo.forEach { + it.addOnchainZap(source, result.txid, claimedSats, result.verifiedSats, OnchainZapStatus.CONFIRMED) + } + // Terminal — the chain has confirmed. Future relay echoes of this + // event id skip the verifier entirely via the `onchainZapResolved` + // gate in `launchVerification`. + source.onchainZapResolved = true + } + + is VerifiedOnchainZap.Pending -> { + repliesTo.forEach { + it.addOnchainZap(source, result.txid, claimedSats, result.verifiedSats, OnchainZapStatus.PENDING) + } + // Not terminal — the tx is in the mempool. A future tip-poll or + // reverify call can upgrade it to CONFIRMED. + } + + is VerifiedOnchainZap.Rejected -> { + if (result.reason == VerifiedOnchainZap.Rejected.Reason.TX_NOT_FOUND) { + // Transient — the tx may not have propagated to the backend's + // indexer yet. Leave the entry as UNVERIFIED so a later + // reverifyOnchainZapsForNote() call (e.g. on chain tip change) + // can promote it. + Log.d("OnchainZap") { "tx not yet indexed for ${event.id} (${result.txid}); will retry" } + } else if (result.txid.isNotEmpty()) { + // Hard rejection — e.g. ZERO_VERIFIED_AMOUNT means this sender's + // tx did not pay the recipient. Drop only entries whose source + // matches THIS event AND that aren't CONFIRMED (the per-target + // CONFIRMED check inside `removeOnchainZapForSource` prevents a + // transient backend hiccup from wiping a previously-verified + // entry on a sibling target). + Log.d("OnchainZap") { "rejected ${result.txid}: ${result.reason}" } + repliesTo.forEach { it.removeOnchainZapForSource(result.txid, event.pubKey) } + // Terminal — don't re-verify on future relay echoes of this + // event id. (Different event ids with the same txid still go + // through their own verifier pass.) + source.onchainZapResolved = true + } else { + // MISSING_TXID etc. — log so we have observability when a + // malformed event reaches the backend. Mark terminal so we + // don't re-verify the same broken event on every echo. + Log.d("OnchainZap") { "rejected ${event.id}: ${result.reason} (no txid)" } + source.onchainZapResolved = true + } + } + } + } catch (t: Throwable) { + // Never swallow cancellation — it must propagate so screen-scoped callers + // (the gallery's reverification driver) can tear down cleanly. + if (t is CancellationException) throw t + Log.w("OnchainZap", "verification failed for ${event.id}", t) + } + } + + companion object { + /** + * Polling interval for the shared on-chain chain-tip flow. Aligned with + * `CachingOnchainBackend.tipHeightTtlSeconds` (60s) — polling faster would + * just hit the cache and do no useful work. + */ + private const val TIP_POLL_INTERVAL_MS = 60_000L + } +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/OnchainZapGallery.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/OnchainZapGallery.kt index 98b67d0a32..5ae2b6c56e 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/OnchainZapGallery.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/note/OnchainZapGallery.kt @@ -111,22 +111,24 @@ private fun DriveOnchainZapReverification( val pendingCount = remember(entries) { entries.count { it.status != OnchainZapStatus.CONFIRMED } } if (pendingCount == 0) return + val resolver = LocalCache.onchainZapResolver + // First-view kick — unconditional, doesn't wait for the tip flow. Keyed on // (idHex, pendingCount) so a brand-new pending entry arriving while the // gallery is still on screen also kicks an immediate reverify instead of // waiting up to a full tip-poll interval. LaunchedEffect(note.idHex, pendingCount) { - LocalCache.reverifyOnchainZapsForNote(note) + resolver.reverifyOnchainZapsForNote(note) } // Tip-change kick — subscribes to the shared poller (lazy, WhileSubscribed // so only one HTTP poller runs across the whole UI no matter how many // galleries are visible). Skips the first emission (null) to avoid // duplicating the first-view kick above. - val tip by LocalCache.onchainTipHeightFlow.collectAsStateWithLifecycle() + val tip by resolver.onchainTipHeightFlow.collectAsStateWithLifecycle() LaunchedEffect(note.idHex, tip) { if (tip != null) { - LocalCache.reverifyOnchainZapsForNote(note) + resolver.reverifyOnchainZapsForNote(note) } } }