Close #28, Implement wallet sync

BlockchainGateway/bc-gateway-app/src/main/kotlin/co/nilin/opex/bcgateway/app/config/AppConfig.kt

@@ -28,17 +28,17 @@ class AppConfig {
 
     @Bean
     fun chainSyncService(
-        syncSchedulerHandler: SyncSchedulerHandler,
+        chainSyncSchedulerHandler: ChainSyncSchedulerHandler,
         chainEndpointProxyFinder: ChainEndpointProxyFinder,
-        syncRecordHandler: SyncRecordHandler,
+        chainSyncRecordHandler: ChainSyncRecordHandler,
         walletSyncRecordHandler: WalletSyncRecordHandler,
         currencyLoader: CurrencyLoader,
         operator: TransactionalOperator
     ): ChainSyncService {
        return ChainSyncServiceImpl(
-           syncSchedulerHandler,
+           chainSyncSchedulerHandler,
            chainEndpointProxyFinder,
-           syncRecordHandler,
+           chainSyncRecordHandler,
            walletSyncRecordHandler,
            currencyLoader,
            operator,
@@ -61,4 +61,4 @@ class AppConfig {
     fun infoService(): InfoService {
       return InfoServiceImpl()
     }
-}
+}

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/Chain.kt

@@ -1,6 +1,5 @@
 package co.nilin.opex.bcgateway.core.model
 
-import java.math.BigDecimal
 import java.time.LocalDateTime
 
 data class Endpoint(val url: String)
@@ -15,4 +14,3 @@ data class ChainSyncRecord(
     val error: String?,
     val records: List<Deposit>
 )
-

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/Deposit.kt

@@ -3,10 +3,11 @@ package co.nilin.opex.bcgateway.core.model
 import java.math.BigDecimal
 
 data class Deposit(
+    val id: Long?,
     val depositor: String,
     val depositorMemo: String?,
     val amount: BigDecimal,
-    val chain: String?,
+    val chain: String,
     val token: Boolean,
     val tokenAddress: String?
 )

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/WalletSync.kt

@@ -5,4 +5,4 @@ import java.time.LocalDateTime
 data class WalletSyncSchedule(val retryTime: LocalDateTime, val delay: Long, val batchSize: Long?)
 data class WalletSyncRecord(
     val time: LocalDateTime, val success: Boolean, val error: String?, val deposit: Deposit
-)
+)

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/service/ChainSyncServiceImpl.kt

@@ -10,9 +10,9 @@ import java.time.temporal.ChronoUnit
 import kotlin.coroutines.coroutineContext
 
 open class ChainSyncServiceImpl(
-    private val syncSchedulerHandler: SyncSchedulerHandler,
+    private val chainSyncSchedulerHandler: ChainSyncSchedulerHandler,
     private val chainEndpointProxyFinder: ChainEndpointProxyFinder,
-    private val syncRecordHandler: SyncRecordHandler,
+    private val chainSyncRecordHandler: ChainSyncRecordHandler,
     private val walletSyncRecordHandler: WalletSyncRecordHandler,
     private val currencyLoader: CurrencyLoader,
     private val operator: TransactionalOperator,
@@ -21,11 +21,11 @@ open class ChainSyncServiceImpl(
 
     override suspend fun startSyncWithChain() {
         withContext(coroutineContext) {
-            val schedules = syncSchedulerHandler.fetchActiveSchedules(currentTime())
+            val schedules = chainSyncSchedulerHandler.fetchActiveSchedules(currentTime())
             schedules.map { syncSchedule ->
                 async(dispatcher) {
                     val syncHandler = chainEndpointProxyFinder.findChainEndpointProxy(syncSchedule.chainName)
-                    val lastSync = syncRecordHandler.loadLastSuccessRecord(syncSchedule.chainName)
+                    val lastSync = chainSyncRecordHandler.loadLastSuccessRecord(syncSchedule.chainName)
                     val tokens = currencyLoader.findImplementationsWithTokenOnChain(syncSchedule.chainName)
                         .map { impl -> impl.tokenAddress!! }
                         .toList()
@@ -37,9 +37,9 @@ open class ChainSyncServiceImpl(
                         )
                     operator.executeAndAwait {
                         walletSyncRecordHandler.saveReadyToSyncTransfers(syncResult.chainName, syncResult.records)
-                        syncRecordHandler.saveSyncRecord(syncResult)
+                        chainSyncRecordHandler.saveSyncRecord(syncResult)
                         if (syncResult.success) {
-                            syncSchedulerHandler.prepareScheduleForNextTry(
+                            chainSyncSchedulerHandler.prepareScheduleForNextTry(
                                 syncSchedule,
                                 currentTime().plus(syncSchedule.delay, ChronoUnit.SECONDS)
                             )
@@ -51,4 +51,4 @@ open class ChainSyncServiceImpl(
     }
 
     protected open fun currentTime() = LocalDateTime.now()
-}
+}

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/service/WalletSyncServiceImpl.kt

@@ -1,6 +1,7 @@
 package co.nilin.opex.bcgateway.core.service
 
 import co.nilin.opex.bcgateway.core.api.WalletSyncService
+import co.nilin.opex.bcgateway.core.model.WalletSyncRecord
 import co.nilin.opex.bcgateway.core.spi.*
 import kotlinx.coroutines.ExecutorCoroutineDispatcher
 import kotlinx.coroutines.async
@@ -27,8 +28,16 @@ class WalletSyncServiceImpl(
                     async(dispatcher) {
                         val uuid = assignedAddressHandler.findUuid(deposit.depositor, deposit.depositorMemo)
                         if (uuid != null) {
-                            val symbol = currencyLoader.findSymbol(deposit.chain!!, deposit.tokenAddress)
+                            val symbol = currencyLoader.findSymbol(deposit.chain, deposit.tokenAddress)
                             if (symbol != null) walletProxy.transfer(uuid, symbol, deposit.amount)
+                            walletSyncRecordHandler.saveWalletSyncRecord(
+                                WalletSyncRecord(
+                                    LocalDateTime.now(),
+                                    true,
+                                    null,
+                                    deposit
+                                )
+                            )
                         }
                     }
                 }

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/spi/SyncRecordHandler.kt → BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/spi/ChainSyncRecordHandler.kt

@@ -2,7 +2,7 @@ package co.nilin.opex.bcgateway.core.spi
 
 import co.nilin.opex.bcgateway.core.model.ChainSyncRecord
 
-interface SyncRecordHandler {
+interface ChainSyncRecordHandler {
     suspend fun loadLastSuccessRecord(chainName: String): ChainSyncRecord?
     suspend fun saveSyncRecord(syncRecord: ChainSyncRecord)
-}
+}

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/spi/SyncSchedulerHandler.kt → BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/spi/ChainSyncSchedulerHandler.kt

@@ -3,7 +3,7 @@ package co.nilin.opex.bcgateway.core.spi
 import co.nilin.opex.bcgateway.core.model.ChainSyncSchedule
 import java.time.LocalDateTime
 
-interface SyncSchedulerHandler {
+interface ChainSyncSchedulerHandler {
     suspend fun fetchActiveSchedules(time: LocalDateTime): List<ChainSyncSchedule>
     suspend fun prepareScheduleForNextTry(syncSchedule: ChainSyncSchedule, time: LocalDateTime)
-}
+}

BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/spi/WalletSyncRecordHandler.kt

@@ -1,8 +1,10 @@
 package co.nilin.opex.bcgateway.core.spi
 
 import co.nilin.opex.bcgateway.core.model.Deposit
+import co.nilin.opex.bcgateway.core.model.WalletSyncRecord
 
 interface WalletSyncRecordHandler {
     suspend fun saveReadyToSyncTransfers(chainName: String, deposits: List<Deposit>)
+    suspend fun saveWalletSyncRecord(syncRecord: WalletSyncRecord)
     suspend fun findReadyToSyncTransfers(count: Long?): List<Deposit>
-}
+}

BlockchainGateway/bc-gateway-core/src/test/kotlin/co/nilin/opex/bcgateway/core/service/ChainSyncServiceImplTest.kt

@@ -6,8 +6,8 @@ import co.nilin.opex.bcgateway.core.model.Endpoint
 import co.nilin.opex.bcgateway.core.spi.ChainEndpointProxy
 import co.nilin.opex.bcgateway.core.spi.ChainEndpointProxyFinder
 import co.nilin.opex.bcgateway.core.spi.CurrencyLoader
-import co.nilin.opex.bcgateway.core.spi.SyncRecordHandler
-import co.nilin.opex.bcgateway.core.spi.SyncSchedulerHandler
+import co.nilin.opex.bcgateway.core.spi.ChainSyncRecordHandler
+import co.nilin.opex.bcgateway.core.spi.ChainSyncSchedulerHandler
 import co.nilin.opex.bcgateway.core.spi.WalletSyncRecordHandler
 import co.nilin.opex.bcgateway.test.OPERATOR
 import java.time.LocalDateTime
@@ -23,7 +23,6 @@ import org.mockito.kotlin.any
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.times
 import org.mockito.kotlin.verify
-import org.mockito.kotlin.verifyNoMoreInteractions
 import org.mockito.kotlin.verifyZeroInteractions
 
 internal class ChainSyncServiceImplTest {
@@ -34,13 +33,13 @@ internal class ChainSyncServiceImplTest {
     val syncService: ChainSyncServiceImpl
 
     @Mock
-    lateinit var syncSchedulerHandler: SyncSchedulerHandler
+    lateinit var chainSyncSchedulerHandler: ChainSyncSchedulerHandler
 
     @Mock
     lateinit var chainEndpointProxyFinder: ChainEndpointProxyFinder
 
     @Mock
-    lateinit var syncRecordHandler: SyncRecordHandler
+    lateinit var chainSyncRecordHandler: ChainSyncRecordHandler
 
     @Mock
     lateinit var walletSyncRecordHandler: WalletSyncRecordHandler
@@ -59,9 +58,9 @@ internal class ChainSyncServiceImplTest {
         }
 
         syncService = object : ChainSyncServiceImpl(
-            syncSchedulerHandler,
+            chainSyncSchedulerHandler,
             chainEndpointProxyFinder,
-            syncRecordHandler,
+            chainSyncRecordHandler,
             walletSyncRecordHandler,
             currencyLoader,
             OPERATOR,
@@ -75,15 +74,15 @@ internal class ChainSyncServiceImplTest {
     fun givenNoActiveSchedules_whenStartSync_thenNoOp() {
         runBlocking {
             //given
-            Mockito.`when`(syncSchedulerHandler.fetchActiveSchedules(any())).thenReturn(emptyList())
+            Mockito.`when`(chainSyncSchedulerHandler.fetchActiveSchedules(any())).thenReturn(emptyList())
 
             //when
             syncService.startSyncWithChain()
 
             //then
             verifyZeroInteractions(
                 chainEndpointProxyFinder,
-                syncRecordHandler,
+                chainSyncRecordHandler,
                 walletSyncRecordHandler,
                 currencyLoader
             )
@@ -96,7 +95,7 @@ internal class ChainSyncServiceImplTest {
             //given
             val delay = 100L
             val syncSchedule = ChainSyncSchedule(ethChain, time, delay)
-            Mockito.`when`(syncSchedulerHandler.fetchActiveSchedules(any()))
+            Mockito.`when`(chainSyncSchedulerHandler.fetchActiveSchedules(any()))
                 .thenReturn(listOf(syncSchedule))
             Mockito.`when`(endpointProxy.syncTransfers(any())).thenReturn(
                 ChainSyncRecord(
@@ -108,9 +107,9 @@ internal class ChainSyncServiceImplTest {
             syncService.startSyncWithChain()
 
             //then
-            verify(syncRecordHandler).saveSyncRecord(any())
+            verify(chainSyncRecordHandler).saveSyncRecord(any())
             verify(walletSyncRecordHandler).saveReadyToSyncTransfers(any(), any())
-            verify(syncSchedulerHandler).prepareScheduleForNextTry(syncSchedule, time.plus(delay, ChronoUnit.SECONDS))
+            verify(chainSyncSchedulerHandler).prepareScheduleForNextTry(syncSchedule, time.plus(delay, ChronoUnit.SECONDS))
         }
     }
 
@@ -120,7 +119,7 @@ internal class ChainSyncServiceImplTest {
             //given
             val delay = 100L
             val syncSchedule = ChainSyncSchedule(ethChain, time, delay)
-            Mockito.`when`(syncSchedulerHandler.fetchActiveSchedules(any()))
+            Mockito.`when`(chainSyncSchedulerHandler.fetchActiveSchedules(any()))
                 .thenReturn(listOf(syncSchedule))
             Mockito.`when`(endpointProxy.syncTransfers(any())).thenReturn(
                 ChainSyncRecord(
@@ -132,11 +131,11 @@ internal class ChainSyncServiceImplTest {
             syncService.startSyncWithChain()
 
             //then
-            verify(syncRecordHandler).saveSyncRecord(any())
+            verify(chainSyncRecordHandler).saveSyncRecord(any())
             verify(walletSyncRecordHandler).saveReadyToSyncTransfers(any(), any())
-            verify(syncSchedulerHandler, times(0)).prepareScheduleForNextTry(any(), any())
+            verify(chainSyncSchedulerHandler, times(0)).prepareScheduleForNextTry(any(), any())
         }
     }
 
 
-}
+}

BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt

@@ -55,7 +55,7 @@ class PostgresConfig(db: DatabaseClient) {
                 CREATE TABLE IF NOT EXISTS chain_sync_schedules (
                     chain VARCHAR(72) PRIMARY KEY,
                     retry_time TIMESTAMP NOT NULL,
-                    delay NUMERIC NOT NULL
+                    delay INTEGER NOT NULL
                 );
                 CREATE TABLE IF NOT EXISTS chain_sync_records (
                     chain VARCHAR(72) PRIMARY KEY,
@@ -65,6 +65,28 @@ class PostgresConfig(db: DatabaseClient) {
                     success BOOLEAN NOT NULL,
                     error VARCHAR(100)
                 );
+                CREATE TABLE IF NOT EXISTS wallet_sync_schedules (
+                    id INTEGER PRIMARY KEY DEFAULT(1) CHECK(id = 1),
+                    retry_time TIMESTAMP NOT NULL,
+                    delay INTEGER NOT NULL,
+                    batch_size INTEGER
+                );
+                CREATE TABLE IF NOT EXISTS wallet_sync_records (
+                    id SERIAL PRIMARY KEY,
+                    time TIMESTAMP NOT NULL,
+                    success BOOLEAN NOT NULL,
+                    error VARCHAR(100)
+                );
+                CREATE TABLE IF NOT EXISTS deposits (
+                    id SERIAL PRIMARY KEY,
+                    wallet_sync_record INTEGER NOT NULL,
+                    chain VARCHAR(72) NOT NULL,
+                    token BOOLEAN NOT NULL,
+                    token_address VARCHAR(72),
+                    amount NUMERIC NOT NULL,
+                    depositor VARCHAR(72) NOT NULL,
+                    depositor_memo VARCHAR(72)
+                );
                 CREATE TABLE IF NOT EXISTS currency (
                     symbol VARCHAR(72) PRIMARY KEY,
                     name VARCHAR(72) NOT NULL
@@ -79,15 +101,6 @@ class PostgresConfig(db: DatabaseClient) {
                     withdraw_fee NUMERIC NOT NULL,
                     withdraw_min NUMERIC NOT NULL
                 );
-                CREATE TABLE IF NOT EXISTS deposits (
-                    id SERIAL PRIMARY KEY,
-                    chain VARCHAR(72),
-                    token BOOLEAN NOT NULL,
-                    token_address VARCHAR(72),
-                    amount NUMERIC NOT NULL,
-                    depositor VARCHAR(72) NOT NULL,
-                    depositorMemo VARCHAR(72)
-                );
             """
         }
         initDb // initialize the database

BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/ChainSyncRecordRepository.kt

@@ -1,11 +1,11 @@
 package co.nilin.opex.port.bcgateway.postgres.dao
 
-import co.nilin.opex.port.bcgateway.postgres.model.SyncRecordModel
+import co.nilin.opex.port.bcgateway.postgres.model.ChainSyncRecordModel
 import org.springframework.data.repository.reactive.ReactiveCrudRepository
 import org.springframework.stereotype.Repository
 import reactor.core.publisher.Mono
 
 @Repository
-interface ChainSyncRecordRepository : ReactiveCrudRepository<SyncRecordModel, String> {
-    fun findByChain(chain: String): Mono<SyncRecordModel>
-}
+interface ChainSyncRecordRepository : ReactiveCrudRepository<ChainSyncRecordModel, String> {
+    fun findByChain(chain: String): Mono<ChainSyncRecordModel>
+}

BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/ChainSyncScheduleRepository.kt

@@ -1,13 +1,16 @@
 package co.nilin.opex.port.bcgateway.postgres.dao
 
-import co.nilin.opex.port.bcgateway.postgres.model.AssignedAddressChainModel
-import co.nilin.opex.port.bcgateway.postgres.model.ChainModel
-import co.nilin.opex.port.bcgateway.postgres.model.SyncScheduleModel
+import co.nilin.opex.port.bcgateway.postgres.model.ChainSyncScheduleModel
+import co.nilin.opex.port.bcgateway.postgres.model.WalletSyncScheduleModel
 import kotlinx.coroutines.flow.Flow
 import org.springframework.data.r2dbc.repository.Query
-import org.springframework.data.repository.query.Param
 import org.springframework.data.repository.reactive.ReactiveCrudRepository
 import org.springframework.stereotype.Repository
+import reactor.core.publisher.Mono
+import java.time.LocalDateTime
 
 @Repository
-interface ChainSyncScheduleRepository : ReactiveCrudRepository<SyncScheduleModel, String>
+interface ChainSyncScheduleRepository : ReactiveCrudRepository<ChainSyncScheduleModel, String> {
+    @Query("select * from chain_sync_schedules where retry_time <= :time")
+    fun findActiveSchedule(time: LocalDateTime): Flow<ChainSyncScheduleModel>
+}

BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt

@@ -2,10 +2,23 @@ package co.nilin.opex.port.bcgateway.postgres.dao
 
 import co.nilin.opex.port.bcgateway.postgres.model.DepositModel
 import kotlinx.coroutines.flow.Flow
+import org.springframework.data.r2dbc.repository.Modifying
+import org.springframework.data.r2dbc.repository.Query
 import org.springframework.data.repository.reactive.ReactiveCrudRepository
 import org.springframework.stereotype.Repository
+import reactor.core.publisher.Mono
 
 @Repository
 interface DepositRepository : ReactiveCrudRepository<DepositModel, Long> {
     fun findByChain(chain: String): Flow<DepositModel>
-}
+
+    @Query("select * from deposits where chain = :chain and wallet_sync_record is null")
+    fun findByChainWhereNotSynced(chain: String): Flow<DepositModel>
+
+    @Query("select * from deposits where wallet_record_id is null limit :count")
+    fun findLimited(count: Long?): Flow<DepositModel>
+
+    @Modifying
+    @Query("update deposits set wallet_sync_record = :walletSyncRecord where id = :id")
+    fun updateWalletSyncRecord(id: Long, walletSyncRecord: Long): Mono<Int>
+}