From b19ec97a5de9419e0d599346a38ed494102fca63 Mon Sep 17 00:00:00 2001 From: ialbert Date: Tue, 20 Jun 2023 19:46:24 +0300 Subject: [PATCH] http://jira.mfd.msk:8088/browse/CLS-357 SDF06 executor --- .../clearing/config/ValidationConfig.java | 12 +- .../spcex/clearing/error/ClearingError.java | 2 + .../service/executors/Sdf06Executor.java | 161 ++++++++++++------ .../validation/Sdf06NewValidationRule.java | 65 +++++-- .../service/validation/ValidationStored.java | 2 +- 5 files changed, 167 insertions(+), 75 deletions(-) diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java index 39b474bf3..309055d9a 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/config/ValidationConfig.java @@ -208,12 +208,14 @@ public class ValidationConfig { context.setValidatedObject(sDf06); context.addImdg(IMDGDistributedNames.Map_Account, imdgAccount); context.addImdg(IMDGDistributedNames.Map_CompanySymbols, imdgCompanySymbols); - context.addImdg(IMDGDistributedNames.Map_Registry, imdgCompany); + context.addImdg(IMDGDistributedNames.Map_Company, imdgCompany); + context.addImdg(IMDGDistributedNames.Map_Registry, imdgRegistry); + context.addImdg(IMDGDistributedNames.Map_Session, imdgSession); return new ValidatorImpl<>(context, - Sdf06NewValidationRule.CompanySymbolPresent, - Sdf06NewValidationRule.AccountPresent -// Sdf06NewValidationRule.MarketIsMkr -// Sdf06NewValidationRule.Balance + Sdf06NewValidationRule.CompanySymbolAndCompanyPresent, + Sdf06NewValidationRule.AccountPresent, + Sdf06NewValidationRule.Session, + Sdf06NewValidationRule.Balance ); }; diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java index 9d23610b2..fa6683504 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/error/ClearingError.java @@ -27,6 +27,8 @@ public enum ClearingError implements IErrorEnumId { AccountNotFoundB(5217L), MarketIsNotMkr(5217L), BalanceInsufficient(5222L), + + ActiveSessionIsPresent(-1L), ; private final Long id; diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java index ea170e1a3..d9c164f9b 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/executors/Sdf06Executor.java @@ -15,9 +15,11 @@ import ru.clearing.classes.statics.data.registry.TradingClearingRegistry; import ru.clearing.classes.statics.data.sdf.SDf06; import ru.clearing.classes.statics.data.sdf.SDf07; import ru.clearing.classes.statics.data.statement.Statement; +import ru.spcex.clearing.error.ClearingError; import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.balance.ExportToFileRequest; import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest; import ru.spcex.clearing.platform.messaging.domain.cud.clearing.AssetOperationRequest; import ru.spcex.clearing.platform.messaging.domain.cud.gateway.AssetOperationApprovalRequest; @@ -26,6 +28,7 @@ import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.clearing.service.validation.ValidationStored; import ru.spcex.platform.enumeration.*; import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.IMessageResolver; @@ -55,12 +58,24 @@ public class Sdf06Executor { private final Imdg companyImdg; private final Imdg sdf06Imdg; private final Imdg sdf07Imdg; + private final ImdgId idGenerator; private final IMessageResolver messageResolver; private final Function sDf06Validator; private final KafkaSender kafkaSender; + private final static BigDecimal successResult = BigDecimal.ZERO; + //ошибка проверок + private final static BigDecimal errorResult2 = new BigDecimal("3"); + //ошибка проверок баланса или наличия сессии - в этом сценарии session создается + private final static BigDecimal errorResult1 = new BigDecimal("1"); + //отказ от gateway + private final static BigDecimal errorResult3 = new BigDecimal("3"); + + //not thread safe: used in EventReceiver single thread executor + //переменная контроля за тем что от гейтвея пришел ответ именно на наш запрос private Long sdf06GroupId; + private Long sdf07GroupId; @Autowired public Sdf06Executor(ImdgProvider imdgProvider, @@ -74,6 +89,7 @@ public class Sdf06Executor { this.depoAccountImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_DepoAccount, DepoAccount.class); this.tradingClearingRegistryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_TradingClearingRegistry, TradingClearingRegistry.class); this.companyImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Company, Company.class); + this.idGenerator = imdgProvider.getImdgIdGenerator(); this.messageResolver = messageResolver; this.sDf06Validator = sDf06Validator; this.kafkaSender = kafkaSender; @@ -87,43 +103,73 @@ public class Sdf06Executor { Collection sdfs = sdfImdg.getCollectionObjectsByFieldValues(Map.of( "generationId", groupId )); + Instant now = Instant.now(); + Long sdf07GroupId = idGenerator.nextId(); + if (sdf06GroupId != null) { + log.warn("currently awaiting gateway response for groupId: {}. skipping groupId {}", sdf06GroupId, groupId); + sdfs.forEach(sdf06 -> { + SDf07 errorSdf07 = createSdf07(sdf06, now, errorResult2); + errorSdf07.setGenerationId(sdf07GroupId); + sdf07Imdg.insert(errorSdf07); + }); + sendToExporter(groupId); + return; + } Collection requests = new ArrayList<>(); + boolean sdf07WasCreated = false; for (SDf06 sDf06 : sdfs) { IValidator validator = sDf06Validator.apply(sDf06); Optional err = validator.tillFirstError(); if (err.isPresent()) { - //todo если ошибка про нехватку баланса - делаем statement с ошибочным статусом - //todo 2 делаем ошибочный SDF07 - log.debug("Validation error: {}", messageResolver.resolve(err.get())); - createErrorSdf07(null, sDf06.getId(), groupId, null); + log.debug("sdf06.id={} validation error: {}", sDf06.getId(), messageResolver.resolve(err.get())); + SDf07 errorSdf07; + BigDecimal errorResult; + Statement stmt = null; + if (needToCreateStatement(err.get())) { + errorResult = errorResult1; + stmt = createStatementBySdf06(sDf06, + ((Company) validator.getStored(ValidationStored.Sdf06Company)).getId(), + validator.getStored(ValidationStored.Sdf06Account)); + } else { + errorResult = errorResult2; + } + errorSdf07 = createSdf07(sDf06, now, errorResult); + errorSdf07.setGenerationId(sdf07GroupId); + sdf07Imdg.insert(errorSdf07); + sdf07WasCreated = true; + if (stmt != null) { + stmt.setOutSDfId(errorSdf07.getId()); + statementImdg.insert(stmt); + log.debug("Statement created: {}", stmt.getId()); + } continue; } - Long companyId = validator.getStored(ValidationStored.Sdf06CompanyId); + Company company = validator.getStored(ValidationStored.Sdf06Company); Account account = validator.getStored(ValidationStored.Sdf06Account); - Company company = companyImdg.getSingleObjectByID(companyId); TradingClearingRegistry tcr = tradingClearingRegistryImdg.getSingleObjectByFieldValues( Map.of("moneyAccountId", account.getId()) ); - if (company == null || tcr == null) { - log.error("Company.id {} {}, TCR by account.id {} {} ", companyId, - company != null ? "present" : "not found", - account.getId(), - tcr != null ? "present" : "not found"); - continue; - } //проверка существует ли statement пока убрал - Statement stmt = createStatementBySdf06(sDf06, companyId, account); + Statement stmt = createStatementBySdf06(sDf06, company.getId(), account); statementImdg.insert(stmt); log.debug("Statement created: {}", stmt.getId()); requests.add(requestFromStatement(stmt, company.getTradingCode(), tcr.getCode())); } if (requests.size() > 0) { - sdf06GroupId = groupId; + this.sdf06GroupId = groupId; + this.sdf07GroupId = sdf07GroupId; kafkaSender.sendRequestToQueue(Consts.ASSET_OPERATION, requests); + } else if (sdf07WasCreated) { + sendToExporter(groupId); } } + private boolean needToCreateStatement(EnumMessage err) { + return ClearingError.ActiveSessionIsPresent.equals(err.getSubject()) || ClearingError.BalanceInsufficient.equals(err.getSubject()); + } + public void processGatewayResponse(BaseRequest req) { + Instant now = Instant.now(); for (SingleAssetResponse gatewayMsg : req.getRequestPayload().getApprovals()) { //получаем запрос для текущей группы sdf06 //находим группу @@ -153,55 +199,65 @@ public class Sdf06Executor { Instant updatedTime = Instant.now(); if (gatewayMsg.isApproved()) { - SDf07 sdf07 = createSuccessSdf07(stmt, updatedTime); + log.trace("statement.id={}, sdf07.id={}, sdf06.id={} executed", + statementId, + sdf06.getGenerationId(), + sdf06.getId()); + SDf07 sdf07 = createSdf07(sdf06, updatedTime, successResult); + sdf07.setGenerationId(sdf07GroupId); sdf07Imdg.insert(sdf07); stmt.setOperationStatus(OperationStatus.Executed.getKey()); stmt.setUpdated(updatedTime); stmt.setOutSDfId(sdf07.getId()); statementImdg.update(stmt); } else { - SDf07 sdf07 = createErrorSdf07(stmt, stmt.getInSDfId(), groupId, null); + log.trace("statement.id={}, sdf07.id={}, sdf06.id={} rejected (by gateway answer)", + statementId, + sdf06.getGenerationId(), + sdf06.getId()); + SDf07 sdf07 = createSdf07(sdf06, now, errorResult3); + sdf07.setGenerationId(sdf07GroupId); sdf07Imdg.insert(sdf07); stmt.setOperationStatus(OperationStatus.Rejected.getKey()); stmt.setUpdated(updatedTime); stmt.setOutSDfId(sdf07.getId()); statementImdg.update(stmt); } - sdf06GroupId = null; - log.debug("All gateway responses received for groupId {}", groupId); } + sendToExporter(sdf07GroupId); + log.debug("All gateway responses received for SDF06 groupId {}. SDF07 groupId {}", sdf06GroupId, sdf07GroupId); + clearContext(); } - private SDf07 createSuccessSdf07(Statement stmt, Instant time) { + private void clearContext() { + this.sdf06GroupId = null; + this.sdf07GroupId = null; + } + + private void sendToExporter(Long generationId) { + ExportToFileRequest exportRequest = new ExportToFileRequest(); + exportRequest.setSdfGroupId(generationId); + exportRequest.setNameOfTable("DF-07"); + kafkaSender.sendRequestToQueue(Consts.EXPORT_PROCESS, exportRequest); + } + + private SDf07 createSdf07(SDf06 sdf06, Instant time, BigDecimal result) { SDf07 sDf07 = new SDf07(); - sDf07.setInSDfId(stmt.getInSDfId()); - sDf07.setResult(BigDecimal.ZERO); - sDf07.setSum(stmt.getAmount()); + sDf07.setInSDfId(sdf06.getId()); sDf07.setGenerationTime(time); + sDf07.setAccount(sdf06.getAccount()); + sDf07.setSum(sdf06.getSum()); + sDf07.setMarket(sdf06.getMarket()); + sDf07.setType(sdf06.getType()); + sDf07.setDeal(sdf06.getDeal()); + sDf07.setClientN(sdf06.getClientN()); + sDf07.setInn(sdf06.getInn()); + sDf07.setBic(sdf06.getBic()); + sDf07.setNumber(sdf06.getNumber()); + sDf07.setResult(result); return sDf07; } - private final static BigDecimal errorResult = new BigDecimal("3"); - private SDf07 createErrorSdf07(Statement stmt, Long sdf06Id, Long sdf06GroupId, EnumMessage err) { - SDf07 sDf07 = new SDf07(); - sDf07.setResult(errorResult); - sDf07.setInSDfId(sdf06Id); - //пока использую такой же groupId как для 06. - //при необходимости определить вместе с gatewayWaitInfo - //(т.к. будут создаваться ошибочные sdf07 в при ошибках валидации Statement, там должен быть такой же generationId) - sDf07.setGenerationId(sdf06GroupId); - if (err != null) { - //fixme нет полей для ошибки - } - if (stmt != null) { - sDf07.setAccount(stmt.getAccount()); - sDf07.setSpec(stmt.getComment()); - } - return sDf07; - } - - - private Statement createStatementBySdf06(SDf06 sdf06, Long companyId, Account account) { Statement stmt = new Statement(); stmt.setAddresseeId(companyId); @@ -210,8 +266,13 @@ public class Sdf06Executor { stmt.setComment(sdf06.getSpec()); stmt.setAccountId(account.getId()); stmt.setAccount(account.getAccount()); - stmt.setInOutDirection(InOutDirection.in.getKey()); - stmt.setAmount(sdf06.getSum()); + BigDecimal amount = BigDecimalUtil.safeBD(sdf06.getSum()); + if (amount.compareTo(BigDecimal.ZERO) >= 0) { + stmt.setInOutDirection(InOutDirection.in.getKey()); + } else { + stmt.setInOutDirection(InOutDirection.out.getKey()); + } + stmt.setAmount(amount.abs()); stmt.setOperationStatus(OperationStatus.Pending.getKey()); stmt.setInSDfId(sdf06.getId()); stmt.setInOutSDfType(InOutSDfType.type6.getKey()); @@ -223,15 +284,11 @@ public class Sdf06Executor { private AssetOperationRequest requestFromStatement(Statement stmt, String tradingCode, String tcrCode) { AssetOperationRequest req = new AssetOperationRequest(); req.setStatementId(stmt.getId()); - req.setAmount(BigDecimalUtil.safeBD(stmt.getAmount()).abs()); + req.setAmount(stmt.getAmount()); req.setSecuritySymbol(CurrencyCode.RUB.getKey()); //fixme retrieve security symbol from validator req.setTradingCode(tradingCode); req.setCode(tcrCode); - if (BigDecimalUtil.safeBD(stmt.getAmount()).compareTo(BigDecimal.ZERO) >= 0) { - req.setDirection(InOutDirection.in.getKey()); - } else { - req.setDirection(InOutDirection.out.getKey()); - } + req.setDirection(stmt.getInOutDirection()); return req; } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/Sdf06NewValidationRule.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/Sdf06NewValidationRule.java index b63e662dc..5e6917981 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/Sdf06NewValidationRule.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/Sdf06NewValidationRule.java @@ -1,14 +1,23 @@ package ru.spcex.clearing.service.validation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import ru.clearing.classes.statics.data.account.Account; +import ru.clearing.classes.statics.data.company.Company; import ru.clearing.classes.statics.data.company.CompanySymbols; +import ru.clearing.classes.statics.data.misc.Session; import ru.clearing.classes.statics.data.registry.Registry; import ru.clearing.classes.statics.data.sdf.SDf06; import ru.spcex.clearing.error.ClearingError; import ru.spcex.clearing.imdg.IMDGDistributedNames; import ru.spcex.platform.enumeration.AccountType; import ru.spcex.platform.enumeration.CompanySymbol; +import ru.spcex.platform.enumeration.RegistryTradingParams; +import ru.spcex.platform.enumeration.WorkflowStatus; import ru.spcex.platform.imdg.api.Imdg; +import ru.spcex.platform.imdg.api.predicate.ImdgPredicate; +import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; +import ru.spcex.platform.imdg.api.predicate.specific.RegistryCodeSqlBuilder; import ru.spcex.platform.imdg.validation.ImdgValidationContext; import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.number.BigDecimalUtil; @@ -16,15 +25,19 @@ import ru.spcex.platform.utils.text.TextUtil; import ru.spcex.platform.utils.validation.IValidationRule; import java.math.BigDecimal; +import java.util.Collection; import java.util.Map; import java.util.Optional; +import java.util.function.Function; public enum Sdf06NewValidationRule implements IValidationRule> { - CompanySymbolPresent() { + CompanySymbolAndCompanyPresent() { @Override public Optional validate(ImdgValidationContext context) { SDf06 sdf06 = context.getValidatedObject(); Imdg companySymbolsImdg = context.obtainMap(IMDGDistributedNames.Map_CompanySymbols, CompanySymbols.class); + Imdg companyImdg = context.obtainMap(IMDGDistributedNames.Map_Company, Company.class); + CompanySymbols companySymbols = null; if (sdf06.getInn() != null) { //TextUtil.isEmpty companySymbols = companySymbolsImdg.getSingleObjectByFieldValues(Map.of( @@ -38,7 +51,13 @@ public enum Sdf06NewValidationRule implements IValidationRule sessionImdg = context.obtainMap(IMDGDistributedNames.Map_Session, Session.class); + Collection activeSessions = sessionImdg.getCollectionObjectsByFieldValues(Map.of( + "workflowStatus", WorkflowStatus.Active.getKey() + )); + if (activeSessions.size() > 0) { + return of(ClearingError.ActiveSessionIsPresent); } return empty(); } @@ -79,24 +101,33 @@ public enum Sdf06NewValidationRule implements IValidationRule= 0) { return empty(); } + Company company = context.getStoredObject(ValidationStored.Sdf06Company); Imdg registryImdg = context.obtainMap(IMDGDistributedNames.Map_Registry, Registry.class); - Registry rgs = registryImdg.getSingleObjectByFieldValues(Map.of( - "account", sdf06.getAccount(), - "accountType", AccountType.Clrn.getKey())); - if (rgs == null) { - return of(ClearingError.BalanceInsufficient, sdf06.getAccount()); - } - //fixme getFreeBalance??? - if (BigDecimalUtil.safeBD(rgs.getCheckBalance()).compareTo(BigDecimalUtil.safeBD(sdf06.getSum()).abs()) < 0) { - return of(ClearingError.BalanceInsufficient, sdf06.getAccount()); + ImdgPredicateBuilder prdctBldr = registryImdg.predicateBuilder(); + Function prdctByRegistryCode = code -> prdctBldr.and( + prdctBldr.equals("account", sdf06.getAccount()), + prdctBldr.equals("companyId", company.getId()), + prdctBldr.sql(RegistryCodeSqlBuilder.getInstance(code).build())); + Registry rgsAmf = registryImdg.getSingleObjectByPredicate(prdctByRegistryCode.apply(RegistryTradingParams.AM_F)); + Registry rgsDmt = registryImdg.getSingleObjectByPredicate(prdctByRegistryCode.apply(RegistryTradingParams.DM_T)); + log.trace("validating sdf06 id={} account={} companyId={} found registries: AM*F.id={} DM*T.id={}", + sdf06.getId(), sdf06.getAccount(), company.getId(), + rgsAmf == null ? null : rgsAmf.getId(), + rgsDmt == null ? null : rgsDmt.getId() + ); + BigDecimal amfBalance = rgsAmf == null ? BigDecimal.ZERO : BigDecimalUtil.safeBD(rgsAmf.getBalance()); + BigDecimal dmtBalance = rgsDmt == null ? BigDecimal.ZERO : BigDecimalUtil.safeBD(rgsDmt.getBalance()); + BigDecimal stmtAmount = BigDecimalUtil.safeBD(sdf06.getSum()); + if (amfBalance.subtract(dmtBalance).compareTo(stmtAmount.abs()) < 0) { + return of(ClearingError.BalanceInsufficient); } return empty(); } } ; - + private final static Logger log = LoggerFactory.getLogger(Sdf06NewValidationRule.class); @Override public String ruleName() { - return "Sdf08NewValidationRule." + name(); + return "Sdf06NewValidationRule." + name(); } } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java index 56057b25a..530f429f9 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/validation/ValidationStored.java @@ -10,5 +10,5 @@ public enum ValidationStored { Sdf08Company, Sdf08Account, - Sdf06CompanyId, Sdf06Account + Sdf06Company, Sdf06Account }