refactor: lock currency through current ownership

chore/issuance-comments-zh
chenyuan 2 months ago
parent 4cacb89813
commit c5e5256395

@ -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<CoinCandidate> 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<BigDecimal> 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) {

@ -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);
}
}

@ -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<PaymentCoin> selectAndLockCoins(PaymentOrder order, PaymentActor actor) {
List<CoinCandidate> 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<PaymentCoin> results = new ArrayList<PaymentCoin>();
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);

@ -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;
}
}

@ -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;

@ -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)
);

@ -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"));
}
}

@ -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<Boolean> first = executor.submit(() -> lockAfterBarrier("PAYMENT_A", ready, start));
Future<Boolean> second = executor.submit(() -> lockAfterBarrier("PAYMENT_B", ready, start));
ready.await();
start.countDown();
List<Boolean> 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));
}
}

@ -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)
);

Loading…
Cancel
Save