Clean /taler-wire-gateway/history

5 files changed, 212 insertions, 258 deletions

diff --git a/bank/src/main/kotlin/tech/libeufin/bank/CorebankApiHandlers.kt b/bank/src/main/kotlin/tech/libeufin/bank/CorebankApiHandlers.kt
index 5cfbfb49..6bc22957 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/CorebankApiHandlers.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/CorebankApiHandlers.kt
@@ -500,7 +500,7 @@ fun Routing.accountsMgmtHandlers(db: Database, ctx: BankApplicationContext) {
         val c = call.authenticateBankRequest(db, TokenScope.readonly) ?: throw unauthorized()
         val resourceName = call.getResourceName("USERNAME")
         if (!resourceName.canI(c, withAdmin = true)) throw forbidden()
-        val historyParams = getHistoryParams(call.request)
+        val historyParams = getHistoryParams(call.request.queryParameters)
         val resourceCustomer = db.customerGetFromLogin(resourceName) ?: throw notFound(
             hint = "Customer '$resourceName' not found in the database",
             talerEc = TalerErrorCode.TALER_EC_END // FIXME: need EC.
diff --git a/bank/src/main/kotlin/tech/libeufin/bank/Database.kt b/bank/src/main/kotlin/tech/libeufin/bank/Database.kt
index 67a4ae95..ba686d06 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/Database.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/Database.kt
@@ -130,7 +130,7 @@ fun resetDatabaseTables(dbConfig: String, sqlDir: String) {
     dbConn.execSQLUpdate(sqlDrop)
 }
 
-class Database(private val dbConfig: String, internal val bankCurrency: String) {
+class Database(private val dbConfig: String, private val bankCurrency: String) {
     private var dbConn: PgConnection? = null
     private var dbCtr: Int = 0
     private val preparedStatements: MutableMap<String, PreparedStatement> = mutableMapOf()
@@ -813,6 +813,142 @@ class Database(private val dbConfig: String, internal val bankCurrency: String)
         }
     }
 
+    suspend fun <T> bankTransactionPoolHistory(
+        params: HistoryParams, 
+        bankAccountId: Long, 
+        direction: TransactionDirection,
+        map: (BankAccountTransaction) -> T?
+    ): List<T> {
+        val conn = conn() ?: throw internalServerError("DB connection down");
+        val channel = "${direction.name}_$bankAccountId";
+        var start = params.start
+        var delta = params.delta
+        var poll_ms = params.poll_ms;
+
+        val (cmpOp, orderBy) = if (delta < 0) Pair("<", "DESC") else Pair(">", "ASC")
+        val stmt = conn.prepareStatement("""
+            SELECT 
+                creditor_payto_uri
+                ,creditor_name
+                ,debtor_payto_uri
+                ,debtor_name
+                ,subject
+                ,(amount).val AS amount_val
+                ,(amount).frac AS amount_frac
+                ,transaction_date
+                ,account_servicer_reference
+                ,payment_information_id
+                ,end_to_end_id
+                ,bank_account_id
+                ,bank_transaction_id
+            FROM bank_account_transactions
+            WHERE bank_transaction_id ${cmpOp} ? 
+                AND bank_account_id=?
+                AND direction=?::direction_enum
+            ORDER BY bank_transaction_id ${orderBy}
+            LIMIT ?
+        """)
+
+        // If going backward with a starting point, it is useless to poll
+        if (delta < 0 && start != Long.MAX_VALUE) {
+            poll_ms = 0;
+        }
+
+        // Only start expensive listening if we intend to poll
+        if (poll_ms > 0) {
+            conn.execSQLUpdate("LISTEN $channel");
+        }
+
+        val items = mutableListOf<T>()
+
+        fun bankTransactionGetHistory(): List<BankAccountTransaction> {
+            stmt.setLong(1, start)
+            stmt.setLong(2, bankAccountId)
+            stmt.setString(3, direction.name)
+            stmt.setLong(4, abs(delta))
+            val rs = stmt.executeQuery()
+            rs.use {
+                val ret = mutableListOf<BankAccountTransaction>()
+                if (!it.next()) return ret
+                do {
+                    ret.add(
+                        BankAccountTransaction(
+                            creditorPaytoUri = it.getString("creditor_payto_uri"),
+                            creditorName = it.getString("creditor_name"),
+                            debtorPaytoUri = it.getString("debtor_payto_uri"),
+                            debtorName = it.getString("debtor_name"),
+                            amount = TalerAmount(
+                                it.getLong("amount_val"),
+                                it.getInt("amount_frac"),
+                                getCurrency()
+                            ),
+                            accountServicerReference = it.getString("account_servicer_reference"),
+                            endToEndId = it.getString("end_to_end_id"),
+                            direction = direction,
+                            bankAccountId = it.getLong("bank_account_id"),
+                            paymentInformationId = it.getString("payment_information_id"),
+                            subject = it.getString("subject"),
+                            transactionDate = it.getLong("transaction_date").microsToJavaInstant() ?: throw faultyTimestampByBank(),
+                            dbRowId = it.getLong("bank_transaction_id")
+                    ))
+                } while (it.next())
+                return ret
+            }
+        }
+
+        fun loadBankHistory() {
+            while (delta != 0L) {
+                val history = bankTransactionGetHistory()
+                if (history.isEmpty())
+                    break;
+                history.forEach {
+                    val item = map(it);
+                    // Advance cursor
+                    start = it.expectRowId()
+        
+                    if (item != null) {
+                        items.add(item)
+                        // Reduce delta
+                        if (delta < 0) delta++ else delta--;
+                    }
+                }
+            }
+        }
+
+        loadBankHistory()
+
+        // Long polling
+        while (delta != 0L && poll_ms > 0) {
+            var remaining = abs(delta);
+            do {
+                val pollStart = System.currentTimeMillis()
+                conn.getNotifications(poll_ms.toInt()).forEach {
+                    val id = it.parameter.toLong()
+                    val new = when {
+                        params.start == Long.MAX_VALUE -> true
+                        delta < 0 -> id < start
+                        else -> id > start
+                    }
+                    if (new) remaining -= 1
+                }
+                val pollEnd = System.currentTimeMillis()
+                poll_ms -= pollEnd - pollStart
+            } while (poll_ms > 0 && remaining > 0L)
+
+            // If going backward without a starting point, we reset loading progress
+            if (params.start == Long.MAX_VALUE) {
+                start = params.start
+                delta = params.delta
+                items.clear()
+            }
+            loadBankHistory()
+        }
+
+        conn.execSQLUpdate("UNLISTEN $channel");
+
+        return items.toList();
+    }
+
     /**
      * The following function returns the list of transactions, according
      * to the history parameters.  The parameters take at least the 'start'
@@ -820,11 +956,6 @@ class Database(private val dbConfig: String, internal val bankCurrency: String)
      * moment, only the TWG uses the direction, to provide the /incoming
      * and /outgoing endpoints.
      */
-    // Helper type to collect the history parameters.
-    private data class HistoryParams(
-        val cmpOp: String, // < or >
-        val orderBy: String // ASC or DESC
-    )
     fun bankTransactionGetHistory(
         start: Long,
         delta: Long,
@@ -832,9 +963,7 @@ class Database(private val dbConfig: String, internal val bankCurrency: String)
         withDirection: TransactionDirection? = null
     ): List<BankAccountTransaction> {
         reconnect()
-        val ops = if (delta < 0)
-            HistoryParams("<", "DESC") else
-                HistoryParams(">", "ASC")
+        val (cmpOp, orderBy) = if (delta < 0) Pair("<", "DESC") else Pair(">", "ASC")
         val stmt = prepare("""
             SELECT 
               creditor_payto_uri
@@ -852,10 +981,10 @@ class Database(private val dbConfig: String, internal val bankCurrency: String)
               ,bank_account_id
               ,bank_transaction_id
             FROM bank_account_transactions
-	        WHERE bank_transaction_id ${ops.cmpOp} ? 
+	        WHERE bank_transaction_id ${cmpOp} ? 
               AND bank_account_id=?
               ${if (withDirection != null) "AND direction=?::direction_enum" else ""}
-            ORDER BY bank_transaction_id ${ops.orderBy}
+            ORDER BY bank_transaction_id ${orderBy}
             LIMIT ?
         """)
         stmt.setLong(1, start)
@@ -1371,143 +1500,4 @@ class Database(private val dbConfig: String, internal val bankCurrency: String)
             )
         }
     }
