Long polling.

Drafting a Taler Wire Gateway testcase and
a helper class to offer methods that abstract
Postgres' LISTEN and NOTIFY.

10 files changed, 220 insertions, 144 deletions

diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt
index 6d3f586b..b1ddbd17 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/DB.kt
@@ -27,7 +27,6 @@ import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.jetbrains.exposed.sql.transactions.transaction
 import tech.libeufin.nexus.iso20022.EntryStatus
 import tech.libeufin.util.EbicsInitState
-import tech.libeufin.util.amount
 import java.sql.Connection
 
 /**
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/bankaccount/BankAccount.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/bankaccount/BankAccount.kt
index b32b814a..416cd4ae 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/bankaccount/BankAccount.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/bankaccount/BankAccount.kt
@@ -27,9 +27,7 @@ import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.transactions.transaction
 import org.w3c.dom.Document
 import tech.libeufin.nexus.*
-import tech.libeufin.nexus.iso20022.CamtParsingError
-import tech.libeufin.nexus.iso20022.CreditDebitIndicator
-import tech.libeufin.nexus.iso20022.parseCamtMessage
+import tech.libeufin.nexus.iso20022.*
 import tech.libeufin.nexus.server.FetchSpecJson
 import tech.libeufin.nexus.server.Pain001Data
 import tech.libeufin.nexus.server.requireBankConnection
@@ -232,7 +230,7 @@ fun processCamtMessage(
                 }
             }
         }
-        val entries = res.reports.map { it.entries }.flatten()
+        val entries: List<CamtBankAccountEntry> = res.reports.map { it.entries }.flatten()
         var newPaymentsLog = ""
         downloadedTransactions = entries.size
         txloop@ for (entry in entries) {
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/server/JSON.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/server/JSON.kt
index 8f87c0e2..38ab4236 100644
--- a/nexus/src/main/kotlin/tech/libeufin/nexus/server/JSON.kt
+++ b/nexus/src/main/kotlin/tech/libeufin/nexus/server/JSON.kt
@@ -440,7 +440,7 @@ class CurrencyAmountSerializer(jc: Class<CurrencyAmount> = CurrencyAmount::class
 @JsonSerialize(using = CurrencyAmountSerializer::class)
 data class CurrencyAmount(
     val currency: String,
-    val value: String // allows calculations
+    val value: String
 )
 fun CurrencyAmount.toPlainString(): String {
     return "${this.currency}:${this.value}"
diff --git a/nexus/src/test/kotlin/DBTest.kt b/nexus/src/test/kotlin/DBTest.kt
index 3b818942..ff044a3e 100644
--- a/nexus/src/test/kotlin/DBTest.kt
+++ b/nexus/src/test/kotlin/DBTest.kt
@@ -1,67 +1,31 @@
 package tech.libeufin.nexus
 
+import kotlinx.coroutines.*
+import org.jetbrains.exposed.dao.flushCache
 import org.jetbrains.exposed.exceptions.ExposedSQLException
 import org.jetbrains.exposed.sql.*
+import org.jetbrains.exposed.sql.statements.api.ExposedConnection
+import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.jetbrains.exposed.sql.transactions.transaction
+import org.jetbrains.exposed.sql.transactions.transactionManager
 import org.junit.Test
+import org.postgresql.PGConnection
+import org.postgresql.jdbc.PgConnection
+import tech.libeufin.util.PostgresListenNotify
 import withTestDatabase
-import java.io.File
-
-object MyTable : Table() {
-    val col1 = text("col1")
-    val col2 = text("col2")
-    override val primaryKey = PrimaryKey(col1, col2)
-}
+import java.sql.Connection
+import java.sql.DriverManager
 
 class DBTest {
-    @Test(expected = ExposedSQLException::class)
-    fun sqlDslTest() {
-        withTestDatabase {
-            transaction {
-                addLogger(StdOutSqlLogger)
-                SchemaUtils.create(MyTable)
-                MyTable.insert {
-                    it[col1] = "foo"
-                    it[col2] = "bar"
-                }
-                // should throw ExposedSQLException
-                MyTable.insert {
-                    it[col1] = "foo"
-                    it[col2] = "bar"
-                }
-                MyTable.insert {  } // shouldn't it fail for non-NULL constraint violation?
-            }
-        }
-    }
 
+    // Testing database notifications (only postgresql)
     @Test
-    fun facadeConfigTest() {
-        withTestDatabase {
-            transaction {
-                addLogger(StdOutSqlLogger)
-                SchemaUtils.create(
-                    FacadesTable,
-                    FacadeStateTable,
-                    NexusUsersTable
-                )
-                val user = NexusUserEntity.new {
-                    username = "testuser"
-                    passwordHash = "x"
-                    superuser = true
-                }
-                val facade = FacadeEntity.new {
-                    facadeName = "testfacade"
-                    type = "any"
-                    creator = user
-                }
-                FacadeStateEntity.new {
-                    bankAccount = "b"
-                    bankConnection = "b"
-                    reserveTransferLevel = "any"
-                    this.facade = facade
-                    currency = "UNUSED"
-                }
-            }
-        }
+    fun notifications() {
+        val genCon = DriverManager.getConnection("jdbc:postgresql://localhost:5432/talercheck?user=job")
+        val pgCon = genCon.unwrap(org.postgresql.jdbc.PgConnection::class.java)
+        val ln = PostgresListenNotify(pgCon, "x")
+        ln.postrgesListen()
+        ln.postgresNotify()
+        runBlocking { ln.postgresWaitNotification(2000L) }
     }
 }
\ No newline at end of file
diff --git a/nexus/src/test/kotlin/MakeEnv.kt b/nexus/src/test/kotlin/MakeEnv.kt
index d2a2b480..8d391151 100644
--- a/nexus/src/test/kotlin/MakeEnv.kt
+++ b/nexus/src/test/kotlin/MakeEnv.kt
@@ -7,6 +7,8 @@ import org.jetbrains.exposed.sql.transactions.transactionManager
 import tech.libeufin.nexus.*
 import tech.libeufin.nexus.dbCreateTables
 import tech.libeufin.nexus.dbDropTables
+import tech.libeufin.nexus.iso20022.*
+import tech.libeufin.nexus.server.CurrencyAmount
 import tech.libeufin.nexus.server.FetchLevel
 import tech.libeufin.nexus.server.FetchSpecAllJson
 import tech.libeufin.sandbox.*
@@ -21,9 +23,9 @@ data class EbicsKeys(
     val sig: CryptoUtil.RsaCrtKeyPair
 )
 const val TEST_DB_FILE = "/tmp/nexus-test.sqlite3"
-const val TEST_DB_CONN = "jdbc:sqlite:$TEST_DB_FILE"
+// const val TEST_DB_CONN = "jdbc:sqlite:$TEST_DB_FILE"
 // Convenience DB connection to switch to Postgresql:
-// const val TEST_DB_CONN = "jdbc:postgresql://localhost:5432/talercheck?user="
+const val TEST_DB_CONN = "jdbc:postgresql://localhost:5432/talercheck?user=job"
 val BANK_IBAN = getIban()
 val FOO_USER_IBAN = getIban()
 val BAR_USER_IBAN = getIban()
@@ -292,4 +294,80 @@ fun withSandboxTestDatabase(f: () -> Unit) {
         }
         f()
     }
-}
\ No newline at end of file
+}
+
+fun talerIncomingForFoo(currency: String, value: String, subject: String) {
+    transaction {
+        val inc = NexusBankTransactionEntity.new {
+            bankAccount = NexusBankAccountEntity.findByName("foo")!!
+            accountTransactionId = "mock"
+            creditDebitIndicator = "CRDT"
+            this.currency = currency
+            this.amount = value
+            status = EntryStatus.BOOK
+            transactionJson = jacksonObjectMapper(
+            ).writerWithDefaultPrettyPrinter(
+            ).writeValueAsString(
+                genNexusIncomingPayment(
+                    amount = CurrencyAmount(currency,value),
+                    subject = subject
+                )
+            )
+        }
+        TalerIncomingPaymentEntity.new {
+            payment = inc
+            reservePublicKey = "mock"
+            timestampMs = 0L
+            debtorPaytoUri = "mock"
+        }
+    }
+}
+
+
+fun genNexusIncomingPayment(
+    amount: CurrencyAmount,
+    subject: String,
+): CamtBankAccountEntry =
+    CamtBankAccountEntry(
+        amount = amount,
+        creditDebitIndicator = CreditDebitIndicator.CRDT,
+        status = EntryStatus.BOOK,
+        bankTransactionCode = "mock",
+        valueDate = null,
+        bookingDate = null,
+        accountServicerRef = null,
+        entryRef = null,
+        currencyExchange = null,
+        counterValueAmount = null,
+        instructedAmount = null,
+        batches = listOf(
+            Batch(
+                paymentInformationId = null,
+                messageId = null,
+                batchTransactions = listOf(
+                    BatchTransaction(
+                        amount = amount,
+                        creditDebitIndicator = CreditDebitIndicator.CRDT,
+                        details = TransactionDetails(
+                            unstructuredRemittanceInformation = subject,
+                            debtor = null,
+                            debtorAccount = null,
+                            debtorAgent = null,
+                            creditor = null,
+                            creditorAccount = null,
+                            creditorAgent = null,
+                            ultimateCreditor = null,
+                            ultimateDebtor = null,
+                            purpose = null,
+                            proprietaryPurpose = null,
+                            currencyExchange = null,
+                            instructedAmount = null,
+                            counterValueAmount = null,
+                            interBankSettlementAmount = null,
+                            returnInfo = null
+                        )
+                    )
+                )
+            )
+        )
+    )
\ No newline at end of file
diff --git a/nexus/src/test/kotlin/TalerTest.kt b/nexus/src/test/kotlin/TalerTest.kt
index bc117307..36c656f1 100644
--- a/nexus/src/test/kotlin/TalerTest.kt
+++ b/nexus/src/test/kotlin/TalerTest.kt
@@ -1,25 +1,50 @@
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import io.ktor.client.plugins.*
 import io.ktor.client.request.*
+import io.ktor.client.statement.*
 import io.ktor.http.*
 import io.ktor.server.testing.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.future.future
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 import org.jetbrains.exposed.sql.transactions.transaction
 import org.junit.Ignore
 import org.junit.Test
 import tech.libeufin.nexus.*
 import tech.libeufin.nexus.bankaccount.fetchBankAccountTransactions
-import tech.libeufin.nexus.server.FetchLevel
-import tech.libeufin.nexus.server.FetchSpecAllJson
-import tech.libeufin.nexus.server.client
-import tech.libeufin.nexus.server.nexusApp
+import tech.libeufin.nexus.iso20022.EntryStatus
+import tech.libeufin.nexus.server.*
 import tech.libeufin.sandbox.sandboxApp
 import tech.libeufin.sandbox.wireTransfer
 
 // This class tests the features related to the Taler facade.
 class TalerTest {
 
+    /**
+     * Tests that a client (normally represented by the wire-watch)
+     * gets incoming transactions.
+     */
+    @Test
+    fun historyIncomingTest() {
+        withNexusAndSandboxUser {
+            testApplication {
+                application(nexusApp)
+                runBlocking {
+                    val future = async {
+                        client.get(
+                            "/facades/taler/taler-wire-gateway/history/incoming?delta=5"
+                        ) {
+                            expectSuccess = true
+                            contentType(ContentType.Application.Json)
+                            basicAuth("foo", "foo")
+                        }
+                    }
+                    talerIncomingForFoo("KUDOS", "10", "Invalid")
+                }
+            }
+        }
+    }
+
     @Ignore // Ignoring because no assert takes place.
     @Test // Triggering a refund because of a duplicate reserve pub.
     fun refundTest() {
diff --git a/util/build.gradle b/util/build.gradle
index a294ff31..4c8da032 100644
--- a/util/build.gradle
+++ b/util/build.gradle
@@ -45,6 +45,8 @@ dependencies {
     // https://mvnrepository.com/artifact/org.jetbrains.kotlin/kotlin-reflect
     implementation group: 'org.jetbrains.kotlin', name: 'kotlin-reflect', version: '1.5.21'
 
+    // Database helper
+    implementation group: 'org.postgresql', name: 'postgresql', version: '42.5.4'
     implementation "org.jetbrains.exposed:exposed-core:$exposed_version"
     implementation "org.jetbrains.exposed:exposed-dao:$exposed_version"
     implementation "io.netty:netty-all:$netty_version"
diff --git a/util/src/main/kotlin/DB.kt b/util/src/main/kotlin/DB.kt
new file mode 100644
index 00000000..63a213e2
--- /dev/null
+++ b/util/src/main/kotlin/DB.kt
@@ -0,0 +1,84 @@
+/*
+ * This file is part of LibEuFin.
+ * Copyright (C) 2020 Taler Systems S.A.
+ *
+ * LibEuFin is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation; either version 3, or
+ * (at your option) any later version.
+ *
+ * LibEuFin is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+ * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Affero General
+ * Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public
+ * License along with LibEuFin; see the file COPYING.  If not, see
+ * <http://www.gnu.org/licenses/>
+ */
+
+package tech.libeufin.util
+import UtilError
+import io.ktor.http.*
+import kotlinx.coroutines.delay
+import logger
+import org.postgresql.jdbc.PgConnection
+import java.lang.Long.max
+
+// This class abstracts the LISTEN/NOTIFY construct supported
+class PostgresListenNotify(
+    private val conn: PgConnection,
+    private val channel: String
+) {
+    fun postrgesListen() {
+        val stmt = conn.createStatement()
+        stmt.execute("LISTEN $channel")
+        stmt.close()
+    }
+    fun postgresNotify() {
+        val stmt = conn.createStatement()
+        stmt.execute("NOTIFY $channel")
+        stmt.close()
+    }
+
+    suspend fun postgresWaitNotification(timeoutMs: Long) {
+        // Splits the checks into 10ms chunks.
+        val sleepTimeMs = 10L
+        var notificationFound = false
+        val iterations = timeoutMs / sleepTimeMs
+        for (i in 0..iterations) {
+            val maybeNotifications = conn.notifications
+            // If a notification arrived, stop fetching for it.
+            if (maybeNotifications.isNotEmpty()) {
+                // Double checking that the channel is correct.
+                // Notification(s) arrived, double-check channel name.
+                maybeNotifications.forEach {
+                    if (it.name != channel) {
+                        throw UtilError(
+                            statusCode = HttpStatusCode.InternalServerError,
+                            reason = "Listener got wrong notification.  Expected: $channel, but got: ${it.name}"
+                        )
+                    }
+                }
+                notificationFound = true
+                break
+            }
+            /* Notification didn't arrive, release the thread and
+             * retry in the next chunk.  */
+            delay(sleepTimeMs)
+        }
+
+        if (!notificationFound) {
+            throw UtilError(
+                statusCode = HttpStatusCode.NotFound,
+                reason = "Timeout expired for notification on channel $channel",
+                ec = LibeufinErrorCode.LIBEUFIN_EC_TIMEOUT_EXPIRED
+            )
+        }
+        /* Notification arrived.  In this current version
+         * we don't pass any data to the caller; the channel
+         * name itself means that the awaited information arrived.
+         * */
+        return
+        }
+    }
\ No newline at end of file
diff --git a/util/src/main/kotlin/DBTypes.kt b/util/src/main/kotlin/DBTypes.kt
deleted file mode 100644
index c38e63cf..00000000
--- a/util/src/main/kotlin/DBTypes.kt
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * This file is part of LibEuFin.
- * Copyright (C) 2020 Taler Systems S.A.
- *
- * LibEuFin is free software; you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation; either version 3, or
- * (at your option) any later version.
- *
- * LibEuFin is distributed in the hope that it will be useful, but
- * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
- * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Affero General
- * Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public
- * License along with LibEuFin; see the file COPYING.  If not, see
- * <http://www.gnu.org/licenses/>
- */
-
-package tech.libeufin.util
-
-import org.jetbrains.exposed.sql.Column
-import org.jetbrains.exposed.sql.ColumnType
-import org.jetbrains.exposed.sql.Table
-import java.math.BigDecimal
-import java.math.RoundingMode
-
-const val SCALE_TWO = 2
-const val NUMBER_MAX_DIGITS = 20
-class BadAmount(badValue: Any?) : Exception("Value '${badValue}' is not a valid amount")
-typealias Amount = BigDecimal
-
-class AmountColumnType : ColumnType() {
-    override fun sqlType(): String  = "DECIMAL(${NUMBER_MAX_DIGITS}, ${SCALE_TWO})"
-        override fun valueFromDB(value: Any): Any {
-            val valueFromDB = super.valueFromDB(value)
-            try {
-                return when (valueFromDB) {
-                    is BigDecimal -> valueFromDB.setScale(SCALE_TWO, RoundingMode.UNNECESSARY)
-                    is Double -> BigDecimal.valueOf(valueFromDB).setScale(SCALE_TWO, RoundingMode.UNNECESSARY)
-                    is Float -> BigDecimal(valueFromDB.toString()).setScale(
-                        SCALE_TWO,
-                        RoundingMode.UNNECESSARY
-                    )
-                    is Int -> BigDecimal(valueFromDB)
-                    is Long -> BigDecimal.valueOf(valueFromDB)
-                    else -> valueFromDB
-                }
-            } catch (e: Exception) {
-                e.printStackTrace()
-                throw BadAmount(value)
-            }
-        }
-
-        override fun valueToDB(value: Any?): Any? {
-            try {
-                (value as BigDecimal).setScale(SCALE_TWO, RoundingMode.UNNECESSARY)
-            } catch (e: Exception) {
-                e.printStackTrace()
-                throw BadAmount(value)
-            }
-
-            if (value.compareTo(BigDecimal.ZERO) == 0) {
-                throw BadAmount(value)
-            }
-            return super.valueToDB(value)
-        }
-}
-
-/**
- * Make sure the number entered by upper layers does not need any rounding
- * to conform to scale == 2
- */
-fun Table.amount(name: String): Column<Amount> {
-    return registerColumn(name, AmountColumnType())
-}
\ No newline at end of file
diff --git a/util/src/main/kotlin/LibeufinErrorCodes.kt b/util/src/main/kotlin/LibeufinErrorCodes.kt
index 6020422a..758292e3 100644
--- a/util/src/main/kotlin/LibeufinErrorCodes.kt
+++ b/util/src/main/kotlin/LibeufinErrorCodes.kt
@@ -71,5 +71,7 @@ enum class LibeufinErrorCode(val code: Int) {
      * A request is using a unsupported currency.  Usually returned
      * along 400 Bad Request
      */
-    LIBEUFIN_EC_BAD_CURRENCY(7)
+    LIBEUFIN_EC_BAD_CURRENCY(7),
+
+    LIBEUFIN_EC_TIMEOUT_EXPIRED(8)
 }
\ No newline at end of file