libeufin

Integration and sandbox testing for FinTech APIs and data formats
Log | Files | Refs | Submodules | README | LICENSE

EbicsFetch.kt (23390B)

      1 /*
      2  * This file is part of LibEuFin.
      3  * Copyright (C) 2024, 2025, 2026 Taler Systems S.A.
      4 
      5  * LibEuFin is free software; you can redistribute it and/or modify
      6  * it under the terms of the GNU Affero General Public License as
      7  * published by the Free Software Foundation; either version 3, or
      8  * (at your option) any later version.
      9 
     10  * LibEuFin is distributed in the hope that it will be useful, but
     11  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
     12  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Affero General
     13  * Public License for more details.
     14 
     15  * You should have received a copy of the GNU Affero General Public
     16  * License along with LibEuFin; see the file COPYING.  If not, see
     17  * <http://www.gnu.org/licenses/>
     18  */
     19 
     20 package tech.libeufin.nexus.cli
     21 
     22 import com.github.ajalt.clikt.core.CliktCommand
     23 import com.github.ajalt.clikt.core.Context
     24 import com.github.ajalt.clikt.core.ProgramResult
     25 import com.github.ajalt.clikt.parameters.arguments.*
     26 import com.github.ajalt.clikt.parameters.groups.provideDelegate
     27 import com.github.ajalt.clikt.parameters.options.*
     28 import com.github.ajalt.clikt.parameters.types.enum
     29 import kotlin.math.min
     30 import kotlinx.coroutines.*
     31 import kotlinx.serialization.Serializable
     32 import kotlinx.serialization.Contextual
     33 import tech.libeufin.common.*
     34 import tech.libeufin.nexus.*
     35 import tech.libeufin.nexus.db.*
     36 import tech.libeufin.nexus.db.PaymentDAO.*
     37 import tech.libeufin.nexus.iso20022.*
     38 import tech.libeufin.ebics.*
     39 import java.io.IOException
     40 import java.io.InputStream
     41 import java.time.*
     42 import org.slf4j.Logger
     43 import org.slf4j.LoggerFactory
     44 import java.time.temporal.*
     45 
     46 /** Register an outgoing [payment] into [db] */
     47 suspend fun registerOutgoingPayment(
     48     db: Database,
     49     payment: OutgoingPayment
     50 ): OutgoingRegistrationResult {
     51     val metadata: Triple<ShortHashCode, BaseURL, String?>? = payment.subject?.let { 
     52         runCatching { parseOutgoingSubject(it) }.getOrNull()
     53     }
     54     val result = db.payment.registerOutgoing(payment, metadata?.first, metadata?.second, metadata?.third)
     55     if (result.new) {
     56         if (result.initiated)
     57             logger.info("$payment")
     58         else 
     59             logger.warn("$payment recovered")
     60     } else {
     61         logger.debug("{} already seen", payment)
     62     }
     63     return result
     64 }
     65 
     66 /** Register an outgoing [payment] into [db] */
     67 suspend fun registerOutgoingBatch(
     68     db: Database,
     69     batch: OutgoingBatch
     70 ) {
     71     logger.info("BATCH ${batch.executionTime.fmtDate()} ${batch.msgId}")
     72     for (it in db.initiated.unsettledTxInBatch(batch.msgId, batch.executionTime)) {
     73         registerOutgoingPayment(db, it)
     74     }
     75 }
     76 
     77 /** 
     78  * Register an incoming [payment] into [db]
     79  * Stores the payment into valid talerable ones or bounces it
     80  */
     81 suspend fun registerIncomingPayment(
     82     db: Database,
     83     cfg: NexusIngestConfig,
     84     payment: IncomingPayment,
     85 ) {
     86     fun logRes(res: InResult, kind: String = "", suffix: String = "") {
     87         val fmt = buildString {
     88             append(payment)
     89             if (kind != "") {
     90                 append(" ")
     91                 append(kind)
     92             }
     93             if (res.new) {
     94                 if (res.bounceId != null) {
     95                     append(" bounced in ${res.bounceId}")
     96                 }
     97             } else {
     98                 if (res.completed) {
     99                     append(" completed")
    100                     if (res.bounceId != null) {
    101                         append(" bounced in ${res.bounceId}")
    102                     }
    103                 } else {
    104                     if (res.bounceId != null) {
    105                         append(" already bounced in ${res.bounceId}")
    106                     } 
    107                 }
    108             }
    109             if (suffix != "") {
    110                 append(" ")
    111                 append(suffix)
    112             }
    113         }
    114         if (res.completed || res.new) {
    115             logger.info(fmt)
    116         } else {
    117             logger.debug(fmt)
    118         }
    119     }
    120     suspend fun bounce(cause: String) {
    121         if (payment.id == null) {
    122             logger.debug("{} ignored: missing bank ID", payment)
    123             return
    124         }
    125         when (cfg.accountType) {
    126             AccountType.exchange -> {
    127                 if (payment.executionTime < cfg.ignoreBouncesBefore) {
    128                     val res = db.payment.registerIncoming(payment)
    129                     logRes(res, suffix = "ignored bounce: $cause")
    130                 } else {
    131                     var bounceAmount = payment.amount
    132                     if (payment.creditFee != null && cfg.bounceDeduceFee) {
    133                         if (payment.creditFee > bounceAmount) {
    134                             val res = db.payment.registerIncoming(payment)
    135                             logRes(res, suffix = "skip bounce (transfer fee higher than amount): $cause")
    136                             return
    137                         }
    138                         bounceAmount -= payment.creditFee
    139                     }
    140                     if (cfg.bounceFee > bounceAmount) {
    141                         val res = db.payment.registerIncoming(payment)
    142                         logRes(res, suffix = "skip bounce (bounce fee higher than amount): $cause")
    143                         return
    144                     }
    145                     bounceAmount -= cfg.bounceFee
    146                     val res = db.payment.registerMalformedIncoming(
    147                         payment,
    148                         bounceAmount, 
    149                         randEbicsId(),
    150                         Instant.now(),
    151                         cause
    152                     )
    153                     when (res) {
    154                         IncomingBounceRegistrationResult.Talerable ->
    155                             logger.warn("{} tried to bounce a talerable transaction", payment)
    156                         is IncomingBounceRegistrationResult.Success -> 
    157                             logRes(res, suffix=": $cause")
    158                     }
    159                 }
    160             }
    161             AccountType.normal -> {
    162                 val res = db.payment.registerIncoming(payment)
    163                 logRes(res)
    164             }
    165         }
    166     }
    167     // Check we have enough info to handle this transaction
    168     if (payment.debtor == null || payment.debtor.receiverName == null) {
    169         val res = db.payment.registerIncoming(payment)
    170         logRes(res, kind = "incomplete")
    171         return
    172     }
    173     if (cfg.restrictionPaytoRegex != null) {
    174         if (!cfg.restrictionPaytoRegex.matches(payment.debtor.toString())) {
    175             bounce("restricted account")
    176             return
    177         }
    178     }
    179     // Else we try to parse the incoming subject
    180     if (payment.subject != null && subjectIsQrBill(payment.subject)) {
    181         when (val res = db.payment.registerQrBillIncoming(payment, payment.subject)) {
    182             IncomingRegistrationResult.ReservePubReuse -> bounce("reverse pub reuse")
    183             IncomingRegistrationResult.MappingReuse -> bounce("mapping reuse")
    184             IncomingRegistrationResult.UnknownMapping -> bounce("unknown mapping")
    185             is IncomingRegistrationResult.Success -> logRes(res)
    186         }
    187     } else {
    188         runCatching { parseIncomingSubject(payment.subject) }.fold(
    189             onSuccess = { metadata -> 
    190                 if (metadata is IncomingSubject.AdminBalanceAdjust) {
    191                     val res = db.payment.registerIncoming(payment)
    192                     logRes(res, kind = "admin balance adjust")
    193                 } else {
    194                     when (val res = db.payment.registerTalerableIncoming(payment, metadata)) {
    195                         IncomingRegistrationResult.ReservePubReuse -> bounce("reverse pub reuse")
    196                         IncomingRegistrationResult.MappingReuse -> bounce("mapping reuse")
    197                         IncomingRegistrationResult.UnknownMapping -> bounce("unknown mapping")
    198                         is IncomingRegistrationResult.Success -> logRes(res)
    199                     }
    200                 }
    201             },
    202             onFailure = { e -> bounce(e.fmt())}
    203         )
    204     }
    205 }
    206 
    207 /** Register a [tx] notification into [db] */
    208 suspend fun registerTransaction(
    209     db: Database,
    210     cfg: NexusIngestConfig,
    211     tx: TxNotification,
    212 ) {
    213     if (tx.executionTime < cfg.ignoreTransactionsBefore) {
    214         logger.debug("IGNORE {}", tx)
    215     } else {
    216         when (tx) {
    217             is IncomingPayment -> registerIncomingPayment(db, cfg, tx)
    218             is OutgoingPayment -> registerOutgoingPayment(db, tx)
    219             is OutgoingBatch -> registerOutgoingBatch(db, tx)
    220             is OutgoingReversal -> {
    221                 logger.error("{}", tx)
    222                 db.initiated.txStatusUpdate(tx.endToEndId, tx.msgId, StatusUpdate.permanent_failure, "Payment bounced: ${tx.reason}")
    223             }
    224         }
    225     }
    226 }
    227 
    228 /** Register a single EBICS [xml] txs [document] into [db] */
    229 suspend fun registerTxs(
    230     db: Database,
    231     cfg: NexusConfig,
    232     xml: InputStream
    233 ): Int {
    234     var nbTx: Int = 0
    235     parseTx(xml).forEach { accountTx ->
    236         if (accountTx.iban == cfg.ebics.account.iban.toString()) {
    237             require(accountTx.currency == null || accountTx.currency == cfg.currency) { "Expected transactions of currency ${cfg.currency} got ${accountTx.currency}" } 
    238             accountTx.txs.forEach { tx ->
    239                 when (tx) {
    240                     is IncomingPayment ->
    241                         require(tx.amount.currency == cfg.currency) { "Expected transactions of currency ${cfg.currency} got ${tx.amount.currency}" } 
    242                     is OutgoingPayment ->
    243                         require(tx.amount.currency == cfg.currency) { "Expected transactions of currency ${cfg.currency} got ${tx.amount.currency}" } 
    244                     is OutgoingBatch, is OutgoingReversal -> {}
    245                 }
    246                 registerTransaction(db, cfg.ingest, tx)
    247                 nbTx += 1
    248             }
    249         } else {
    250             logger.debug("Skip transaction for unknown account ${accountTx.iban}")
    251         }
    252     }
    253     return nbTx
    254 }
    255 
    256 /** Register a single EBICS [xml] [document] into [db] */
    257 suspend fun registerFile(
    258     db: Database,
    259     cfg: NexusConfig,
    260     xml: InputStream,
    261     doc: OrderDoc
    262 ) {
    263     when (doc) {
    264         OrderDoc.report, OrderDoc.statement, OrderDoc.notification -> {
    265             try {
    266                 registerTxs(db, cfg, xml)
    267             } catch (e: Exception) {
    268                 throw Exception("Ingesting transactions files failed", e)
    269             }
    270         }
    271         OrderDoc.acknowledgement -> {
    272             val acks = parseCustomerAck(xml)
    273             for (ack in acks) {
    274                 when (ack.actionType) {
    275                     HacAction.ORDER_HAC_FINAL_POS -> {
    276                         logger.debug("{}", ack)
    277                         db.initiated.orderSuccess(ack.orderId!!)?.let { messageId ->
    278                             logger.info("Batch $messageId order ${ack.orderId} accepted at ${ack.timestamp.fmtDateTime()}")
    279                         }
    280                     }
    281                     HacAction.ORDER_HAC_FINAL_NEG -> {
    282                         logger.debug("{}", ack)
    283                         db.initiated.orderFailure(ack.orderId!!)?.let { (messageId, msg) ->
    284                             logger.error("Batch $messageId order ${ack.orderId} refused at ${ack.timestamp.fmtDateTime()}${if (msg != null) ": $msg" else ""}")
    285                         }
    286                     }
    287                     else -> {
    288                         logger.debug("{}", ack)
    289                         if (ack.orderId != null) {
    290                             db.initiated.orderStep(ack.orderId, ack.msg())
    291                         }
    292                     }
    293                 }
    294             }
    295         }
    296         OrderDoc.status -> {
    297             val msgStatus = parseCustomerPaymentStatusReport(xml)
    298             logger.debug("{}", msgStatus)
    299             if (msgStatus.code != null) {
    300                 val msg = msgStatus.msg()
    301                 val state = when (msgStatus.code) {
    302                     ExternalPaymentGroupStatusCode.ACSC -> StatusUpdate.success
    303                     ExternalPaymentGroupStatusCode.RJCT -> {
    304                         logger.error("Batch ${msgStatus.id} failed: $msg")
    305                         StatusUpdate.permanent_failure
    306                     }
    307                     else -> StatusUpdate.pending
    308                 }
    309                 db.initiated.batchStatusUpdate(msgStatus.id, state, msg)
    310             }
    311             for (pmtStatus in msgStatus.payments) {
    312                 if (pmtStatus.id != "NOTPROVIDED") {
    313                     logger.warn("Unexpected payment status for ${msgStatus.id}.${pmtStatus.id}")
    314                 } else if (pmtStatus.code != null) {
    315                     val msg = pmtStatus.msg()
    316                     val state = when (pmtStatus.code) {
    317                         ExternalPaymentGroupStatusCode.ACSC -> StatusUpdate.success
    318                         ExternalPaymentGroupStatusCode.RJCT -> {
    319                             logger.error("Batch ${msgStatus.id} failed: $msg")
    320                             StatusUpdate.permanent_failure
    321                         }
    322                         else -> StatusUpdate.pending
    323                     }
    324                     db.initiated.batchStatusUpdate(msgStatus.id, state, msg)
    325                 }
    326                 for (txStatus in pmtStatus.transactions) {
    327                     val msg = txStatus.msg()
    328                     val state = when (txStatus.code) {
    329                         ExternalPaymentTransactionStatusCode.RJCT,
    330                         ExternalPaymentTransactionStatusCode.BLCK -> {
    331                             logger.error("Transaction ${txStatus.endToEndId} failed: $msg")
    332                             StatusUpdate.permanent_failure
    333                         }
    334                         else -> StatusUpdate.pending
    335                     }
    336                     db.initiated.txStatusUpdate(txStatus.endToEndId, null, state, msg)
    337                 }
    338             }
    339         }
    340     }
    341 }
    342 
    343 /** Register an EBICS [payload] of [doc] into [db] */
    344 private suspend fun registerPayload(
    345     db: Database,
    346     cfg: NexusConfig,
    347     payload: InputStream,
    348     doc: OrderDoc
    349 ) { 
    350     // Unzip payload if necessary
    351     when (doc) {
    352         OrderDoc.status,
    353         OrderDoc.report,
    354         OrderDoc.statement, 
    355         OrderDoc.notification -> {
    356             try {
    357                 payload.unzipEach { fileName, xml ->
    358                     logger.trace("parse $fileName")
    359                     registerFile(db, cfg, xml, doc)
    360                 }
    361             } catch (e: IOException) {
    362                 throw Exception("Could not open any ZIP archive", e)
    363             }
    364         }
    365         OrderDoc.acknowledgement -> registerFile(db, cfg, payload, doc)
    366     }
    367 }
    368 
    369 /** 
    370  * Fetch and register banking records from [orders] using EBICS [client] starting from [pinnedStart]
    371  * 
    372  * If [pinnedStart] is null fetch new records.
    373  */
    374 private suspend fun fetchEbicsDocuments(
    375     client: EbicsClient,
    376     db: Database,
    377     cfg: NexusConfig,
    378     orders: Collection<EbicsOrder>,
    379     pinnedStart: Instant?,
    380     peek: Boolean
    381 ): Boolean {
    382     val lastExecutionTime: Instant? = pinnedStart
    383     var success = true
    384     for ((doc, orders) in orders.groupBy { it.doc() }) {
    385         if (doc == null) {
    386             logger.debug("Skip unsupported orders {}", orders)
    387         } else {
    388             if (lastExecutionTime == null) {
    389                 logger.info("Fetching new '${doc.fullDescription()}'")
    390             } else {
    391                 logger.info("Fetching '${doc.fullDescription()}' from timestamp: $lastExecutionTime")
    392             }
    393             for (order in orders) {
    394                 try {
    395                     client.download(
    396                         order,
    397                         lastExecutionTime,
    398                         null,
    399                         peek
    400                     ) { payload ->
    401                         registerPayload(db, cfg, payload, doc)
    402                     }
    403                 } catch (e: EbicsError.Code) {
    404                     when (e.bankCode) {
    405                         EbicsReturnCode.EBICS_NO_DOWNLOAD_DATA_AVAILABLE -> continue
    406                         EbicsReturnCode.EBICS_AUTHORISATION_ORDER_IDENTIFIER_FAILED -> {
    407                             e.fmtLog(logger)
    408                             success = false
    409                             continue
    410                         }
    411                         else -> throw e
    412                     }
    413                 }
    414             }
    415         }
    416     }
    417     return success
    418 }
    419 
    420 @Serializable
    421 data class Checkpoint(
    422     @Contextual
    423     val last_successfull: Instant? = null,
    424     @Contextual
    425     val last_trial: Instant? = null
    426 )
    427 
    428 class EbicsFetch: EbicsCmd() {
    429     override fun help(context: Context) = "Downloads and parse EBICS files from the bank and register them into the database"
    430 
    431     private val documents: Set<OrderDoc> by argument(
    432         help = "Which documents should be fetched? If none are specified, all supported documents will be fetched",
    433         helpTags = OrderDoc.entries.associate { Pair(it.name, it.shortDescription()) },
    434     ).enum<OrderDoc>().multiple().unique()
    435     private val pinnedStart by option(
    436         help = "Only supported in --transient mode, this option lets specify the earliest timestamp of the downloaded documents",
    437         metavar = "YYYY-MM-DD"
    438     ).convert { dateToInstant(it) }
    439     private val peek by option("--peek",
    440         help = "Only supported in --transient mode, do not consume fetched documents"
    441     ).flag()
    442     private val transientCheckpoint by option("--checkpoint",
    443         help = "Only supported in --transient mode, run a checkpoint"
    444     ).flag()
    445 
    446     override fun run() = cliCmd(logger) {
    447         nexusConfig(config).withDb { db, cfg ->
    448             val (clientKeys, bankKeys) = expectFullKeys(cfg.ebics)
    449             val client = EbicsClient(
    450                 cfg.ebics.host,
    451                 httpClient(),
    452                 db.ebics,
    453                 EbicsLogger(ebicsLog),
    454                 clientKeys,
    455                 bankKeys
    456             )
    457             val docs = if (documents.isEmpty()) OrderDoc.entries else documents.toList()
    458 
    459             // EBICS order than should be fetched
    460             val selectedOrder = docs.flatMap { cfg.ebics.dialect.downloadDoc(it) }
    461 
    462             // Try to obtain real-time notification channel if not transient
    463             val wssNotification = if (transient) {
    464                 logger.info("Transient mode: fetching once and returning")
    465                 null
    466             } else {
    467                 val tmp = listenForNotification(client)
    468                 logger.info("Running with a frequency of ${cfg.fetch.frequencyRaw}")
    469                 tmp
    470             }
    471 
    472             var lastFetch = Instant.EPOCH
    473             
    474             while (true) {
    475                 val checkpoint = db.kv.get<TaskStatus>(CHECKPOINT_KEY) ?: TaskStatus()
    476                 var nextFetch = lastFetch + cfg.fetch.frequency
    477                 var nextCheckpoint = run {
    478                     // We never ran, we must checkpoint now
    479                     if (checkpoint.last_trial == null) {
    480                         Instant.EPOCH
    481                     } else {
    482                         // We run today at checkpointTime
    483                         val checkpointDate = OffsetDateTime.now().with(cfg.fetch.checkpointTime)
    484                         val checkpointToday = checkpointDate.toInstant()
    485                         // If we already ran today we ruAn tomorrow
    486                         if (checkpoint.last_trial > checkpointToday) {
    487                             checkpointDate.plusDays(1).toInstant()
    488                         } else {
    489                             checkpointToday
    490                         }
    491                     }
    492                 }
    493 
    494                 val now = Instant.now()
    495                 var success: Boolean = true
    496                 if (
    497                     // Run transient checkpoint at request
    498                     (transient && transientCheckpoint) ||
    499                     // Or run recurrent checkpoint
    500                     (!transient && now > nextCheckpoint)
    501                 ) {
    502                     logger.info("Running checkpoint")
    503                     
    504                     val since = if (transient && pinnedStart != null && (checkpoint.last_successfull == null || pinnedStart!!.isBefore(checkpoint.last_successfull))) {
    505                         pinnedStart
    506                     } else {
    507                         checkpoint.last_successfull
    508                     }
    509                     success = try {
    510                         /// We fetch HKD to only fetch supported EBICS orders and get the document versions
    511                         val orders = client.download(EbicsOrder.V3.HKD) { stream ->
    512                             val hkd = EbicsAdministrative.parseHKD(stream)
    513                             val supportedOrder = hkd.partner.orders.map { it.order }
    514                             logger.debug {
    515                                 val fmt = supportedOrder.map(EbicsOrder::description).joinToString(" ")
    516                                 "HKD: ${fmt}"
    517                             }
    518                             selectedOrder select supportedOrder
    519                         }
    520                         fetchEbicsDocuments(client, db, cfg, orders, since, transient && peek)
    521                     } catch (e: Exception) {
    522                         e.fmtLog(logger)
    523                         false
    524                     }
    525                     db.kv.updateTaskStatus(CHECKPOINT_KEY, now, success)
    526                     lastFetch = now
    527                 } else if (transient || now > nextFetch) {
    528                     if (!transient) logger.info("Running at frequency")
    529                     success = try {
    530                         /// We fetch HAA to only fetch pending & supported EBICS orders and get the document versions
    531                         val orders = client.download(EbicsOrder.V3.HAA) { stream ->
    532                             val haa = EbicsAdministrative.parseHAA(stream)
    533                             logger.debug {
    534                                 val orders = haa.orders.map(EbicsOrder::description).joinToString(" ")
    535                                 "HAA: ${orders}"
    536                             }
    537                             selectedOrder select haa.orders
    538                         }
    539                         fetchEbicsDocuments(client, db, cfg, orders, if (transient) pinnedStart else null, transient && peek)
    540                     } catch (e: Exception) {
    541                         e.fmtLog(logger)
    542                         false
    543                     }
    544                     lastFetch = now
    545                 }
    546                 db.kv.updateTaskStatus(SUBMIT_TASK_KEY, now, success)
    547 
    548                 if (transient) throw ProgramResult(if (!success) 1 else 0)
    549 
    550                 val delay = min(ChronoUnit.MILLIS.between(now, nextFetch), ChronoUnit.MILLIS.between(now, nextCheckpoint))
    551                 if (wssNotification == null) {
    552                     delay(delay)
    553                 } else {
    554                     val notifications = withTimeoutOrNull(delay) {
    555                         wssNotification.receive()
    556                     }
    557                     if (notifications != null) {
    558                         // Only fetch requested and supported orders
    559                         val orders = selectedOrder select notifications
    560                         if (orders.isNotEmpty()) {
    561                             logger.info("Running at real-time notifications reception")
    562                             fetchEbicsDocuments(client, db, cfg, notifications, null, false)
    563                         }
    564                     }
    565                 }
    566             }
    567         }
    568     }
    569 }