From c5e52563956d7e2764404990439a1579314b7360 Mon Sep 17 00:00:00 2001 From: chenyuan Date: Tue, 18 Aug 2026 14:28:15 +0800 Subject: [PATCH] refactor: lock currency through current ownership --- .../JdbcExchangeResourceRepository.java | 34 ++++---- ...igitalCurrencyGenerationModuleGateway.java | 4 +- .../JdbcPaymentResourceRepository.java | 22 +++--- .../CurrencyOwnershipLockRepository.java | 41 ++++++++++ .../module_2_4_5_training_attempt_upgrade.sql | 12 +++ src/main/resources/schema.sql | 6 +- .../rest/PaymentControllerTest.java | 63 +++++++++++++-- .../CurrencyOwnershipConcurrencyTest.java | 77 +++++++++++++++++++ src/test/resources/schema.sql | 7 +- 9 files changed, 227 insertions(+), 39 deletions(-) create mode 100644 src/main/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipLockRepository.java create mode 100644 src/test/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipConcurrencyTest.java diff --git a/src/main/java/com/yau/digitalrmb/exchange/infrastructure/persistence/JdbcExchangeResourceRepository.java b/src/main/java/com/yau/digitalrmb/exchange/infrastructure/persistence/JdbcExchangeResourceRepository.java index a41f2b7..f90b164 100644 --- a/src/main/java/com/yau/digitalrmb/exchange/infrastructure/persistence/JdbcExchangeResourceRepository.java +++ b/src/main/java/com/yau/digitalrmb/exchange/infrastructure/persistence/JdbcExchangeResourceRepository.java @@ -16,6 +16,7 @@ import com.yau.digitalrmb.shared.exception.BusinessException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Repository; import com.yau.digitalrmb.shared.wallet.WalletPrerequisiteProjectionService; +import com.yau.digitalrmb.shared.currency.CurrencyOwnershipLockRepository; import java.math.BigDecimal; import java.time.LocalDate; @@ -29,15 +30,18 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor private final WalletPrivateKeyCipher cipher; private final InstitutionKeyService keyService; private final WalletPrerequisiteProjectionService walletPrerequisites; + private final CurrencyOwnershipLockRepository ownershipLocks; public JdbcExchangeResourceRepository(JdbcTemplate jdbc, InstitutionIdentityCryptography cryptography, WalletPrivateKeyCipher cipher, InstitutionKeyService keyService, - WalletPrerequisiteProjectionService walletPrerequisites) { + WalletPrerequisiteProjectionService walletPrerequisites, + CurrencyOwnershipLockRepository ownershipLocks) { this.jdbc = jdbc; this.cryptography = cryptography; this.cipher = cipher; this.keyService = keyService; this.walletPrerequisites = walletPrerequisites; + this.ownershipLocks = ownershipLocks; } @Override @@ -51,7 +55,8 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor "c.balance_limit,c.valid_until," + "CASE WHEN c.daily_counter_date=CURRENT_DATE THEN c.daily_used_amount ELSE 0 END used_today," + "cert.public_key,a.account_id,a.bank_code,a.bank_name,a.card_number,a.card_last4," + - "a.balance bank_balance,COALESCE(SUM(co.denomination),0) inventory_balance " + + "a.balance bank_balance,COALESCE(SUM(CASE WHEN own.currency_id IS NOT NULL " + + "THEN co.denomination ELSE 0 END),0) inventory_balance " + "FROM digital_wallet w " + "JOIN wallet_certificate cert ON cert.wallet_id=w.wallet_id " + "AND cert.status='VALID' AND cert.filing_status='REGISTERED' " + @@ -62,7 +67,8 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor "JOIN simulated_bank_account a ON a.account_id=b.bank_account_id AND a.status='ACTIVE' " + "LEFT JOIN commercial_bank_currency co ON co.bank_code=a.bank_code " + "AND co.organization_id=? AND co.received_by_user_id=? " + - "AND co.status NOT IN ('EXCHANGE_LOCKED','TRANSFERRED') " + + "LEFT JOIN central_bank_currency_ownership own ON own.currency_id=co.currency_id " + + "AND own.owner_type='BANK' AND own.owner_id=a.bank_code AND own.status='AVAILABLE' " + "WHERE w.user_id=? AND w.status='ACTIVE' AND w.central_bank_confirmation_signature IS NOT NULL " + "AND a.bank_code=? " + "GROUP BY w.wallet_id,w.wallet_type,w.status,w.balance,c.contract_id,c.single_payment_limit," + @@ -182,7 +188,7 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor List candidates = jdbc.query("SELECT c.currency_id,c.denomination,c.status,c.complete_currency " + "FROM commercial_bank_currency c JOIN central_bank_currency_ownership o ON o.currency_id=c.currency_id " + "WHERE c.bank_code=? AND c.organization_id=? AND c.received_by_user_id=? " + - "AND c.status NOT IN ('EXCHANGE_LOCKED','TRANSFERRED') AND o.owner_type='BANK' " + + "AND o.owner_type='BANK' " + "AND o.owner_id=? AND o.status='AVAILABLE' ORDER BY c.denomination DESC,c.currency_id FOR UPDATE", (rs, row) -> new CoinCandidate(rs.getString("currency_id"), rs.getBigDecimal("denomination"), rs.getString("status"), rs.getString("complete_currency")), order.getBankCode(), @@ -199,8 +205,8 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor } if (remaining.signum() != 0) throw validation("发行模块形成的商业银行币串库存无法精确组成兑换金额"); for (CoinCandidate coin : selected) { - int locked = jdbc.update("UPDATE commercial_bank_currency SET status='EXCHANGE_LOCKED' " + - "WHERE currency_id=? AND status=?", coin.id, coin.status); + int locked = ownershipLocks.tryLock(coin.id, "BANK", order.getBankCode(), + "EXCHANGE", order.getId().toString()) ? 1 : 0; if (locked != 1) throw validation("币串库存已被其他兑换请求占用，请重试"); jdbc.update("INSERT INTO exchange_coin_reservation (exchange_id,currency_id,denomination,source_status,status," + "reserved_at,reserved_by) VALUES (?,?,?,?, 'RESERVED',CURRENT_TIMESTAMP,?)", @@ -229,12 +235,12 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor "JOIN exchange_coin_reservation r ON r.currency_id=c.currency_id " + "JOIN central_bank_currency_ownership o ON o.currency_id=c.currency_id " + "WHERE r.exchange_id=? AND c.currency_id=? AND c.bank_code=? AND c.organization_id=? " + - "AND c.received_by_user_id=? AND c.status='EXCHANGE_LOCKED' AND r.status='RESERVED' " + + "AND c.received_by_user_id=? AND r.status='RESERVED' " + "AND c.complete_currency IS NOT NULL AND c.issuance_ownership_original_text IS NOT NULL " + "AND c.issuance_ownership_signature IS NOT NULL AND o.owner_type='BANK' AND o.owner_id=? " + - "AND o.status='AVAILABLE'", + "AND o.status='LOCKED' AND o.lock_business_type='EXCHANGE' AND o.lock_business_id=?", Integer.class, order.getId().toString(), coin.getCurrencyId(), order.getBankCode(), - order.getOrganizationId(), actor.getUserId(), order.getBankCode()); + order.getOrganizationId(), actor.getUserId(), order.getBankCode(), order.getId().toString()); if (valid == null || valid != 1) { throw validation("币串真实性、当前权属或状态校验失败：" + coin.getCurrencyId()); } @@ -256,14 +262,10 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor transactionId, order.getId().toString(), order.getBankCode(), order.getWalletId(), order.getAmount(), coins.size(), confirmationDigest, centralBankSignature, actor.getUsername()); for (ReservedCoin coin : coins) { - jdbc.update("UPDATE commercial_bank_currency SET status='TRANSFERRED' WHERE currency_id=? AND status='EXCHANGE_LOCKED'", - coin.getCurrencyId()); jdbc.update("UPDATE exchange_coin_reservation SET status='TRANSFERRED',transferred_at=CURRENT_TIMESTAMP " + "WHERE exchange_id=? AND currency_id=?", order.getId().toString(), coin.getCurrencyId()); - int changed = jdbc.update("UPDATE central_bank_currency_ownership SET owner_type='WALLET',owner_id=?," + - "status='AVAILABLE',last_transaction_id=?,updated_at=CURRENT_TIMESTAMP " + - "WHERE currency_id=? AND owner_type='BANK' AND owner_id=? AND status='AVAILABLE'", - order.getWalletId(), transactionId, coin.getCurrencyId(), order.getBankCode()); + int changed = ownershipLocks.transferLocked(coin.getCurrencyId(), "BANK", order.getBankCode(), + "EXCHANGE", order.getId().toString(), "WALLET", order.getWalletId(), transactionId) ? 1 : 0; if (changed != 1) throw validation("央行登记中心币串权属已变化：" + coin.getCurrencyId()); } List balances = jdbc.query("SELECT balance FROM digital_wallet WHERE wallet_id=? AND status='ACTIVE' FOR UPDATE", @@ -300,7 +302,7 @@ public class JdbcExchangeResourceRepository implements ExchangeResourceRepositor BigDecimal total = jdbc.queryForObject("SELECT COALESCE(SUM(c.denomination),0) " + "FROM commercial_bank_currency c JOIN central_bank_currency_ownership o ON o.currency_id=c.currency_id " + "WHERE c.bank_code=? AND c.organization_id=? AND c.received_by_user_id=? " + - "AND c.status NOT IN ('EXCHANGE_LOCKED','TRANSFERRED') AND o.owner_type='BANK' " + + "AND o.owner_type='BANK' " + "AND o.owner_id=? AND o.status='AVAILABLE'", BigDecimal.class, order.getBankCode(), order.getOrganizationId(), actor.getUserId(), order.getBankCode()); if (total == null || total.compareTo(order.getAmount()) < 0) { diff --git a/src/main/java/com/yau/digitalrmb/issuance/application/service/DigitalCurrencyGenerationModuleGateway.java b/src/main/java/com/yau/digitalrmb/issuance/application/service/DigitalCurrencyGenerationModuleGateway.java index 8353bb8..a7e99e3 100644 --- a/src/main/java/com/yau/digitalrmb/issuance/application/service/DigitalCurrencyGenerationModuleGateway.java +++ b/src/main/java/com/yau/digitalrmb/issuance/application/service/DigitalCurrencyGenerationModuleGateway.java @@ -223,8 +223,8 @@ public class DigitalCurrencyGenerationModuleGateway { coin.getOwnershipOriginalText(), coin.getOwnershipSignature(), now, actor.getUserId(), actor.getUsername()); jdbcTemplate.update("INSERT INTO central_bank_currency_ownership " - + "(currency_id,owner_type,owner_id,status,last_transaction_id,updated_at) " - + "VALUES (?,'BANK',?,'AVAILABLE',NULL,?)", + + "(currency_id,owner_type,owner_id,status,lock_business_type,lock_business_id,last_transaction_id,updated_at) " + + "VALUES (?,'BANK',?,'AVAILABLE',NULL,NULL,NULL,?)", coin.getCoinId(), requiredBankCode, now); } } diff --git a/src/main/java/com/yau/digitalrmb/payment/infrastructure/persistence/JdbcPaymentResourceRepository.java b/src/main/java/com/yau/digitalrmb/payment/infrastructure/persistence/JdbcPaymentResourceRepository.java index bd21dcf..0aad7e3 100644 --- a/src/main/java/com/yau/digitalrmb/payment/infrastructure/persistence/JdbcPaymentResourceRepository.java +++ b/src/main/java/com/yau/digitalrmb/payment/infrastructure/persistence/JdbcPaymentResourceRepository.java @@ -18,6 +18,7 @@ import com.yau.digitalrmb.shared.api.ErrorCode; import com.yau.digitalrmb.shared.exception.BusinessException; import com.yau.digitalrmb.shared.wallet.WalletPrerequisiteProjectionService; import com.yau.digitalrmb.shared.wallet.WalletPrerequisiteReference; +import com.yau.digitalrmb.shared.currency.CurrencyOwnershipLockRepository; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Repository; @@ -35,12 +36,15 @@ public class JdbcPaymentResourceRepository implements PaymentResourceRepository private final WalletPrivateKeyCipher walletCipher; private final InstitutionKeyService keyService; private final WalletPrerequisiteProjectionService walletProjection; + private final CurrencyOwnershipLockRepository ownershipLocks; public JdbcPaymentResourceRepository(JdbcTemplate jdbc, InstitutionIdentityCryptography cryptography, WalletPrivateKeyCipher walletCipher, InstitutionKeyService keyService, - WalletPrerequisiteProjectionService walletProjection) { + WalletPrerequisiteProjectionService walletProjection, + CurrencyOwnershipLockRepository ownershipLocks) { this.jdbc = jdbc; this.cryptography = cryptography; this.walletCipher = walletCipher; this.keyService = keyService; this.walletProjection = walletProjection; + this.ownershipLocks = ownershipLocks; } @Override @@ -115,14 +119,10 @@ public class JdbcPaymentResourceRepository implements PaymentResourceRepository if (!keyService.verifyCentralBank(subject, digest, signature)) throw validation("中央银行结算确认签名验证失败"); for (PaymentCoin coin : coins) { - int ownership = jdbc.update("UPDATE central_bank_currency_ownership SET owner_type='WALLET',owner_id=?," + - "status='AVAILABLE',last_transaction_id=NULL,updated_at=CURRENT_TIMESTAMP WHERE currency_id=? " + - "AND owner_type='WALLET' AND owner_id=? AND status='AVAILABLE'", - order.getPayeeWalletId(), coin.getCurrencyId(), order.getPayerWalletId()); + int ownership = ownershipLocks.transferLocked(coin.getCurrencyId(), "WALLET", + order.getPayerWalletId(), "PAYMENT", order.getId().toString(), "WALLET", + order.getPayeeWalletId(), transactionId) ? 1 : 0; if (ownership != 1) throw validation("中央银行登记中心币串权属已变化：" + coin.getCurrencyId()); - int currency = jdbc.update("UPDATE commercial_bank_currency SET status='TRANSFERRED' " + - "WHERE currency_id=? AND status='PAYMENT_LOCKED'", coin.getCurrencyId()); - if (currency != 1) throw validation("付款币串锁定状态异常：" + coin.getCurrencyId()); jdbc.update("UPDATE payment_coin_reservation SET status='TRANSFERRED',transferred_at=CURRENT_TIMESTAMP " + "WHERE payment_id=? AND currency_id=? AND status='RESERVED'", order.getId().toString(), coin.getCurrencyId()); @@ -252,7 +252,7 @@ public class JdbcPaymentResourceRepository implements PaymentResourceRepository private List selectAndLockCoins(PaymentOrder order, PaymentActor actor) { List candidates = jdbc.query("SELECT c.currency_id,c.denomination,c.status FROM commercial_bank_currency c " + "JOIN central_bank_currency_ownership o ON o.currency_id=c.currency_id WHERE o.owner_type='WALLET' " + - "AND o.owner_id=? AND o.status='AVAILABLE' AND c.status='TRANSFERRED' ORDER BY c.denomination DESC,c.currency_id FOR UPDATE", + "AND o.owner_id=? AND o.status='AVAILABLE' ORDER BY c.denomination DESC,c.currency_id FOR UPDATE", (rs, row) -> new CoinCandidate(rs.getString("currency_id"), rs.getBigDecimal("denomination"), rs.getString("status")), order.getPayerWalletId()); BigDecimal remaining = order.getAmount(); @@ -266,8 +266,8 @@ public class JdbcPaymentResourceRepository implements PaymentResourceRepository if (remaining.signum() != 0) throw validation("付款钱包中的数字货币无法组成支付金额"); List results = new ArrayList(); for (CoinCandidate coin : selected) { - if (jdbc.update("UPDATE commercial_bank_currency SET status='PAYMENT_LOCKED' WHERE currency_id=? AND status='TRANSFERRED'", - coin.currencyId) != 1) throw validation("付款币串已被其他支付占用"); + if (!ownershipLocks.tryLock(coin.currencyId, "WALLET", order.getPayerWalletId(), + "PAYMENT", order.getId().toString())) throw validation("付款币串已被其他支付占用"); jdbc.update("INSERT INTO payment_coin_reservation (payment_id,currency_id,denomination,source_status,status,reserved_at) " + "VALUES (?,?,?,?, 'RESERVED',CURRENT_TIMESTAMP)", order.getId().toString(), coin.currencyId, coin.denomination, coin.status); diff --git a/src/main/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipLockRepository.java b/src/main/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipLockRepository.java new file mode 100644 index 0000000..8367ecd --- /dev/null +++ b/src/main/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipLockRepository.java @@ -0,0 +1,41 @@ +package com.yau.digitalrmb.shared.currency; + +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Repository; + +@Repository +public class CurrencyOwnershipLockRepository { + private final JdbcTemplate jdbc; + + public CurrencyOwnershipLockRepository(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + public boolean tryLock(String currencyId, String ownerType, String ownerId, + String businessType, String businessId) { + return jdbc.update("UPDATE central_bank_currency_ownership " + + "SET status='LOCKED',lock_business_type=?,lock_business_id=?,updated_at=CURRENT_TIMESTAMP " + + "WHERE currency_id=? AND owner_type=? AND owner_id=? AND status='AVAILABLE'", + businessType, businessId, currencyId, ownerType, ownerId) == 1; + } + + public boolean transferLocked(String currencyId, String ownerType, String ownerId, + String businessType, String businessId, + String newOwnerType, String newOwnerId, String transactionId) { + return jdbc.update("UPDATE central_bank_currency_ownership SET owner_type=?,owner_id=?,status='AVAILABLE'," + + "lock_business_type=NULL,lock_business_id=NULL,last_transaction_id=?,updated_at=CURRENT_TIMESTAMP " + + "WHERE currency_id=? AND owner_type=? AND owner_id=? AND status='LOCKED' " + + "AND lock_business_type=? AND lock_business_id=?", + newOwnerType, newOwnerId, transactionId, currencyId, ownerType, ownerId, + businessType, businessId) == 1; + } + + public boolean release(String currencyId, String ownerType, String ownerId, + String businessType, String businessId) { + return jdbc.update("UPDATE central_bank_currency_ownership SET status='AVAILABLE'," + + "lock_business_type=NULL,lock_business_id=NULL,updated_at=CURRENT_TIMESTAMP " + + "WHERE currency_id=? AND owner_type=? AND owner_id=? AND status='LOCKED' " + + "AND lock_business_type=? AND lock_business_id=?", + currencyId, ownerType, ownerId, businessType, businessId) == 1; + } +} diff --git a/src/main/resources/db/manual/module_2_4_5_training_attempt_upgrade.sql b/src/main/resources/db/manual/module_2_4_5_training_attempt_upgrade.sql index be4e5b2..6e0111f 100644 --- a/src/main/resources/db/manual/module_2_4_5_training_attempt_upgrade.sql +++ b/src/main/resources/db/manual/module_2_4_5_training_attempt_upgrade.sql @@ -56,3 +56,15 @@ CREATE TABLE IF NOT EXISTS issuance_source_currency_usage ( CONSTRAINT uk_issuance_source_currency UNIQUE (source_currency_id), INDEX idx_issuance_source_request (issuance_request_id, status) ); + +ALTER TABLE central_bank_currency_ownership + ADD COLUMN lock_business_type VARCHAR(32) NULL AFTER status, + ADD COLUMN lock_business_id VARCHAR(96) NULL AFTER lock_business_type; + +ALTER TABLE central_bank_currency_ownership + DROP FOREIGN KEY fk_currency_ownership_transaction, + ADD INDEX idx_currency_ownership_available (owner_type, owner_id, status), + ADD INDEX idx_currency_ownership_lock_business (lock_business_type, lock_business_id); + +ALTER TABLE payment_coin_reservation + DROP INDEX uk_payment_reserved_currency; diff --git a/src/main/resources/schema.sql b/src/main/resources/schema.sql index df1e3d7..4c7595f 100644 --- a/src/main/resources/schema.sql +++ b/src/main/resources/schema.sql @@ -501,10 +501,13 @@ CREATE TABLE IF NOT EXISTS central_bank_currency_ownership ( owner_type VARCHAR(16) NOT NULL, owner_id VARCHAR(128) NOT NULL, status VARCHAR(16) NOT NULL, + lock_business_type VARCHAR(32) NULL, + lock_business_id VARCHAR(96) NULL, last_transaction_id VARCHAR(128) NULL, updated_at TIMESTAMP NOT NULL, CONSTRAINT fk_currency_ownership_currency FOREIGN KEY (currency_id) REFERENCES commercial_bank_currency(currency_id), - CONSTRAINT fk_currency_ownership_transaction FOREIGN KEY (last_transaction_id) REFERENCES currency_ownership_transfer(transaction_id) + INDEX idx_currency_ownership_available (owner_type, owner_id, status), + INDEX idx_currency_ownership_lock_business (lock_business_type, lock_business_id) ); CREATE TABLE IF NOT EXISTS wallet_balance_ledger ( @@ -769,7 +772,6 @@ CREATE TABLE IF NOT EXISTS payment_coin_reservation ( reserved_at TIMESTAMP NOT NULL, transferred_at TIMESTAMP NULL, PRIMARY KEY (payment_id, currency_id), - UNIQUE KEY uk_payment_reserved_currency (currency_id), FOREIGN KEY (payment_id) REFERENCES payment_order(id), FOREIGN KEY (currency_id) REFERENCES commercial_bank_currency(currency_id) ); diff --git a/src/test/java/com/yau/digitalrmb/payment/interfaces/rest/PaymentControllerTest.java b/src/test/java/com/yau/digitalrmb/payment/interfaces/rest/PaymentControllerTest.java index 9511d9f..8735485 100644 --- a/src/test/java/com/yau/digitalrmb/payment/interfaces/rest/PaymentControllerTest.java +++ b/src/test/java/com/yau/digitalrmb/payment/interfaces/rest/PaymentControllerTest.java @@ -65,6 +65,9 @@ class PaymentControllerTest { InstitutionKeySubject subject = new InstitutionKeySubject(PAYER_USER, 3001L, 2001L); keyService.commercialBankKey(subject, "test"); keyService.centralBankKey(subject, "test"); + InstitutionKeySubject payeeSubject = new InstitutionKeySubject(PAYEE_USER, 3001L, 2001L); + keyService.commercialBankKey(payeeSubject, "test"); + keyService.centralBankKey(payeeSubject, "test"); InstitutionSm2KeyPair payerWalletKey = cryptography.generateSm2KeyPair(); InstitutionSm2KeyPair payeeWalletKey = cryptography.generateSm2KeyPair(); WalletOpeningTestData.insertCompletedWallet(jdbc, 991200L, PAYER_USER, 3001L, 2001L, @@ -75,7 +78,10 @@ class PaymentControllerTest { "BKCHCNBJ00002", "6222020000000001", BigDecimal.ZERO); insertWallet(PAYER_USER, PAYER_WALLET, "CERT_991200", "BKCHCNBJ00001", "ACCOUNT_PAYMENT_A", new BigDecimal("200.00"), payerWalletKey); - insertPayerContract(); + insertWallet(PAYEE_USER, PAYEE_WALLET, "CERT_991300", "BKCHCNBJ00002", + "ACCOUNT_PAYMENT_B", BigDecimal.ZERO, payeeWalletKey); + insertContract(PAYER_WALLET, "CONTRACT_991200"); + insertContract(PAYEE_WALLET, "CONTRACT_991300"); for (int index = 1; index <= 2; index++) { jdbc.update("INSERT INTO commercial_bank_currency (currency_id,source_currency_id,source_batch_id,issuance_request_id," + "bank_code,organization_id,denomination,currency,complete_currency,issuance_ownership_original_text," + @@ -90,8 +96,8 @@ class PaymentControllerTest { @AfterEach void cleanUp() { cleanPaymentData(); - jdbc.update("DELETE FROM institution_sm2_key_audit WHERE user_id=?", PAYER_USER); - jdbc.update("DELETE FROM institution_sm2_key WHERE user_id=?", PAYER_USER); + jdbc.update("DELETE FROM institution_sm2_key_audit WHERE user_id IN (?,?)", PAYER_USER, PAYEE_USER); + jdbc.update("DELETE FROM institution_sm2_key WHERE user_id IN (?,?)", PAYER_USER, PAYEE_USER); jdbc.update("DELETE FROM institution_identifier_application WHERE id IN (991000,991001,991002)"); cleanWalletState(); jdbc.update("DELETE FROM commercial_bank_currency WHERE issuance_request_id='PAYMENT_TEST_ISSUANCE'"); @@ -154,6 +160,12 @@ class PaymentControllerTest { assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM payment_wallet_ledger WHERE payment_id=?", Integer.class, id)).isEqualTo(2); assertThat(jdbc.queryForObject("SELECT owner_id FROM central_bank_currency_ownership WHERE currency_id='DC_PAYMENT_1'", String.class)) .isEqualTo(PAYEE_WALLET); + assertThat(jdbc.queryForObject("SELECT lock_business_id FROM central_bank_currency_ownership " + + "WHERE currency_id='DC_PAYMENT_1'", String.class)).isNull(); + assertThat(jdbc.queryForObject("SELECT last_transaction_id FROM central_bank_currency_ownership " + + "WHERE currency_id='DC_PAYMENT_1'", String.class)).startsWith("PAY_TXN_"); + assertThat(jdbc.queryForObject("SELECT status FROM commercial_bank_currency WHERE currency_id='DC_PAYMENT_1'", + String.class)).isEqualTo("TRANSFERRED"); } @Test @@ -178,6 +190,42 @@ class PaymentControllerTest { BigDecimal.class, PAYER_WALLET)).isEqualByComparingTo("0.00"); } + @Test + void transferredCoinsCanBeReservedByThePayeeInALaterPayment() throws Exception { + String firstId = createSignedPayment(PAYER_WALLET, PAYEE_WALLET, payer()); + mockMvc.perform(post("/api/v1/payments/{id}/payer-bank-process", firstId).with(payer())) + .andExpect(status().isOk()); + mockMvc.perform(post("/api/v1/payments/{id}/central-bank-settle", firstId).with(payer())) + .andExpect(status().isOk()); + mockMvc.perform(post("/api/v1/payments/{id}/payee-bank-credit", firstId).with(payer())) + .andExpect(status().isOk()); + + String secondId = createSignedPayment(PAYEE_WALLET, PAYER_WALLET, payee()); + mockMvc.perform(post("/api/v1/payments/{id}/payer-bank-process", secondId).with(payee())) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.data.coinCount").value(2)); + + assertThat(jdbc.queryForObject("SELECT status FROM central_bank_currency_ownership " + + "WHERE currency_id='DC_PAYMENT_1'", String.class)).isEqualTo("LOCKED"); + assertThat(jdbc.queryForObject("SELECT lock_business_type FROM central_bank_currency_ownership " + + "WHERE currency_id='DC_PAYMENT_1'", String.class)).isEqualTo("PAYMENT"); + assertThat(jdbc.queryForObject("SELECT lock_business_id FROM central_bank_currency_ownership " + + "WHERE currency_id='DC_PAYMENT_1'", String.class)).isEqualTo(secondId); + } + + private String createSignedPayment(String payerWalletId, String payeeWalletId, + RequestPostProcessor actor) throws Exception { + String created = mockMvc.perform(post("/api/v1/payments").with(actor).contentType(MediaType.APPLICATION_JSON) + .content("{\"payerWalletId\":\"" + payerWalletId + "\",\"payeeWalletId\":\"" + payeeWalletId + + "\",\"amount\":200.00,\"note\":\"reuse coins\"}")) + .andExpect(status().isOk()) + .andReturn().getResponse().getContentAsString(); + String id = JsonPath.read(created, "$.data.id"); + mockMvc.perform(post("/api/v1/payments/{id}/sign", id).with(actor)) + .andExpect(status().isOk()); + return id; + } + private void insertUser(String id, String account, String schoolId, String classId) { jdbc.update("INSERT INTO sys_user (user_id,student_id,password,user_name,class_id,class_name,school_id,school_name,role_id,is_deleted,zy_user_id) " + "VALUES (?,?,?,?,?,'支付测试班',?,'延安大学',4,0,?)", id, account, "unused", account, classId, schoolId, account); @@ -194,9 +242,10 @@ class PaymentControllerTest { jdbc.update("INSERT INTO wallet_bank_binding (wallet_id,bank_account_id,status,bound_at) VALUES (?,?,'BOUND',CURRENT_TIMESTAMP)", walletId, accountId); } - private void insertPayerContract() { + private void insertContract(String walletId, String contractId) { jdbc.update("INSERT INTO wallet_contract (contract_id,wallet_id,wallet_type,single_payment_limit,daily_payment_limit,annual_payment_limit,balance_limit,valid_until,original_text,digest,status,daily_used_amount,daily_counter_date,annual_used_amount,annual_counter_year,created_at,updated_at) " + - "VALUES ('CONTRACT_991200',?,'TYPE_II',50000,100000,500000,500000,NULL,'CONTRACT','DIGEST','ACTIVE',0,CURRENT_DATE,0,2026,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)", PAYER_WALLET); + "VALUES (?,?,'TYPE_II',50000,100000,500000,500000,NULL,'CONTRACT','DIGEST','ACTIVE',0,CURRENT_DATE,0,2026,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)", + contractId, walletId); } private void cleanWalletState() { @@ -224,4 +273,8 @@ class PaymentControllerTest { private RequestPostProcessor payer() { return jwt().jwt(jwt -> jwt.subject(PAYER_USER).claim("userId", PAYER_USER).claim("preferred_username", "payment-a")); } + + private RequestPostProcessor payee() { + return jwt().jwt(jwt -> jwt.subject(PAYEE_USER).claim("userId", PAYEE_USER).claim("preferred_username", "payment-b")); + } } diff --git a/src/test/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipConcurrencyTest.java b/src/test/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipConcurrencyTest.java new file mode 100644 index 0000000..ae028ab --- /dev/null +++ b/src/test/java/com/yau/digitalrmb/shared/currency/CurrencyOwnershipConcurrencyTest.java @@ -0,0 +1,77 @@ +package com.yau.digitalrmb.shared.currency; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest(properties = "file.path=target/test-files") +@ActiveProfiles("test") +class CurrencyOwnershipConcurrencyTest { + private static final String CURRENCY_ID = "DC_OWNERSHIP_CONCURRENCY"; + + @Autowired private JdbcTemplate jdbc; + @Autowired private TransactionTemplate transactions; + @Autowired private CurrencyOwnershipLockRepository ownershipLocks; + + @BeforeEach + void seedCurrencyOwnership() { + jdbc.update("INSERT INTO commercial_bank_currency (currency_id,source_currency_id,source_batch_id," + + "issuance_request_id,bank_code,organization_id,denomination,currency,complete_currency," + + "issuance_ownership_original_text,issuance_ownership_signature,status,received_at," + + "received_by_user_id,received_by) VALUES (?,998801,998800,?,'BK_TEST','ORG_TEST',100,'DC'," + + "'COMPLETE','OWNERSHIP','SIGNATURE','ISSUED',CURRENT_TIMESTAMP,'test-user','test')", + CURRENCY_ID, "99880000-0000-0000-0000-000000000001"); + jdbc.update("INSERT INTO central_bank_currency_ownership " + + "(currency_id,owner_type,owner_id,status,lock_business_type,lock_business_id,last_transaction_id,updated_at) " + + "VALUES (?,'WALLET','WALLET_TEST','AVAILABLE',NULL,NULL,NULL,CURRENT_TIMESTAMP)", CURRENCY_ID); + } + + @AfterEach + void cleanCurrencyOwnership() { + jdbc.update("DELETE FROM central_bank_currency_ownership WHERE currency_id=?", CURRENCY_ID); + jdbc.update("DELETE FROM commercial_bank_currency WHERE currency_id=?", CURRENCY_ID); + } + + @Test + void allowsExactlyOneConcurrentBusinessToLockAnAvailableOwnershipRow() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + try { + Future first = executor.submit(() -> lockAfterBarrier("PAYMENT_A", ready, start)); + Future second = executor.submit(() -> lockAfterBarrier("PAYMENT_B", ready, start)); + ready.await(); + start.countDown(); + + List results = Arrays.asList(first.get(), second.get()); + assertThat(results).containsExactlyInAnyOrder(true, false); + assertThat(jdbc.queryForObject("SELECT status FROM central_bank_currency_ownership WHERE currency_id=?", + String.class, CURRENCY_ID)).isEqualTo("LOCKED"); + assertThat(jdbc.queryForObject("SELECT lock_business_id FROM central_bank_currency_ownership WHERE currency_id=?", + String.class, CURRENCY_ID)).isIn("PAYMENT_A", "PAYMENT_B"); + } finally { + executor.shutdownNow(); + } + } + + private Boolean lockAfterBarrier(String businessId, CountDownLatch ready, CountDownLatch start) throws Exception { + ready.countDown(); + start.await(); + return transactions.execute(status -> ownershipLocks.tryLock( + CURRENCY_ID, "WALLET", "WALLET_TEST", "PAYMENT", businessId)); + } +} diff --git a/src/test/resources/schema.sql b/src/test/resources/schema.sql index e16cdb1..d1cbefe 100644 --- a/src/test/resources/schema.sql +++ b/src/test/resources/schema.sql @@ -571,9 +571,11 @@ CREATE TABLE IF NOT EXISTS currency_ownership_transfer ( ); CREATE TABLE IF NOT EXISTS central_bank_currency_ownership ( currency_id VARCHAR(96) PRIMARY KEY, owner_type VARCHAR(16) NOT NULL, owner_id VARCHAR(128) NOT NULL, - status VARCHAR(16) NOT NULL, last_transaction_id VARCHAR(128) NULL, updated_at TIMESTAMP NOT NULL, + status VARCHAR(16) NOT NULL, lock_business_type VARCHAR(32) NULL, lock_business_id VARCHAR(96) NULL, + last_transaction_id VARCHAR(128) NULL, updated_at TIMESTAMP NOT NULL, CONSTRAINT fk_currency_ownership_currency FOREIGN KEY(currency_id) REFERENCES commercial_bank_currency(currency_id), - CONSTRAINT fk_currency_ownership_transaction FOREIGN KEY(last_transaction_id) REFERENCES currency_ownership_transfer(transaction_id) + INDEX idx_currency_ownership_available(owner_type,owner_id,status), + INDEX idx_currency_ownership_lock_business(lock_business_type,lock_business_id) ); CREATE TABLE IF NOT EXISTS wallet_balance_ledger ( ledger_no VARCHAR(128) PRIMARY KEY, wallet_id VARCHAR(96) NOT NULL, exchange_id CHAR(36) NOT NULL UNIQUE, @@ -742,7 +744,6 @@ CREATE TABLE IF NOT EXISTS payment_coin_reservation ( reserved_at TIMESTAMP NOT NULL, transferred_at TIMESTAMP NULL, PRIMARY KEY (payment_id, currency_id), - UNIQUE KEY uk_payment_reserved_currency (currency_id), FOREIGN KEY (payment_id) REFERENCES payment_order(id), FOREIGN KEY (currency_id) REFERENCES commercial_bank_currency(currency_id) );