[ANCHOR-1407]: Enhance underpayment handling and transaction recording logic

platform/src/main/java/org/stellar/anchor/platform/component/observer/PaymentObserverBeans.java

@@ -9,6 +9,7 @@
 import org.stellar.anchor.apiclient.PlatformApiClient;
 import org.stellar.anchor.asset.AssetService;
 import org.stellar.anchor.config.StellarNetworkConfig;
+import org.stellar.anchor.event.EventService;
 import org.stellar.anchor.ledger.Horizon;
 import org.stellar.anchor.ledger.LedgerClient;
 import org.stellar.anchor.ledger.StellarRpc;
@@ -95,7 +96,9 @@ public DefaultPaymentListener paymentListener(
       JdbcSep6TransactionStore sep6TransactionStore,
       PlatformApiClient platformApiClient,
       RpcConfig rpcConfig,
-      AssetService assetService) {
+      AssetService assetService,
+      EventService eventService,
+      SacToAssetMapper sacToAssetMapper) {
 
     return new DefaultPaymentListener(
         paymentObservingAccountsManager,
@@ -104,7 +107,9 @@ public DefaultPaymentListener paymentListener(
         sep6TransactionStore,
         platformApiClient,
         rpcConfig,
-        assetService);
+        assetService,
+        eventService,
+        sacToAssetMapper);
   }
 
   /**

platform/src/main/java/org/stellar/anchor/platform/observer/stellar/DefaultPaymentListener.java

@@ -17,15 +17,19 @@
 import java.math.RoundingMode;
 import java.util.List;
 import java.util.Objects;
+import java.util.UUID;
 import org.springframework.dao.DataAccessException;
 import org.springframework.jdbc.CannotGetJdbcConnectionException;
 import org.springframework.transaction.CannotCreateTransactionException;
 import org.springframework.transaction.TransactionException;
 import org.stellar.anchor.api.asset.AssetInfo;
+import org.stellar.anchor.api.event.AnchorEvent;
 import org.stellar.anchor.api.exception.AnchorException;
+import org.stellar.anchor.api.platform.PlatformTransactionData.Sep;
 import org.stellar.anchor.api.sep.SepTransactionStatus;
 import org.stellar.anchor.apiclient.PlatformApiClient;
 import org.stellar.anchor.asset.AssetService;
+import org.stellar.anchor.event.EventService;
 import org.stellar.anchor.ledger.LedgerTransaction;
 import org.stellar.anchor.ledger.LedgerTransaction.LedgerInvokeHostFunctionOperation;
 import org.stellar.anchor.ledger.LedgerTransaction.LedgerPayment;
@@ -34,6 +38,8 @@
 import org.stellar.anchor.platform.data.*;
 import org.stellar.anchor.platform.observer.PaymentListener;
 import org.stellar.anchor.platform.service.AnchorMetrics;
+import org.stellar.anchor.platform.utils.PaymentHelper;
+import org.stellar.anchor.platform.utils.PlatformTransactionHelper;
 import org.stellar.anchor.util.AssetHelper;
 import org.stellar.anchor.util.GsonUtils;
 import org.stellar.sdk.Memo;
@@ -57,6 +63,8 @@ public class DefaultPaymentListener implements PaymentListener {
   private final PlatformApiClient platformApiClient;
   private final RpcConfig rpcConfig;
   private final AssetService assetService;
+  private final EventService.Session eventSession;
+  private final SacToAssetMapper sacToAssetMapper;
 
   public DefaultPaymentListener(
       PaymentObservingAccountsManager paymentObservingAccountsManager,
@@ -65,14 +73,19 @@ public DefaultPaymentListener(
       JdbcSep6TransactionStore sep6TransactionStore,
       PlatformApiClient platformApiClient,
       RpcConfig rpcConfig,
-      AssetService assetService) {
+      AssetService assetService,
+      EventService eventService,
+      SacToAssetMapper sacToAssetMapper) {
     this.paymentObservingAccountsManager = paymentObservingAccountsManager;
     this.sep31TransactionStore = sep31TransactionStore;
     this.sep24TransactionStore = sep24TransactionStore;
     this.sep6TransactionStore = sep6TransactionStore;
     this.platformApiClient = platformApiClient;
     this.rpcConfig = rpcConfig;
     this.assetService = assetService;
+    this.eventSession =
+        eventService.createSession(this.getClass().getName(), EventService.EventQueue.TRANSACTION);
+    this.sacToAssetMapper = sacToAssetMapper;
   }
 
   @Override
@@ -719,7 +732,8 @@ boolean checkAssetAmountSufficient(
       LedgerTransaction ledgerTransaction,
       LedgerPayment ledgerPayment,
       JdbcSepTransaction sepTransaction,
-      boolean enforceAssetMatch) {
+      boolean enforceAssetMatch)
+      throws IOException {
     String paymentAssetName = "stellar:" + getSep11AssetName(ledgerPayment.getAsset());
     if (!sepTransaction.getAmountInAsset().equals(paymentAssetName)) {
       if (enforceAssetMatch) {
@@ -753,6 +767,7 @@ boolean checkAssetAmountSufficient(
           expectedAmount.toPlainString(),
           gotAmount.toPlainString());
       Metrics.counter(AnchorMetrics.PAYMENT_OBSERVER_AMOUNT_INSUFFICIENT.toString()).increment();
+      recordUnderpayment(ledgerTransaction, sepTransaction);
       return false;
     }
 
@@ -763,4 +778,66 @@ boolean checkAssetAmountSufficient(
         ledgerTransaction.getHash());
     return true;
   }
+
+  /**
+   * Gives a payment rejected for its amount somewhere to go: the Stellar transaction is added to
+   * the transaction's {@code stellar_transactions} and the business server is told with one {@code
+   * transaction_status_changed} event. The status and the amounts stay as they are, so the payment
+   * is not credited; the business server decides whether to credit it through {@code
+   * notify_onchain_funds_received}, ask for the rest, or refund it. A failure to save or to publish
+   * is rethrown as an {@link IOException}, which holds the observer's cursor so the payment is
+   * retried, as for a failed Platform API notification.
+   */
+  private void recordUnderpayment(
+      LedgerTransaction ledgerTransaction, JdbcSepTransaction sepTransaction) throws IOException {
+    // A redelivered payment is already in the list: it is not saved again, but the event is sent
+    // again, because the first delivery may have failed after the save.
+    if (!isRecorded(ledgerTransaction, sepTransaction)) {
+      try {
+        PaymentHelper.recordStellarTransaction(sacToAssetMapper, ledgerTransaction, sepTransaction);
+        switch (Sep.from(sepTransaction.getProtocol())) {
+          case SEP_6 -> sep6TransactionStore.save((JdbcSep6Transaction) sepTransaction);
+          case SEP_24 -> sep24TransactionStore.save((JdbcSep24Transaction) sepTransaction);
+          case SEP_31 -> sep31TransactionStore.save((JdbcSep31Transaction) sepTransaction);
+          default ->
+              throw new IllegalStateException(
+                  "Unsupported protocol: " + sepTransaction.getProtocol());
+        }
+      } catch (AnchorException | DataAccessException | TransactionException ex) {
+        throw holdUnderpayment(ledgerTransaction, sepTransaction, ex);
+      }
+    }
+
+    AnchorEvent event =
+        AnchorEvent.builder()
+            .id(UUID.randomUUID().toString())
+            .sep(sepTransaction.getProtocol())
+            .type(AnchorEvent.Type.TRANSACTION_STATUS_CHANGED)
+            .transaction(
+                PlatformTransactionHelper.toGetTransactionResponse(sepTransaction, assetService))
+            .build();
+    try {
+      eventSession.publish(event);
+    } catch (AnchorException | RuntimeException ex) {
+      throw holdUnderpayment(ledgerTransaction, sepTransaction, ex);
+    }
+  }
+
+  private IOException holdUnderpayment(
+      LedgerTransaction ledgerTransaction, JdbcSepTransaction sepTransaction, Exception cause) {
+    errorF(
+        "Recording the underpayment failed, so the payment will be retried. sepTxn.id={}, ledgerTxn.id={}",
+        sepTransaction.getId(),
+        ledgerTransaction.getHash());
+    errorEx(cause);
+    return new IOException(
+        "Recording the underpayment of transaction " + sepTransaction.getId() + " failed", cause);
+  }
+
+  private static boolean isRecorded(
+      LedgerTransaction ledgerTransaction, JdbcSepTransaction sepTransaction) {
+    return sepTransaction.getStellarTransactions() != null
+        && sepTransaction.getStellarTransactions().stream()
+            .anyMatch(tx -> Objects.equals(tx.getId(), ledgerTransaction.getHash()));
+  }
 }