-}
-
-// TODO perf, merge with Database to reuse connection and prepared statement
-suspend fun <T> bankTransactionPoolHistory(
-    db: Database, 
-    params: HistoryParams, 
-    bankAccountId: Long, 
-    direction: TransactionDirection,
-    map: (BankAccountTransaction) -> T?
-): List<T> {
-    val conn = db.conn() ?: throw internalServerError("DB connection down");
-    val channel = "${direction.name}_$bankAccountId";
-    var start = params.start
-    var delta = params.delta
-    var poll_ms = params.poll_ms;
-
-    val (cmpOp, orderBy) = if (delta < 0) Pair("<", "DESC") else Pair(">", "ASC")
-    val stmt = conn.prepareStatement("""
-        SELECT 
-            creditor_payto_uri
-            ,creditor_name
-            ,debtor_payto_uri
-            ,debtor_name
-            ,subject
-            ,(amount).val AS amount_val
-            ,(amount).frac AS amount_frac
-            ,transaction_date
-            ,account_servicer_reference
-            ,payment_information_id
-            ,end_to_end_id
-            ,bank_account_id
-            ,bank_transaction_id
-        FROM bank_account_transactions
-        WHERE bank_transaction_id ${cmpOp} ? 
-            AND bank_account_id=?
-            AND direction=?::direction_enum
-        ORDER BY bank_transaction_id ${orderBy}
-        LIMIT ?
-    """)
-
-    // If going backward with a starting point, it is useless to poll
-    if (delta < 0 && start != Long.MAX_VALUE) {
-        poll_ms = 0;
-    }
-
-    // Only start expensive listening if we intend to poll
-    if (poll_ms > 0) {
-        conn.execSQLUpdate("LISTEN $channel");
-    }
-
-    val items = mutableListOf<T>()
-
-    fun bankTransactionGetHistory(): List<BankAccountTransaction> {
-        stmt.setLong(1, start)
-        stmt.setLong(2, bankAccountId)
-        stmt.setString(3, direction.name)
-        stmt.setLong(4, abs(delta))
-        val rs = stmt.executeQuery()
-        rs.use {
-            val ret = mutableListOf<BankAccountTransaction>()
-            if (!it.next()) return ret
-            do {
-                ret.add(
-                    BankAccountTransaction(
-                        creditorPaytoUri = it.getString("creditor_payto_uri"),
-                        creditorName = it.getString("creditor_name"),
-                        debtorPaytoUri = it.getString("debtor_payto_uri"),
-                        debtorName = it.getString("debtor_name"),
-                        amount = TalerAmount(
-                            it.getLong("amount_val"),
-                            it.getInt("amount_frac"),
-                            db.bankCurrency
-                        ),
-                        accountServicerReference = it.getString("account_servicer_reference"),
-                        endToEndId = it.getString("end_to_end_id"),
-                        direction = direction,
-                        bankAccountId = it.getLong("bank_account_id"),
-                        paymentInformationId = it.getString("payment_information_id"),
-                        subject = it.getString("subject"),
-                        transactionDate = it.getLong("transaction_date").microsToJavaInstant() ?: throw faultyTimestampByBank(),
-                        dbRowId = it.getLong("bank_transaction_id")
-                ))
-            } while (it.next())
-            return ret
-        }
-    }
-
-    fun loadBankHistory() {
-        while (delta != 0L) {
-            val history = bankTransactionGetHistory()
-            if (history.isEmpty())
-                break;
-            history.forEach {
-                val item = map(it);
-                // Advance cursor
-                start = it.expectRowId()
-    
-                if (item != null) {
-                    items.add(item)
-                    // Reduce delta
-                    if (delta < 0) delta++ else delta--;
-                }
-            }
-        }
-    }
-
-    loadBankHistory()
-
-    // Long polling
-    while (delta != 0L && poll_ms > 0) {
-        var remaining = abs(delta);
-        do {
-            val pollStart = System.currentTimeMillis()
-            conn.getNotifications(poll_ms.toInt()).forEach {
-                val id = it.parameter.toLong()
-                val new = when {
-                    params.start == Long.MAX_VALUE -> true
-                    delta < 0 -> id < start
-                    else -> id > start
-                }
-                if (new) remaining -= 1
-            }
-            val pollEnd = System.currentTimeMillis()
-            poll_ms -= pollEnd - pollStart
-        } while (poll_ms > 0 && remaining > 0L)
-
-        // If going backward without a starting point, we reset loading progress
-        if (params.start == Long.MAX_VALUE) {
-            start = params.start
-            delta = params.delta
-            items.clear()
-        }
-        loadBankHistory()
-    }
-
-    conn.execSQLUpdate("UNLISTEN $channel");
-    conn.close()
-
-    return items.toList();
 }
