Support IGNORE_TRANSACTIONS_BEFORE

6 files changed, 67 insertions, 40 deletions

diff --git a/common/src/main/kotlin/TalerConfig.kt b/common/src/main/kotlin/TalerConfig.kt
index c0148278..8961ff47 100644
--- a/common/src/main/kotlin/TalerConfig.kt
+++ b/common/src/main/kotlin/TalerConfig.kt
@@ -24,8 +24,9 @@ import org.slf4j.LoggerFactory
 import java.nio.file.AccessDeniedException
 import java.nio.file.NoSuchFileException
 import java.nio.file.Path
+import java.time.*
 import java.time.temporal.ChronoUnit
-import java.time.Duration
+import java.time.format.DateTimeParseException
 import kotlin.io.path.*
 
 private val logger: Logger = LoggerFactory.getLogger("libeufin-config")
@@ -491,6 +492,20 @@ class TalerConfig internal constructor(
     fun requireDuration(section: String, option: String): Duration =
         lookupDuration(section, option) ?: throw TalerConfigError.missing("temporal", section, option)
 
+    fun lookupDate(section: String, option: String): Instant? {
+        val raw = lookupString(section, option) ?: return null
+        val date = try {
+            LocalDate.parse(raw)
+        } catch (e: DateTimeParseException ) {
+            throw TalerConfigError.invalid("date", section, option, "'$raw' not a valid date at index ${e.errorIndex}")
+        }
+        return date.atStartOfDay(ZoneId.of("UTC")).toInstant()
+    }
+
+    fun requireDate(section: String, option: String): Instant =
+        lookupDate(section, option) ?: throw TalerConfigError.missing("date", section, option)
+    
+
     companion object {
         private val TIME_AMOUNT_PATTERN = Regex("([0-9]+) ?([a-z'\"]+)")
     }
diff --git a/contrib/nexus.conf b/contrib/nexus.conf
index 098f0402..b6403eac 100644
--- a/contrib/nexus.conf
+++ b/contrib/nexus.conf
@@ -49,6 +49,8 @@ CONFIG = postgres:///libeufin
 
 [nexus-fetch]
 FREQUENCY = 30m
+# Ignore all transactions prior to a certain date, useful when you want to use an existing account with old transactions that should not be bounced.
+# IGNORE_TRANSACTIONS_BEFORE = YYYY-MM-DD
 
 [nexus-submit]
 FREQUENCY = 30m
diff --git a/nexus/conf/test.conf b/nexus/conf/test.conf
index 1b52e17f..e6d52fff 100644
--- a/nexus/conf/test.conf
+++ b/nexus/conf/test.conf
@@ -12,4 +12,7 @@ BIC = BIC
 NAME = myname
 
 [libeufin-nexusdb-postgres]
-CONFIG = postgres:///libeufincheck
\ No newline at end of file
+CONFIG = postgres:///libeufincheck
+
+[nexus-fetch]
+IGNORE_TRANSACTIONS_BEFORE = 2024-04-04
\ 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
index f25934cd..fd6be2d9 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
@@ -144,20 +144,24 @@ suspend fun ingestIncomingPayment(
 
 private suspend fun ingestDocument(
     db: Database,
-    currency: String,
+    cfg: NexusConfig,
     xml: InputStream,
     whichDocument: SupportedDocument
 ) {
     when (whichDocument) {
         SupportedDocument.CAMT_054 -> {
             try {
-                parseTxNotif(xml, currency).forEach {
-                    when (it) {
-                        is TxNotification.Incoming -> ingestIncomingPayment(db, it.payment)
-                        is TxNotification.Outgoing -> ingestOutgoingPayment(db, it.payment)
-                        is TxNotification.Reversal -> {
-                            logger.error("BOUNCE '${it.msgId}': ${it.reason}")
-                            db.initiated.reversal(it.msgId, "Payment bounced: ${it.reason}")
+                parseTxNotif(xml, cfg.currency).forEach {
+                    if (cfg.fetch.ignoreBefore != null && it.executionTime < cfg.fetch.ignoreBefore) {
+                        logger.debug("IGNORE $it")
+                    } else {
+                        when (it) {
+                            is IncomingPayment -> ingestIncomingPayment(db, it)
+                            is OutgoingPayment -> ingestOutgoingPayment(db, it)
+                            is TxNotification.Reversal -> {
+                                logger.error("BOUNCE '${it.msgId}': ${it.reason}")
+                                db.initiated.reversal(it.msgId, "Payment bounced: ${it.reason}")
+                            }
                         }
                     }
                 }
@@ -211,7 +215,7 @@ private suspend fun ingestDocument(
 
 private suspend fun ingestDocuments(
     db: Database,
-    currency: String,
+    cfg: NexusConfig,
     content: InputStream,
     whichDocument: SupportedDocument
 ) {
@@ -223,13 +227,13 @@ private suspend fun ingestDocuments(
             try {
                 content.unzipEach { fileName, xmlContent ->
                     logger.trace("parse $fileName")
-                    ingestDocument(db, currency, xmlContent, whichDocument)
+                    ingestDocument(db, cfg, xmlContent, whichDocument)
                 }
             } catch (e: IOException) {
                 throw Exception("Could not open any ZIP archive", e)
             }
         }
-        SupportedDocument.PAIN_002_LOGS -> ingestDocument(db, currency, content, whichDocument)
+        SupportedDocument.PAIN_002_LOGS -> ingestDocument(db, cfg, content, whichDocument)
     }
 }
 
@@ -282,7 +286,7 @@ private suspend fun fetchDocuments(
                     stream,
                     doc == SupportedDocument.PAIN_002_LOGS
                 )
-                ingestDocuments(db, ctx.cfg.currency, loggedStream, doc)
+                ingestDocuments(db, ctx.cfg, loggedStream, doc)
             }
             true
         } catch (e: Exception) {
@@ -391,16 +395,15 @@ class EbicsFetch: CliktCommand("Fetches EBICS files") {
                     throw Exception("Failed to fetch documents")
                 }
             } else {
-                var frequency: Duration = cfg.config.requireDuration("nexus-fetch", "frequency")
                 val raw = cfg.config.requireString("nexus-fetch", "frequency")
                 logger.debug("Running with a frequency of $raw")
-                if (frequency == Duration.ZERO) {
+                if (cfg.fetch.frequency == Duration.ZERO) {
                     logger.warn("Long-polling not implemented, running therefore in transient mode")
                 }
                 do {
                     fetchDocuments(db, ctx, docs)
-                    delay(frequency.toKotlinDuration())
-                } while (frequency != Duration.ZERO)
+                    delay(cfg.fetch.frequency.toKotlinDuration())
+                } while (cfg.fetch.frequency != Duration.ZERO)
             }
         }
     }
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt
index 103795d6..09fdf0b0 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/Iso20022.kt
@@ -244,11 +244,11 @@ fun parseCustomerPaymentStatusReport(xml: InputStream): PaymentStatus {
 }
 
 sealed interface TxNotification {
-    data class Incoming(val payment: IncomingPayment): TxNotification
-    data class Outgoing(val payment: OutgoingPayment): TxNotification
+    val executionTime: Instant
     data class Reversal(
         val msgId: String,
-        val reason: String?
+        val reason: String?,
+        override val executionTime: Instant
     ): TxNotification
 }
 
@@ -257,10 +257,10 @@ data class IncomingPayment(
     val amount: TalerAmount,
     val wireTransferSubject: String,
     val debitPaytoUri: String,
-    val executionTime: Instant,
+    override val executionTime: Instant,
     /** ISO20022 AccountServicerReference */
     val bankId: String
-)  {
+): TxNotification {
     override fun toString(): String {
         return "IN ${executionTime.fmtDate()} $amount '$bankId' debitor=$debitPaytoUri subject=$wireTransferSubject"
     }
@@ -269,12 +269,12 @@ data class IncomingPayment(
 /** ISO20022 outgoing payment */
 data class OutgoingPayment(
     val amount: TalerAmount,
-    val executionTime: Instant,
+    override val executionTime: Instant,
     /** ISO20022 MessageIdentification */
     val messageId: String,
     val creditPaytoUri: String? = null, // not showing in camt.054
     val wireTransferSubject: String? = null // not showing in camt.054
-) {
+): TxNotification {
     override fun toString(): String {
         return "OUT ${executionTime.fmtDate()} $amount '$messageId' creditor=$creditPaytoUri subject=$wireTransferSubject"
     }
@@ -331,7 +331,8 @@ fun parseTxNotif(
             } else {
                 notifications.add(TxNotification.Reversal(
                     msgId = msgId,
-                    reason = info
+                    reason = info,
+                    executionTime = bookDate
                 ))
             }
             return@notificationForEachTx
@@ -358,24 +359,20 @@ fun parseTxNotif(
                         debtorPayto.append("?receiver-name=$urlEncName")
                     }
                 }
-                notifications.add(TxNotification.Incoming(
-                    IncomingPayment(
-                        amount = amount,
-                        bankId = bankId,
-                        debitPaytoUri = debtorPayto.toString(),
-                        executionTime = bookDate,
-                        wireTransferSubject = subject.toString()
-                    )
+                notifications.add(IncomingPayment(
+                    amount = amount,
+                    bankId = bankId,
+                    debitPaytoUri = debtorPayto.toString(),
+                    executionTime = bookDate,
+                    wireTransferSubject = subject.toString()
                 ))
             }
             "DBIT" -> {
                 val messageId = one("Refs").one("MsgId").text()
-                notifications.add(TxNotification.Outgoing(
-                    OutgoingPayment(
-                        amount = amount,
-                        messageId = messageId,
-                        executionTime = bookDate
-                    )
+                notifications.add(OutgoingPayment(
+                    amount = amount,
+                    messageId = messageId,
+                    executionTime = bookDate
                 ))
             }
             else -> throw Exception("Unknown transaction notification kind '$kind'")
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
index 72f64f58..496a13cf 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
@@ -62,6 +62,11 @@ fun Instant.fmtDate(): String =
 fun Instant.fmtDateTime(): String =
     DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.of("UTC")).format(this)
 
+class NexusFetchConfig(config: TalerConfig) {
+    val frequency = config.requireDuration("nexus-fetch", "frequency")
+    val ignoreBefore = config.lookupDate("nexus-fetch", "ignore_transactions_before")
+}
+
 /** Configuration for libeufin-nexus */
 class NexusConfig(val config: TalerConfig) {
     private fun requireString(option: String): String = config.requireString("nexus-ebics", option)
@@ -96,6 +101,8 @@ class NexusConfig(val config: TalerConfig) {
         if (this != "postfinance") throw Exception("Only 'postfinance' dialect is supported.")
         return@run this
     }
+
+    val fetch = NexusFetchConfig(config)
 }