Add withdrawal support for notify_refund_pending RPC method (#1353)

### Description

Add support for withdrawals for existing `notify_refund_pending` RPC
method

### Context

I noticed refunds are not available for `pending_user_transfer_complete`
and `pending_external statuses`.

Generally speaking, in the regular flow transaction goes from
`pending_user_transfer_start` -> `pending_anchor` -> `completed`
However, when the anchor takes times to process a transaction and it's
still in `pending_anchor` , user wants to refund it and can do so.
In the cases where status is `pending_user_transfer_complete` (funds are
available for pickup) or `pending_external` (funds are processed by
external entity) we want to fist move it to `pending_anchor` to indicate
that funds are not available for pickup anymore, (or that bank transfer
was cancelled), i.e. indicate that refund has been initiated.
In this case, if we have issues with Stellar transfer or Circle, etc.
transaction won't be in status that shows user wrong information.
(Especially in `pending_user_transfer_complete` case, where user funds
is technically no longer available for pickup anymore, but status
doesn't show that)
After that, we can use existing `notify_refund_send` RPC.

### Testing

- `./gradlew test`
- Added new test cases 

### Documentation

Flow charts should be updated
(https://stellarorg.atlassian.net/browse/ANCHOR-688)

### Known limitations

N/A

api-schema/src/main/java/org/stellar/anchor/api/rpc/method/NotifyRefundPendingRequest.java

@@ -14,7 +14,7 @@
 @EqualsAndHashCode(callSuper = false)
 public class NotifyRefundPendingRequest extends RpcMethodParamsRequest {
 
-  @NotNull private Refund refund;
+  private Refund refund;
 
   @Data
   @Builder

api-schema/src/main/java/org/stellar/anchor/api/shared/Refunds.java

@@ -9,7 +9,7 @@
 @Data
 @Builder
 @AllArgsConstructor
-public class Refunds {
+public class Refunds implements SepRefunds {
 
   @JsonProperty("amount_refunded")
   @SerializedName("amount_refunded")
@@ -22,4 +22,9 @@ public class Refunds {
   RefundPayment[] payments;
 
   public Refunds() {}
+
+  @Override
+  public boolean hasRefundPayments() {
+    return payments != null;
+  }
 }

api-schema/src/main/java/org/stellar/anchor/api/shared/SepRefunds.java

@@ -0,0 +1,5 @@
+package org.stellar.anchor.api.shared;
+
+public interface SepRefunds {
+  boolean hasRefundPayments();
+}

core/src/main/java/org/stellar/anchor/sep24/Sep24Refunds.java

@@ -12,9 +12,10 @@
 import java.util.stream.Collectors;
 import org.stellar.anchor.api.sep.AssetInfo;
 import org.stellar.anchor.api.shared.Refunds;
+import org.stellar.anchor.api.shared.SepRefunds;
 
 @SuppressWarnings("unused")
-public interface Sep24Refunds {
+public interface Sep24Refunds extends SepRefunds {
   String getAmountRefunded();
 
   void setAmountRefunded(String amountRefunded);

core/src/test/java/org/stellar/anchor/sep24/PojoSep24Refunds.java

@@ -8,4 +8,9 @@ public class PojoSep24Refunds implements Sep24Refunds {
   String amountRefunded;
   String amountFee;
   List<Sep24RefundPayment> refundPayments;
+
+  @Override
+  public boolean hasRefundPayments() {
+    return refundPayments != null;
+  }
 }

essential-tests/src/testFixtures/kotlin/org/stellar/anchor/platform/integrationtest/PlatformApiTests.kt

@@ -27,7 +27,8 @@ import org.stellar.anchor.platform.AbstractIntegrationTests
 import org.stellar.anchor.platform.TestConfig
 import org.stellar.anchor.util.GsonUtils
 
-@Disabled
+@Disabled // TODO re-enable tests: https://stellarorg.atlassian.net/browse/ANCHOR-693
+// TODO add refund flow test for withdrawal: https://stellarorg.atlassian.net/browse/ANCHOR-694
 class PlatformApiTests : AbstractIntegrationTests(TestConfig()) {
   private val gson = GsonUtils.getInstance()
 

platform/src/main/java/org/stellar/anchor/platform/data/JdbcSep24Refunds.java

@@ -41,4 +41,9 @@ public void setRefundPayments(List<Sep24RefundPayment> refundPayments) {
             String.format("Error casting %s to JdbcSep24RefundPayment", rp.getClass()));
     }
   }
+
+  @Override
+  public boolean hasRefundPayments() {
+    return payments != null;
+  }
 }

platform/src/main/java/org/stellar/anchor/platform/rpc/NotifyRefundPendingHandler.java

@@ -1,13 +1,11 @@
 package org.stellar.anchor.platform.rpc;
 
 import static java.util.Collections.emptySet;
-import static org.stellar.anchor.api.platform.PlatformTransactionData.Kind.DEPOSIT;
-import static org.stellar.anchor.api.platform.PlatformTransactionData.Kind.DEPOSIT_EXCHANGE;
+import static org.stellar.anchor.api.platform.PlatformTransactionData.Kind.*;
 import static org.stellar.anchor.api.platform.PlatformTransactionData.Sep.SEP_24;
 import static org.stellar.anchor.api.platform.PlatformTransactionData.Sep.SEP_6;
 import static org.stellar.anchor.api.rpc.method.RpcMethod.NOTIFY_REFUND_PENDING;
-import static org.stellar.anchor.api.sep.SepTransactionStatus.PENDING_ANCHOR;
-import static org.stellar.anchor.api.sep.SepTransactionStatus.PENDING_EXTERNAL;
+import static org.stellar.anchor.api.sep.SepTransactionStatus.*;
 import static org.stellar.anchor.util.AssetHelper.getAssetCode;
 import static org.stellar.anchor.util.MathHelper.*;
 
@@ -31,6 +29,7 @@
 import org.stellar.anchor.api.shared.Amount;
 import org.stellar.anchor.api.shared.RefundPayment;
 import org.stellar.anchor.api.shared.Refunds;
+import org.stellar.anchor.api.shared.SepRefunds;
 import org.stellar.anchor.asset.AssetService;
 import org.stellar.anchor.event.EventService;
 import org.stellar.anchor.metrics.MetricsService;
@@ -69,29 +68,72 @@ protected void validate(JdbcSepTransaction txn, NotifyRefundPendingRequest reque
       throws InvalidParamsException, InvalidRequestException, BadRequestException {
     super.validate(txn, request);
 
-    AssetValidationUtils.validateAsset(
-        "refund.amount",
-        AmountAssetRequest.builder()
-            .amount(request.getRefund().getAmount().getAmount())
-            .asset(txn.getAmountInAsset())
-            .build(),
-        assetService);
-    AssetValidationUtils.validateAsset(
-        "refund.amountFee",
-        AmountAssetRequest.builder()
-            .amount(request.getRefund().getAmountFee().getAmount())
-            .asset(txn.getAmountInAsset())
-            .build(),
-        true,
-        assetService);
-
-    if (!txn.getAmountInAsset().equals(request.getRefund().getAmount().getAsset())) {
-      throw new InvalidParamsException(
-          "refund.amount.asset does not match transaction amount_in_asset");
+    Sep sep = Sep.from(txn.getProtocol());
+    SepRefunds refunds;
+    PlatformTransactionData.Kind kind;
+    if (sep == SEP_6) {
+      kind = Kind.from(((JdbcSep6Transaction) txn).getKind());
+      refunds = ((JdbcSep6Transaction) txn).getRefunds();
+    } else if (sep == SEP_24) {
+      kind = Kind.from(((JdbcSep24Transaction) txn).getKind());
+      refunds = ((JdbcSep24Transaction) txn).getRefunds();
+    } else {
+      throw new InvalidRequestException(
+          String.format(
+              "RPC method[%s] is not supported for protocol[%s]",
+              getRpcMethod(), txn.getProtocol()));
     }
-    if (!txn.getAmountFeeAsset().equals(request.getRefund().getAmountFee().getAsset())) {
-      throw new InvalidParamsException(
-          "refund.amount_fee.asset does not match match transaction amount_fee_asset");
+
+    if (ImmutableSet.of(DEPOSIT, DEPOSIT_EXCHANGE).contains(kind)) {
+      if (request.getRefund() == null) {
+        throw new InvalidParamsException("refund must not be null");
+      }
+
+      AssetValidationUtils.validateAsset(
+          "refund.amount",
+          AmountAssetRequest.builder()
+              .amount(request.getRefund().getAmount().getAmount())
+              .asset(txn.getAmountInAsset())
+              .build(),
+          assetService);
+      AssetValidationUtils.validateAsset(
+          "refund.amountFee",
+          AmountAssetRequest.builder()
+              .amount(request.getRefund().getAmountFee().getAmount())
+              .asset(txn.getAmountInAsset())
+              .build(),
+          true,
+          assetService);
+
+      if (!txn.getAmountInAsset().equals(request.getRefund().getAmount().getAsset())) {
+        throw new InvalidParamsException(
+            "refund.amount.asset does not match transaction amount_in_asset");
+      }
+      if (!txn.getAmountFeeAsset().equals(request.getRefund().getAmountFee().getAsset())) {
+        throw new InvalidParamsException(
+            "refund.amount_fee.asset does not match match transaction amount_fee_asset");
+      }
+
+      AssetInfo assetInfo = assetService.getAsset(getAssetCode(txn.getAmountInAsset()));
+
+      var amount = request.getRefund().getAmount().getAmount();
+      var amountFee = request.getRefund().getAmountFee().getAmount();
+
+      BigDecimal totalRefunded;
+      if (refunds == null || !refunds.hasRefundPayments()) {
+        totalRefunded = sum(assetInfo, amount, amountFee);
+      } else {
+        var amountRefunded =
+            refunds instanceof Refunds
+                ? ((Refunds) refunds).getAmountRefunded().getAmount()
+                : ((Sep24Refunds) refunds).getAmountRefunded();
+        totalRefunded = sum(assetInfo, amountRefunded, amount, amountFee);
+      }
+
+      BigDecimal amountIn = decimal(txn.getAmountIn(), assetInfo);
+      if (totalRefunded.compareTo(amountIn) > 0) {
+        throw new InvalidParamsException("Refund amount exceeds amount_in");
+      }
     }
   }
 
@@ -104,56 +146,16 @@ public RpcMethod getRpcMethod() {
   protected SepTransactionStatus getNextStatus(
       JdbcSepTransaction txn, NotifyRefundPendingRequest request)
       throws InvalidParamsException, InvalidRequestException {
-    String amount;
-    String amountFee;
-    switch (PlatformTransactionData.Sep.from(txn.getProtocol())) {
-      case SEP_6:
-        JdbcSep6Transaction txn6 = (JdbcSep6Transaction) txn;
-        AssetInfo assetInfo6 = assetService.getAsset(getAssetCode(txn.getAmountInAsset()));
+    var kind =
+        txn instanceof JdbcSep6Transaction
+            ? Kind.from(((JdbcSep6Transaction) txn).getKind())
+            : Kind.from(((JdbcSep24Transaction) txn).getKind());
 
-        Refunds refunds = txn6.getRefunds();
-        amount = request.getRefund().getAmount().getAmount();
-        amountFee = request.getRefund().getAmountFee().getAmount();
-
-        BigDecimal totalRefunded6;
-        if (refunds == null || refunds.getPayments() == null) {
-          totalRefunded6 = sum(assetInfo6, amount, amountFee);
-        } else {
-          totalRefunded6 =
-              sum(assetInfo6, refunds.getAmountRefunded().getAmount(), amount, amountFee);
-        }
-
-        BigDecimal amountIn6 = decimal(txn.getAmountIn(), assetInfo6);
-        if (totalRefunded6.compareTo(amountIn6) > 0) {
-          throw new InvalidParamsException("Refund amount exceeds amount_in");
-        }
-
-        return PENDING_EXTERNAL;
-      case SEP_24:
-        JdbcSep24Transaction txn24 = (JdbcSep24Transaction) txn;
-        AssetInfo assetInfo = assetService.getAsset(getAssetCode(txn.getAmountInAsset()));
-
-        Sep24Refunds sep24Refunds = txn24.getRefunds();
-        amount = request.getRefund().getAmount().getAmount();
-        amountFee = request.getRefund().getAmountFee().getAmount();
-
-        BigDecimal totalRefunded;
-        if (sep24Refunds == null || sep24Refunds.getRefundPayments() == null) {
-          totalRefunded = sum(assetInfo, amount, amountFee);
-        } else {
-          totalRefunded = sum(assetInfo, sep24Refunds.getAmountRefunded(), amount, amountFee);
-        }
-
-        BigDecimal amountIn = decimal(txn.getAmountIn(), assetInfo);
-        if (totalRefunded.compareTo(amountIn) > 0) {
-          throw new InvalidParamsException("Refund amount exceeds amount_in");
-        }
-
-        return PENDING_EXTERNAL;
+    if (ImmutableSet.of(WITHDRAWAL, WITHDRAWAL_EXCHANGE).contains(kind)) {
+      return PENDING_ANCHOR;
     }
-    throw new InvalidRequestException(
-        String.format(
-            "RPC method[%s] is not supported for protocol[%s]", getRpcMethod(), txn.getProtocol()));
+
+    return PENDING_EXTERNAL;
   }
 
   @Override
@@ -163,14 +165,15 @@ protected Set<SepTransactionStatus> getSupportedStatuses(JdbcSepTransaction txn)
       if (ImmutableSet.of(DEPOSIT, DEPOSIT_EXCHANGE).contains(Kind.from(txn6.getKind()))) {
         return Set.of(PENDING_ANCHOR);
       }
-      return emptySet();
+      return Set.of(PENDING_USR_TRANSFER_COMPLETE, PENDING_EXTERNAL);
     }
 
     if (SEP_24 == Sep.from(txn.getProtocol())) {
       JdbcSep24Transaction txn24 = (JdbcSep24Transaction) txn;
       if (DEPOSIT == Kind.from(txn24.getKind())) {
         return Set.of(PENDING_ANCHOR);
       }
+      return Set.of(PENDING_USR_TRANSFER_COMPLETE, PENDING_EXTERNAL);
     }
     return emptySet();
   }
@@ -185,6 +188,13 @@ protected void updateTransactionWithRpcRequest(
       case SEP_6:
         JdbcSep6Transaction txn6 = (JdbcSep6Transaction) txn;
 
+        // For withdrawals only update status to pending_anchor to mark that refund has been
+        // initiated . Actual Refund
+        // object will be passed in the next update (notify_refund_sent), as it's not yet known.
+        if (ImmutableSet.of(WITHDRAWAL, WITHDRAWAL_EXCHANGE).contains(Kind.from(txn6.getKind()))) {
+          return;
+        }
+
         RefundPayment requestPayment =
             RefundPayment.builder()
                 .id(requestRefund.getId())
@@ -243,6 +253,10 @@ protected void updateTransactionWithRpcRequest(
       case SEP_24:
         JdbcSep24Transaction txn24 = (JdbcSep24Transaction) txn;
 
+        if (WITHDRAWAL == Kind.from(txn24.getKind())) {
+          return;
+        }
+
         Sep24RefundPayment refundPayment =
             JdbcSep24RefundPayment.builder()
                 .id(requestRefund.getId())

platform/src/main/java/org/stellar/anchor/platform/rpc/RpcMethodHandler.java

@@ -40,6 +40,7 @@
 import org.stellar.anchor.sep31.Sep31TransactionStore;
 import org.stellar.anchor.sep6.Sep6TransactionStore;
 import org.stellar.anchor.util.GsonUtils;
+import org.stellar.anchor.util.Log;
 
 public abstract class RpcMethodHandler<T extends RpcMethodParamsRequest> {
 
@@ -75,7 +76,9 @@ public RpcMethodHandler(
 
   public GetTransactionResponse handle(Object requestParams) throws AnchorException {
     T request = gson.fromJson(gson.toJson(requestParams), requestType);
+    Log.infoF("Processing RPC request {}", request);
     JdbcSepTransaction txn = getTransaction(request.getTransactionId());
+    Log.debugF("SEP transaction before request is executed {}", txn);
 
     if (txn == null) {
       throw new InvalidRequestException(
@@ -106,6 +109,8 @@ public GetTransactionResponse handle(Object requestParams) throws AnchorExceptio
 
     updateTransaction(txn, request);
 
+    Log.debugF("Transaction after update is executed {}", txn);
+
     GetTransactionResponse txResponse = toGetTransactionResponse(txn, assetService);
 
     eventSession.publish(