\ No newline at end of file
diff --git a/bank/src/main/kotlin/tech/libeufin/bank/WireGatewayApiHandlers.kt b/bank/src/main/kotlin/tech/libeufin/bank/WireGatewayApiHandlers.kt
index e52828f6..b3b9603c 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/WireGatewayApiHandlers.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/WireGatewayApiHandlers.kt
@@ -122,11 +122,11 @@ fun Routing.talerWireGatewayHandlers(db: Database, ctx: BankApplicationContext)
 
     suspend fun <T> historyEndpoint(call: ApplicationCall, direction: TransactionDirection, reduce: (List<T>, String) -> Any, map: (BankAccountTransaction) -> T?) {
         call.authCheck(TokenScope.readonly, true)
-        val params = getHistoryParams(call.request)
+        val params = getHistoryParams(call.request.queryParameters)
         val bankAccount = call.bankAccount()
         if (!bankAccount.isTalerExchange) throw forbidden("History is not related to a Taler exchange.")
 
-        val items = bankTransactionPoolHistory(db, params, bankAccount.expectRowId(), direction, map);
+        val items = db.bankTransactionPoolHistory(params, bankAccount.expectRowId(), direction, map);
     
         if (items.isEmpty()) {
             call.respond(HttpStatusCode.NoContent)
diff --git a/bank/src/main/kotlin/tech/libeufin/bank/helpers.kt b/bank/src/main/kotlin/tech/libeufin/bank/helpers.kt
index 9b8e02e6..6514e0ef 100644
--- a/bank/src/main/kotlin/tech/libeufin/bank/helpers.kt
+++ b/bank/src/main/kotlin/tech/libeufin/bank/helpers.kt
@@ -372,9 +372,9 @@ data class HistoryParams(
  * Extracts the query parameters from "history-like" endpoints,
  * providing the defaults according to the API specification.
  */
-fun getHistoryParams(req: ApplicationRequest): HistoryParams {
+fun getHistoryParams(params: Parameters): HistoryParams {
     val deltaParam: String =
-        req.queryParameters["delta"] ?: throw MissingRequestParameterException(parameterName = "delta")
+        params["delta"] ?: throw MissingRequestParameterException(parameterName = "delta")
     val delta: Long = try {
         deltaParam.toLong()
     } catch (e: Exception) {
@@ -382,7 +382,7 @@ fun getHistoryParams(req: ApplicationRequest): HistoryParams {
         throw badRequest("Param 'delta' not a number")
     }
     // Note: minimum 'start' is zero, as database IDs start from 1.
-    val start: Long = when (val param = req.queryParameters["start"]) {
+    val start: Long = when (val param = params["start"]) {
         null -> if (delta >= 0) 0L else Long.MAX_VALUE
         else -> try {
             param.toLong()
@@ -391,7 +391,7 @@ fun getHistoryParams(req: ApplicationRequest): HistoryParams {
             throw badRequest("Param 'start' not a number")
         }
     }
-    val poll_ms: Long = when (val param = req.queryParameters["long_poll_ms"]) {
+    val poll_ms: Long = when (val param = params["long_poll_ms"]) {
         null -> 0
         else -> try {
             param.toLong()
diff --git a/bank/src/test/kotlin/TalerApiTest.kt b/bank/src/test/kotlin/TalerApiTest.kt
index 894e94b8..fd1fc38f 100644
--- a/bank/src/test/kotlin/TalerApiTest.kt
+++ b/bank/src/test/kotlin/TalerApiTest.kt
@@ -222,6 +222,30 @@ class TalerApiTest {
             )
         )
 
+        suspend fun HttpResponse.assertHistory(size: Int) {
+            assertOk()
+            val txt = this.bodyAsText()
+            val history = Json.decodeFromString<IncomingHistory>(txt)
+            val params = getHistoryParams(this.call.request.url.parameters)
+       
+            // testing the size is like expected.
+            assert(history.incoming_transactions.size == size) {
+                println("incoming_transactions has wrong size: ${history.incoming_transactions.size}")
+                println("Response was: ${txt}")
+            }
+            if (params.delta < 0) {
+                // testing that the first row_id is at most the 'start' query param.
+                assert(history.incoming_transactions[0].row_id <= params.start)
+                // testing that the row_id decreases.
+                assert(history.incoming_transactions.windowed(2).all { (a, b) -> a.row_id > b.row_id })
+            } else {
+                // testing that the first row_id is at least the 'start' query param.
+                assert(history.incoming_transactions[0].row_id >= params.start)
+                // testing that the row_id increases.
+                assert(history.incoming_transactions.windowed(2).all { (a, b) -> a.row_id < b.row_id })
+            }
+        }
+
         testApplication {
             application {
                 corebankWebApp(db, ctx)
@@ -250,34 +274,22 @@ class TalerApiTest {
             // Check ignore bogus subject
             client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=7") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: IncomingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.incoming_transactions.size)
-            }
+            }.assertHistory(5)
            
             // Check skip bogus subject
             client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=5") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: IncomingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.incoming_transactions.size)
-            }
+            }.assertHistory(5)
             
             // Check no useless polling
             client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=-6&start=20&long_poll_ms=6000000") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: IncomingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.incoming_transactions.size)
-            }
+            }.assertHistory(5)
 
             // Check polling end
             client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=6&long_poll_ms=60") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: IncomingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.incoming_transactions.size)
-            }
+            }.assertHistory(5)
 
             // Check polling succedd
             runBlocking {
@@ -287,10 +299,7 @@ class TalerApiTest {
                 }
                 client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=-6&long_poll_ms=6000000") {
                     basicAuth("bar", "secret")
-                }.assertOk().run {
-                    val j: IncomingHistory = Json.decodeFromString(this.bodyAsText())
-                    assertEquals(6, j.incoming_transactions.size)
-                }
+                }.assertHistory(6)
             }
 
             // Check polling timeout
@@ -301,10 +310,7 @@ class TalerApiTest {
                 }
                 client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=8&long_poll_ms=300") {
                     basicAuth("bar", "secret")
-                }.assertOk().run {
-                    val j: IncomingHistory = Json.decodeFromString(this.bodyAsText())
-                    assertEquals(7, j.incoming_transactions.size)
-                }
+                }.assertHistory(7)
             }
 
             // Testing ranges.
