commit 553ce68abdb4c73e486994594bae2bcc6cfdef2d
parent 297228f904554b09e320fa490035a0d68450d746
Author: MS <ms@taler.net>
Date: Mon, 23 Oct 2023 03:26:46 +0200
nexus db: creating & getting payment initiations.
Diffstat:
11 files changed, 211 insertions(+), 49 deletions(-)
diff --git a/.idea/kotlinc.xml b/.idea/kotlinc.xml
@@ -1,6 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
- <component name="KotlinJpsPluginSettings">
- <option name="version" value="1.7.22" />
- </component>
-</project>
-\ No newline at end of file
diff --git a/Makefile b/Makefile
@@ -53,7 +53,6 @@ install-nexus:
install contrib/libeufin-nexus.conf $(nexus_config_dir)/
install -D database-versioning/libeufin-nexus*.sql -t $(nexus_sql_dir)
install -D database-versioning/versioning.sql -t $(nexus_sql_dir)
- install -D database-versioning/procedures.sql -t $(nexus_sql_dir)
./gradlew -q -Pprefix=$(abs_destdir)$(prefix) nexus:installToPrefix
.PHONY: assemble
diff --git a/contrib/libeufin-nexus.conf b/contrib/libeufin-nexus.conf
@@ -38,6 +38,9 @@ BANK_DIALECT = postfinance
[nexus-postgres]
CONFIG = postgres:///libeufin-nexus
+[libeufin-nexusdb-postgres]
+SQL_DIR = $DATADIR/sql/
+
[nexus-ebics-fetch]
FREQUENCY = 30s # used when long-polling is not supported
STATEMENT_LOG_DIRECTORY = /tmp/ebics-messages/
diff --git a/database-versioning/libeufin-nexus-0001.sql b/database-versioning/libeufin-nexus-0001.sql
@@ -51,7 +51,7 @@ CREATE TABLE IF NOT EXISTS initiated_outgoing_transactions
(initiated_outgoing_transaction_id INT8 GENERATED BY DEFAULT AS IDENTITY UNIQUE -- used as our ID in PAIN
,amount taler_amount NOT NULL
,wire_transfer_subject TEXT
- ,execution_time INT8 NOT NULL
+ ,initiation_time INT8 NOT NULL
,credit_payto_uri TEXT NOT NULL
,outgoing_transaction_id INT8 REFERENCES outgoing_transactions (outgoing_transaction_id)
,submitted BOOL DEFAULT FALSE
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Database.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Database.kt
@@ -5,6 +5,7 @@ import kotlinx.coroutines.withContext
import org.postgresql.jdbc.PgConnection
import tech.libeufin.util.pgDataSource
import com.zaxxer.hikari.*
+import tech.libeufin.util.microsToJavaInstant
import tech.libeufin.util.stripIbanPayto
import tech.libeufin.util.toDbMicros
import java.sql.PreparedStatement
@@ -26,8 +27,8 @@ data class TalerAmount(
data class InitiatedPayment(
val amount: TalerAmount,
val wireTransferSubject: String,
- val executionTime: Instant,
val creditPaytoUri: String,
+ val initiationTime: Instant,
val clientRequestUuid: String? = null
)
@@ -36,7 +37,6 @@ data class InitiatedPayment(
* into the database.
*/
enum class PaymentInitiationOutcome {
- BAD_TIMESTAMP,
BAD_CREDIT_PAYTO,
UNIQUE_CONSTRAINT_VIOLATION,
SUCCESS
@@ -97,42 +97,95 @@ class Database(dbConfig: String): java.io.Closeable {
}
/**
+ * Sets payment initiation as submitted.
+ *
+ * @param rowId row ID of the record to set.
+ * @return true on success, false otherwise.
+ */
+ suspend fun initiatedPaymentSetSubmitted(rowId: Long): Boolean {
+ throw NotImplementedError()
+ }
+
+ /**
+ * Gets any initiated payment that was not submitted to the
+ * bank yet.
+ *
+ * @param currency in which currency should the payment be submitted to the bank.
+ * @return potentially empty list of initiated payments.
+ */
+ suspend fun initiatedPaymentsUnsubmittedGet(currency: String): Map<Long, InitiatedPayment> = runConn { conn ->
+ val stmt = conn.prepareStatement("""
+ SELECT
+ initiated_outgoing_transaction_id
+ ,(amount).val as amount_val
+ ,(amount).frac as amount_frac
+ ,wire_transfer_subject
+ ,credit_payto_uri
+ ,initiation_time
+ ,client_request_uuid
+ FROM initiated_outgoing_transactions
+ WHERE submitted=false;
+ """)
+ val maybeMap = mutableMapOf<Long, InitiatedPayment>()
+ stmt.executeQuery().use {
+ if (!it.next()) return@use
+ do {
+ val rowId = it.getLong("initiated_outgoing_transaction_id")
+ val initiationTime = it.getLong("initiation_time").microsToJavaInstant()
+ if (initiationTime == null) { // nexus fault
+ throw Exception("Found invalid timestamp at initiated payment with ID: $rowId")
+ }
+ maybeMap[rowId] = InitiatedPayment(
+ amount = TalerAmount(
+ value = it.getLong("amount_val"),
+ fraction = it.getInt("amount_frac"),
+ currency = currency
+ ),
+ creditPaytoUri = it.getString("credit_payto_uri"),
+ wireTransferSubject = it.getString("wire_transfer_subject"),
+ initiationTime = initiationTime,
+ clientRequestUuid = it.getString("client_request_uuid")
+ )
+ } while (it.next())
+ }
+ return@runConn maybeMap
+ }
+ /**
* Initiate a payment in the database. The "submit"
* command is then responsible to pick it up and submit
- * it at the bank.
+ * it to the bank.
*
* @param paymentData any data that's used to prepare the payment.
* @return true if the insertion went through, false in case of errors.
*/
- suspend fun initiatePayment(paymentData: InitiatedPayment): PaymentInitiationOutcome = runConn { conn ->
+ suspend fun initiatedPaymentCreate(paymentData: InitiatedPayment): PaymentInitiationOutcome = runConn { conn ->
val stmt = conn.prepareStatement("""
INSERT INTO initiated_outgoing_transactions (
amount
,wire_transfer_subject
- ,execution_time
,credit_payto_uri
+ ,initiation_time
,client_request_uuid
) VALUES (
(?,?)::taler_amount
,?
,?
,?
- ,?
+ ,?
)
""")
stmt.setLong(1, paymentData.amount.value)
stmt.setInt(2, paymentData.amount.fraction)
stmt.setString(3, paymentData.wireTransferSubject)
- val executionTime = paymentData.executionTime.toDbMicros() ?: run {
- logger.error("Execution time could not be converted to microseconds for the database.")
- return@runConn PaymentInitiationOutcome.BAD_TIMESTAMP // nexus fault.
- }
- stmt.setLong(4, executionTime)
val paytoOnlyIban = stripIbanPayto(paymentData.creditPaytoUri) ?: run {
logger.error("Credit Payto address is invalid.")
return@runConn PaymentInitiationOutcome.BAD_CREDIT_PAYTO // client fault.
}
- stmt.setString(5, paytoOnlyIban)
+ stmt.setString(4, paytoOnlyIban)
+ val initiationTime = paymentData.initiationTime.toDbMicros() ?: run {
+ throw Exception("Initiation time could not be converted to microseconds for the database.")
+ }
+ stmt.setLong(5, initiationTime)
stmt.setString(6, paymentData.clientRequestUuid) // can be null.
if (stmt.maybeUpdate())
return@runConn PaymentInitiationOutcome.SUCCESS
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/DbInit.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/DbInit.kt
@@ -0,0 +1,43 @@
+package tech.libeufin.nexus
+
+import com.github.ajalt.clikt.core.CliktCommand
+import com.github.ajalt.clikt.parameters.options.flag
+import com.github.ajalt.clikt.parameters.options.option
+import tech.libeufin.util.initializeDatabaseTables
+import tech.libeufin.util.resetDatabaseTables
+import kotlin.system.exitProcess
+
+fun doOrFail(doLambda: () -> Unit) {
+ try {
+ doLambda()
+ } catch (e: Exception) {
+ logger.error(e.message)
+ exitProcess(1)
+ }
+}
+
+/**
+ * This subcommand tries to load the SQL files that define
+ * the Nexus DB schema. Admits the --reset option to delete
+ * the data first.
+ */
+class DbInit : CliktCommand("Initialize the libeufin-nexus database", name = "dbinit") {
+ private val configFile by option(
+ "--config", "-c",
+ help = "set the configuration file"
+ )
+ private val requestReset by option(
+ "--reset", "-r",
+ help = "reset database (DANGEROUS: All existing data is lost)"
+ ).flag()
+
+ override fun run() {
+ val cfg = loadConfigOrFail(configFile).extractDbConfigOrFail()
+ doOrFail {
+ if (requestReset) {
+ resetDatabaseTables(cfg, sqlFilePrefix = "libeufin-nexus")
+ }
+ initializeDatabaseTables(cfg, sqlFilePrefix = "libeufin-nexus")
+ }
+ }
+}
+\ No newline at end of file
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSetup.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsSetup.kt
@@ -311,14 +311,8 @@ private fun findBic(maybeList: List<EbicsTypes.AccountInfo>?): String? {
* @param configFile location of the configuration entry point.
* @return internal representation of the configuration.
*/
-private fun extractConfig(configFile: String?): EbicsSetupConfig {
- val config = TalerConfig(NEXUS_CONFIG_SOURCE)
- try {
- config.load(configFile)
- } catch (e: Exception) {
- logger.error("Could not load configuration from ${configFile}, detail: ${e.message}")
- exitProcess(1)
- }
+private fun extractEbicsConfig(configFile: String?): EbicsSetupConfig {
+ val config = loadConfigOrFail(configFile)
// Checking the config.
val cfg = try {
EbicsSetupConfig(config)
@@ -355,7 +349,7 @@ private fun makePdf(privs: ClientPrivateKeysFile, cfg: EbicsSetupConfig) {
/**
* CLI class implementing the "ebics-setup" subcommand.
*/
-class EbicsSetup: CliktCommand() {
+class EbicsSetup: CliktCommand("Set up the EBICS subscriber") {
private val configFile by option(
"--config", "-c",
help = "set the configuration file"
@@ -380,7 +374,7 @@ class EbicsSetup: CliktCommand() {
* This function collects the main steps of setting up an EBICS access.
*/
override fun run() {
- val cfg = extractConfig(this.configFile)
+ val cfg = extractEbicsConfig(this.configFile)
if (checkFullConfig) {
throw NotImplementedError("--check-full-config flag not implemented")
}
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/Main.kt
@@ -25,15 +25,11 @@
package tech.libeufin.nexus
import ConfigSource
import TalerConfig
-import TalerConfigError
import com.github.ajalt.clikt.core.CliktCommand
import com.github.ajalt.clikt.core.subcommands
-import com.github.ajalt.clikt.parameters.options.flag
-import com.github.ajalt.clikt.parameters.options.option
import com.github.ajalt.clikt.parameters.options.versionOption
import io.ktor.client.*
import io.ktor.util.*
-import kotlinx.coroutines.runBlocking
import kotlinx.serialization.Contextual
import kotlinx.serialization.KSerializer
import org.slf4j.Logger
@@ -44,20 +40,15 @@ import kotlinx.serialization.Serializable
import kotlinx.serialization.descriptors.PrimitiveKind
import kotlinx.serialization.descriptors.PrimitiveSerialDescriptor
import kotlinx.serialization.descriptors.SerialDescriptor
-import kotlinx.serialization.encodeToString
import kotlinx.serialization.encoding.Decoder
import kotlinx.serialization.encoding.Encoder
import kotlinx.serialization.json.Json
import kotlinx.serialization.modules.SerializersModule
import net.taler.wallet.crypto.Base32Crockford
-import org.slf4j.event.Level
import tech.libeufin.nexus.ebics.*
import tech.libeufin.util.*
-import tech.libeufin.util.ebics_h004.EbicsTypes
import java.security.interfaces.RSAPrivateCrtKey
import java.security.interfaces.RSAPublicKey
-import java.time.Instant
-import kotlin.reflect.typeOf
val NEXUS_CONFIG_SOURCE = ConfigSource("libeufin-nexus", "libeufin-nexus")
val logger: Logger = LoggerFactory.getLogger("tech.libeufin.nexus.Main")
@@ -268,12 +259,43 @@ fun loadPrivateKeysFromDisk(location: String): ClientPrivateKeysFile? {
}
/**
+ * Abstracts the config loading and exception handling.
+ *
+ * @param configFile potentially NULL configuration file location.
+ * @return the configuration handle.
+ */
+fun loadConfigOrFail(configFile: String?): TalerConfig {
+ val config = TalerConfig(NEXUS_CONFIG_SOURCE)
+ try {
+ config.load(configFile)
+ } catch (e: Exception) {
+ logger.error("Could not load configuration from ${configFile}, detail: ${e.message}")
+ exitProcess(1)
+ }
+ return config
+}
+
+/**
+ * Abstracts fetching the DB config values to set up Nexus.
+ */
+fun TalerConfig.extractDbConfigOrFail(): DatabaseConfig =
+ try {
+ DatabaseConfig(
+ dbConnStr = requireString("nexus-postgres", "config"),
+ sqlDir = requirePath("libeufin-nexusdb-postgres", "sql_dir")
+ )
+ } catch (e: Exception) {
+ logger.error("Could not load config options for Nexus DB, detail: ${e.message}.")
+ exitProcess(1)
+ }
+
+/**
* Main CLI class that collects all the subcommands.
*/
class LibeufinNexusCommand : CliktCommand() {
init {
versionOption(getVersion())
- subcommands(EbicsSetup())
+ subcommands(EbicsSetup(), DbInit())
}
override fun run() = Unit
}
diff --git a/nexus/src/test/kotlin/DatabaseTest.kt b/nexus/src/test/kotlin/DatabaseTest.kt
@@ -4,24 +4,73 @@ import tech.libeufin.nexus.InitiatedPayment
import tech.libeufin.nexus.NEXUS_CONFIG_SOURCE
import tech.libeufin.nexus.PaymentInitiationOutcome
import tech.libeufin.nexus.TalerAmount
+import tech.libeufin.util.connectWithSchema
import java.time.Instant
import kotlin.test.assertEquals
+import kotlin.test.assertTrue
class DatabaseTest {
@Test
fun paymentInitiation() {
val db = prepDb(TalerConfig(NEXUS_CONFIG_SOURCE))
+ runBlocking {
+ val beEmpty = db.initiatedPaymentsUnsubmittedGet("KUDOS")// expect no records.
+ assertEquals(beEmpty.size, 0)
+ }
val initPay = InitiatedPayment(
amount = TalerAmount(44, 0, "KUDOS"),
creditPaytoUri = "payto://iban/not-used",
- executionTime = Instant.now(),
wireTransferSubject = "test",
- clientRequestUuid = "unique"
+ clientRequestUuid = "unique",
+ initiationTime = Instant.now()
)
runBlocking {
- assertEquals(db.initiatePayment(initPay), PaymentInitiationOutcome.SUCCESS)
- assertEquals(db.initiatePayment(initPay), PaymentInitiationOutcome.UNIQUE_CONSTRAINT_VIOLATION)
+ assertEquals(db.initiatedPaymentCreate(initPay), PaymentInitiationOutcome.SUCCESS)
+ assertEquals(db.initiatedPaymentCreate(initPay), PaymentInitiationOutcome.UNIQUE_CONSTRAINT_VIOLATION)
+ val haveOne = db.initiatedPaymentsUnsubmittedGet("KUDOS")
+ assertTrue {
+ haveOne.size == 1
+ && haveOne.containsKey(1)
+ && haveOne[1]?.clientRequestUuid == "unique"
+ }
+ }
+ }
+
+ /**
+ * Tests how the fetch method gets the list of
+ * multiple unsubmitted payment initiations.
+ */
+ @Test
+ fun paymentInitiationsMultiple() {
+ val db = prepDb(TalerConfig(NEXUS_CONFIG_SOURCE))
+ fun genInitPay(subject: String, rowUuid: String? = null) =
+ InitiatedPayment(
+ amount = TalerAmount(44, 0, "KUDOS"),
+ creditPaytoUri = "payto://iban/not-used",
+ wireTransferSubject = subject,
+ initiationTime = Instant.now(),
+ clientRequestUuid = rowUuid
+ )
+ runBlocking {
+ assertEquals(db.initiatedPaymentCreate(genInitPay("#1")), PaymentInitiationOutcome.SUCCESS)
+ assertEquals(db.initiatedPaymentCreate(genInitPay("#2")), PaymentInitiationOutcome.SUCCESS)
+ assertEquals(db.initiatedPaymentCreate(genInitPay("#3")), PaymentInitiationOutcome.SUCCESS)
+ assertEquals(db.initiatedPaymentCreate(genInitPay("#4")), PaymentInitiationOutcome.SUCCESS)
+ // Marking one as submitted, hence not expecting it in the results.
+ db.runConn { conn ->
+ conn.execSQLUpdate("""
+ UPDATE initiated_outgoing_transactions
+ SET submitted = true
+ WHERE initiated_outgoing_transaction_id=3;
+ """.trimIndent())
+ }
+ db.initiatedPaymentsUnsubmittedGet("KUDOS").apply {
+ assertEquals(3, this.size)
+ assertEquals("#1", this[1]?.wireTransferSubject)
+ assertEquals("#2", this[2]?.wireTransferSubject)
+ assertEquals("#4", this[4]?.wireTransferSubject)
+ }
}
}
}
\ No newline at end of file
diff --git a/nexus/src/test/kotlin/Keys.kt b/nexus/src/test/kotlin/Keys.kt
@@ -23,9 +23,9 @@ class PublicKeys {
bank_encryption_public_key = CryptoUtil.generateRsaKeyPair(2028).public
)
// storing them on disk.
- assertTrue(syncJsonToDisk(fileContent, config.bankPublicKeysFilename))
+ assertTrue(syncJsonToDisk(fileContent, "/tmp/nexus-tests-bank-keys.json"))
// loading them and check that values are the same.
- val fromDisk = loadBankKeys(config.bankPublicKeysFilename)
+ val fromDisk = loadBankKeys("/tmp/nexus-tests-bank-keys.json")
assertNotNull(fromDisk)
assertTrue {
fromDisk.accepted &&
diff --git a/util/src/main/kotlin/DB.kt b/util/src/main/kotlin/DB.kt
@@ -400,8 +400,13 @@ fun initializeDatabaseTables(cfg: DatabaseConfig, sqlFilePrefix: String) {
val sqlPatchText = path.readText()
conn.execSQLUpdate(sqlPatchText)
}
- val sqlProcedures = File("${cfg.sqlDir}/procedures.sql").readText()
- conn.execSQLUpdate(sqlProcedures)
+ val sqlProcedures = File("${cfg.sqlDir}/procedures.sql")
+ // Nexus doesn't have any procedures.
+ if (!sqlProcedures.exists()) {
+ logger.info("No procedures.sql for the SQL collection: $sqlFilePrefix")
+ return@transaction
+ }
+ conn.execSQLUpdate(sqlProcedures.readText())
}
}
}
