feat: support incoming paykit requests #1098
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,251 @@ | ||
| @file:OptIn(ExperimentalTime::class) | ||
|
|
||
| package to.bitkit.repositories | ||
|
|
||
| import com.synonym.paykit.OutboundPrivateCounterpartySendReport | ||
| import com.synonym.paykit.PaymentRequestLifecycleState | ||
| import com.synonym.paykit.PaymentRequestLocalRole | ||
| import com.synonym.paykit.PaymentRequestRecord | ||
| import com.synonym.paykit.PrivateStreamCounterpartyIntakeReport | ||
| import kotlinx.coroutines.CoroutineDispatcher | ||
| import kotlinx.coroutines.CoroutineScope | ||
| import kotlinx.coroutines.Job | ||
| import kotlinx.coroutines.SupervisorJob | ||
| import kotlinx.coroutines.delay | ||
| import kotlinx.coroutines.flow.MutableStateFlow | ||
| import kotlinx.coroutines.flow.StateFlow | ||
| import kotlinx.coroutines.flow.asStateFlow | ||
| import kotlinx.coroutines.flow.update | ||
| import kotlinx.coroutines.launch | ||
| import kotlinx.coroutines.sync.Mutex | ||
| import kotlinx.coroutines.sync.withLock | ||
| import kotlinx.coroutines.withContext | ||
| import to.bitkit.di.IoDispatcher | ||
| import to.bitkit.ext.runSuspendCatching | ||
| import to.bitkit.models.PubkyPublicKeyFormat | ||
| import to.bitkit.services.PaykitSdkService | ||
| import to.bitkit.utils.AppError | ||
| import to.bitkit.utils.Logger | ||
| import java.math.BigDecimal | ||
| import java.util.concurrent.atomic.AtomicLong | ||
| import javax.inject.Inject | ||
| import javax.inject.Singleton | ||
| import kotlin.time.Clock | ||
| import kotlin.time.Duration | ||
| import kotlin.time.ExperimentalTime | ||
| import kotlin.time.Instant | ||
|
|
||
| data class PaykitPaymentRequestId( | ||
| val paymentRequestId: String, | ||
| val counterparty: String, | ||
| val counterpartyReceiverPath: String, | ||
| ) | ||
|
|
||
| data class PaykitPaymentRequest( | ||
| val paymentRequestId: String, | ||
| val counterparty: String, | ||
| val counterpartyReceiverPath: String, | ||
| val amountValue: String, | ||
| val amountSats: ULong, | ||
| val paymentReference: String, | ||
| val expiresAt: Instant?, | ||
| val acceptedPaymentEndpointIdentifiers: List<String>, | ||
| val metadata: String, | ||
| ) { | ||
| val id: PaykitPaymentRequestId | ||
| get() = PaykitPaymentRequestId(paymentRequestId, counterparty, counterpartyReceiverPath) | ||
|
|
||
| fun isExpired(now: Instant): Boolean = expiresAt?.let { it <= now } == true | ||
|
|
||
| fun acceptsLightningInvoiceAmountMsats(amountMsats: ULong?): Boolean = | ||
| amountMsats == null || amountSats <= ULong.MAX_VALUE / 1000uL && amountMsats == amountSats * 1000uL | ||
|
|
||
| fun acceptsLightningInvoiceAmountSats(amountSats: ULong): Boolean = | ||
| amountSats == 0uL || acceptsPaymentAmount(amountSats) | ||
|
|
||
| fun acceptsPaymentAmount(amountSats: ULong): Boolean = amountSats == this.amountSats | ||
| } | ||
|
|
||
| sealed class PaykitPaymentRequestError(message: String) : AppError(message) { | ||
| data object RequestUnavailable : PaykitPaymentRequestError("Payment request is unavailable") | ||
| data object RequestExpired : PaykitPaymentRequestError("Payment request has expired") | ||
| } | ||
|
|
||
| @Singleton | ||
| class PaykitPaymentRequestRepo @Inject constructor( | ||
| @IoDispatcher private val ioDispatcher: CoroutineDispatcher, | ||
| private val paykitSdkService: PaykitSdkService, | ||
| private val clock: Clock, | ||
| ) { | ||
| companion object { | ||
| private const val TAG = "PaykitPaymentRequestRepo" | ||
| } | ||
|
|
||
| private val operationMutex = Mutex() | ||
| private val stateGeneration = AtomicLong() | ||
| private val repoScope = CoroutineScope(SupervisorJob() + ioDispatcher) | ||
| private var expirationJob: Job? = null | ||
| private val _pendingRequests = MutableStateFlow<List<PaykitPaymentRequest>>(emptyList()) | ||
| val pendingRequests: StateFlow<List<PaykitPaymentRequest>> = _pendingRequests.asStateFlow() | ||
|
|
||
| suspend fun refresh(): Result<Unit> { | ||
| val generation = stateGeneration.get() | ||
| return withContext(ioDispatcher) { | ||
| runSuspendCatching { | ||
| operationMutex.withLock { | ||
| runSuspendCatching { synchronizeLocked(generation) } | ||
| .onFailure { discardExpiredRequestsLocked() } | ||
| .getOrThrow() | ||
| } | ||
| }.onFailure { | ||
| Logger.warn("Failed to refresh incoming Paykit payment requests", it, context = TAG) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| suspend fun accept(request: PaykitPaymentRequest): Result<Unit> = updateRequest(request) { | ||
| paykitSdkService.acceptPaymentRequest( | ||
| counterparty = it.counterparty, | ||
| counterpartyReceiverPath = it.counterpartyReceiverPath, | ||
| paymentRequestId = it.paymentRequestId, | ||
| ) | ||
| }.onFailure { | ||
| Logger.warn("Failed to accept incoming Paykit payment request", it, context = TAG) | ||
| } | ||
|
|
||
| fun isPending(request: PaykitPaymentRequest): Boolean = | ||
| !request.isExpired(clock.now()) && _pendingRequests.value.any { it.id == request.id } | ||
|
|
||
| suspend fun clear() { | ||
| stateGeneration.incrementAndGet() | ||
| withContext(ioDispatcher) { | ||
| operationMutex.withLock { | ||
| expirationJob?.cancel() | ||
| expirationJob = null | ||
| _pendingRequests.update { emptyList() } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private suspend fun synchronizeLocked(generation: Long) { | ||
| processPendingMessages() | ||
| paykitSdkService.receivePrivateMessagesFromLinkedPeers().also(::logIntakeFailures) | ||
| val now = clock.now() | ||
| val requests = paykitSdkService.actionableReceivedPaymentRequests().mapNotNull { | ||
| it.toPaykitPaymentRequest(now) | ||
| } | ||
| if (stateGeneration.get() != generation) return | ||
| _pendingRequests.update { requests } | ||
| scheduleExpirationLocked() | ||
| } | ||
|
|
||
| private suspend fun updateRequest( | ||
| request: PaykitPaymentRequest, | ||
| operation: suspend (PaykitPaymentRequest) -> Unit, | ||
| ): Result<Unit> = withContext(ioDispatcher) { | ||
| runSuspendCatching { | ||
| operationMutex.withLock { | ||
| if (request.isExpired(clock.now())) { | ||
| discardExpiredRequestsLocked() | ||
| throw PaykitPaymentRequestError.RequestExpired | ||
| } | ||
| val current = _pendingRequests.value.firstOrNull { it.id == request.id } | ||
| ?: throw PaykitPaymentRequestError.RequestUnavailable | ||
|
|
||
| operation(current) | ||
| _pendingRequests.update { requests -> requests.filterNot { it.id == current.id } } | ||
| discardExpiredRequestsLocked() | ||
| processPendingMessages() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private suspend fun processPendingMessages() { | ||
| runSuspendCatching { paykitSdkService.processPendingPrivateMessages() } | ||
| .onSuccess(::logOutboundFailures) | ||
| .onFailure { Logger.warn("Failed to deliver pending Paykit private messages", it, context = TAG) } | ||
| } | ||
|
|
||
| private fun logOutboundFailures(reports: List<OutboundPrivateCounterpartySendReport>) { | ||
| reports.forEach { | ||
| val error = it.error ?: return@forEach | ||
| Logger.warn( | ||
| "Failed to deliver Paykit private messages to '${PubkyPublicKeyFormat.redacted(it.counterparty)}': " + | ||
| "'${error.redactedContext()}'", | ||
| context = TAG, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| private fun logIntakeFailures(reports: List<PrivateStreamCounterpartyIntakeReport>) { | ||
| reports.forEach { | ||
| val error = it.error ?: return@forEach | ||
| Logger.warn( | ||
| "Failed to receive Paykit private messages from '${PubkyPublicKeyFormat.redacted(it.counterparty)}': " + | ||
| "'${error.redactedContext()}'", | ||
| context = TAG, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| private fun discardExpiredRequestsLocked() { | ||
| val now = clock.now() | ||
| _pendingRequests.update { requests -> requests.filterNot { it.isExpired(now) } } | ||
| scheduleExpirationLocked() | ||
| } | ||
|
|
||
| private fun scheduleExpirationLocked() { | ||
| expirationJob?.cancel() | ||
| expirationJob = null | ||
|
|
||
| val nextExpiration = _pendingRequests.value.mapNotNull { it.expiresAt }.minOrNull() ?: return | ||
| val delayDuration = (nextExpiration - clock.now()).coerceAtLeast(Duration.ZERO) | ||
| expirationJob = repoScope.launch { | ||
| delay(delayDuration) | ||
| operationMutex.withLock { | ||
| expirationJob = null | ||
| discardExpiredRequestsLocked() | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private val bitcoinAmountPattern = Regex("(?:[0-9]+(?:\\.[0-9]*)?|\\.[0-9]+)") | ||
|
|
||
| @Suppress("ReturnCount") | ||
| private fun PaymentRequestRecord.toPaykitPaymentRequest(now: Instant): PaykitPaymentRequest? { | ||
| if (localRole != PaymentRequestLocalRole.PAYER || state != PaymentRequestLifecycleState.PROPOSED) return null | ||
| val requestTerms = terms ?: return null | ||
| if (requestTerms.recurrence != null || requestTerms.amount.asset != "btc") return null | ||
| val amountSats = requestTerms.amount.value.toSats() | ||
| ?.takeIf { it <= ULong.MAX_VALUE / 1000uL } | ||
| ?: return null | ||
| val endpoints = requestTerms.acceptedPaymentEndpointIdentifiers | ||
| .filter { MethodId.fromRawValue(it) != null } | ||
| .distinct() | ||
| if (endpoints.isEmpty()) return null | ||
|
|
||
| val expiresAt = requestTerms.proposalExpiresAt?.let { | ||
| runCatching { Instant.parse(it) }.getOrNull() ?: return null | ||
| } | ||
| if (expiresAt != null && expiresAt <= now) return null | ||
|
|
||
| return PaykitPaymentRequest( | ||
| paymentRequestId = paymentRequestId, | ||
| counterparty = counterparty, | ||
| counterpartyReceiverPath = counterpartyReceiverPath, | ||
| amountValue = requestTerms.amount.value, | ||
| amountSats = amountSats, | ||
| paymentReference = requestTerms.paymentReference.exportText(), | ||
| expiresAt = expiresAt, | ||
| acceptedPaymentEndpointIdentifiers = endpoints, | ||
| metadata = requestTerms.metadata.exportText(), | ||
|
Comment on lines
+239
to
+242
Member
There was a problem hiding this comment.
|
||
| ) | ||
| } | ||
|
|
||
| private fun String.toSats(): ULong? { | ||
| if (!bitcoinAmountPattern.matches(this)) return null | ||
| return runCatching { | ||
| BigDecimal(this).movePointRight(8).toBigIntegerExact().toString().toULong() | ||
| }.getOrNull()?.takeIf { it > 0uL } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
use
appScope(ioDispatcher, TAG)