commit 34c6a65baf6aadf3519e6ae69fb2a65f032c7d4d
parent afba15574c709cd8ff14672a4026f8e392cf8703
Author: MS <ms@taler.net>
Date: Thu, 16 Nov 2023 15:34:48 +0100
nexus fetch: parsing outgoing payments.
Diffstat:
6 files changed, 289 insertions(+), 85 deletions(-)
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Database.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Database.kt
@@ -94,7 +94,7 @@ enum class PaymentInitiationOutcome {
data class OutgoingPayment(
val amount: TalerAmount,
- val wireTransferSubject: String?,
+ val wireTransferSubject: String,
val executionTime: Instant,
val creditPaytoUri: String,
val bankTransferId: String
@@ -223,6 +223,27 @@ class Database(dbConfig: String): java.io.Closeable {
return@runConn OutgoingPaymentOutcome.SUCCESS
}
+ /**
+ * Checks if the outgoing payment was already processed by Nexus.
+ *
+ * @param bankUid unique identifier assigned by the bank to the payment.
+ * Normally, that's the <UETR> value found in camt.05x records. Outgoing
+ * payment have been observed to _lack_ the <AcctSvcrRef> element.
+ * @return true if found, false otherwise
+ */
+ suspend fun isOutgoingPaymentSeen(bankUid: String): Boolean = runConn { conn ->
+ val stmt = conn.prepareStatement("""
+ SELECT 1
+ FROM outgoing_transactions
+ WHERE bank_transfer_id = ?;
+ """)
+ stmt.setString(1, bankUid)
+ val res = stmt.executeQuery()
+ res.use {
+ return@runConn it.next()
+ }
+ }
+
// INCOMING PAYMENTS METHODS
/**
@@ -577,4 +598,30 @@ class Database(dbConfig: String): java.io.Closeable {
*/
return@runConn PaymentInitiationOutcome.UNIQUE_CONSTRAINT_VIOLATION
}
+
+ /**
+ * Gets the ID of an initiated payment. Useful to link it to its
+ * outgoing payment witnessed in a bank record.
+ *
+ * @param uid UID as given by Nexus when it initiated the payment.
+ * This value then gets specified as the MsgId of pain.001,
+ * and it gets associated by the bank to the booked entries
+ * in camt.05x reports.
+ * @retrun the initiated payment row ID, or null if not found. NOTE:
+ * null gets returned even when the initiated payment exists,
+ * *but* it was NOT flagged as submitted.
+ */
+ suspend fun initiatedPaymentGetFromUid(uid: String): Long? = runConn { conn ->
+ val stmt = conn.prepareStatement("""
+ SELECT initiated_outgoing_transaction_id
+ FROM initiated_outgoing_transactions
+ WHERE request_uid = ? AND submitted = 'success';
+ """)
+ stmt.setString(1, uid)
+ val res = stmt.executeQuery()
+ res.use {
+ if (!it.next()) return@runConn null
+ return@runConn it.getLong("initiated_outgoing_transaction_id")
+ }
+ }
}
\ No newline at end of file
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
@@ -12,7 +12,6 @@ import tech.libeufin.util.*
import tech.libeufin.util.ebics_h005.Ebics3Request
import java.io.File
import java.io.IOException
-import java.lang.StringBuilder
import java.net.URLEncoder
import java.nio.file.Path
import java.time.Instant
@@ -22,6 +21,7 @@ import java.util.UUID
import kotlin.concurrent.fixedRateTimer
import kotlin.io.path.createDirectories
import kotlin.system.exitProcess
+import kotlin.text.StringBuilder
/**
* Necessary data to perform a download.
@@ -203,6 +203,131 @@ fun getTalerAmount(
)
}
+private fun XmlElementDestructor.extractOutgoingTxNotif(
+ acceptedCurrency: String,
+ bookDate: Instant
+): OutgoingPaymentWithLink {
+ // Obtaining the amount.
+ val amount: TalerAmount = requireUniqueChildNamed("Amt") {
+ val currency = focusElement.getAttribute("Ccy")
+ if (currency != acceptedCurrency) throw Exception("Currency $currency not supported")
+ getTalerAmount(focusElement.textContent, currency)
+ }
+ /**
+ * Obtaining payment UID. Outgoing tx don't get any AcctSvcrRef,
+ * but UETR. They however echo the MsgId that was used in the original
+ * pain.001, but that's given by us, rather than by the bank.
+ */
+ val uidFromBank = StringBuilder()
+ val link = StringBuilder()
+ requireUniqueChildNamed("Refs") {
+ requireUniqueChildNamed("UETR") {
+ uidFromBank.append(focusElement.textContent)
+ }
+ requireUniqueChildNamed("MsgId") {
+ link.append(focusElement.textContent)
+ }
+ }
+ // Obtaining payment subject.
+ val subject = StringBuilder()
+ requireUniqueChildNamed("RmtInf") {
+ this.mapEachChildNamed("Ustrd") {
+ val piece = this.focusElement.textContent
+ subject.append(piece)
+ }
+ }
+
+ // Obtaining the payer's details
+ val creditorPayto = StringBuilder("payto://iban/")
+ requireUniqueChildNamed("RltdPties") {
+ requireUniqueChildNamed("CdtrAcct") {
+ requireUniqueChildNamed("Id") {
+ requireUniqueChildNamed("IBAN") {
+ creditorPayto.append(focusElement.textContent)
+ }
+ }
+ }
+ requireUniqueChildNamed("Cdtr") {
+ requireUniqueChildNamed("Pty") {
+ requireUniqueChildNamed("Nm") {
+ val urlEncName = URLEncoder.encode(focusElement.textContent, "utf-8")
+ creditorPayto.append("?receiver-name=$urlEncName")
+ }
+ }
+ }
+ }
+ val payment = OutgoingPayment(
+ amount = amount,
+ bankTransferId = uidFromBank.toString(),
+ creditPaytoUri = creditorPayto.toString(),
+ executionTime = bookDate,
+ wireTransferSubject = subject.toString()
+ )
+ return OutgoingPaymentWithLink(payment, link.toString())
+}
+
+private fun XmlElementDestructor.extractIncomingTxNotif(
+ acceptedCurrency: String,
+ bookDate: Instant
+): IncomingPayment {
+ // Obtaining the amount.
+ val amount: TalerAmount = requireUniqueChildNamed("Amt") {
+ val currency = focusElement.getAttribute("Ccy")
+ if (currency != acceptedCurrency) throw Exception("Currency $currency not supported")
+ getTalerAmount(focusElement.textContent, currency)
+ }
+ // Obtaining payment UID.
+ val uidFromBank: String = requireUniqueChildNamed("Refs") {
+ requireUniqueChildNamed("AcctSvcrRef") {
+ focusElement.textContent
+ }
+ }
+ // Obtaining payment subject.
+ val subject = StringBuilder()
+ requireUniqueChildNamed("RmtInf") {
+ this.mapEachChildNamed("Ustrd") {
+ val piece = this.focusElement.textContent
+ subject.append(piece)
+ }
+ }
+
+ // Obtaining the payer's details
+ val debtorPayto = StringBuilder("payto://iban/")
+ requireUniqueChildNamed("RltdPties") {
+ requireUniqueChildNamed("DbtrAcct") {
+ requireUniqueChildNamed("Id") {
+ requireUniqueChildNamed("IBAN") {
+ debtorPayto.append(focusElement.textContent)
+ }
+ }
+ }
+ // warn: it might need the postal address too..
+ requireUniqueChildNamed("Dbtr") {
+ requireUniqueChildNamed("Nm") {
+ val urlEncName = URLEncoder.encode(focusElement.textContent, "utf-8")
+ debtorPayto.append("?receiver-name=$urlEncName")
+ }
+ }
+ }
+ return IncomingPayment(
+ amount = amount,
+ bankTransferId = uidFromBank,
+ debitPaytoUri = debtorPayto.toString(),
+ executionTime = bookDate,
+ wireTransferSubject = subject.toString()
+ )
+}
+
+data class OutgoingPaymentWithLink(
+ val outgoingPayment: OutgoingPayment,
+ val initiatedPaymentLink: String
+)
+
+data class Camt054Result(
+ val incoming: MutableList<IncomingPayment> = mutableListOf(),
+ val outgoing: MutableList<OutgoingPaymentWithLink> = mutableListOf()
+)
+
/**
* Searches for incoming transactions in a camt.054 document, that
* was downloaded via EBICS notification.
@@ -210,77 +335,33 @@ fun getTalerAmount(
* @param notifXml the input document.
* @return any incoming payment as a list of [IncomingPayment]
*/
-fun findIncomingTxInNotification(
+fun parseNotification(
notifXml: String,
acceptedCurrency: String
-): List<IncomingPayment> {
+): Camt054Result {
val notifDoc = XMLUtil.parseStringIntoDom(notifXml)
- val ret = mutableListOf<IncomingPayment>()
+ val ret = Camt054Result()
destructXml(notifDoc) {
requireRootElement("Document") {
requireUniqueChildNamed("BkToCstmrDbtCdtNtfctn") {
mapEachChildNamed("Ntfctn") {
mapEachChildNamed("Ntry") {
+ val bookDate: Instant = requireUniqueChildNamed("BookgDt") {
+ requireUniqueChildNamed("Dt") {
+ parseBookDate(focusElement.textContent)
+ }
+ }
mapEachChildNamed("NtryDtls") {
- mapEachChildNamed("TxDtls") maybeDbit@{
-
- // currently, only incoming payments are considered.
+ mapEachChildNamed("TxDtls") {
if (requireUniqueChildNamed("CdtDbtInd") {
focusElement.textContent == "DBIT"
- }) return@maybeDbit
-
- // Obtaining the amount.
- val amount: TalerAmount = requireUniqueChildNamed("Amt") {
- val currency = focusElement.getAttribute("Ccy")
- if (currency != acceptedCurrency) throw Exception("Currency $currency not supported")
- getTalerAmount(focusElement.textContent, currency)
- }
- // Obtaining payment UID.
- val uidFromBank: String = requireUniqueChildNamed("Refs") {
- requireUniqueChildNamed("AcctSvcrRef") {
- focusElement.textContent
- }
+ }) {
+ val outgoingPayment = extractOutgoingTxNotif(acceptedCurrency, bookDate)
+ ret.outgoing.add(outgoingPayment)
+ } else {
+ val incomingPayment = extractIncomingTxNotif(acceptedCurrency, bookDate)
+ ret.incoming.add(incomingPayment)
}
- // Obtaining payment subject.
- val subject = StringBuilder()
- requireUniqueChildNamed("RmtInf") {
- this.mapEachChildNamed("Ustrd") {
- val piece = this.focusElement.textContent
- subject.append(piece)
- }
- }
- // Obtaining the execution time.
- val executionTime: Instant = requireUniqueChildNamed("RltdDts") {
- requireUniqueChildNamed("AccptncDtTm") {
- parseCamtTime(focusElement.textContent)
- }
- }
- // Obtaining the payer's details
- val debtorPayto = StringBuilder("payto://iban/")
- requireUniqueChildNamed("RltdPties") {
- requireUniqueChildNamed("DbtrAcct") {
- requireUniqueChildNamed("Id") {
- requireUniqueChildNamed("IBAN") {
- debtorPayto.append(focusElement.textContent)
- }
- }
- }
- // warn: it might need the postal address too..
- requireUniqueChildNamed("Dbtr") {
- requireUniqueChildNamed("Nm") {
- val urlEncName = URLEncoder.encode(focusElement.textContent, "utf-8")
- debtorPayto.append("?receiver-name=$urlEncName")
- }
- }
- }
- val incomingPayment = IncomingPayment(
- amount = amount,
- bankTransferId = uidFromBank,
- debitPaytoUri = debtorPayto.toString(),
- executionTime = executionTime,
- wireTransferSubject = subject.toString()
- )
- ret.add(incomingPayment)
}
}
}
@@ -325,7 +406,7 @@ fun isReservePub(maybeReservePub: String): ByteArray? {
* @return [ByteArray] as the reserve public key, or null if the
* payment cannot lead to a Taler withdrawal.
*/
-suspend fun isTalerable(
+suspend fun getTalerReservePub(
db: Database,
payment: IncomingPayment
): ByteArray? {
@@ -341,6 +422,66 @@ suspend fun isTalerable(
}
/**
+ * Ingests any outgoing payment that was NOT ingested yet. It
+ * links it to the initiated outgoing transaction that originated
+ * it.
+ *
+ * @param db database handle.
+ * @param payment payment to (maybe) ingest.
+ */
+private suspend fun ingestOutgoingPayment(
+ db: Database,
+ payment: OutgoingPaymentWithLink
+) {
+ // Check if the payment was ingested already.
+ if (db.isOutgoingPaymentSeen(payment.outgoingPayment.bankTransferId)) {
+ logger.debug("Outgoing payment with UID '${payment.outgoingPayment.bankTransferId}' already seen.")
+ return
+ }
+ // Get the initiate payment to link to this.
+ val initId: Long? = db.initiatedPaymentGetFromUid(payment.initiatedPaymentLink)
+ if (initId == null) {
+ throw Exception("Outgoing payment lacks (submitted) initiated " +
+ "counterpart with UID ${payment.initiatedPaymentLink}"
+ )
+ }
+ // store the payment and its linked init
+ val insertionResult = db.outgoingPaymentCreate(payment.outgoingPayment, initId)
+ if (insertionResult != OutgoingPaymentOutcome.SUCCESS) {
+ throw Exception("Could not store outgoing payment with bank-given" +
+ "UID '${payment.outgoingPayment.bankTransferId}' " +
+ "and update its related initiation. DB result: $insertionResult"
+ )
+ }
+}
+
+/**
+ * Ingests any incoming payment that was NOT ingested yet. Stores
+ * the payment into valid talerable ones or bounces it, according
+ * to the subject.
+ *
+ * @param db database handle.
+ * @param incomingPayment payment to (maybe) ingest.
+ */
+private suspend fun ingestIncomingPayment(
+ db: Database,
+ incomingPayment: IncomingPayment
+) {
+ if (db.isIncomingPaymentSeen(incomingPayment.bankTransferId)) {
+ logger.debug("Incoming payment with UID '${incomingPayment.bankTransferId}' already seen.")
+ return
+ }
+ val reservePub = getTalerReservePub(db, incomingPayment)
+ if (reservePub == null) {
+ db.incomingPaymentCreateBounced(
+ incomingPayment, UUID.randomUUID().toString().take(35)
+ )
+ return
+ }
+ db.incomingTalerablePaymentCreate(incomingPayment, reservePub)
+}
+
+/**
* Parses the response of an EBICS notification looking for
* incoming payments. As a result, it either creates a Taler
* withdrawal or bounces the incoming payment. In detail, this
@@ -360,12 +501,14 @@ fun ingestNotification(
content: ByteArray
): Boolean {
val incomingPayments = mutableListOf<IncomingPayment>()
+ val outgoingPayments = mutableListOf<OutgoingPaymentWithLink>()
val filenamePrefix = "camt.054_P_" // Only these files have all the details.
try {
content.unzipForEach { fileName, xmlContent ->
if (!fileName.startsWith(filenamePrefix)) return@unzipForEach
- val found = findIncomingTxInNotification(xmlContent, ctx.cfg.currency)
- incomingPayments += found
+ val found = parseNotification(xmlContent, ctx.cfg.currency)
+ incomingPayments += found.incoming
+ outgoingPayments += found.outgoing
}
} catch (e: IOException) {
logger.error("Could not open any ZIP archive")
@@ -374,23 +517,14 @@ fun ingestNotification(
logger.error(e.message)
return false
}
- // Distinguishing now valid and invalid payments.
- // Any error at this point is only due to Nexus.
+
try {
- incomingPayments.forEach {
- runBlocking {
- if (db.isIncomingPaymentSeen(it.bankTransferId)) {
- logger.debug("Incoming payment with UID '${it.bankTransferId}' already seen.")
- return@runBlocking
- }
- val reservePub = isTalerable(db, it)
- if (reservePub == null) {
- db.incomingPaymentCreateBounced(
- it, UUID.randomUUID().toString().take(35)
- )
- return@runBlocking
- }
- db.incomingTalerablePaymentCreate(it, reservePub)
+ runBlocking {
+ incomingPayments.forEach {
+ ingestIncomingPayment(db, it)
+ }
+ outgoingPayments.forEach {
+ ingestOutgoingPayment(db, it)
}
}
} catch (e: Exception) {
@@ -526,8 +660,8 @@ class EbicsFetch: CliktCommand("Fetches bank records. Defaults to camt.054 noti
val maybeStdin = generateSequence(::readLine).joinToString("\n")
when(whichDoc) {
SupportedDocument.CAMT_054 -> {
- val incoming = findIncomingTxInNotification(maybeStdin, cfg.currency)
- incoming.forEach {
+ val res = parseNotification(maybeStdin, cfg.currency)
+ res.incoming.forEach {
println(it)
}
}
diff --git a/nexus/src/test/kotlin/Common.kt b/nexus/src/test/kotlin/Common.kt
@@ -97,7 +97,7 @@ fun genIncPay(subject: String = "test wire transfer") =
)
// Generates an outgoing payment, given its subject.
-fun genOutPay(subject: String? = null) =
+fun genOutPay(subject: String = "outgoing payment") =
OutgoingPayment(
amount = TalerAmount(44, 0, "KUDOS"),
creditPaytoUri = "payto://iban/TEST-IBAN?receiver-name=Test",
diff --git a/nexus/src/test/kotlin/DatabaseTest.kt b/nexus/src/test/kotlin/DatabaseTest.kt
@@ -3,9 +3,8 @@ import org.junit.Test
import tech.libeufin.nexus.*
import java.time.Instant
import kotlin.random.Random
+import kotlin.test.*
import kotlin.test.assertEquals
-import kotlin.test.assertFalse
-import kotlin.test.assertTrue
class OutgoingPaymentsTest {
@@ -19,10 +18,12 @@ class OutgoingPaymentsTest {
val db = prepDb(TalerConfig(NEXUS_CONFIG_SOURCE))
runBlocking {
// inserting without reconciling
+ assertFalse(db.isOutgoingPaymentSeen("entropic"))
assertEquals(
OutgoingPaymentOutcome.SUCCESS,
db.outgoingPaymentCreate(genOutPay("paid by nexus"))
)
+ assertTrue(db.isOutgoingPaymentSeen("entropic"))
// inserting trying to reconcile with a non-existing initiated payment.
assertEquals(
OutgoingPaymentOutcome.INITIATED_COUNTERPART_NOT_FOUND,
@@ -171,6 +172,7 @@ class PaymentInitiationsTest {
initiationTime = Instant.now()
)
runBlocking {
+ assertNull(db.initiatedPaymentGetFromUid("unique"))
assertEquals(db.initiatedPaymentCreate(initPay), PaymentInitiationOutcome.SUCCESS)
assertEquals(db.initiatedPaymentCreate(initPay), PaymentInitiationOutcome.UNIQUE_CONSTRAINT_VIOLATION)
val haveOne = db.initiatedPaymentsUnsubmittedGet("KUDOS")
@@ -179,6 +181,8 @@ class PaymentInitiationsTest {
&& haveOne.containsKey(1)
&& haveOne[1]?.requestUid == "unique"
}
+ db.initiatedPaymentSetSubmittedState(1, DatabaseSubmissionState.success)
+ assertNotNull(db.initiatedPaymentGetFromUid("unique"))
}
}
diff --git a/nexus/src/test/kotlin/Parsing.kt b/nexus/src/test/kotlin/Parsing.kt
@@ -4,6 +4,7 @@ import tech.libeufin.nexus.TalerAmount
import tech.libeufin.nexus.getAmountNoCurrency
import tech.libeufin.nexus.getTalerAmount
import tech.libeufin.nexus.isReservePub
+import tech.libeufin.util.parseBookDate
import tech.libeufin.util.parseCamtTime
import java.lang.StringBuilder
import kotlin.test.assertEquals
@@ -19,6 +20,12 @@ class Parsing {
assertThrows<Exception> { parseCamtTime("2023-11-06T20:00:00+01:00") }
assertThrows<Exception> { parseCamtTime("2023-11-06T20:00:00Z") }
}
+
+ @Test
+ fun bookDateTest() {
+ parseBookDate("1970-01-01")
+ }
+
@Test // Could be moved in a dedicated Amounts.kt test module.
fun generateCurrencyAgnosticAmount() {
assertThrows<Exception> {
diff --git a/util/src/main/kotlin/time.kt b/util/src/main/kotlin/time.kt
@@ -92,4 +92,16 @@ fun parseCamtTime(timeFromCamt: String): Instant {
val t = LocalDateTime.parse(timeFromCamt)
val utc = ZoneId.of("UTC")
return t.toInstant(utc.rules.getOffset(t))
+}
+
+/**
+ * Parses a date string as found in the booking date of
+ * camt.054 reports. They have this format: yyyy-MM-dd.
+ *
+ * @param bookDate input to parse
+ * @return [Instant] to the UTC.
+ */
+fun parseBookDate(bookDate: String): Instant {
+ val l = LocalDate.parse(bookDate)
+ return Instant.from(l.atStartOfDay(ZoneId.of("UTC")))
}
\ No newline at end of file