platform/src/main/java/org/stellar/anchor/platform/utils/PaymentHelper.java

@@ -4,7 +4,10 @@
 import static org.stellar.anchor.util.Log.warnF;
 
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import org.stellar.anchor.api.platform.PlatformTransactionData.Sep;
@@ -47,8 +50,25 @@ static Asset resolveSacAsset(SacToAssetMapper sacToAssetMapper, String contractI
     }
   }
 
+  /**
+   * Adds the Stellar transaction to the transaction's {@code stellar_transactions} and marks the
+   * funds as received.
+   */
   public static void addStellarTransaction(
       SacToAssetMapper sacToAssetMapper, LedgerTransaction ledgerTxn, JdbcSepTransaction sepTxn) {
+    recordStellarTransaction(sacToAssetMapper, ledgerTxn, sepTxn);
+
+    sepTxn.setTransferReceivedAt(
+        Optional.ofNullable(ledgerTxn.getCreatedAt()).orElse(Instant.now()));
+    sepTxn.setStellarTransactionId(ledgerTxn.getHash());
+  }
+
+  /**
+   * Adds the Stellar transaction to the transaction's {@code stellar_transactions} only, leaving
+   * the received-funds fields alone, for a payment that is recorded but not credited.
+   */
+  public static void recordStellarTransaction(
+      SacToAssetMapper sacToAssetMapper, LedgerTransaction ledgerTxn, JdbcSepTransaction sepTxn) {
     String memo = null;
     String memoType = null;
 
@@ -115,14 +135,32 @@ public static void addStellarTransaction(
                     .collect(toList()))
             .build();
 
-    if (sepTxn.getStellarTransactions() == null) {
-      sepTxn.setStellarTransactions(List.of(stellarTransaction));
-    } else {
-      sepTxn.getStellarTransactions().add(stellarTransaction);
-    }
+    sepTxn.setStellarTransactions(addOrMerge(sepTxn.getStellarTransactions(), stellarTransaction));
+  }
 
