From ff3f0a8b709aece8405b3c4bd9fe88e0ae344b6b Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sat, 11 Sep 2021 13:34:06 +0430 Subject: [PATCH 01/11] Add SyncRecordHandler, WalletSyncRecordHandler raw implementations --- .../postgres/impl/SyncRecordHandlerImpl.kt | 19 +++++++++++++++++++ .../impl/WalletSyncRecordHandlerImpl.kt | 19 +++++++++++++++++++ 2 files changed, 38 insertions(+) create mode 100644 BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt create mode 100644 BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt new file mode 100644 index 000000000..912f10e63 --- /dev/null +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -0,0 +1,19 @@ +package co.nilin.opex.port.bcgateway.postgres.impl + +import co.nilin.opex.bcgateway.core.model.ChainSyncRecord +import co.nilin.opex.bcgateway.core.model.ChainSyncSchedule +import co.nilin.opex.bcgateway.core.spi.SyncRecordHandler +import co.nilin.opex.bcgateway.core.spi.SyncSchedulerHandler +import org.springframework.stereotype.Component +import java.time.LocalDateTime + +@Component +class SyncRecordHandlerImpl : SyncRecordHandler { + override suspend fun loadLastSuccessRecord(chainName: String): ChainSyncRecord? { + TODO("Not yet implemented") + } + + override suspend fun saveSyncRecord(syncRecord: ChainSyncRecord) { + TODO("Not yet implemented") + } +} \ No newline at end of file diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt new file mode 100644 index 000000000..1d10ce26f --- /dev/null +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt @@ -0,0 +1,19 @@ +package co.nilin.opex.port.bcgateway.postgres.impl + +import co.nilin.opex.bcgateway.core.model.Deposit +import co.nilin.opex.bcgateway.core.model.WalletSyncSchedule +import co.nilin.opex.bcgateway.core.spi.WalletSyncRecordHandler +import co.nilin.opex.bcgateway.core.spi.WalletSyncSchedulerHandler +import org.springframework.stereotype.Component +import java.time.LocalDateTime + +@Component +class WalletSyncRecordHandlerImpl : WalletSyncRecordHandler { + override suspend fun saveReadyToSyncTransfers(chainName: String, deposits: List) { + TODO("Not yet implemented") + } + + override suspend fun findReadyToSyncTransfers(count: Long?): List { + TODO("Not yet implemented") + } +} \ No newline at end of file From 4d5e31e7b57f5f897627969d9a3762a55d0baed4 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 11:43:28 +0430 Subject: [PATCH 02/11] Fix some optional columns in postgres module --- .../opex/port/bcgateway/postgres/config/PostgresConfig.kt | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt index 0804207c9..ae042a6cd 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt @@ -54,14 +54,14 @@ class PostgresConfig(db: DatabaseClient) { ); CREATE TABLE IF NOT EXISTS chain_sync_schedules ( chain VARCHAR(72) PRIMARY KEY, - retry_time TIMESTAMP, - delay NUMERIC + retry_time TIMESTAMP NOT NULL, + delay NUMERIC NOT NULL ); CREATE TABLE IF NOT EXISTS chain_sync_records ( chain VARCHAR(72) PRIMARY KEY, time TIMESTAMP NOT NULL, endpoint_url VARCHAR(72) NOT NULL, - latest_block INTEGER NOT NULL, + latest_block INTEGER, success BOOLEAN NOT NULL, error VARCHAR(100) ); From ea4be346a7ab676e7abf7613944d227887762700 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 11:57:32 +0430 Subject: [PATCH 03/11] Implement SyncRecordHandlerImpl.loadLastSuccessRecord() without records list --- .../opex/bcgateway/core/model/WalletSync.kt | 5 +--- .../postgres/dao/ChainSyncRecordRepository.kt | 12 ++++------ .../postgres/impl/SyncRecordHandlerImpl.kt | 23 +++++++++++++++---- .../bcgateway/postgres/model/SyncModel.kt | 2 +- 4 files changed, 25 insertions(+), 17 deletions(-) diff --git a/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/WalletSync.kt b/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/WalletSync.kt index 3f4ec00fa..81fc4115d 100644 --- a/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/WalletSync.kt +++ b/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/WalletSync.kt @@ -4,8 +4,5 @@ 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 + val time: LocalDateTime, val success: Boolean, val error: String?, val deposit: Deposit ) \ No newline at end of file diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/ChainSyncRecordRepository.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/ChainSyncRecordRepository.kt index b3f4ee2b0..b0fa51f46 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/ChainSyncRecordRepository.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/ChainSyncRecordRepository.kt @@ -1,13 +1,11 @@ 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.SyncRecord -import kotlinx.coroutines.flow.Flow -import org.springframework.data.r2dbc.repository.Query -import org.springframework.data.repository.query.Param +import co.nilin.opex.port.bcgateway.postgres.model.SyncRecordModel import org.springframework.data.repository.reactive.ReactiveCrudRepository import org.springframework.stereotype.Repository +import reactor.core.publisher.Mono @Repository -interface ChainSyncRecordRepository : ReactiveCrudRepository \ No newline at end of file +interface ChainSyncRecordRepository : ReactiveCrudRepository { + fun findByChain(chain: String): Mono +} \ No newline at end of file diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt index 912f10e63..4594b48a3 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -1,16 +1,29 @@ package co.nilin.opex.port.bcgateway.postgres.impl import co.nilin.opex.bcgateway.core.model.ChainSyncRecord -import co.nilin.opex.bcgateway.core.model.ChainSyncSchedule +import co.nilin.opex.bcgateway.core.model.Endpoint import co.nilin.opex.bcgateway.core.spi.SyncRecordHandler -import co.nilin.opex.bcgateway.core.spi.SyncSchedulerHandler +import co.nilin.opex.port.bcgateway.postgres.dao.ChainSyncRecordRepository +import kotlinx.coroutines.reactive.awaitSingleOrNull import org.springframework.stereotype.Component -import java.time.LocalDateTime @Component -class SyncRecordHandlerImpl : SyncRecordHandler { +class SyncRecordHandlerImpl(private val chainSyncRecordRepository: ChainSyncRecordRepository) : SyncRecordHandler { override suspend fun loadLastSuccessRecord(chainName: String): ChainSyncRecord? { - TODO("Not yet implemented") + val dao = chainSyncRecordRepository.findByChain(chainName).awaitSingleOrNull() + return if (dao !== null) { + ChainSyncRecord( + dao.chain, + dao.time, + Endpoint(dao.endpointUrl), + dao.latestBlock, + dao.success, + dao.error, + emptyList() + ) + } else { + null + } } override suspend fun saveSyncRecord(syncRecord: ChainSyncRecord) { diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt index a131264ce..161b6af57 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt @@ -11,7 +11,7 @@ data class SyncScheduleModel( ) @Table("chain_sync_record") -data class SyncRecord( +data class SyncRecordModel( @Id @Column("chain") val chain: String, val time: LocalDateTime, @Column("endpoint_url") val endpointUrl: String, From 8ae1fbaf319e7c62366ef57e2f699cca3a995a18 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 12:36:48 +0430 Subject: [PATCH 04/11] Add deposits table and repository --- .../bcgateway/postgres/config/PostgresConfig.kt | 8 ++++++++ .../bcgateway/postgres/dao/DepositRepository.kt | 12 ++++++++++++ .../bcgateway/postgres/model/DepositModel.kt | 16 ++++++++++++++++ 3 files changed, 36 insertions(+) create mode 100644 BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt create mode 100644 BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt index ae042a6cd..6388680d7 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt @@ -79,6 +79,14 @@ 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_address VARCHAR(72), + amount NUMERIC NOT NULL, + depositor VARCHAR(72) NOT NULL, + depositorMemo VARCHAR(72) + ); """ } initDb // initialize the database diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt new file mode 100644 index 000000000..2f7bbb9df --- /dev/null +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt @@ -0,0 +1,12 @@ +package co.nilin.opex.port.bcgateway.postgres.dao + +import co.nilin.opex.port.bcgateway.postgres.model.AddressTypeModel +import co.nilin.opex.port.bcgateway.postgres.model.DepositModel +import kotlinx.coroutines.flow.Flow +import org.springframework.data.repository.reactive.ReactiveCrudRepository +import org.springframework.stereotype.Repository + +@Repository +interface DepositRepository : ReactiveCrudRepository { + fun findByChain(chain: String): Flow +} \ No newline at end of file diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt new file mode 100644 index 000000000..9b76488df --- /dev/null +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt @@ -0,0 +1,16 @@ +package co.nilin.opex.port.bcgateway.postgres.model + +import org.springframework.data.annotation.Id +import org.springframework.data.relational.core.mapping.Column +import org.springframework.data.relational.core.mapping.Table +import java.math.BigDecimal + +@Table("deposits") +data class DepositModel( + @Id val id: Long?, + val depositor: String, + val depositorMemo: String?, + val amount: BigDecimal, + val chain: String?, + val tokenAddress: String? +) \ No newline at end of file From d7b60b4e6dc611d1e1b8ba58bb3df51738145b91 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 12:41:20 +0430 Subject: [PATCH 05/11] Add records to sync record handler --- .../postgres/dao/DepositRepository.kt | 1 - .../postgres/impl/SyncRecordHandlerImpl.kt | 30 ++++++++++++------- 2 files changed, 20 insertions(+), 11 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt index 2f7bbb9df..f5e21069b 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/dao/DepositRepository.kt @@ -1,6 +1,5 @@ package co.nilin.opex.port.bcgateway.postgres.dao -import co.nilin.opex.port.bcgateway.postgres.model.AddressTypeModel import co.nilin.opex.port.bcgateway.postgres.model.DepositModel import kotlinx.coroutines.flow.Flow import org.springframework.data.repository.reactive.ReactiveCrudRepository diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt index 4594b48a3..e98ee312d 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -1,25 +1,35 @@ package co.nilin.opex.port.bcgateway.postgres.impl import co.nilin.opex.bcgateway.core.model.ChainSyncRecord +import co.nilin.opex.bcgateway.core.model.Deposit import co.nilin.opex.bcgateway.core.model.Endpoint import co.nilin.opex.bcgateway.core.spi.SyncRecordHandler import co.nilin.opex.port.bcgateway.postgres.dao.ChainSyncRecordRepository +import co.nilin.opex.port.bcgateway.postgres.dao.DepositRepository +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.reactive.awaitSingleOrNull import org.springframework.stereotype.Component @Component -class SyncRecordHandlerImpl(private val chainSyncRecordRepository: ChainSyncRecordRepository) : SyncRecordHandler { +class SyncRecordHandlerImpl( + private val chainSyncRecordRepository: ChainSyncRecordRepository, + private val depositRepository: DepositRepository +) : SyncRecordHandler { override suspend fun loadLastSuccessRecord(chainName: String): ChainSyncRecord? { - val dao = chainSyncRecordRepository.findByChain(chainName).awaitSingleOrNull() - return if (dao !== null) { + val chainSyncRecordDao = chainSyncRecordRepository.findByChain(chainName).awaitSingleOrNull() + return if (chainSyncRecordDao !== null) { + val deposits = depositRepository.findByChain(chainName).map { + Deposit(it.depositor, it.depositorMemo, it.amount, it.chain, it.tokenAddress) + } ChainSyncRecord( - dao.chain, - dao.time, - Endpoint(dao.endpointUrl), - dao.latestBlock, - dao.success, - dao.error, - emptyList() + chainSyncRecordDao.chain, + chainSyncRecordDao.time, + Endpoint(chainSyncRecordDao.endpointUrl), + chainSyncRecordDao.latestBlock, + chainSyncRecordDao.success, + chainSyncRecordDao.error, + deposits.toList() ) } else { null From 26e30b68825ffdebe858eb95be6ea2264918b094 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 13:20:10 +0430 Subject: [PATCH 06/11] Add token flag to deposit --- .../kotlin/co/nilin/opex/bcgateway/core/model/Deposit.kt | 9 ++++++++- .../port/bcgateway/postgres/config/PostgresConfig.kt | 1 + .../opex/port/bcgateway/postgres/model/DepositModel.kt | 1 + 3 files changed, 10 insertions(+), 1 deletion(-) diff --git a/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/Deposit.kt b/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/Deposit.kt index 579803a87..9cdaee869 100644 --- a/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/Deposit.kt +++ b/BlockchainGateway/bc-gateway-core/src/main/kotlin/co/nilin/opex/bcgateway/core/model/Deposit.kt @@ -2,4 +2,11 @@ package co.nilin.opex.bcgateway.core.model import java.math.BigDecimal -data class Deposit(val depositor: String, val depositorMemo: String?, val amount: BigDecimal, val chain: String?, val tokenAddress: String?) +data class Deposit( + val depositor: String, + val depositorMemo: String?, + val amount: BigDecimal, + val chain: String?, + val token: Boolean, + val tokenAddress: String? +) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt index 6388680d7..14c3890df 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/config/PostgresConfig.kt @@ -82,6 +82,7 @@ class PostgresConfig(db: DatabaseClient) { 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, diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt index 9b76488df..d79321a48 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/DepositModel.kt @@ -12,5 +12,6 @@ data class DepositModel( val depositorMemo: String?, val amount: BigDecimal, val chain: String?, + val token: Boolean, val tokenAddress: String? ) \ No newline at end of file From 865ddbf5f2f7b438e87649f7f41029f1b3c005a6 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 13:21:46 +0430 Subject: [PATCH 07/11] Implement SyncRecordHandlerImpl.saveSyncRecord() --- .../postgres/impl/SyncRecordHandlerImpl.kt | 29 +++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt index e98ee312d..2a708fba3 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -6,10 +6,14 @@ import co.nilin.opex.bcgateway.core.model.Endpoint import co.nilin.opex.bcgateway.core.spi.SyncRecordHandler import co.nilin.opex.port.bcgateway.postgres.dao.ChainSyncRecordRepository import co.nilin.opex.port.bcgateway.postgres.dao.DepositRepository +import co.nilin.opex.port.bcgateway.postgres.model.DepositModel +import co.nilin.opex.port.bcgateway.postgres.model.SyncRecordModel import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.reactive.awaitFirst import kotlinx.coroutines.reactive.awaitSingleOrNull import org.springframework.stereotype.Component +import reactor.core.publisher.Mono @Component class SyncRecordHandlerImpl( @@ -20,7 +24,7 @@ class SyncRecordHandlerImpl( val chainSyncRecordDao = chainSyncRecordRepository.findByChain(chainName).awaitSingleOrNull() return if (chainSyncRecordDao !== null) { val deposits = depositRepository.findByChain(chainName).map { - Deposit(it.depositor, it.depositorMemo, it.amount, it.chain, it.tokenAddress) + Deposit(it.depositor, it.depositorMemo, it.amount, it.chain, it.token, it.tokenAddress) } ChainSyncRecord( chainSyncRecordDao.chain, @@ -37,6 +41,27 @@ class SyncRecordHandlerImpl( } override suspend fun saveSyncRecord(syncRecord: ChainSyncRecord) { - TODO("Not yet implemented") + val chainSyncRecordDao = + SyncRecordModel( + syncRecord.chainName, + syncRecord.time, + syncRecord.endpoint.url, + syncRecord.latestBlock, + syncRecord.success, + syncRecord.error + ) + val deposits = syncRecord.records.map { + DepositModel( + null, + it.depositor, + it.depositorMemo, + it.amount, + it.chain, + it.token, + it.tokenAddress + ) + } + Mono.`when`(chainSyncRecordRepository.save(chainSyncRecordDao), depositRepository.saveAll(deposits)) + .awaitFirst() } } \ No newline at end of file From 0630720b3089c73cdfbc483d62413fabd79f4ae2 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 14:00:01 +0430 Subject: [PATCH 08/11] Add transactional saving ability to sync records --- .../bcgateway/postgres/impl/SyncRecordHandlerImpl.kt | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt index 2a708fba3..b456fe05b 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -8,12 +8,14 @@ import co.nilin.opex.port.bcgateway.postgres.dao.ChainSyncRecordRepository import co.nilin.opex.port.bcgateway.postgres.dao.DepositRepository import co.nilin.opex.port.bcgateway.postgres.model.DepositModel import co.nilin.opex.port.bcgateway.postgres.model.SyncRecordModel +import kotlinx.coroutines.async import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.toList import kotlinx.coroutines.reactive.awaitFirst import kotlinx.coroutines.reactive.awaitSingleOrNull +import kotlinx.coroutines.runBlocking import org.springframework.stereotype.Component -import reactor.core.publisher.Mono +import org.springframework.transaction.annotation.Transactional @Component class SyncRecordHandlerImpl( @@ -22,6 +24,7 @@ class SyncRecordHandlerImpl( ) : SyncRecordHandler { override suspend fun loadLastSuccessRecord(chainName: String): ChainSyncRecord? { val chainSyncRecordDao = chainSyncRecordRepository.findByChain(chainName).awaitSingleOrNull() + return if (chainSyncRecordDao !== null) { val deposits = depositRepository.findByChain(chainName).map { Deposit(it.depositor, it.depositorMemo, it.amount, it.chain, it.token, it.tokenAddress) @@ -40,6 +43,7 @@ class SyncRecordHandlerImpl( } } + @Transactional override suspend fun saveSyncRecord(syncRecord: ChainSyncRecord) { val chainSyncRecordDao = SyncRecordModel( @@ -50,6 +54,7 @@ class SyncRecordHandlerImpl( syncRecord.success, syncRecord.error ) + chainSyncRecordRepository.save(chainSyncRecordDao).awaitFirst() val deposits = syncRecord.records.map { DepositModel( null, @@ -61,7 +66,6 @@ class SyncRecordHandlerImpl( it.tokenAddress ) } - Mono.`when`(chainSyncRecordRepository.save(chainSyncRecordDao), depositRepository.saveAll(deposits)) - .awaitFirst() + depositRepository.saveAll(deposits).awaitFirst() } } \ No newline at end of file From 075b90830a239a8311263b33b8393ae57219f897 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 14:01:04 +0430 Subject: [PATCH 09/11] Organize sync record imports --- .../opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt | 2 -- 1 file changed, 2 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt index b456fe05b..dac616392 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -8,12 +8,10 @@ import co.nilin.opex.port.bcgateway.postgres.dao.ChainSyncRecordRepository import co.nilin.opex.port.bcgateway.postgres.dao.DepositRepository import co.nilin.opex.port.bcgateway.postgres.model.DepositModel import co.nilin.opex.port.bcgateway.postgres.model.SyncRecordModel -import kotlinx.coroutines.async import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.toList import kotlinx.coroutines.reactive.awaitFirst import kotlinx.coroutines.reactive.awaitSingleOrNull -import kotlinx.coroutines.runBlocking import org.springframework.stereotype.Component import org.springframework.transaction.annotation.Transactional From ba26622e94a86adb4beebf2e46a6ef3ad3171ad3 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 14:48:15 +0430 Subject: [PATCH 10/11] Implement saveReadyToSyncTransfers --- .../postgres/impl/SyncRecordHandlerImpl.kt | 5 ++-- .../impl/WalletSyncRecordHandlerImpl.kt | 25 +++++++++++++++---- 2 files changed, 22 insertions(+), 8 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt index dac616392..319b8e4c0 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/SyncRecordHandlerImpl.kt @@ -22,7 +22,6 @@ class SyncRecordHandlerImpl( ) : SyncRecordHandler { override suspend fun loadLastSuccessRecord(chainName: String): ChainSyncRecord? { val chainSyncRecordDao = chainSyncRecordRepository.findByChain(chainName).awaitSingleOrNull() - return if (chainSyncRecordDao !== null) { val deposits = depositRepository.findByChain(chainName).map { Deposit(it.depositor, it.depositorMemo, it.amount, it.chain, it.token, it.tokenAddress) @@ -53,7 +52,7 @@ class SyncRecordHandlerImpl( syncRecord.error ) chainSyncRecordRepository.save(chainSyncRecordDao).awaitFirst() - val deposits = syncRecord.records.map { + val depositsDao = syncRecord.records.map { DepositModel( null, it.depositor, @@ -64,6 +63,6 @@ class SyncRecordHandlerImpl( it.tokenAddress ) } - depositRepository.saveAll(deposits).awaitFirst() + depositRepository.saveAll(depositsDao).awaitFirst() } } \ No newline at end of file diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt index 1d10ce26f..3addb0f6f 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/impl/WalletSyncRecordHandlerImpl.kt @@ -1,16 +1,31 @@ package co.nilin.opex.port.bcgateway.postgres.impl import co.nilin.opex.bcgateway.core.model.Deposit -import co.nilin.opex.bcgateway.core.model.WalletSyncSchedule import co.nilin.opex.bcgateway.core.spi.WalletSyncRecordHandler -import co.nilin.opex.bcgateway.core.spi.WalletSyncSchedulerHandler +import co.nilin.opex.port.bcgateway.postgres.dao.DepositRepository +import co.nilin.opex.port.bcgateway.postgres.model.DepositModel +import kotlinx.coroutines.reactive.awaitFirst import org.springframework.stereotype.Component -import java.time.LocalDateTime +import org.springframework.transaction.annotation.Transactional @Component -class WalletSyncRecordHandlerImpl : WalletSyncRecordHandler { +class WalletSyncRecordHandlerImpl( + private val depositRepository: DepositRepository +) : WalletSyncRecordHandler { + @Transactional override suspend fun saveReadyToSyncTransfers(chainName: String, deposits: List) { - TODO("Not yet implemented") + val depositsDao = deposits.map { + DepositModel( + null, + it.depositor, + it.depositorMemo, + it.amount, + it.chain, + it.token, + it.tokenAddress + ) + } + depositRepository.saveAll(depositsDao).awaitFirst() } override suspend fun findReadyToSyncTransfers(count: Long?): List { From 91c9cb3327771d4f69093333e57818ab962aead9 Mon Sep 17 00:00:00 2001 From: metalicn20 Date: Sun, 12 Sep 2021 15:16:40 +0430 Subject: [PATCH 11/11] Fix SyncModel table names --- .../co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt index 161b6af57..d7093e80a 100644 --- a/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt +++ b/BlockchainGateway/bc-gateway-ports/bc-persister-postgres/src/main/kotlin/co/nilin/opex/port/bcgateway/postgres/model/SyncModel.kt @@ -5,12 +5,12 @@ import org.springframework.data.relational.core.mapping.Column import org.springframework.data.relational.core.mapping.Table import java.time.LocalDateTime -@Table("chain_sync_schedule") +@Table("chain_sync_schedules") data class SyncScheduleModel( @Id @Column("chain") val chain: String, @Column("retry_time") val retryTime: LocalDateTime, val delay: Long ) -@Table("chain_sync_record") +@Table("chain_sync_records") data class SyncRecordModel( @Id @Column("chain") val chain: String, val time: LocalDateTime,