@@ -315,36 +321,12 @@ class TalerApiTest {
             // forward range:
             client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=10&start=30") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val txt = this.bodyAsText()
-                val history = Json.decodeFromString<IncomingHistory>(txt)
-                // testing the size is like expected.
-                assert(history.incoming_transactions.size == 10) {
-                    println("incoming_transaction has wrong size: ${history.incoming_transactions.size}")
-                    println("Response was: ${txt}")
-                }
-                // testing that the first row_id is at least the 'start' query param.
-                assert(history.incoming_transactions[0].row_id >= 30)
-                // testing that the row_id increases.
-                assert(history.incoming_transactions.windowed(2).all { (a, b) -> a.row_id < b.row_id })
-            }
+            }.assertHistory(10)
 
             // backward range:
             client.get("/accounts/bar/taler-wire-gateway/history/incoming?delta=-10&start=300") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val txt = this.bodyAsText()
-                val history = Json.decodeFromString<IncomingHistory>(txt)
-                // testing the size is like expected.
-                assert(history.incoming_transactions.size == 10) {
-                    println("incoming_transaction has wrong size: ${history.incoming_transactions.size}")
-                    println("Response was: ${txt}")
-                }
-                // testing that the first row_id is at most the 'start' query param.
-                assert(history.incoming_transactions[0].row_id <= 300)
-                // testing that the row_id decreases.
-                assert(history.incoming_transactions.windowed(2).all { (a, b) -> a.row_id > b.row_id })
-            } 
+            }.assertHistory(10)
         }
     }
 
