fix: require accepted onchain broadcasts

app/src/main/java/to/bitkit/repositories/LightningRepo.kt

@@ -42,6 +42,7 @@ import org.lightningdevkit.ldknode.Address
 import org.lightningdevkit.ldknode.BalanceDetails
 import org.lightningdevkit.ldknode.BestBlock
 import org.lightningdevkit.ldknode.Bolt11Invoice
+import org.lightningdevkit.ldknode.BroadcastOutcome
 import org.lightningdevkit.ldknode.ChannelConfig
 import org.lightningdevkit.ldknode.ChannelDataMigration
 import org.lightningdevkit.ldknode.ChannelDetails
@@ -1449,6 +1450,16 @@ class LightningRepo @Inject constructor(
         Result.success(payments)
     }
 
+    suspend fun getOnchainBroadcastOutcome(txid: Txid): Result<BroadcastOutcome?> =
+        executeWhenNodeRunning("getOnchainBroadcastOutcome") {
+            runSuspendCatching { lightningService.getOnchainBroadcastOutcome(txid) }
+        }
+
+    suspend fun acknowledgeOnchainBroadcastOutcome(txid: Txid): Result<Unit> =
+        executeWhenNodeRunning("acknowledgeOnchainBroadcastOutcome") {
+            runSuspendCatching { lightningService.acknowledgeOnchainBroadcastOutcome(txid) }
+        }
+
     suspend fun getAddressBalance(address: String): Result<ULong> = executeWhenNodeRunning("getAddressBalance") {
         runCatching {
             lightningService.getAddressBalance(address)

app/src/main/java/to/bitkit/repositories/PaykitPaymentProofRepo.kt

@@ -12,6 +12,8 @@ import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.contentOrNull
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
+import org.lightningdevkit.ldknode.BroadcastOutcome
+import org.lightningdevkit.ldknode.BroadcastOutcomeStatus
 import org.lightningdevkit.ldknode.PaymentDetails
 import org.lightningdevkit.ldknode.PaymentDirection
 import org.lightningdevkit.ldknode.PaymentKind
@@ -24,6 +26,7 @@ import to.bitkit.models.PubkyPublicKeyFormat
 import to.bitkit.services.PaykitSdkService
 import to.bitkit.utils.Logger
 import java.security.MessageDigest
+import java.util.UUID
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -41,6 +44,10 @@ data class PendingPaykitPaymentProof(
     val kind: PaykitPaymentProofKind,
     val paymentIdentifier: String? = null,
     val proofData: String? = null,
+    val preparationId: String? = null,
+    val requiresBroadcastOutcome: Boolean = false,
+    val broadcastLineage: List<String> = emptyList(),
+    val broadcastOutcomeHandled: Boolean = false,
 )
 
 @Singleton
@@ -61,19 +68,12 @@ class PaykitPaymentProofRepo @Inject constructor(
         request: PaykitPaymentRequest,
         paymentEndpointIdentifier: String,
         kind: PaykitPaymentProofKind,
-    ): Result<Unit> = withContext(ioDispatcher) {
+    ): Result<String> = withContext(ioDispatcher) {
         runSuspendCatching {
             operationMutex.withLock {
                 val proof = pendingProof(request, paymentEndpointIdentifier, kind)
-                val proofs = loadProofs()
-                    .filterNot {
-                        PubkyPublicKeyFormat.matches(it.identity, proof.identity) &&
-                            it.requestId == request.id &&
-                            it.paymentIdentifier == null &&
-                            it.proofData == null
-                    } +
-                    proof
-                persist(proofs)
+                persist(loadProofs() + proof)
+                requireNotNull(proof.preparationId)
             }
         }.onFailure { Logger.warn("Failed to prepare a Paykit payment proof", it, context = TAG) }
     }
@@ -105,6 +105,50 @@ class PaykitPaymentProofRepo @Inject constructor(
         }.onFailure { Logger.warn("Failed to associate a Paykit Lightning payment proof", it, context = TAG) }
     }
 
+    suspend fun associateOnchainPayment(
+        request: PaykitPaymentRequest,
+        txid: String,
+        paymentEndpointIdentifier: String,
+        preparationId: String?,
+    ): Result<Unit> = withContext(ioDispatcher) {
+        runSuspendCatching {
+            if (!txid.isHex(HASH_BYTE_COUNT)) throw PaykitPaymentRequestError.RequestUnavailable
+            operationMutex.withLock {
+                val proofs = loadProofs().toMutableList()
+                val index = proofs.indexOfLast {
+                    if (preparationId != null) {
+                        it.preparationId == preparationId &&
+                            it.requestId == request.id &&
+                            it.kind == PaykitPaymentProofKind.Onchain &&
+                            it.paymentIdentifier == null &&
+                            it.proofData == null
+                    } else {
+                        it.requestId == request.id &&
+                            it.kind == PaykitPaymentProofKind.Onchain &&
+                            it.paymentIdentifier == null &&
+                            it.proofData == null
+                    }
+                }
+                val proof = if (index >= 0) {
+                    proofs[index].copy(
+                        paymentIdentifier = txid.lowercase(),
+                        requiresBroadcastOutcome = true,
+                        broadcastLineage = listOf(txid.lowercase()),
+                    )
+                } else {
+                    pendingProof(request, paymentEndpointIdentifier, PaykitPaymentProofKind.Onchain)
+                        .copy(
+                            paymentIdentifier = txid.lowercase(),
+                            requiresBroadcastOutcome = true,
+                            broadcastLineage = listOf(txid.lowercase()),
+                        )
+                }
+                if (index >= 0) proofs[index] = proof else proofs += proof
+                persist(proofs)
+            }
+        }.onFailure { Logger.warn("Failed to associate a Paykit on-chain payment proof", it, context = TAG) }
+    }
+
     suspend fun completeLightningPayment(paymentHash: String, preimage: String?) = withContext(ioDispatcher) {
         if (preimage == null) return@withContext
         if (!preimage.matchesPaymentHash(paymentHash)) {
@@ -195,8 +239,8 @@ class PaykitPaymentProofRepo @Inject constructor(
         it.kind == PaykitPaymentProofKind.Lightning && it.paymentIdentifier.equals(paymentHash, ignoreCase = true)
     }
 
-    suspend fun cancelPreparation(request: PaykitPaymentRequest) = removeProofs {
-        it.requestId == request.id && it.paymentIdentifier == null && it.proofData == null
+    suspend fun cancelPreparation(preparationId: String) = removeProofs {
+        it.preparationId == preparationId && it.paymentIdentifier == null && it.proofData == null
     }
 
     suspend fun reconcile() = withContext(ioDispatcher) {
@@ -238,6 +282,17 @@ class PaykitPaymentProofRepo @Inject constructor(
         proof: PendingPaykitPaymentProof,
         payments: List<PaymentDetails>,
     ) {
+        if (proof.broadcastOutcomeHandled) {
+            acknowledgeHandledBroadcastOutcome(proof)
+            return
+        }
+        if (
+            proof.kind == PaykitPaymentProofKind.Onchain &&
+            (proof.requiresBroadcastOutcome || proof.proofData == null)
+        ) {
+            reconcileOnchainProof(proof)
+            return
+        }
         if (proof.proofData != null) {
             submitReady(proof)
             return
@@ -268,21 +323,87 @@ class PaykitPaymentProofRepo @Inject constructor(
         }
     }
 
+    private suspend fun reconcileOnchainProof(proof: PendingPaykitPaymentProof) {
+        val txid = proof.paymentIdentifier ?: return
+        val outcome = lightningRepo.getOnchainBroadcastOutcome(txid).getOrThrow() ?: return
+        when (outcome.status) {
+            BroadcastOutcomeStatus.PENDING -> persistBroadcastOutcome(proof, outcome, proofData = null)
+            BroadcastOutcomeStatus.ACCEPTED -> handleAcceptedBroadcastOutcome(proof, outcome)
+            BroadcastOutcomeStatus.ABANDONED -> handleAbandonedBroadcastOutcome(proof, outcome)
+        }
+    }
+
+    private suspend fun handleAcceptedBroadcastOutcome(
+        proof: PendingPaykitPaymentProof,
+        outcome: BroadcastOutcome,
+    ) {
+        val completed = persistBroadcastOutcome(proof, outcome, proofData = outcome.txid) ?: return
+        if (!queueReadyProof(completed)) return
+        val handled = completed.copy(
+            proofData = null,
+            broadcastOutcomeHandled = true,
+        )
+        persistHandledBroadcastOutcome(handled)
+        acknowledgeHandledBroadcastOutcome(handled)
+    }
+
+    private suspend fun handleAbandonedBroadcastOutcome(
+        proof: PendingPaykitPaymentProof,
+        outcome: BroadcastOutcome,
+    ) {
+        val canonical = persistBroadcastOutcome(proof, outcome, proofData = null) ?: return
+        val handled = canonical.copy(broadcastOutcomeHandled = true)
+        persistHandledBroadcastOutcome(handled)
+        acknowledgeHandledBroadcastOutcome(handled)
+    }
+
+    private suspend fun persistBroadcastOutcome(
+        proof: PendingPaykitPaymentProof,
+        outcome: BroadcastOutcome,
+        proofData: String?,
+    ): PendingPaykitPaymentProof? {
+        val updated = proof.copy(
+            paymentIdentifier = outcome.txid.lowercase(),
+            proofData = proofData?.lowercase(),
+            requiresBroadcastOutcome = true,
+            broadcastLineage = outcome.lineage.map { it.lowercase() },
+        )
+        if (!replaceProofLocked(proof, updated)) return null
+        return updated
+    }
+
+    private suspend fun persistHandledBroadcastOutcome(proof: PendingPaykitPaymentProof) {
+        val proofs = loadProofs().filterNot { it.matchesRequest(proof) } + proof
+        persist(proofs)
+    }
+
+    private suspend fun acknowledgeHandledBroadcastOutcome(proof: PendingPaykitPaymentProof) {
+        val txid = proof.paymentIdentifier ?: return
+        lightningRepo.acknowledgeOnchainBroadcastOutcome(txid).getOrThrow()
+        removeProofsLocked { it == proof }
+    }
+
     private suspend fun submitReady(proof: PendingPaykitPaymentProof) {
-        val proofData = proof.proofData ?: return
+        if (!queueReadyProof(proof)) return
+        runSuspendCatching { removeProofsLocked { it.matchesRequest(proof) } }
+            .onFailure { Logger.warn("Failed to clear a submitted Paykit payment proof", it, context = TAG) }
+    }
+
+    private suspend fun queueReadyProof(proof: PendingPaykitPaymentProof): Boolean {
+        val proofData = proof.proofData ?: return false
         val identityStatus = paykitSdkService.identityStatus()
         if (
             identityStatus?.liveSessionAvailable != true ||
             !PubkyPublicKeyFormat.matches(identityStatus.publicKey, proof.identity)
         ) {
-            return
+            return false
         }
 
         val record = paykitSdkService.paymentRequests().firstOrNull {
             it.paymentRequestId == proof.requestId.paymentRequestId &&
                 PubkyPublicKeyFormat.matches(it.counterparty, proof.requestId.counterparty) &&
                 it.counterpartyReceiverPath == proof.requestId.counterpartyReceiverPath
-        } ?: return
+        } ?: return false
         val proofJson = proofJson(proof.kind, proofData)
         val alreadyQueued = record.paymentProofs.any {
             it.billingPeriod == null &&
@@ -307,11 +428,7 @@ class PaykitPaymentProofRepo @Inject constructor(
                     )
                 }
         }
-        runSuspendCatching {
-            removeProofsLocked {
-                PubkyPublicKeyFormat.matches(it.identity, proof.identity) && it.requestId == proof.requestId
-            }
-        }.onFailure { Logger.warn("Failed to clear a submitted Paykit payment proof", it, context = TAG) }
+        return true
     }
 
     private suspend fun removeProofs(predicate: (PendingPaykitPaymentProof) -> Boolean) = withContext(ioDispatcher) {
@@ -327,6 +444,18 @@ class PaykitPaymentProofRepo @Inject constructor(
         if (remaining != current) persist(remaining)
     }
 
+    private suspend fun replaceProofLocked(
+        currentProof: PendingPaykitPaymentProof,
+        updatedProof: PendingPaykitPaymentProof,
+    ): Boolean {
+        val proofs = loadProofs().toMutableList()
+        val index = proofs.indexOf(currentProof)
+        if (index < 0) return false
+        proofs[index] = updatedProof
+        persist(proofs)
+        return true
+    }
+
     private suspend fun persistAndSubmit(
         completedProofs: List<PendingPaykitPaymentProof>,
         allProofs: List<PendingPaykitPaymentProof>,
@@ -363,6 +492,7 @@ class PaykitPaymentProofRepo @Inject constructor(
             requestId = request.id,
             paymentEndpointIdentifier = paymentEndpointIdentifier,
             kind = kind,
+            preparationId = UUID.randomUUID().toString(),
         )
     }
 
@@ -381,6 +511,9 @@ private fun endpointSupports(identifier: String, kind: PaykitPaymentProofKind):
     }
 }
 
+private fun PendingPaykitPaymentProof.matchesRequest(other: PendingPaykitPaymentProof): Boolean =
+    PubkyPublicKeyFormat.matches(identity, other.identity) && requestId == other.requestId
+
 private fun proofJson(kind: PaykitPaymentProofKind, data: String): String = buildJsonObject {
     put("data", JsonPrimitive(data))
     put("type", JsonPrimitive(kind.type))

app/src/main/java/to/bitkit/services/LightningService.kt

@@ -23,6 +23,7 @@ import org.lightningdevkit.ldknode.BackgroundSyncConfig
 import org.lightningdevkit.ldknode.BalanceDetails
 import org.lightningdevkit.ldknode.Bolt11Invoice
 import org.lightningdevkit.ldknode.Bolt11InvoiceDescription
+import org.lightningdevkit.ldknode.BroadcastOutcome
 import org.lightningdevkit.ldknode.BuildException
 import org.lightningdevkit.ldknode.Builder
 import org.lightningdevkit.ldknode.ChannelConfig
@@ -71,6 +72,7 @@ import to.bitkit.utils.LdkError
 import to.bitkit.utils.LdkLogWriter
 import to.bitkit.utils.Logger
 import to.bitkit.utils.LoggerLdk
+import to.bitkit.utils.PendingOnchainBroadcastError
 import to.bitkit.utils.ServiceError
 import to.bitkit.utils.jsonLogOf
 import java.io.File
@@ -915,20 +917,23 @@ class LightningService @Inject constructor(
     ): Txid {
         val node = this.node ?: throw ServiceError.NodeNotSetup()
 
-        Logger.info(
-            "Sending $sats sats to $address, satsPerVByte=$satsPerVByte, isMaxAmount = $isMaxAmount",
-            context = TAG,
-        )
-
         return ServiceQueue.LDK.background {
+            val onchainPayment = node.onchainPayment()
+            onchainPayment.listPendingBroadcasts().firstOrNull()?.let {
+                throw PendingOnchainBroadcastError(it.txid)
+            }
+            Logger.info(
+                "Sending '$sats' sats to '$address', satsPerVByte='$satsPerVByte', isMaxAmount='$isMaxAmount'",
+                context = TAG,
+            )
             if (isMaxAmount) {
-                node.onchainPayment().sendAllToAddress(
+                onchainPayment.sendAllToAddress(
                     address = address,
                     retainReserve = true,
                     feeRate = FeeRate.fromSatPerVbUnchecked(satsPerVByte),
                 )
             } else {
-                node.onchainPayment().sendToAddress(
+                onchainPayment.sendToAddress(
                     address = address,
                     amountSats = sats,
                     feeRate = FeeRate.fromSatPerVbUnchecked(satsPerVByte),
@@ -1331,6 +1336,20 @@ class LightningService @Inject constructor(
             node.listPayments()
         }
     }
+
+    suspend fun getOnchainBroadcastOutcome(txid: Txid): BroadcastOutcome? {
+        val node = this.node ?: throw ServiceError.NodeNotSetup()
+        return ServiceQueue.LDK.background {
+            node.onchainPayment().broadcastOutcome(txid)
+        }
+    }
+
+    suspend fun acknowledgeOnchainBroadcastOutcome(txid: Txid) {
+        val node = this.node ?: throw ServiceError.NodeNotSetup()
+        ServiceQueue.LDK.background {
+            node.onchainPayment().acknowledgeBroadcastOutcome(txid)
+        }
+    }
     // endregion
 
     // region debug