commit 6aaf661c073d5ffd74f4407fd386f7df4d0702de
parent 71050ab44fbccb970ec530383cdeef42ca0cf928
Author: Antoine A <>
Date: Fri, 26 Apr 2024 18:10:22 +0900
nexus: wire gateway /history/outgoing
Diffstat:
10 files changed, 145 insertions(+), 38 deletions(-)
diff --git a/common/src/main/kotlin/TxMedatada.kt b/common/src/main/kotlin/TxMedatada.kt
@@ -18,10 +18,16 @@
*/
package tech.libeufin.common
-private val PATTERN = Regex("[a-z0-9A-Z]{52}")
+private val BASE32_32B_PATTERN = Regex("[a-z0-9A-Z]{52}")
/** Extract the reserve public key from an incoming Taler transaction subject */
fun parseIncomingTxMetadata(subject: String): EddsaPublicKey {
- val match = PATTERN.find(subject)?.value ?: throw Exception("Missing reserve public key")
+ val match = BASE32_32B_PATTERN.find(subject)?.value ?: throw Exception("Missing reserve public key")
return EddsaPublicKey(match)
+}
+
+/** Extract the reserve public key from an incoming Taler transaction subject */
+fun parseOutgoingTxMetadata(subject: String): Pair<ShortHashCode, ExchangeUrl> {
+ val (wtid, baseUrl) = subject.splitOnce(" ") ?: throw Exception("Malformed outgoing subject")
+ return Pair(EddsaPublicKey(wtid), ExchangeUrl(baseUrl))
}
\ No newline at end of file
diff --git a/database-versioning/libeufin-nexus-0003.sql b/database-versioning/libeufin-nexus-0003.sql
@@ -23,7 +23,7 @@ CREATE TABLE IF NOT EXISTS talerable_outgoing_transactions
(talerable_outgoing_transaction_id INT8 GENERATED BY DEFAULT AS IDENTITY UNIQUE
,initiated_outgoing_transaction_id INT8 UNIQUE REFERENCES initiated_outgoing_transactions(initiated_outgoing_transaction_id) ON DELETE CASCADE
,outgoing_transaction_id INT8 UNIQUE REFERENCES outgoing_transactions(outgoing_transaction_id) ON DELETE CASCADE
- ,CONSTRAINT tx_link CHECK ((initiated_outgoing_transaction_id IS NOT NULL) OR (outgoing_transaction_id IS NOT NULL))
+ ,CONSTRAINT tx_link CHECK (initiated_outgoing_transaction_id IS NOT NULL OR outgoing_transaction_id IS NOT NULL)
,request_uid BYTEA UNIQUE CHECK (LENGTH(request_uid)=64)
,wtid BYTEA NOT NULL UNIQUE CHECK (LENGTH(wtid)=32)
,exchange_base_url TEXT NOT NULL
@@ -31,7 +31,7 @@ CREATE TABLE IF NOT EXISTS talerable_outgoing_transactions
COMMENT ON COLUMN talerable_outgoing_transactions.initiated_outgoing_transaction_id
IS 'If the transaction have been initiated';
COMMENT ON COLUMN talerable_outgoing_transactions.outgoing_transaction_id
- IS 'If the transaction have been recovered or done';
+ IS 'If the transaction have been recovered';
COMMENT ON CONSTRAINT tx_link ON talerable_outgoing_transactions
IS 'A transaction is either initiated or recovered';
diff --git a/database-versioning/libeufin-nexus-procedures.sql b/database-versioning/libeufin-nexus-procedures.sql
@@ -33,6 +33,8 @@ CREATE FUNCTION register_outgoing(
,IN in_execution_time INT8
,IN in_credit_payto_uri TEXT
,IN in_message_id TEXT
+ ,IN in_wtid BYTEA
+ ,IN in_exchange_url TEXT
,OUT out_tx_id INT8
,OUT out_found BOOLEAN
,OUT out_initiated BOOLEAN
@@ -40,6 +42,7 @@ CREATE FUNCTION register_outgoing(
LANGUAGE plpgsql AS $$
DECLARE
init_id INT8;
+talerable_id INT8;
BEGIN
-- Check if already registered
SELECT outgoing_transaction_id INTO out_tx_id
@@ -79,6 +82,20 @@ ELSE
WHERE request_uid = in_message_id
RETURNING true INTO out_initiated;
END IF;
+
+-- Register as talerable if contains wtid and exchange URL
+IF in_wtid IS NOT NULL OR in_exchange_url IS NOT NULL THEN
+ INSERT INTO talerable_outgoing_transactions (
+ outgoing_transaction_id,
+ wtid,
+ exchange_base_url
+ ) VALUES (out_tx_id, in_wtid, in_exchange_url)
+ ON CONFLICT (wtid) DO NOTHING
+ RETURNING talerable_outgoing_transaction_id INTO talerable_id;
+ IF talerable_id IS NOT NULL THEN
+ PERFORM pg_notify('outgoing_tx', talerable_id::text);
+ END IF;
+END IF;
END $$;
COMMENT ON FUNCTION register_outgoing
IS 'Register an outgoing transaction and optionally reconciles the related initiated transaction with it';
@@ -307,5 +324,5 @@ INSERT INTO talerable_outgoing_transactions(
,in_exchange_base_url
) RETURNING talerable_outgoing_transaction_id INTO out_tx_row_id;
out_timestamp = in_timestamp;
--- TODO notification
+PERFORM pg_notify('outgoing_tx', out_tx_row_id::text);
END $$;
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/EbicsFetch.kt
@@ -95,7 +95,10 @@ suspend fun ingestOutgoingPayment(
db: Database,
payment: OutgoingPayment
) {
- val result = db.payment.registerOutgoing(payment)
+ val metadata: Pair<ShortHashCode, ExchangeUrl>? = payment.wireTransferSubject?.let {
+ runCatching { parseOutgoingTxMetadata(it) }.getOrNull()
+ }
+ val result = db.payment.registerOutgoing(payment, metadata?.first, metadata?.second)
if (result.new) {
if (result.initiated)
logger.info("$payment")
@@ -106,8 +109,6 @@ suspend fun ingestOutgoingPayment(
}
}
-private val PATTERN = Regex("[a-z0-9A-Z]{52}")
-
/**
* Ingests an incoming payment. Stores the payment into valid talerable ones
* or bounces it, according to the subject.
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/api/WireGatewayApi.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/api/WireGatewayApi.kt
@@ -81,9 +81,9 @@ fun Routing.wireGatewayApi(db: Database, cfg: NexusConfig) {
get("/taler-wire-gateway/history/incoming") {
historyEndpoint(::IncomingHistory, ExchangeDAO::incomingHistory)
}
- /*get("/taler-wire-gateway/history/outgoing") {
+ get("/taler-wire-gateway/history/outgoing") {
historyEndpoint(::OutgoingHistory, ExchangeDAO::outgoingHistory)
- }*/
+ }
post("/taler-wire-gateway/admin/add-incoming") {
val req = call.receive<AddIncomingRequest>()
cfg.checkCurrency(req.amount)
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/db/ExchangeDAO.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/db/ExchangeDAO.kt
@@ -52,6 +52,38 @@ class ExchangeDAO(private val db: Database) {
)
}
+ /** Query [exchangeId] history of taler outgoing transactions */
+ suspend fun outgoingHistory(
+ params: HistoryParams
+ ): List<OutgoingTransaction>
+ // Outgoing transactions can be initiated or recovered. We take the first data to
+ // reach database : the initiation first else the recovered transaction.
+ = db.poolHistoryGlobal(params, db::listenOutgoing, """
+ SELECT
+ talerable_outgoing_transaction_id
+ ,COALESCE(iot.initiation_time, ot.execution_time) AS execution_time
+ ,(COALESCE(iot.amount, ot.amount)).val AS amount_val
+ ,(COALESCE(iot.amount, ot.amount)).frac AS amount_frac
+ ,COALESCE(iot.credit_payto_uri, ot.credit_payto_uri) AS credit_payto_uri
+ ,wtid
+ ,exchange_base_url
+ FROM talerable_outgoing_transactions AS tot
+ LEFT OUTER JOIN outgoing_transactions AS ot
+ ON tot.outgoing_transaction_id=ot.outgoing_transaction_id
+ LEFT OUTER JOIN initiated_outgoing_transactions AS iot
+ ON tot.initiated_outgoing_transaction_id=iot.initiated_outgoing_transaction_id
+ WHERE
+ """, "talerable_outgoing_transaction_id") {
+ OutgoingTransaction(
+ row_id = it.getLong("talerable_outgoing_transaction_id"),
+ date = it.getTalerTimestamp("execution_time"),
+ amount = it.getAmount("amount", db.bankCurrency),
+ credit_account = it.getString("credit_payto_uri"),
+ wtid = ShortHashCode(it.getBytes("wtid")),
+ exchange_base_url = it.getString("exchange_base_url")
+ )
+ }
+
/** Result of taler transfer transaction creation */
sealed interface TransferResult {
/** Transaction [id] and wire transfer [timestamp] */
diff --git a/nexus/src/main/kotlin/tech/libeufin/nexus/db/PaymentDAO.kt b/nexus/src/main/kotlin/tech/libeufin/nexus/db/PaymentDAO.kt
@@ -19,10 +19,8 @@
package tech.libeufin.nexus.db
-import tech.libeufin.common.EddsaPublicKey
-import tech.libeufin.common.TalerAmount
+import tech.libeufin.common.*
import tech.libeufin.common.db.one
-import tech.libeufin.common.micros
import tech.libeufin.nexus.IncomingPayment
import tech.libeufin.nexus.OutgoingPayment
import java.time.Instant
@@ -37,10 +35,14 @@ class PaymentDAO(private val db: Database) {
)
/** Register an outgoing payment reconciling it with its initiated payment counterpart if present */
- suspend fun registerOutgoing(paymentData: OutgoingPayment): OutgoingRegistrationResult = db.conn {
+ suspend fun registerOutgoing(
+ paymentData: OutgoingPayment,
+ wtid: ShortHashCode?,
+ baseUrl: ExchangeUrl?,
+ ): OutgoingRegistrationResult = db.conn {
val stmt = it.prepareStatement("""
SELECT out_tx_id, out_initiated, out_found
- FROM register_outgoing((?,?)::taler_amount,?,?,?,?)
+ FROM register_outgoing((?,?)::taler_amount,?,?,?,?,?,?)
""")
val executionTime = paymentData.executionTime.micros()
stmt.setLong(1, paymentData.amount.value)
@@ -49,6 +51,17 @@ class PaymentDAO(private val db: Database) {
stmt.setLong(4, executionTime)
stmt.setString(5, paymentData.creditPaytoUri)
stmt.setString(6, paymentData.messageId)
+ if (wtid != null) {
+ stmt.setBytes(7, wtid.raw)
+ } else {
+ stmt.setNull(7, java.sql.Types.NULL)
+ }
+ if (baseUrl != null) {
+ stmt.setString(8, baseUrl.url)
+ } else {
+ stmt.setNull(8, java.sql.Types.NULL)
+ }
+
stmt.one {
OutgoingRegistrationResult(
it.getLong("out_tx_id"),
diff --git a/nexus/src/test/kotlin/DatabaseTest.kt b/nexus/src/test/kotlin/DatabaseTest.kt
@@ -18,8 +18,9 @@
*/
import org.junit.Test
-import tech.libeufin.common.TalerAmount
+import tech.libeufin.common.*
import tech.libeufin.nexus.db.InitiatedDAO.PaymentInitiationResult
+import tech.libeufin.nexus.*
import java.time.Instant
import kotlin.test.assertEquals
import kotlin.test.assertIs
@@ -31,31 +32,43 @@ class OutgoingPaymentsTest {
@Test
fun register() = setup { db, _ ->
// With reconciling
- genOutPay("paid by nexus", "first").run {
+ genOutPay("paid by nexus").run {
assertIs<PaymentInitiationResult.Success>(
- db.initiated.create(genInitPay("waiting for reconciliation", "first"))
+ db.initiated.create(genInitPay("waiting for reconciliation", messageId))
)
- db.payment.registerOutgoing(this).run {
- assertTrue(new,)
+ db.payment.registerOutgoing(this, null, null).run {
+ assertTrue(new)
assertTrue(initiated)
}
- db.payment.registerOutgoing(this).run {
+ db.payment.registerOutgoing(this, null, null).run {
assertFalse(new)
assertTrue(initiated)
}
}
// Without reconciling
- genOutPay("not paid by nexus", "second").run {
- db.payment.registerOutgoing(this).run {
+ genOutPay("not paid by nexus").run {
+ db.payment.registerOutgoing(this, null, null).run {
assertTrue(new)
assertFalse(initiated)
}
- db.payment.registerOutgoing(this).run {
+ db.payment.registerOutgoing(this, null, null).run {
assertFalse(new)
assertFalse(initiated)
}
}
}
+
+ @Test
+ fun talerable() = setup { db, _ ->
+ val wtid = ShortHashCode.rand()
+ val url = "https://exchange.com"
+ genOutPay("$wtid $url").run {
+ assertIs<PaymentInitiationResult.Success>(
+ db.initiated.create(genInitPay("waiting for reconciliation", messageId))
+ )
+ ingestOutgoingPayment(db, this)
+ }
+ }
}
class IncomingPaymentsTest {
diff --git a/nexus/src/test/kotlin/WireGatewayApiTest.kt b/nexus/src/test/kotlin/WireGatewayApiTest.kt
@@ -123,7 +123,7 @@ class WireGatewayApiTest {
},
{
// Transactions using raw bank transaction logic
- ingestIncomingPayment(db, genInPay("history test with ${ShortHashCode.rand()} reserve pub"))
+ talerableIn(db)
}
),
ignored = listOf(
@@ -133,45 +133,45 @@ class WireGatewayApiTest {
},
{
// Ignore outgoing transaction
- ingestOutgoingPayment(db, genOutPay("ignored"))
+ talerableOut(db)
}
)
)
}
- /*
/**
* Testing the /history/outgoing call from the TWG API.
*/
@Test
- fun historyOutgoing() = serverSetup {
- setMaxDebt("exchange", "KUDOS:1000000")
- authRoutine(HttpMethod.Get, "/accounts/merchant/taler-wire-gateway/history/outgoing")
+ fun historyOutgoing() = serverSetup { db ->
+ //authRoutine(HttpMethod.Get, "/accounts/merchant/taler-wire-gateway/history/outgoing")
historyRoutine<OutgoingHistory>(
- url = "/accounts/exchange/taler-wire-gateway/history/outgoing",
+ url = "/taler-wire-gateway/history/outgoing",
ids = { it.outgoing_transactions.map { it.row_id } },
registered = listOf(
{
- // Transactions using clean add incoming logic
- transfer("KUDOS:10")
+ transfer()
+ },
+ {
+ talerableOut(db)
}
),
ignored = listOf(
{
- // gnore manual incoming transaction
- tx("exchange", "KUDOS:10", "merchant", "${ShortHashCode.rand()} http://exchange.example.com/")
+ // Ignore manual incoming transaction
+ talerableIn(db)
},
{
// Ignore malformed incoming transaction
- tx("merchant", "KUDOS:10", "exchange", "ignored")
+ ingestIncomingPayment(db, genInPay("ignored"))
},
{
// Ignore malformed outgoing transaction
- tx("exchange", "KUDOS:10", "merchant", "ignored")
+ ingestOutgoingPayment(db, genOutPay("ignored"))
}
)
)
- }*/
+ }
// Testing the /admin/add-incoming call from the TWG API.
@Test
diff --git a/nexus/src/test/kotlin/helpers.kt b/nexus/src/test/kotlin/helpers.kt
@@ -125,4 +125,29 @@ fun genOutPay(subject: String, messageId: String? = null): OutgoingPayment {
executionTime = Instant.now(),
messageId = id
)
+}
+
+/** Perform a taler outgoing transaction */
+suspend fun ApplicationTestBuilder.transfer() {
+ client.post("/taler-wire-gateway/transfer") {
+ json {
+ "request_uid" to HashCode.rand()
+ "amount" to "CHF:55"
+ "exchange_base_url" to "http://exchange.example.com/"
+ "wtid" to ShortHashCode.rand()
+ "credit_account" to grothoffPayto
+ }
+ }.assertOk()
+}
+
+/** Ingest a talerable outgoing transaction */
+suspend fun talerableOut(db: Database) {
+ val wtid = ShortHashCode.rand()
+ ingestOutgoingPayment(db, genOutPay("$wtid http://exchange.example.com/"))
+}
+
+/** Ingest a talerable incoming transaction */
+suspend fun talerableIn(db: Database) {
+ val reserve_pub = ShortHashCode.rand()
+ ingestIncomingPayment(db, genInPay("history test with $reserve_pub reserve pub"))
}
\ No newline at end of file