@@ -363,6 +345,30 @@ class TalerApiTest {
             )
         )
 
+        suspend fun HttpResponse.assertHistory(size: Int) {
+            assertOk()
+            val txt = this.bodyAsText()
+            val history = Json.decodeFromString<OutgoingHistory>(txt)
+            val params = getHistoryParams(this.call.request.url.parameters)
+       
+            // testing the size is like expected.
+            assert(history.outgoing_transactions.size == size) {
+                println("outgoing_transactions has wrong size: ${history.outgoing_transactions.size}")
+                println("Response was: ${txt}")
+            }
+            if (params.delta < 0) {
+                // testing that the first row_id is at most the 'start' query param.
+                assert(history.outgoing_transactions[0].row_id <= params.start)
+                // testing that the row_id decreases.
+                assert(history.outgoing_transactions.windowed(2).all { (a, b) -> a.row_id > b.row_id })
+            } else {
+                // testing that the first row_id is at least the 'start' query param.
+                assert(history.outgoing_transactions[0].row_id >= params.start)
+                // testing that the row_id increases.
+                assert(history.outgoing_transactions.windowed(2).all { (a, b) -> a.row_id < b.row_id })
+            }
+        }
+
         testApplication {
             application {
                 corebankWebApp(db, ctx)
@@ -391,34 +397,22 @@ class TalerApiTest {
             // Check ignore bogus subject
             client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=7") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: OutgoingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.outgoing_transactions.size)
-            }
+            }.assertHistory(5)
            
             // Check skip bogus subject
             client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=5") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: OutgoingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.outgoing_transactions.size)
-            }
+            }.assertHistory(5)
 
             // Check no useless polling
             client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=-6&start=20&long_poll_ms=6000000") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: OutgoingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.outgoing_transactions.size)
