[ANCHOR-439] Remove transaction store from reference server

api-schema/src/main/java/org/stellar/anchor/api/platform/GetTransactionsRequest.java

@@ -0,0 +1,34 @@
+package org.stellar.anchor.api.platform;
+
+import com.google.gson.annotations.SerializedName;
+import java.util.List;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NonNull;
+import org.stellar.anchor.api.sep.SepTransactionStatus;
+
+/**
+ * The request body of the GET /transactions endpoint of the Platform API.
+ *
+ * @see <a
+ *     href="https://github.com/stellar/stellar-docs/blob/main/openapi/anchor-platform/Platform%20API.yml">Platform
+ *     API</a>
+ */
+@Data
+@Builder
+public class GetTransactionsRequest {
+  @NonNull private TransactionsSeps sep;
+
+  @SerializedName("order_by")
+  private TransactionsOrderBy orderBy;
+
+  private TransactionsOrder order;
+
+  private List<SepTransactionStatus> statuses;
+
+  @SerializedName("page_size")
+  private Integer pageSize;
+
+  @SerializedName("page_number")
+  private Integer pageNumber;
+}

api-schema/src/main/java/org/stellar/anchor/api/platform/TransactionsOrder.java

@@ -0,0 +1,10 @@
+package org.stellar.anchor.api.platform;
+
+import com.google.gson.annotations.SerializedName;
+
+public enum TransactionsOrder {
+  @SerializedName("asc")
+  ASC,
+  @SerializedName("desc")
+  DESC
+}

core/src/main/java/org/stellar/anchor/apiclient/TransactionsOrderBy.java → api-schema/src/main/java/org/stellar/anchor/api/platform/TransactionsOrderBy.java

