libeufin

helpers.kt (5241B)

---

 /*
  * This file is part of LibEuFin.
  * Copyright (C) 2024-2025 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.common.db
 
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull
 import tech.libeufin.common.HistoryParams
 import tech.libeufin.common.PageParams
 import java.sql.PreparedStatement
 import java.sql.ResultSet
 import kotlin.math.abs
 import kotlin.math.min
 
 /**
  * Hard upper bound on the number of records returned by a single
  * query, regardless of the limit requested by the client.
  */
 private const val MAX_RECORDS: Long = 50_000
 
 /** Apply paging logic to a sql query */
 suspend fun <T> DbPool.page(
     params: PageParams,
     idName: String,
     query: String,
     args: TalerStatement.() -> Unit = {},
     map: (ResultSet) -> T
 ): List<T> {
     val backward = params.limit < 0
     val pageQuery = """
         $query
         $idName ${if (backward) '<' else '>'} ?
         ORDER BY $idName ${if (backward) "DESC" else "ASC"}
         LIMIT ?
     """
     return serializable(pageQuery) {
         args()
         bind(params.offset)
         // Widen before abs(): abs(Int.MIN_VALUE) is Int.MIN_VALUE, which would
         // reach Postgres as a negative LIMIT.
         bind(min(MAX_RECORDS, abs(params.limit.toLong())))
         all { map(it) }
     }
 }
 
 /**
 * The following function returns the list of transactions, according
 * to the history parameters and perform long polling when necessary
 */
 suspend fun <T> DbPool.poolHistory(
     params: HistoryParams, 
     bankAccountId: Long,
     listen: suspend (Long, suspend (Flow<Long>) -> List<T>) -> List<T>,
     query: String,
     accountColumn: String = "bank_account_id",
     map: (ResultSet) -> T
 ): List<T> {
 
     suspend fun load(): List<T> = page(
         params.page, 
         "bank_transaction_id", 
         "$query $accountColumn=? AND", 
         {
             bind(bankAccountId)
         },
         map
     )
     
     // When going backward there is always at least one transaction or none
     return if (params.page.limit >= 0 && params.polling.timeout_ms > 0) {
         listen(bankAccountId) { flow ->
             coroutineScope {
                 // Start buffering notification before loading transactions to not miss any
                 val polling = launch {
                     withTimeoutOrNull(params.polling.timeout_ms) {
                         flow.first { it > params.page.offset } // Always forward so >
                     }
                 }    
                 // Initial loading
                 val init = load()
                 // Long polling if we found no transactions
                 if (init.isEmpty()) {
                     if (polling.join() != null) {
                         load()
                     } else {
                         init
                     }
                 } else {
                     polling.cancel()
                     init
                 }
             }
         }
     } else {
         load()
     }
 }
 
 /**
 * The following function returns the list of transactions, according
 * to the history parameters and perform long polling when necessary
 */
 suspend fun <T> DbPool.poolHistoryGlobal(
     params: HistoryParams, 
     listen: suspend (suspend (Flow<Long>) -> List<T>) -> List<T>,
     query: String,
     idColumnValue: String,
     map: (ResultSet) -> T
 ): List<T> {
 
     suspend fun load(): List<T> = page(
         params.page, 
         idColumnValue,
         query,
         map=map
     )
 
     // When going backward there is always at least one transaction or none
     return if (params.page.limit >= 0 && params.polling.timeout_ms > 0) {
         listen { flow ->
             coroutineScope {
                 // Start buffering notification before loading transactions to not miss any
                 val polling = launch {
                     withTimeoutOrNull(params.polling.timeout_ms) {
                         flow.first { it > params.page.offset } // Always forward so >
                     }
                 }    
                 // Initial loading
                 val init = load()
                 // Long polling if we found no transactions
                 if (init.isEmpty()) {
                     if (polling.join() != null) {
                         load()
                     } else {
                         init
                     }
                 } else {
                     polling.cancel()
                     init
                 }
             }
         }
     } else {
         load()
     }
 }