-            }
+            }.assertHistory(5)
 
             // Check polling end
             client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=6&long_poll_ms=60") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val j: OutgoingHistory = Json.decodeFromString(this.bodyAsText())
-                assertEquals(5, j.outgoing_transactions.size)
-            }
+            }.assertHistory(5)
 
             // Check polling succedd
             runBlocking {
@@ -428,10 +422,7 @@ class TalerApiTest {
                 }
                 client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=-6&long_poll_ms=6000000") {
                     basicAuth("bar", "secret")
-                }.assertOk().run {
-                    val j: OutgoingHistory = Json.decodeFromString(this.bodyAsText())
-                    assertEquals(6, j.outgoing_transactions.size)
-                }
+                }.assertHistory(6)
             }
 
             // Check polling timeout
@@ -442,10 +433,7 @@ class TalerApiTest {
                 }
                 client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=8&long_poll_ms=300") {
                     basicAuth("bar", "secret")
-                }.assertOk().run {
-                    val j: OutgoingHistory = Json.decodeFromString(this.bodyAsText())
-                    assertEquals(7, j.outgoing_transactions.size)
-                }
+                }.assertHistory(7)
             }
 
             // Testing ranges.
@@ -455,36 +443,12 @@ class TalerApiTest {
             // forward range:
             client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=10&start=30") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val txt = this.bodyAsText()
-                val history = Json.decodeFromString<OutgoingHistory>(txt)
-                // testing the size is like expected.
-                assert(history.outgoing_transactions.size == 10) {
-                    println("outgoing_transactions has wrong size: ${history.outgoing_transactions.size}")
-                    println("Response was: ${txt}")
-                }
-                // testing that the first row_id is at least the 'start' query param.
-                assert(history.outgoing_transactions[0].row_id >= 30)
-                // testing that the row_id increases.
-                assert(history.outgoing_transactions.windowed(2).all { (a, b) -> a.row_id < b.row_id })
-            }
+            }.assertHistory(10)
 
             // backward range:
             client.get("/accounts/bar/taler-wire-gateway/history/outgoing?delta=-10&start=300") {
                 basicAuth("bar", "secret")
-            }.assertOk().run {
-                val txt = this.bodyAsText()
-                val history = Json.decodeFromString<OutgoingHistory>(txt)
-                // testing the size is like expected.
-                assert(history.outgoing_transactions.size == 10) {
-                    println("outgoing_transactions has wrong size: ${history.outgoing_transactions.size}")
-                    println("Response was: ${txt}")
-                }
-                // testing that the first row_id is at most the 'start' query param.
-                assert(history.outgoing_transactions[0].row_id <= 300)
-                // testing that the row_id decreases.
-                assert(history.outgoing_transactions.windowed(2).all { (a, b) -> a.row_id > b.row_id })
-            } 
+            }.assertHistory(10)
         }
     }
 
@@ -689,4 +653,4 @@ class TalerApiTest {
         )
         assertEquals(withPort, "taler://withdraw/www.example.com:9876/taler-integration/my-id")
     }
-}
+}
\ No newline at end of file