-    sepTxn.setTransferReceivedAt(
-        Optional.ofNullable(ledgerTxn.getCreatedAt()).orElse(Instant.now()));
-    sepTxn.setStellarTransactionId(ledgerTxn.getHash());
+  /**
+   * Returns the transactions with the new one in place of the entry that has the same hash, or
+   * appended when there is none, so a hash never appears twice. The payments of both are kept,
+   * ordered by first appearance.
+   */
+  private static List<StellarTransaction> addOrMerge(
+      List<StellarTransaction> transactions, StellarTransaction added) {
+    List<StellarTransaction> merged =
+        transactions == null ? new ArrayList<>() : new ArrayList<>(transactions);
+    for (int i = 0; i < merged.size(); i++) {
+      StellarTransaction existing = merged.get(i);
+      if (Objects.equals(existing.getId(), added.getId())) {
+        Map<String, StellarPayment> payments = new LinkedHashMap<>();
+        if (existing.getPayments() != null) {
+          existing.getPayments().forEach(p -> payments.put(p.getId(), p));
+        }
+        added.getPayments().forEach(p -> payments.put(p.getId(), p));
+        added.setPayments(new ArrayList<>(payments.values()));
+        merged.set(i, added);
+        return merged;
+      }
+    }
+    merged.add(added);
+    return merged;
   }
 }