integrate new CAMT parser, make TWG work again

9 files changed, 259 insertions, 208 deletions

diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt
index 1f1b57fd..f5aab033 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt
@@ -51,6 +51,8 @@ object TalerRequestedPayments : LongIdTable() {
     /**
      * This column gets a value only after the bank acknowledges the payment via
      * a camt.05x entry.  The "crunch" logic is responsible for assigning such value.
+     *
+     * FIXME(dold): Shouldn't this happen at the level of the PreparedPaymentsTable?
      */
     val rawConfirmed = reference("raw_confirmed", RawBankTransactionsTable).nullable()
 }
@@ -73,14 +75,18 @@ class TalerRequestedPaymentEntity(id: EntityID<Long>) : LongEntity(id) {
  */
 object TalerIncomingPayments : LongIdTable() {
     val payment = reference("payment", RawBankTransactionsTable)
-    val valid = bool("valid")
+    val reservePublicKey = text("reservePublicKey")
+    val timestampMs = long("timestampMs")
+    val incomingPaytoUri = text("incomingPaytoUri")
 }
 
 class TalerIncomingPaymentEntity(id: EntityID<Long>) : LongEntity(id) {
     companion object : LongEntityClass<TalerIncomingPaymentEntity>(TalerIncomingPayments)
 
     var payment by RawBankTransactionEntity referencedOn TalerIncomingPayments.payment
-    var valid by TalerIncomingPayments.valid
+    var reservePublicKey by TalerIncomingPayments.reservePublicKey
+    var timestampMs by TalerIncomingPayments.timestampMs
+    var incomingPaytoUri by TalerIncomingPayments.incomingPaytoUri
 }
 
 /**
@@ -109,33 +115,52 @@ class NexusBankMessageEntity(id: EntityID<Int>) : IntEntity(id) {
  * CAMT message.
  */
 object RawBankTransactionsTable : LongIdTable() {
-    val unstructuredRemittanceInformation = text("unstructuredRemittanceInformation")
-    val transactionType = text("transactionType") /* DBIT or CRDT */
+    /**
+     * Identifier for the transaction that is unique among all transactions of the account.
+     * The scheme for this identifier is the accounts transaction identification scheme.
+     *
+     * Note that this is *not* a unique ID per account, as the same underlying
+     * transaction can show up multiple times with a different status.
+     */
+    val accountTransactionId = text("accountTransactionId")
+
+    /**
+     * Bank account that this transaction happened on.
+     */
+    val bankAccount = reference("bankAccount", NexusBankAccountsTable)
+
+    /**
+     * Direction of the amount.
+     */
+    val creditDebitIndicator = text("creditDebitIndicator")
+
+    /**
+     * Currency of the amount.
+     */
     val currency = text("currency")
     val amount = text("amount")
-    val counterpartIban = text("counterpartIban")
-    val counterpartBic = text("counterpartBic")
-    val counterpartName = text("counterpartName")
-    val bookingDate = long("bookingDate")
-    val status = text("status") // BOOK or other.
-    val uid = text("uid") // AcctSvcrRef code, given by the bank.
-    val bankAccount = reference("bankAccount", NexusBankAccountsTable)
+
+    /**
+     * Booked / pending / informational.
+     */
+    val status = text("status")
+
+    /**
+     * Full details of the transaction in JSON format.
+     */
+    val transactionJson = text("transactionJson")
 }
 
 class RawBankTransactionEntity(id: EntityID<Long>) : LongEntity(id) {
     companion object : LongEntityClass<RawBankTransactionEntity>(RawBankTransactionsTable)
 
-    var unstructuredRemittanceInformation by RawBankTransactionsTable.unstructuredRemittanceInformation
-    var transactionType by RawBankTransactionsTable.transactionType
     var currency by RawBankTransactionsTable.currency
     var amount by RawBankTransactionsTable.amount
-    var counterpartIban by RawBankTransactionsTable.counterpartIban
-    var counterpartBic by RawBankTransactionsTable.counterpartBic
-    var counterpartName by RawBankTransactionsTable.counterpartName
-    var bookingDate by RawBankTransactionsTable.bookingDate
     var status by RawBankTransactionsTable.status
-    var uid by RawBankTransactionsTable.uid
+    var creditDebitIndicator by RawBankTransactionsTable.creditDebitIndicator
     var bankAccount by NexusBankAccountEntity referencedOn RawBankTransactionsTable.bankAccount
+    var transactionJson by RawBankTransactionsTable.transactionJson
+    var accountTransactionId by RawBankTransactionsTable.accountTransactionId
 }
 
 /**
@@ -160,11 +185,6 @@ object PreparedPaymentsTable : IdTable<String>() {
 
     /* Indicates whether the PAIN message was sent to the bank. */
     val submitted = bool("submitted").default(false)
-
-    /* Indicates whether the bank didn't perform the payment: note that
-     * this state can be reached when the payment gets listed in a CRZ
-     * response OR when the payment doesn't show up in a C52/C53 response */
-    val invalid = bool("invalid").default(false)
 }
 
 class PreparedPaymentEntity(id: EntityID<String>) : Entity<String>(id) {
@@ -184,11 +204,11 @@ class PreparedPaymentEntity(id: EntityID<String>) : Entity<String>(id) {
     var creditorBic by PreparedPaymentsTable.creditorBic
     var creditorName by PreparedPaymentsTable.creditorName
     var submitted by PreparedPaymentsTable.submitted
-    var invalid by PreparedPaymentsTable.invalid
 }
 
 /**
  * This table holds triples of <iban, bic, holder name>.
+ * FIXME(dold):  Allow other account and bank identifications than IBAN and BIC
  */
 object NexusBankAccountsTable : IdTable<String>() {
     override val id = text("id").entityId()
@@ -198,7 +218,7 @@ object NexusBankAccountsTable : IdTable<String>() {
     val defaultBankConnection = reference("defaultBankConnection", NexusBankConnectionsTable).nullable()
 
     // Highest bank message ID that this bank account is aware of.
-    val highestSeenBankMessageId = integer("")
+    val highestSeenBankMessageId = integer("highestSeenBankMessageId")
 }
 
 class NexusBankAccountEntity(id: EntityID<String>) : Entity<String>(id) {
@@ -284,7 +304,7 @@ class FacadeEntity(id: EntityID<String>) : Entity<String>(id) {
     var creator by NexusUserEntity referencedOn FacadesTable.creator
 }
 
-object TalerFacadeStatesTable : IntIdTable() {
+object TalerFacadeStateTable : IntIdTable() {
     val bankAccount = text("bankAccount")
     val bankConnection = text("bankConnection")
 
@@ -296,16 +316,16 @@ object TalerFacadeStatesTable : IntIdTable() {
 }
 
 class TalerFacadeStateEntity(id: EntityID<Int>) : IntEntity(id) {
-    companion object : IntEntityClass<TalerFacadeStateEntity>(TalerFacadeStatesTable)
+    companion object : IntEntityClass<TalerFacadeStateEntity>(TalerFacadeStateTable)
 
-    var bankAccount by TalerFacadeStatesTable.bankAccount
-    var bankConnection by TalerFacadeStatesTable.bankConnection
+    var bankAccount by TalerFacadeStateTable.bankAccount
+    var bankConnection by TalerFacadeStateTable.bankConnection
 
     /* "statement", "report", "notification" */
-    var reserveTransferLevel by TalerFacadeStatesTable.reserveTransferLevel
-    var intervalIncrement by TalerFacadeStatesTable.intervalIncrement
-    var facade by FacadeEntity referencedOn TalerFacadeStatesTable.facade
-    var highestSeenMsgID by TalerFacadeStatesTable.highestSeenMsgID
+    var reserveTransferLevel by TalerFacadeStateTable.reserveTransferLevel
+    var intervalIncrement by TalerFacadeStateTable.intervalIncrement
+    var facade by FacadeEntity referencedOn TalerFacadeStateTable.facade
+    var highestSeenMsgID by TalerFacadeStateTable.highestSeenMsgID
 }
 
 fun dbCreateTables(dbName: String) {
@@ -324,7 +344,7 @@ fun dbCreateTables(dbName: String) {
             NexusBankConnectionsTable,
             NexusBankMessagesTable,
             FacadesTable,
-            TalerFacadeStatesTable
+            TalerFacadeStateTable
         )
     }
-}
\ No newline at end of file
+}
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Helpers.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Helpers.kt
index ec02eeb9..246ad9ce 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/Helpers.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/Helpers.kt
@@ -19,6 +19,7 @@
 
 package tech.libeufin.nexus
 
+import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import io.ktor.client.HttpClient
 import io.ktor.http.HttpStatusCode
 import io.ktor.request.ApplicationRequest
@@ -123,62 +124,52 @@ fun getEbicsSubscriberDetails(userId: String, transportId: String): EbicsClientS
     return getEbicsSubscriberDetailsInternal(subscriber)
 }
 
-// returns true if the payment is found in the database.
-fun isDuplicate(camt: Document, acctSvcrRef: String): Boolean {
-    val foundWithStatus = transaction {
+/**
+ * Check if the transaction is already found in the database.
+ */
+private fun isDuplicate(acctSvcrRef: String): Boolean {
+    // FIXME: make this generic depending on transaction identification scheme
+    val ati = "AcctSvcrRef:$acctSvcrRef"
+    return transaction {
         val res = RawBankTransactionEntity.find {
-            RawBankTransactionsTable.uid eq acctSvcrRef
+            RawBankTransactionsTable.accountTransactionId eq ati
         }.firstOrNull()
-        if (res != null) {
-            Pair(true, res.status)
-        } else {
-            Pair(false, null)
-        }
+        res != null
     }
-    if (!foundWithStatus.first)
-        return false
-
-    // ignore if status if the same as the one stored previously
-    val givenStatus = camt.pickString("//*[local-name()='Ntry']//*[local-name()='Sts']")
-    if (givenStatus == foundWithStatus.second)
-        return true
-
-    // at this point, the message has neither a known status, or it is itself known.
-    return false
 }
 
 fun processCamtMessage(
     bankAccountId: String,
-    camt53doc: Document
+    camtDoc: Document
 ) {
+    logger.info("processing CAMT message")
     transaction {
         val acct = NexusBankAccountEntity.findById(bankAccountId)
         if (acct == null) {
             throw NexusError(HttpStatusCode.NotFound, "user not found")
         }
-        val bookingDate = parseDashedDate(
-            camt53doc.pickString("//*[local-name()='BookgDt']//*[local-name()='Dt']")
-        )
-        val acctSvcrRef = camt53doc.pickString("//*[local-name()='AcctSvcrRef']")
-        if (isDuplicate(camt53doc, acctSvcrRef)) {
-            logger.info("Processing a duplicate, not storing it.")
-            return@transaction
-        }
-        RawBankTransactionEntity.new {
-            bankAccount = acct
-            uid = acctSvcrRef
-            unstructuredRemittanceInformation =
-                camt53doc.pickString("//*[local-name()='Ntry']//*[local-name()='Ustrd']")
-            transactionType = camt53doc.pickString("//*[local-name()='Ntry']//*[local-name()='CdtDbtInd']")
-            currency = camt53doc.pickString("//*[local-name()='Ntry']//*[local-name()='Amt']/@Ccy")
-            amount = camt53doc.pickString("//*[local-name()='Ntry']//*[local-name()='Amt']")
-            status = camt53doc.pickString("//*[local-name()='Ntry']//*[local-name()='Sts']")
-            this.bookingDate = LocalDateTime.from(bookingDate).millis()
-            counterpartIban =
-                camt53doc.pickString("//*[local-name()='${if (this.transactionType == "DBIT") "CdtrAcct" else "DbtrAcct"}']//*[local-name()='IBAN']")
-            counterpartName =
-                camt53doc.pickString("//*[local-name()='RltdPties']//*[local-name()='${if (this.transactionType == "DBIT") "Cdtr" else "Dbtr"}']//*[local-name()='Nm']")
-            counterpartBic = camt53doc.pickString("//*[local-name()='RltdAgts']//*[local-name()='BIC']")
+        val transactions = getTransactions(camtDoc)
+        logger.info("found ${transactions.size} transactions")
+        for (tx in transactions) {
+            val acctSvcrRef = tx.accountServicerReference
+            if (acctSvcrRef == null) {
+                // FIXME(dold): Report this!
+                logger.error("missing account servicer reference in transaction")
+                continue
+            }
+            if (isDuplicate(acctSvcrRef)) {
+                logger.info("Processing a duplicate, not storing it.")
+                return@transaction
+            }
+            RawBankTransactionEntity.new {
+                bankAccount = acct
+                accountTransactionId = "AcctSvcrRef:$acctSvcrRef"
+                amount = tx.amount
+                currency = tx.currency
+                transactionJson = jacksonObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(tx)
+                creditDebitIndicator = tx.creditDebitIndicator.name
+                status = tx.status.name
+            }
         }
     }
 }
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt
index 95cbae85..e1eb5a7c 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt
@@ -24,6 +24,8 @@ package tech.libeufin.nexus
  */
 
 import com.fasterxml.jackson.annotation.JsonInclude
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import com.fasterxml.jackson.annotation.JsonTypeInfo
 import org.w3c.dom.Document
 import tech.libeufin.util.XmlElementDestructor
 import tech.libeufin.util.destructXml
@@ -76,12 +78,12 @@ data class TransactionDetails(
     val unstructuredRemittanceInformation: String
 )
 
-abstract class AccountIdentification(type: String) : TypedEntity(type)
+abstract class AccountIdentification() : TypedEntity()
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 data class AccountIdentificationIban(
     val iban: String
-) : AccountIdentification("account-identification-iban")
+) : AccountIdentification()
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 data class AccountIdentificationGeneric(
@@ -89,7 +91,7 @@ data class AccountIdentificationGeneric(
     val issuer: String?,
     val code: String?,
     val proprietary: String?
-) : AccountIdentification("account-identification-generic")
+) : AccountIdentification()
 
 data class BankTransaction(
     val account: AccountIdentification,
@@ -124,32 +126,44 @@ data class BankTransaction(
     val details: List<TransactionDetails>,
     val valueDate: DateOrDateTime?,
     val bookingDate: DateOrDateTime?,
-    val accountServicerReference: String
+    val accountServicerReference: String?
 )
 
-abstract class TypedEntity(val type: String)
+@JsonTypeInfo(
+    use = JsonTypeInfo.Id.NAME,
+    include = JsonTypeInfo.As.PROPERTY,
+    property = "type"
+)
+@JsonSubTypes(
+    JsonSubTypes.Type(value = Agent::class, name = "agent"),
+    JsonSubTypes.Type(value = Party::class, name = "party"),
+    JsonSubTypes.Type(value = Date::class, name = "date"),
+    JsonSubTypes.Type(value = DateTime::class, name = "datetime"),
+    JsonSubTypes.Type(value = AccountIdentificationIban::class, name = "account-identification-iban"),
+    JsonSubTypes.Type(value = AccountIdentificationGeneric::class, name = "account-identification-generic")
+)
+abstract class TypedEntity()
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 class Agent(
     val name: String?,
     val bic: String
-) : TypedEntity("agent")
+) : TypedEntity()
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 class Party(
     val name: String?
-) : TypedEntity("party")
-
+) : TypedEntity()
 
-abstract class DateOrDateTime(type: String) : TypedEntity(type)
+abstract class DateOrDateTime() : TypedEntity()
 
 class Date(
     val date: String
-) : DateOrDateTime("date")
+) : DateOrDateTime()
 
 class DateTime(
     val date: String
-) : DateOrDateTime("datetime")
+) : DateOrDateTime()
 
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/JSON.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/JSON.kt
index da38273a..c75efe15 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/JSON.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/JSON.kt
@@ -188,19 +188,8 @@ data class PaymentStatus(
     val preparationDate: String
 )
 
-/** Response type of "GET /collected-transactions" */
-data class Transaction(
-    val account: String,
-    val counterpartIban: String,
-    val counterpartBic: String,
-    val counterpartName: String,
-    val amount: String,
-    val subject: String,
-    val date: String
-)
-
 data class Transactions(
-    val transactions: MutableList<Transaction> = mutableListOf()
+    val transactions: MutableList<BankTransaction> = mutableListOf()
 )
 
 /** Request type of "POST /collected-transactions" */
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
index 875d19d3..96c27494 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
@@ -263,20 +263,23 @@ fun ApplicationRequest.hasBody(): Boolean {
     return false
 }
 
+inline fun reportAndIgnoreErrors(f: () -> Unit) {
+    try {
+        f()
+    } catch (e: java.lang.Exception) {
+        logger.error("ignoring exception", e)
+    }
+}
+
 fun moreFrequentBackgroundTasks(httpClient: HttpClient) {
     GlobalScope.launch {
         while (true) {
-            logger.debug("More frequent background job")
-            ingestTalerTransactions()
-            submitPreparedPaymentsViaEbics()
-            try {
-                downloadTalerFacadesTransactions(httpClient, "C52")
-            } catch (e: Exception) {
-                val sw = StringWriter()
-                val pw = PrintWriter(sw)
-                e.printStackTrace(pw)
-                logger.info("==== Frequent background task exception ====\n${sw}======")
-            }
+            logger.debug("Running more frequent background jobs")
+            reportAndIgnoreErrors { downloadTalerFacadesTransactions(httpClient, "C53") }
+            reportAndIgnoreErrors { downloadTalerFacadesTransactions(httpClient, "C52")  }
+            reportAndIgnoreErrors { ingestTalerTransactions() }
+            reportAndIgnoreErrors { submitPreparedPaymentsViaEbics() }
+            logger.debug("More frequent background jobs done")
             delay(Duration.ofSeconds(1))
         }
     }
@@ -287,7 +290,7 @@ fun lessFrequentBackgroundTasks(httpClient: HttpClient) {
         while (true) {
             logger.debug("Less frequent background job")
             try {
-                downloadTalerFacadesTransactions(httpClient, "C53")
+                //downloadTalerFacadesTransactions(httpClient, "C53")
             } catch (e: Exception) {
                 val sw = StringWriter()
                 val pw = PrintWriter(sw)
@@ -702,25 +705,10 @@ fun serverMain(dbName: String) {
                 val end = call.request.queryParameters["end"]
                 val ret = Transactions()
                 transaction {
-                    val userId = transaction { authenticateRequest(call.request).id.value }
-                    RawBankTransactionEntity.find {
-                        (RawBankTransactionsTable.bankAccount eq bankAccount) and
-                                RawBankTransactionsTable.bookingDate.between(
-                                    parseDashedDate(start ?: "1970-01-01").millis(),
-                                    parseDashedDate(end ?: LocalDateTime.now().toDashedDate()).millis()
-                                )
-                    }.forEach {
-                        ret.transactions.add(
-                            Transaction(
-                                account = it.bankAccount.id.value,
-                                counterpartBic = it.counterpartBic,
-                                counterpartIban = it.counterpartIban,
-                                counterpartName = it.counterpartName,
-                                date = importDateFromMillis(it.bookingDate).toDashedDate(),
-                                subject = it.unstructuredRemittanceInformation,
-                                amount = "${it.currency}:${it.amount}"
-                            )
-                        )
+                    authenticateRequest(call.request).id.value
+                    RawBankTransactionEntity.all().map {
+                        val tx = jacksonObjectMapper().readValue(it.transactionJson, BankTransaction::class.java)
+                        ret.transactions.add(tx)
                     }
                 }
                 call.respond(ret)
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/taler.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/taler.kt
index 7513e120..aa5b1a3d 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/taler.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/taler.kt
@@ -38,7 +38,6 @@ import org.jetbrains.exposed.dao.Entity
 import org.jetbrains.exposed.dao.id.IdTable
 import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.transactions.transaction
-import org.w3c.dom.Document
 import tech.libeufin.util.*
 import kotlin.math.abs
 import kotlin.math.min
@@ -129,13 +128,11 @@ fun parsePayto(paytoUri: String): Payto {
      * maps to a "bic".
      */
 
-
     /**
      * payto://iban/BIC/IBAN?name=<name>
      * payto://x-taler-bank/<bank hostname>/<plain account number>
      */
-
-    val ibanMatch = Regex("payto://iban/([A-Z0-9]+)/([A-Z0-9]+)\\?name=(\\w+)").find(paytoUri)
+    val ibanMatch = Regex("payto://iban/([A-Z0-9]+)/([A-Z0-9]+)\\?receiver-name=(\\w+)").find(paytoUri)
     if (ibanMatch != null) {
         val (bic, iban, name) = ibanMatch.destructured
         return Payto(name, iban, bic.replace("/", ""))
@@ -159,16 +156,10 @@ fun <T : Entity<Long>> SizedIterable<T>.orderTaler(delta: Int): List<T> {
 }
 
 /**
- * NOTE: those payto-builders default all to the x-taler-bank transport.
- * A mechanism to easily switch transport is needed, as production needs
- * 'iban'.
+ * Build an IBAN payto URI.
  */
-fun buildPaytoUri(name: String, iban: String, bic: String): String {
-    return "payto://iban/$bic/$iban?name=$name"
-}
-
-fun buildPaytoUri(iban: String, bic: String): String {
-    return "payto://iban/$bic/$iban"
+fun buildIbanPaytoUri(iban: String, bic: String, name: String): String {
+    return "payto://iban/$bic/$iban?receiver-name=$name"
 }
 
 /** Builds the comparison operator for history entries based on the sign of 'delta'  */
@@ -240,14 +231,22 @@ fun paymentFailed(entry: RawBankTransactionEntity): Boolean {
     return false
 }
 
-// Tries to extract a valid PUB from the raw subject line
-fun normalizeSubject(rawSubject: String): String {
+/**
+ * Tries to extract a valid reserve public key from the raw subject line
+ */
+fun extractReservePubFromSubject(rawSubject: String): String? {
     val re = "\\b[a-z0-9A-Z]{52}\\b".toRegex()
-    val result = re.find("1ENVZ6EYGB6Z509KRJ6E59GK1EQXZF8XXNY9SN33C2KDGSHV9KA0")
-    if (result == null) throw NexusError(
-        HttpStatusCode.BadRequest, "Reserve pub not found in subject: ${rawSubject}"
-    )
-    return result.value
+    val result = re.find(rawSubject) ?: return null
+    return result.value.toUpperCase()
+}
+
+/**
+ * Tries to extract a valid wire transfer id from the subject.
+ */
+fun extractWtidFromSubject(rawSubject: String): String? {
+    val re = "\\b[a-z0-9A-Z]{52}\\b".toRegex()
+    val result = re.find(rawSubject) ?: return null
+    return result.value.toUpperCase()
 }
 
 fun getTalerFacadeState(fcid: String): TalerFacadeStateEntity {
@@ -256,7 +255,7 @@ fun getTalerFacadeState(fcid: String): TalerFacadeStateEntity {
         "Could not find facade '${fcid}'"
     )
     val facadeState = TalerFacadeStateEntity.find {
-        TalerFacadeStatesTable.facade eq facade.id.value
+        TalerFacadeStateTable.facade eq facade.id.value
     }.firstOrNull() ?: throw NexusError(
         HttpStatusCode.NotFound,
         "Could not find any state for facade: ${fcid}"
@@ -270,7 +269,7 @@ fun getTalerFacadeBankAccount(fcid: String): NexusBankAccountEntity {
         "Could not find facade '${fcid}'"
     )
     val facadeState = TalerFacadeStateEntity.find {
-        TalerFacadeStatesTable.facade eq facade.id.value
+        TalerFacadeStateTable.facade eq facade.id.value
     }.firstOrNull() ?: throw NexusError(
         HttpStatusCode.NotFound,
         "Could not find any state for facade: ${fcid}"
@@ -424,12 +423,15 @@ suspend fun submitPreparedPaymentsViaEbics() {
                 HttpStatusCode.InternalServerError,
                 "Such facade '${it.facade.id.value}' doesn't map to any Ebics subscriber"
             )
-            val bankAccount: NexusBankAccountEntity = NexusBankAccountEntity.findById(it.bankAccount) ?: throw NexusError(
-                HttpStatusCode.InternalServerError,
-                "Bank account '${it.bankAccount}' not found for facade '${it.id.value}'"
-            )
-            PreparedPaymentEntity.find { PreparedPaymentsTable.debitorIban eq bankAccount.iban and
-                    not(PreparedPaymentsTable.submitted) }.forEach {
+            val bankAccount: NexusBankAccountEntity =
+                NexusBankAccountEntity.findById(it.bankAccount) ?: throw NexusError(
+                    HttpStatusCode.InternalServerError,
+                    "Bank account '${it.bankAccount}' not found for facade '${it.id.value}'"
+                )
+            PreparedPaymentEntity.find {
+                PreparedPaymentsTable.debitorIban eq bankAccount.iban and
+                        not(PreparedPaymentsTable.submitted)
+            }.forEach {
                 val pain001document = createPain001document(it)
                 logger.debug("Preparing payment: ${pain001document}")
                 val subscriberDetails = getEbicsSubscriberDetailsInternal(subscriberEntity)
@@ -452,6 +454,69 @@ suspend fun submitPreparedPaymentsViaEbics() {
     }
 }
 
+private fun ingestIncoming(payment: RawBankTransactionEntity, txDtls: TransactionDetails) {
+    val subject = txDtls.unstructuredRemittanceInformation
+    val debtorName = txDtls.relatedParties.debtor?.name
+    if (debtorName == null) {
+        logger.warn("empty debtor name")
+        return
+    }
+    val debtorAcct = txDtls.relatedParties.debtorAccount
+    if (debtorAcct == null) {
+        // FIXME: Report payment, we can't even send it back
+        logger.warn("empty debitor account")
+        return
+    }
+    if (debtorAcct !is AccountIdentificationIban) {
+        // FIXME: Report payment, we can't even send it back
+        logger.warn("non-iban debitor account")
+        return
+    }
+    val debtorAgent = txDtls.relatedParties.debtorAgent
+    if (debtorAgent == null) {
+        // FIXME: Report payment, we can't even send it back
+        logger.warn("missing debitor agent")
+        return
+    }
+    val reservePub = extractReservePubFromSubject(subject)
+    if (reservePub == null) {
+        // FIXME: send back!
+        logger.warn("could not find reserve pub in remittance information")
+        return
+    }
+    if (!CryptoUtil.checkValidEddsaPublicKey(reservePub)) {
+        // FIXME: send back!
+        logger.warn("invalid public key")
+        return
+    }
+    TalerIncomingPaymentEntity.new {
+        this.payment = payment
+        reservePublicKey = reservePub
+        timestampMs = System.currentTimeMillis()
+        incomingPaytoUri = buildIbanPaytoUri(debtorAcct.iban, debtorAgent.bic, debtorName)
+    }
+    return
+}
+
+private fun ingestOutgoing(payment: RawBankTransactionEntity, txDtls: TransactionDetails) {
+    val subject = txDtls.unstructuredRemittanceInformation
+    logger.debug("Ingesting outgoing payment: subject")
+    val wtid = extractWtidFromSubject(subject)
+    if (wtid == null) {
+        logger.warn("did not find wire transfer ID in outgoing payment")
+        return
+    }
+    val talerRequested = TalerRequestedPaymentEntity.find {
+        TalerRequestedPayments.wtid eq subject
+    }.firstOrNull()
+    if (talerRequested == null) {
+        logger.info("Payment '${subject}' shows in history, but was never requested!")
+        return
+    }
+    logger.debug("Payment: ${subject} was requested, and gets now marked as 'confirmed'")
+    talerRequested.rawConfirmed = payment
+}
+
 /**
  * Crawls the database to find ALL the users that have a Taler
  * facade and process their histories respecting the TWG policy.
@@ -462,7 +527,7 @@ suspend fun submitPreparedPaymentsViaEbics() {
  */
 fun ingestTalerTransactions() {
     fun ingest(subscriberAccount: NexusBankAccountEntity, facade: FacadeEntity) {
-        logger.debug("Ingesting transactions for Taler facade: ${facade.id.value}")
+        logger.debug("Ingesting transactions for Taler facade ${facade.id.value}")
         val facadeState = getTalerFacadeState(facade.id.value)
         var lastId = facadeState.highestSeenMsgID
         RawBankTransactionEntity.find {
@@ -474,32 +539,15 @@ fun ingestTalerTransactions() {
                     (RawBankTransactionsTable.id.greater(lastId))
         }.orderBy(Pair(RawBankTransactionsTable.id, SortOrder.ASC)).forEach {
             // Incoming payment.
-            if (it.transactionType == "CRDT") {
-                val normalizedSubject = normalizeSubject(it.unstructuredRemittanceInformation)
-                if (CryptoUtil.checkValidEddsaPublicKey(normalizedSubject)) {
-                    TalerIncomingPaymentEntity.new {
-                        payment = it
-                        valid = true
-                    }
-                } else {
-                    TalerIncomingPaymentEntity.new {
-                        payment = it
-                        valid = false
-                    }
-                }
-            }
-            // Outgoing payment
-            if (it.transactionType == "DBIT") {
-                logger.debug("Ingesting outgoing payment: ${it.unstructuredRemittanceInformation}")
-                val talerRequested = TalerRequestedPaymentEntity.find {
-                    TalerRequestedPayments.wtid eq it.unstructuredRemittanceInformation
-                }.firstOrNull()
-                if (talerRequested == null){
-                    logger.info("Payment '${it.unstructuredRemittanceInformation}' shows in history, but was never requested!")
-                    return@forEach
+            val tx = jacksonObjectMapper().readValue(it.transactionJson, BankTransaction::class.java)
+            if (tx.isBatch) {
+                // We don't support batch transactions at the moment!
+                logger.warn("batch transactions not supported")
+            } else {
+                when (tx.creditDebitIndicator) {
+                    CreditDebitIndicator.DBIT -> ingestOutgoing(it, txDtls = tx.details[0])
+                    CreditDebitIndicator.CRDT -> ingestIncoming(it, txDtls = tx.details[0])
                 }
-                logger.debug("Payment: ${it.unstructuredRemittanceInformation} was requested, and gets now marked as 'confirmed'")
-                talerRequested.rawConfirmed = it
             }
             lastId = it.id.value
         }
@@ -529,6 +577,7 @@ suspend fun historyOutgoing(call: ApplicationCall): Unit {
     val history = TalerOutgoingHistory()
     transaction {
         val user = authenticateRequest(call.request)
+
         /** Retrieve all the outgoing payments from the _clean Taler outgoing table_ */
         val subscriberBankAccount = getTalerFacadeBankAccount(expectNonNull(call.parameters["fcid"]))
         val reqPayments = TalerRequestedPaymentEntity.find {
@@ -541,13 +590,13 @@ suspend fun historyOutgoing(call: ApplicationCall): Unit {
                         row_id = it.id.value,
                         amount = it.amount,
                         wtid = it.wtid,
-                        date = GnunetTimestamp(
-                            it.rawConfirmed?.bookingDate?.div(1000) ?: throw NexusError(
-                                HttpStatusCode.InternalServerError, "Null value met after check, VERY strange."
-                            )
-                        ),
+                        date = GnunetTimestamp(it.preparedPayment.preparationDate),
                         credit_account = it.creditAccount,
-                        debit_account = buildPaytoUri(subscriberBankAccount.iban, subscriberBankAccount.bankCode),
+                        debit_account = buildIbanPaytoUri(
+                            subscriberBankAccount.iban,
+                            subscriberBankAccount.bankCode,
+                            subscriberBankAccount.accountHolder
+                        ),
                         exchange_base_url = "FIXME-to-request-along-subscriber-registration"
                     )
                 )
@@ -573,26 +622,22 @@ suspend fun historyIncoming(call: ApplicationCall): Unit {
     val startCmpOp = getComparisonOperator(delta, start, TalerIncomingPayments)
     transaction {
         val orderedPayments = TalerIncomingPaymentEntity.find {
-            TalerIncomingPayments.valid eq true and startCmpOp
+            startCmpOp
         }.orderTaler(delta)
         if (orderedPayments.isNotEmpty()) {
             orderedPayments.subList(0, min(abs(delta), orderedPayments.size)).forEach {
                 history.incoming_transactions.add(
                     TalerIncomingBankTransaction(
-                        date = GnunetTimestamp(it.payment.bookingDate / 1000),
+                        date = GnunetTimestamp(it.timestampMs),
                         row_id = it.id.value,
                         amount = "${it.payment.currency}:${it.payment.amount}",
-                        reserve_pub = it.payment.unstructuredRemittanceInformation,
-                        credit_account = buildPaytoUri(
-                            it.payment.bankAccount.accountHolder,
+                        reserve_pub = it.reservePublicKey,
+                        credit_account = buildIbanPaytoUri(
                             it.payment.bankAccount.iban,
-                            it.payment.bankAccount.bankCode
+                            it.payment.bankAccount.bankCode,
+                            it.payment.bankAccount.accountHolder
                         ),
-                        debit_account = buildPaytoUri(
-                            it.payment.counterpartName,
-                            it.payment.counterpartIban,
-                            it.payment.counterpartBic
-                        )
+                        debit_account = it.incomingPaytoUri
                     )
                 )
             }
diff --git a/nexus/src/test/kotlin/DBTest.kt b/nexus/src/test/kotlin/DBTest.kt
index 4f1078c7..2dc2ebc9 100644
--- a/nexus/src/test/kotlin/DBTest.kt
+++ b/nexus/src/test/kotlin/DBTest.kt
@@ -41,7 +41,7 @@ class DBTest {
                 addLogger(StdOutSqlLogger)
                 SchemaUtils.create(
                     FacadesTable,
-                    TalerFacadeStatesTable,
+                    TalerFacadeStateTable,
                     NexusUsersTable
                 )
                 val user = NexusUserEntity.new("u") {
diff --git a/nexus/src/test/kotlin/Iso20022Test.kt b/nexus/src/test/kotlin/Iso20022Test.kt
index 377c9779..79a2e137 100644
--- a/nexus/src/test/kotlin/Iso20022Test.kt
+++ b/nexus/src/test/kotlin/Iso20022Test.kt
@@ -20,6 +20,10 @@ class Iso20022Test {
     fun testTransactionsImport() {
         val camt53 = loadXmlResource("iso20022-samples/camt.053.001.02.gesamtbeispiel.xml")
         val txs = getTransactions(camt53)
-        println(jacksonObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(txs))
+        for (tx in txs) {
+            val txStr = jacksonObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(tx)
+            println(txStr)
+            val tx2 = jacksonObjectMapper().readValue(txStr, BankTransaction::class.java)
+        }
     }
 }
\ No newline at end of file
diff --git a/nexus/src/test/kotlin/SubjectNormalization.kt b/nexus/src/test/kotlin/SubjectNormalization.kt
index 63f02d23..4c244cae 100644
--- a/nexus/src/test/kotlin/SubjectNormalization.kt
+++ b/nexus/src/test/kotlin/SubjectNormalization.kt
@@ -1,13 +1,13 @@
 import org.junit.Test
 
-import tech.libeufin.nexus.normalizeSubject
+import tech.libeufin.nexus.extractReservePubFromSubject
 
 class SubjectNormalization {
 
     @Test
     fun testBeforeAndAfter() {
         val mereValue = "1ENVZ6EYGB6Z509KRJ6E59GK1EQXZF8XXNY9SN33C2KDGSHV9KA0"
-        assert(mereValue == normalizeSubject(mereValue))
-        assert(mereValue == normalizeSubject("noise before ${mereValue} noise after"))
+        assert(mereValue == extractReservePubFromSubject(mereValue))
+        assert(mereValue == extractReservePubFromSubject("noise before ${mereValue} noise after"))
     }
 }
\ No newline at end of file