mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-10 00:16:59 +00:00
refactor(onchain-zaps): extract verification coordinator from LocalCache
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.
This commit is contained in:
Claude
@@ -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<Note>? = 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<Note>,
|
||||
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<HexKey> = 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<HexKey> = 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<Long?> by lazy {
|
||||
val scope =
|
||||
runCatching { Amethyst.instance.applicationIOScope }.getOrNull()
|
||||
?: return@lazy MutableStateFlow<Long?>(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,
|
||||
|
||||
+295
@@ -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<HexKey> = 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<HexKey> = 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<Long?> by lazy {
|
||||
val scope =
|
||||
runCatching { Amethyst.instance.applicationIOScope }.getOrNull()
|
||||
?: return@lazy MutableStateFlow<Long?>(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<Note>,
|
||||
) {
|
||||
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<Note>,
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user