libeufin

InitiatedDAO.kt (14430B)

---

 /*
  * This file is part of LibEuFin.
  * Copyright (C) 2024, 2025, 2026 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.nexus.db
 
 import tech.libeufin.common.asInstant
 import tech.libeufin.common.db.*
 import tech.libeufin.common.micros
 import tech.libeufin.nexus.iso20022.*
 import java.time.Instant
 
 /** Data access logic for initiated outgoing payments */
 class InitiatedDAO(private val db: Database) {
     private val UNSETTLED_FILTER = 
         "status NOT IN (${SubmissionState.SETTLED.joinToString(",") { "'$it'" }})"
     private val PENDING_FILTER =
         "status IN (${SubmissionState.PENDING.joinToString(",") { "'$it'" }})"
 
     /** Outgoing payments initiation result */
     sealed interface PaymentInitiationResult {
         data class Success(val id: Long): PaymentInitiationResult
         data object RequestUidReuse: PaymentInitiationResult
     }
 
     /** Initiate a new payment */
     suspend fun create(payment: InitiatedPayment): PaymentInitiationResult = db.serializable(
         """
         INSERT INTO initiated_outgoing_transactions (
              amount
             ,subject
             ,credit_payto
             ,initiation_time
             ,end_to_end_id
         ) VALUES ((?,?)::taler_amount,?,?,?,?)
         RETURNING initiated_outgoing_transaction_id
         """
     ) { 
         // TODO check payto uri
         bind(payment.amount)
         bind(payment.subject)
         bind(payment.creditor.toString())
         bind(payment.initiationTime)
         bind(payment.endToEndId)
         oneUniqueViolation(PaymentInitiationResult.RequestUidReuse) {
             PaymentInitiationResult.Success(it.getLong("initiated_outgoing_transaction_id"))
         }
     }
 
     /** Register submission success of order [orderId] for batch [id] at [timestamp] */
     suspend fun batchSubmissionSuccess(
         id: Long,
         timestamp: Instant,
         orderId: String?
     ) = db.serializableTransaction { tx ->
         // Update batch status
         val updated = tx.withStatement(
             """
             UPDATE initiated_outgoing_batches 
             SET  status = 'pending'
                 ,submission_date = ?
                 ,status_msg = NULL
                 ,order_id = ?
                 ,submission_counter = submission_counter + 1
             WHERE initiated_outgoing_batch_id = ? AND order_id IS NULL
             """
         ) {
             bind(timestamp)
             bind(orderId)
             bind(id)
             executeUpdate()
         }
         if (updated > 0) {
             // Update unsettled batch's transaction status
             tx.withStatement(
                 """
                 UPDATE initiated_outgoing_transactions 
                 SET status = 'pending', status_msg = NULL
                 WHERE initiated_outgoing_batch_id = ? AND $UNSETTLED_FILTER
                 """
             ) {
                 bind(id)
                 executeUpdate()
             }
         }
     }
 
     /** Register submission failure with [msg] for batch [id] at [timestamp]*/
     suspend fun batchSubmissionFailure(
         id: Long,
         timestamp: Instant,
         msg: String?,
         permanent: Boolean = false
     ) = db.serializableTransaction { tx ->
         // Update batch status
         tx.withStatement(
             """
             UPDATE initiated_outgoing_batches
             SET  status = ?::submission_state
                 ,submission_date = ?
                 ,status_msg = ?
                 ,submission_counter = submission_counter + 1
             WHERE initiated_outgoing_batch_id = ?
             """
         ) {
             if (permanent) {
                 bind(StatusUpdate.permanent_failure)
             } else {
                 bind(StatusUpdate.transient_failure)
             }
             bind(timestamp)
             bind(msg)
             bind(id)
             executeUpdate()
         }
         // Update unsettled batch's transaction status
         tx.withStatement(
             """
             UPDATE initiated_outgoing_transactions 
             SET status = ?::submission_state, status_msg = ?
             WHERE initiated_outgoing_batch_id = ? AND $UNSETTLED_FILTER
             """
         ) {
             if (permanent) {
                 bind(StatusUpdate.permanent_failure)
             } else {
                 bind(StatusUpdate.transient_failure)
             }
             bind(msg)
             bind(id)
             executeUpdate()
         }
     }
 
     /** Register order step [msg] for [orderId] */
     suspend fun orderStep(orderId: String, msg: String) = db.serializableTransaction { tx ->
         // Update pending batch status
         val batchId = tx.withStatement(
             """
             UPDATE initiated_outgoing_batches 
             SET status = 'pending', status_msg = ?
             WHERE order_id = ? AND $PENDING_FILTER
             RETURNING initiated_outgoing_batch_id
             """
         ) {
             bind(msg)
             bind(orderId)
             oneOrNull { it.getLong(1)  }
         }
         if (batchId != null) {
             // Update pending batch's transaction status
             tx.withStatement(
                 """
                 UPDATE initiated_outgoing_transactions 
                 SET status = 'pending', status_msg = ?
                 WHERE initiated_outgoing_batch_id = ? AND $PENDING_FILTER
                 """
             ) {
                 bind(msg)
                 bind(batchId)
                 executeUpdate()
             }
         }
     }
 
     /** Register order success for [orderId] and return message_id if found */
     suspend fun orderSuccess(orderId: String): String? = db.serializableTransaction { tx ->
        // Update batch status
         val result = tx.withStatement(
             """
             UPDATE initiated_outgoing_batches 
             SET status = 'success'
             WHERE order_id = ?
             RETURNING initiated_outgoing_batch_id, message_id
             """
         ) {
             bind(orderId)
             oneOrNull { 
                 Pair(
                     it.getLong("initiated_outgoing_batch_id"),
                     it.getString("message_id")
                 )
             }
         }
         if (result == null) return@serializableTransaction null
         val (batchId, messageId) = result
         // Update unsettled batch's transaction status
         tx.withStatement(
             """
             UPDATE initiated_outgoing_transactions 
             SET status = 'pending'
             WHERE initiated_outgoing_batch_id = ? AND $UNSETTLED_FILTER
             """
         ) {
             bind(batchId)
             executeUpdate()
         }
         messageId
     }
 
     /** Register order failure for [orderId] and return message_id and previous status_msg if found */
     suspend fun orderFailure(orderId: String): Pair<String, String?>? = db.serializableTransaction { tx ->
         // Update batch status
         val result = tx.withStatement(
             """
             UPDATE initiated_outgoing_batches 
             SET status = 'permanent_failure'
             WHERE order_id = ?
             RETURNING initiated_outgoing_batch_id, message_id, status_msg
             """
         ) {
             bind(orderId)
             oneOrNull { 
                 Triple(
                     it.getLong("initiated_outgoing_batch_id"),
                     it.getString("message_id"),
                     it.getString("status_msg")
                 )
             }
         }
         if (result == null) return@serializableTransaction null
         val (batchId, messageId, msg) = result
         // Update batch's transaction status
         tx.withStatement(
             """
             UPDATE initiated_outgoing_transactions 
             SET status = 'permanent_failure'
             WHERE initiated_outgoing_batch_id = ?
             """
         ) {
             bind(batchId)
             executeUpdate()
         }
         Pair(messageId, msg)
     }
 
     /** Register payment status [state] with [msg] for batch [msgId] */
     suspend fun batchStatusUpdate(msgId: String, state: StatusUpdate, msg: String?) = db.serializable(
         "SELECT out_ok FROM batch_status_update(?,?::submission_state,?)"
     ) { 
         bind(msgId)
         bind(state)
         bind(msg)
         one {
             it.getBoolean(1)
         }
     }
 
     /** Register payment status [state] with [msg] for transaction [endToEndId] in batch [msgId] */
     suspend fun txStatusUpdate(endToEndId: String, msgId: String?, state: StatusUpdate, msg: String?) = db.serializable(
         "SELECT out_ok FROM tx_status_update(?,?,?::submission_state,?)"
     ) {
         bind(endToEndId)
         bind(msgId)
         bind(state)
         bind(msg)
         one {
             it.getBoolean(1)
         }
     }
 
     /** Unsettled initiated payment in batch [msgId] */
     suspend fun unsettledTxInBatch(msgId: String, executionTime: Instant) = db.serializable(
         """
         SELECT end_to_end_id
               ,(amount).val as amount_val
               ,(amount).frac as amount_frac
               ,subject
               ,credit_payto
         FROM initiated_outgoing_transactions
             JOIN initiated_outgoing_batches USING (initiated_outgoing_batch_id)
         WHERE message_id = ?
             AND initiated_outgoing_transactions.$UNSETTLED_FILTER
         """
     ) { 
         bind(msgId)
         all {
             OutgoingPayment(
                 id = OutgoingId(
                     msgId = msgId,
                     endToEndId = it.getString("end_to_end_id"),
                     acctSvcrRef = null
                 ),
                 amount = it.getAmount("amount", db.currency),
                 subject = it.getString("subject"),
                 executionTime = executionTime,
                 creditor = it.getIbanPayto("credit_payto")
             )
         }
     }
 
     suspend fun ack(id: Long): Boolean {
         return db.serializable("UPDATE initiated_outgoing_transactions SET awaiting_ack=false WHERE initiated_outgoing_transaction_id = ?") {
             bind(id)
             executeUpdateCheck()
         }
     }
 
     /** Group unbatched transaction into a single batch */
     suspend fun batch(timestamp: Instant, ebicsId: String, requireAck: Boolean) {
         db.serializable("SELECT FROM batch_outgoing_transactions(?, ?, ?)") {
             bind(timestamp)
             bind(ebicsId)
             bind(requireAck)
             executeQuery()
         }
     }
 
     /** List every initiated payment pending submission in the order they should be submitted */
     suspend fun submittable(): List<PaymentBatch> {
         val selectPart = """
             SELECT initiated_outgoing_batch_id, message_id, creation_date, (sum).val as sum_val, (sum).frac as sum_frac
             FROM initiated_outgoing_batches
         """
         return db.serializableTransaction { tx ->
             // We want to maximize the number of successfully submitted batches in the event 
             // of a malformed transaction or a persistent error classified as transient. We send 
             // the unsubmitted batches first, starting with the oldest by creation time.
             // This is the happy path, giving every batch a chance while being fair on the
             // basis of creation date. 
             // Then we retry the failed batches, starting with the oldest by submission time.
             // This the bad path retrying each failed batch applying a rotation based on 
             // resubmission time.
             val batches = tx.withStatement(
                 """
                 ($selectPart WHERE status='unsubmitted' ORDER BY creation_date)
                 UNION ALL
                 ($selectPart WHERE status='transient_failure' ORDER BY submission_date)
                 """
             ) {
                 all {
                     PaymentBatch(
                         id = it.getLong("initiated_outgoing_batch_id"),
                         messageId = it.getString("message_id"),
                         creationDate = it.getLong("creation_date").asInstant(),
                         sum = it.getAmount("sum", db.currency),
                         payments = emptyList()
                     )
                 }
             }.associate { it.id to Pair(it, mutableListOf<InitiatedPayment>()) }
 
             // Then load transactions
             tx.withStatement(
                 """
                 SELECT
                     initiated_outgoing_transaction_id
                     ,(amount).val as amount_val
                     ,(amount).frac as amount_frac
                     ,subject
                     ,credit_payto
                     ,initiated_outgoing_transactions.initiation_time
                     ,end_to_end_id
                     ,initiated_outgoing_batch_id
                 FROM initiated_outgoing_transactions
                     JOIN initiated_outgoing_batches USING (initiated_outgoing_batch_id)
                 WHERE initiated_outgoing_batches.status IN ('unsubmitted', 'transient_failure')
                 """
             ) {
                 all {
                     val payment = InitiatedPayment(
                         id = it.getLong("initiated_outgoing_transaction_id"),
                         amount = it.getAmount("amount", db.currency),
                         creditor = it.getIbanPayto("credit_payto"),
                         subject = it.getString("subject"),
                         initiationTime = it.getLong("initiation_time").asInstant(),
                         endToEndId = it.getString("end_to_end_id")
                     )
                     val batchId = it.getLong("initiated_outgoing_batch_id")
                     batches[batchId]!!.second.add(payment)
                     Unit
                 }
             }
 
             batches.values.map { (it, payments) -> it.copy(payments = payments) }
         }
     }
 }