@@ -1,4 +1,4 @@
-package org.stellar.anchor.apiclient;
+package org.stellar.anchor.api.platform;
 
 public enum TransactionsOrderBy {
   CREATED_AT("started_at"),

core/src/main/java/org/stellar/anchor/apiclient/TransactionsSeps.java → api-schema/src/main/java/org/stellar/anchor/api/platform/TransactionsSeps.java

@@ -1,4 +1,4 @@
-package org.stellar.anchor.apiclient;
+package org.stellar.anchor.api.platform;
 
 import com.google.gson.annotations.SerializedName;
 

core/src/main/java/org/stellar/anchor/util/TransactionsParams.java

@@ -5,8 +5,8 @@
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import org.springframework.data.domain.Sort;
+import org.stellar.anchor.api.platform.TransactionsOrderBy;
 import org.stellar.anchor.api.sep.SepTransactionStatus;
-import org.stellar.anchor.apiclient.TransactionsOrderBy;
 
 @Data
 @AllArgsConstructor

integration-tests/src/test/kotlin/org/stellar/anchor/platform/test/Sep31Tests.kt

@@ -12,6 +12,8 @@ import org.stellar.anchor.api.exception.SepException
 import org.stellar.anchor.api.platform.*
 import org.stellar.anchor.api.platform.PlatformTransactionData.Sep.SEP_31
 import org.stellar.anchor.api.platform.PlatformTransactionData.builder
+import org.stellar.anchor.api.platform.TransactionsOrderBy
+import org.stellar.anchor.api.platform.TransactionsSeps
 import org.stellar.anchor.api.sep.SepTransactionStatus
 import org.stellar.anchor.api.sep.sep12.Sep12PutCustomerRequest
 import org.stellar.anchor.api.sep.sep12.Sep12PutCustomerResponse
@@ -20,8 +22,6 @@ import org.stellar.anchor.api.sep.sep31.Sep31GetTransactionResponse
 import org.stellar.anchor.api.sep.sep31.Sep31PostTransactionRequest
 import org.stellar.anchor.api.sep.sep31.Sep31PostTransactionResponse
 import org.stellar.anchor.apiclient.PlatformApiClient
-import org.stellar.anchor.apiclient.TransactionsOrderBy
-import org.stellar.anchor.apiclient.TransactionsSeps
 import org.stellar.anchor.auth.AuthHelper
 import org.stellar.anchor.platform.*
 import org.stellar.anchor.util.GsonUtils

kotlin-reference-server/src/main/kotlin/org/stellar/reference/client/PlatformClient.kt

@@ -4,9 +4,13 @@ import io.ktor.client.*
 import io.ktor.client.call.*
 import io.ktor.client.request.*
 import io.ktor.http.*
+import io.ktor.util.*
 import org.stellar.anchor.api.platform.GetTransactionResponse
+import org.stellar.anchor.api.platform.GetTransactionsRequest
+import org.stellar.anchor.api.platform.GetTransactionsResponse
 import org.stellar.anchor.api.platform.PatchTransactionsRequest
 import org.stellar.anchor.api.platform.PatchTransactionsResponse
+import org.stellar.anchor.api.sep.SepTransactionStatus
 import org.stellar.anchor.util.GsonUtils
 
 class PlatformClient(private val httpClient: HttpClient, private val endpoint: String) {
@@ -32,4 +36,34 @@ class PlatformClient(private val httpClient: HttpClient, private val endpoint: S
     return GsonUtils.getInstance()
       .fromJson(response.body<String>(), PatchTransactionsResponse::class.java)
   }
+
+  suspend fun getTransactions(request: GetTransactionsRequest): GetTransactionsResponse {
+    val response =
+      httpClient.request("$endpoint/transactions") {
+        method = HttpMethod.Get
+        url {
+          parameters.append("sep", request.sep.name.toLowerCasePreservingASCIIRules())
+          if (request.orderBy != null) {
+            parameters.append("order_by", request.orderBy.name.toLowerCasePreservingASCIIRules())
+          }
+          if (request.order != null) {
+            parameters.append("order", request.order.name.toLowerCasePreservingASCIIRules())
+          }
+          if (request.statuses != null) {
+            parameters.append("statuses", SepTransactionStatus.mergeStatusesList(request.statuses))
+          }
+          if (request.pageSize != null) {
+            parameters.append("page_size", request.pageSize.toString())
+          }
+          if (request.pageNumber != null) {
+            parameters.append("page_number", request.pageNumber.toString())
+          }
+        }
+      }
+    if (response.status != HttpStatusCode.OK) {
+      throw Exception("Error getting transactions: ${response.status}")
+    }
+    return GsonUtils.getInstance()
+      .fromJson(response.body<String>(), GetTransactionsResponse::class.java)
+  }
 }

kotlin-reference-server/src/main/kotlin/org/stellar/reference/di/EventConsumerContainer.kt

@@ -5,7 +5,6 @@ import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.stellar.reference.event.EventConsumer
 import org.stellar.reference.event.processor.AnchorEventProcessor
-import org.stellar.reference.event.processor.InMemoryTransactionStore
 import org.stellar.reference.event.processor.NoOpEventProcessor
 import org.stellar.reference.event.processor.Sep6EventProcessor
 
@@ -24,15 +23,8 @@ object EventConsumerContainer {
     KafkaConsumer<String, String>(consumerConfig).also {
       it.subscribe(listOf(config.eventSettings.topic))
     }
-  private val activeTransactionStore = InMemoryTransactionStore()
   private val sep6EventProcessor =
-    Sep6EventProcessor(
-      config,
-      ServiceContainer.horizon,
-      ServiceContainer.platform,
-      ServiceContainer.customerService,
-      activeTransactionStore
-    )
+    Sep6EventProcessor(config, ServiceContainer.horizon, ServiceContainer.platform)
   private val noOpEventProcessor = NoOpEventProcessor()
   private val processor = AnchorEventProcessor(sep6EventProcessor, noOpEventProcessor)
   val eventConsumer = EventConsumer(kafkaConsumer, processor)

kotlin-reference-server/src/main/kotlin/org/stellar/reference/event/processor/Sep6EventProcessor.kt

@@ -3,17 +3,13 @@ package org.stellar.reference.event.processor
 import java.lang.RuntimeException
 import java.time.Instant
 import kotlinx.coroutines.runBlocking
-import org.stellar.anchor.api.callback.GetCustomerRequest
 import org.stellar.anchor.api.event.AnchorEvent
-import org.stellar.anchor.api.platform.PatchTransactionRequest
-import org.stellar.anchor.api.platform.PatchTransactionsRequest
-import org.stellar.anchor.api.platform.PlatformTransactionData
+import org.stellar.anchor.api.platform.*
 import org.stellar.anchor.api.sep.SepTransactionStatus
 import org.stellar.anchor.api.shared.Amount
 import org.stellar.anchor.api.shared.InstructionField
 import org.stellar.anchor.api.shared.StellarPayment
 import org.stellar.anchor.api.shared.StellarTransaction
-import org.stellar.reference.callbacks.customer.CustomerService
 import org.stellar.reference.client.PlatformClient
 import org.stellar.reference.data.Config
 import org.stellar.reference.log
@@ -28,8 +24,6 @@ class Sep6EventProcessor(
   private val config: Config,
   private val server: Server,
   private val platformClient: PlatformClient,
-  private val customerService: CustomerService,
-  private val activeTransactionStore: ActiveTransactionStore
 ) : SepAnchorEventProcessor {
   override fun onQuoteCreated(event: AnchorEvent) {
     TODO("Not yet implemented")
@@ -52,10 +46,6 @@ class Sep6EventProcessor(
       )
       return
     }
-    val customer =
-      customerService.getCustomer(
-        GetCustomerRequest.builder().account(event.transaction.destinationAccount).build()
-      )
     runBlocking {
       patchTransaction(
         PlatformTransactionData.builder()
@@ -64,10 +54,6 @@ class Sep6EventProcessor(
           .build()
       )
     }
-    activeTransactionStore.addTransaction(customer.id, event.transaction.id)
-    log.info(
-      "Added transaction ${event.transaction.id} to active transaction store for customer ${customer.id}"
-    )
   }
 
   override fun onTransactionError(event: AnchorEvent) {
@@ -89,14 +75,7 @@ class Sep6EventProcessor(
         }
       }
       SepTransactionStatus.COMPLETED -> {
-        val customer =
-          customerService.getCustomer(
-            GetCustomerRequest.builder().account(transaction.destinationAccount).build()
-          )
-        activeTransactionStore.removeTransaction(customer.id, transaction.id)
-        log.info(
-          "Removed transaction ${transaction.id} from active transaction store for customer ${customer.id}"
-        )
+        log.info("Transaction ${transaction.id} completed")
       }
       else -> {
         log.warn("Received transaction status changed event with unsupported status: $status")
@@ -105,51 +84,58 @@ class Sep6EventProcessor(
   }
 
   override fun onCustomerUpdated(event: AnchorEvent) {
-    val observedAccount = event.customer.id
-    val transactionIds = activeTransactionStore.getTransactions(observedAccount)
-    log.info(
-      "Found ${transactionIds.size} transactions for customer $observedAccount in active transaction store"
-    )
+    val transactionIds = runBlocking {
+      platformClient
+        .getTransactions(
+          GetTransactionsRequest.builder()
+            .sep(TransactionsSeps.SEP_6)
+            .orderBy(TransactionsOrderBy.CREATED_AT)
+            .order(TransactionsOrder.ASC)
+            .statuses(listOf(SepTransactionStatus.PENDING_CUSTOMER_INFO_UPDATE))
+            .build()
+        )
+        .records
+        .map { it.id }
+    }
+    log.info("Found ${transactionIds.size} transactions pending customer info update")
     transactionIds.forEach { id ->
       val transaction = runBlocking { platformClient.getTransaction(id) }
-      if (transaction.status == SepTransactionStatus.PENDING_CUSTOMER_INFO_UPDATE) {
-        val keypair = KeyPair.fromSecretSeed(config.appSettings.secret)
-        val assetCode = transaction.amountExpected.asset.toAssetId()
+      val keypair = KeyPair.fromSecretSeed(config.appSettings.secret)
+      val assetCode = transaction.amountExpected.asset.toAssetId()
 
-        val asset = Asset.create(assetCode)
-        val amount = transaction.amountExpected.amount
-        val destination = transaction.destinationAccount
+      val asset = Asset.create(assetCode)
+      val amount = transaction.amountExpected.amount
+      val destination = transaction.destinationAccount
 
-        val stellarTxn = submitStellarTransaction(keypair.accountId, destination, asset, amount)
-        runBlocking {
-          patchTransaction(
-            PlatformTransactionData.builder()
-              .id(transaction.id)
-              .status(SepTransactionStatus.COMPLETED)
-              .updatedAt(Instant.now())
-              .completedAt(Instant.now())
-              .requiredInfoMessage(null)
-              .requiredInfoUpdates(null)
-              .requiredCustomerInfoUpdates(null)
-              .requiredCustomerInfoUpdates(null)
-              .instructions(
-                mapOf(
-                  "organization.bank_number" to
-                    InstructionField.builder()
-                      .value("121122676")
-                      .description("US Bank routing number")
-                      .build(),
-                  "organization.bank_account_number" to
-                    InstructionField.builder()
-                      .value("13719713158835300")
-                      .description("US Bank account number")
-                      .build(),
-                )
+      val stellarTxn = submitStellarTransaction(keypair.accountId, destination, asset, amount)
+      runBlocking {
+        patchTransaction(
+          PlatformTransactionData.builder()
+            .id(transaction.id)
+            .status(SepTransactionStatus.COMPLETED)
+            .updatedAt(Instant.now())
+            .completedAt(Instant.now())
+            .requiredInfoMessage(null)
+            .requiredInfoUpdates(null)
+            .requiredCustomerInfoUpdates(null)
+            .requiredCustomerInfoUpdates(null)
+            .instructions(
+              mapOf(
+                "organization.bank_number" to
+                  InstructionField.builder()
+                    .value("121122676")
+                    .description("US Bank routing number")
+                    .build(),
+                "organization.bank_account_number" to
+                  InstructionField.builder()
+                    .value("13719713158835300")
+                    .description("US Bank account number")
+                    .build(),
               )
-              .stellarTransactions(listOf(stellarTxn))
-              .build()
-          )
-        }
+            )
+            .stellarTransactions(listOf(stellarTxn))
+            .build()
+        )
       }
     }
   }

platform/src/main/java/org/stellar/anchor/platform/controller/platform/PlatformController.java

@@ -7,9 +7,9 @@
 import org.springframework.web.bind.annotation.*;
 import org.stellar.anchor.api.exception.AnchorException;
 import org.stellar.anchor.api.platform.*;
+import org.stellar.anchor.api.platform.TransactionsOrderBy;
+import org.stellar.anchor.api.platform.TransactionsSeps;
 import org.stellar.anchor.api.sep.SepTransactionStatus;
-import org.stellar.anchor.apiclient.TransactionsOrderBy;
-import org.stellar.anchor.apiclient.TransactionsSeps;
 import org.stellar.anchor.platform.service.TransactionService;
 import org.stellar.anchor.util.TransactionsParams;
 

platform/src/main/java/org/stellar/anchor/platform/service/TransactionService.java

@@ -25,10 +25,10 @@
 import org.stellar.anchor.api.exception.InternalServerErrorException;
 import org.stellar.anchor.api.exception.NotFoundException;
 import org.stellar.anchor.api.platform.*;
+import org.stellar.anchor.api.platform.TransactionsSeps;
 import org.stellar.anchor.api.sep.AssetInfo;
 import org.stellar.anchor.api.sep.SepTransactionStatus;
 import org.stellar.anchor.api.shared.Amount;
-import org.stellar.anchor.apiclient.TransactionsSeps;
 import org.stellar.anchor.asset.AssetService;
 import org.stellar.anchor.event.EventService;
 import org.stellar.anchor.event.EventService.Session;

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

@@ -9,9 +9,9 @@
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.domain.Sort;
 import org.stellar.anchor.api.exception.BadRequestException;
+import org.stellar.anchor.api.platform.TransactionsOrderBy;
+import org.stellar.anchor.api.platform.TransactionsSeps;
 import org.stellar.anchor.api.sep.SepTransactionStatus;
-import org.stellar.anchor.apiclient.TransactionsOrderBy;
-import org.stellar.anchor.apiclient.TransactionsSeps;
 
 // Abstract class because https://github.com/spring-projects/spring-boot/pull/22885
 public abstract class StringEnumConverter<T extends Enum<T>> implements Converter<String, T> {

platform/src/test/kotlin/org/stellar/anchor/platform/util/EnumConverterTest.kt

@@ -5,7 +5,7 @@ import kotlin.test.assertEquals
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import org.stellar.anchor.api.exception.BadRequestException
-import org.stellar.anchor.apiclient.TransactionsOrderBy
+import org.stellar.anchor.api.platform.TransactionsOrderBy
 import org.stellar.anchor.platform.utils.StringEnumConverter
 import org.stellar.anchor.platform.utils.StringEnumConverter.TransactionsOrderByConverter