From 9a9ba6b803dfc0ab2516a4327c03c559f7f87dea Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Fri, 3 Jul 2026 09:32:25 +0300 Subject: [PATCH] LP: added last rejecting date calculating --- credit-tracker.properties | 5 ++ .../settings/CreditTrackerSettings.java | 2 + .../settings/RejectedDateUpdaterSettings.java | 9 +++ .../entities/indic/RuApplication.java | 13 ++++ .../credit_tracker/facade/SignalFacade.java | 4 +- .../mapper/indic/IndicFlagsMapper.java | 7 +- .../mapper/signal/LegalsMapMapper.java | 5 ++ .../nbch/credit_tracker/model/FlagData.java | 3 + .../service/indic/FlagsService.java | 6 ++ .../common/UpdatingLastRejectDateService.java | 54 ++++++++++++++ .../service/signal/PackageRecordsService.java | 23 +++++- .../service/task/impl/FlagsEnricher.java | 3 +- .../task/impl/LastRejectedDateEnricher.java | 74 +++++++++++++++++++ .../mybatis/mapper/indic/FlagsMapper.xml | 17 +++++ .../mybatis/mapper/signal/LegalsMapMapper.xml | 41 +++++++++- 15 files changed, 256 insertions(+), 10 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/config/settings/RejectedDateUpdaterSettings.java create mode 100644 src/main/java/ru/nbch/credit_tracker/entities/indic/RuApplication.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingLastRejectDateService.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/LastRejectedDateEnricher.java diff --git a/credit-tracker.properties b/credit-tracker.properties index 1bb5e9e..eacf555 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -122,6 +122,11 @@ ct.pipeline.name-updater-queue-capacity=${PIPELINE_NAME_UPDATER_QUEUE_CAPACITY:2 ct.flag-updater.concurrency=${FLAG_UPDATER_CONCURRENCY:10} ct.flag-updater.queue-capacity=${FLAG_UPDATER_QUEUE_CAPACITY:100} +#select and calc last rejected date +ct.update-rejected-date-cron-expression=${UPDATE_REJECTED_DATE_CRON_EXPRESSION:0 25 18 * * *} +ct.rejected-date-updater.concurrency=${REJECTED_DATE_UPDATER_CONCURRENCY:10} +ct.rejected-date-updater.queue-capacity=${REJECTED_DATE_UPDATER_QUEUE_CAPACITY:100} + #Backoff settings ct.pipeline.error-attempt-count=${PIPELINE_ERROR_ATTEMPT_COUNT:5} ct.pipeline.error-attempt-delay=${PIPELINE_ERROR_ATTEMPT_DELAY:2S} diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java index 0ef27f5..e3fa7dc 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java @@ -30,6 +30,8 @@ public class CreditTrackerSettings { @NestedConfigurationProperty private FlagUpdaterSettings flagUpdater; @NestedConfigurationProperty + private RejectedDateUpdaterSettings rejectedDateUpdater; + @NestedConfigurationProperty private KafkaConsumerSettings kafkaConsumerSettings; private Long maxXmlFileSize; diff --git a/src/main/java/ru/nbch/credit_tracker/config/settings/RejectedDateUpdaterSettings.java b/src/main/java/ru/nbch/credit_tracker/config/settings/RejectedDateUpdaterSettings.java new file mode 100644 index 0000000..1fc98d6 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/config/settings/RejectedDateUpdaterSettings.java @@ -0,0 +1,9 @@ +package ru.nbch.credit_tracker.config.settings; + +import org.springframework.boot.context.properties.bind.DefaultValue; + +public record RejectedDateUpdaterSettings( + @DefaultValue("10") Integer concurrency, + @DefaultValue("100") Integer queueCapacity +) { +} diff --git a/src/main/java/ru/nbch/credit_tracker/entities/indic/RuApplication.java b/src/main/java/ru/nbch/credit_tracker/entities/indic/RuApplication.java new file mode 100644 index 0000000..dae80a6 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/entities/indic/RuApplication.java @@ -0,0 +1,13 @@ +package ru.nbch.credit_tracker.entities.indic; + +import lombok.Getter; +import lombok.Setter; + +import java.time.LocalDate; + +@Getter +@Setter +public class RuApplication { + private Integer fid; + private LocalDate rejectedDate; +} diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java index aeea8f9..d4e7e1e 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -153,13 +153,13 @@ public class SignalFacade { log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name()); if (subjectType == 1) { - Pipeline pipeline = npPipelinePackageFabric.build(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT)); + Pipeline pipeline = npPipelinePackageFabric.build(new SetupParam(packageId, 1, PipelineMode.DEFAULT)); pipelineOrchestrator.startOrDelay(() -> { saver.readAndSaveNpPackage(packageId, parsedFilePath); CtFileUtils.deleteQuietly(parsedFilePath); }, pipeline, packageId); } else { - Pipeline pipeline = lpPipelinePackageFabric.build(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT)); + Pipeline pipeline = lpPipelinePackageFabric.build(new SetupParam(packageId, 2, PipelineMode.DEFAULT)); pipelineOrchestrator.startOrDelay(() -> { saver.readAndSaveLpPackage(packageId, parsedFilePath); CtFileUtils.deleteQuietly(parsedFilePath); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/indic/IndicFlagsMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/indic/IndicFlagsMapper.java index 9a68aea..ff9d016 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/indic/IndicFlagsMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/indic/IndicFlagsMapper.java @@ -1,10 +1,12 @@ package ru.nbch.credit_tracker.mapper.indic; -import java.time.LocalDate; -import java.util.List; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import ru.nbch.credit_tracker.entities.indic.PaymentsFlag; +import ru.nbch.credit_tracker.entities.indic.RuApplication; + +import java.time.LocalDate; +import java.util.List; @Mapper public interface IndicFlagsMapper { @@ -16,4 +18,5 @@ public interface IndicFlagsMapper { List findFlag90ByRuAllArrears(@Param("fids") List fids, @Param("reportedDate") LocalDate reportedDate); + List findLastRejecteDates(@Param("fids") List fids); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java index 05cfab5..9a22a05 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java @@ -15,10 +15,15 @@ public interface LegalsMapMapper { void updateFlagCachedData(@Param("json") String json); + void updateLastRejectedDateCache(@Param("json") String json); + List findEligibleForFlagsUpdate(@Param("lastId") Long lastId, @Param("interval") Integer interval, @Param("batchSize") Integer batchSize); + List findEligibleForRejectedDatesToUpdate(@Param("lastId") Long lastId, + @Param("batchSize") Integer batchSize); + List findEligibleForFlagsInsert(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); diff --git a/src/main/java/ru/nbch/credit_tracker/model/FlagData.java b/src/main/java/ru/nbch/credit_tracker/model/FlagData.java index d8f7a5a..b726e30 100644 --- a/src/main/java/ru/nbch/credit_tracker/model/FlagData.java +++ b/src/main/java/ru/nbch/credit_tracker/model/FlagData.java @@ -5,6 +5,8 @@ import lombok.Builder; import lombok.Getter; import lombok.Setter; +import java.time.LocalDate; + @Getter @Setter @Builder @@ -14,4 +16,5 @@ public class FlagData { private Long regNum; private Integer flag_debt5000; private Integer flag_90_12; + private LocalDate lastRejectedDate; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService.java index e699a6f..3455b6f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/FlagsService.java @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.service.indic; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.entities.indic.PaymentsFlag; +import ru.nbch.credit_tracker.entities.indic.RuApplication; import ru.nbch.credit_tracker.mapper.indic.IndicFlagsMapper; import java.time.LocalDate; @@ -38,4 +39,9 @@ public class FlagsService { return flagsMapper.findFlag90ByRuAllArrears(fids, date).stream() .collect(Collectors.toMap(PaymentsFlag::getFid, PaymentsFlag::getFlag)); } + + public Map findLastRejecteDates(List fids) { + return flagsMapper.findLastRejecteDates(fids).stream() + .collect(Collectors.toMap(RuApplication::getFid, RuApplication::getRejectedDate)); + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingLastRejectDateService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingLastRejectDateService.java new file mode 100644 index 0000000..2f6793e --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingLastRejectDateService.java @@ -0,0 +1,54 @@ +package ru.nbch.credit_tracker.service.mode.common; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.entities.app.LegalsMap; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; +import ru.nbch.credit_tracker.service.task.impl.LastRejectedDateEnricher; + +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +@Slf4j +@Service +public class UpdatingLastRejectDateService { + private final static int BATCH_SIZE = 1000; + private final LegalsMapMapper legalsMapMapper; + private final LastRejectedDateEnricher lastRejectedDateEnricher; + private final AtomicBoolean updateIsRunning = new AtomicBoolean(false); + + public UpdatingLastRejectDateService(LegalsMapMapper legalsMapMapper, + LastRejectedDateEnricher lastRejectedDateEnricher) { + this.legalsMapMapper = legalsMapMapper; + this.lastRejectedDateEnricher = lastRejectedDateEnricher; + } + + @Scheduled(cron = "${ct.update-rejected-date-cron-expression}") + private void updateFlags() { + if (!updateIsRunning.compareAndSet(false, true)) { + log.error("update flags processor already running... skipping scheduled update"); + return; + } + try { + log.info("start update flags processor"); + long lastLegalId = 0L; + int batchNumber = 1; + List legalsMaps; + while (true) { + legalsMaps = legalsMapMapper.findEligibleForRejectedDatesToUpdate(lastLegalId, BATCH_SIZE); + log.trace("update batch {}, signals_legals_map records loaded: {}.", batchNumber, legalsMaps.size()); + if (legalsMaps.isEmpty()) { + break; + } + + lastRejectedDateEnricher.accept(legalsMaps); + lastLegalId = legalsMaps.getLast().getId(); + log.trace("update batch {}, lastLegalId {}", batchNumber++, lastLegalId); + } + log.info("end update flags processor"); + } finally { + updateIsRunning.set(false); + } + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java index 92a4258..3c558b8 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java @@ -158,10 +158,10 @@ public class PackageRecordsService { try { String json = entityToJsonMapper.writeValueAsString(flagData.stream() .filter(Objects::nonNull) - .map(phoneMap -> new HashMap() {{ - put("inn", phoneMap.getInn()); - put("reg_num", phoneMap.getRegNum()); - put("flag_90_12", phoneMap.getFlag_90_12()); + .map(flag -> new HashMap() {{ + put("inn", flag.getInn()); + put("reg_num", flag.getRegNum()); + put("flag_90_12", flag.getFlag_90_12()); }}).toList()); legalsMapMapper.updateFlagCachedData(json); } catch (JsonProcessingException e) { @@ -169,6 +169,21 @@ public class PackageRecordsService { } } + public void updateLpLastRejectedDateCache(List flagData) { + try { + String json = entityToJsonMapper.writeValueAsString(flagData.stream() + .filter(Objects::nonNull) + .map(flag -> new HashMap() {{ + put("inn", flag.getInn()); + put("reg_num", flag.getRegNum()); + put("last_rejected_date", flag.getLastRejectedDate()); + }}).toList()); + legalsMapMapper.updateLastRejectedDateCache(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public PhonesMap findCachedByPhone(Long phone) { return phonesMapMapper.findByPhone(phone); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java index 2c8ed1f..b119764 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricher.java @@ -46,7 +46,8 @@ public class FlagsEnricher { } public void process(List phonesMaps, List legalsMaps) { - log.debug("FlagsEnricher handle: size: {}", phonesMaps.size()); + log.debug("FlagsEnricher handle: np size: {}", phonesMaps.size()); + log.debug("FlagsEnricher handle: lp size: {}", legalsMaps.size()); List noFlagDataFids = phonesMaps.stream() .map(PhonesMap::getFid) diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LastRejectedDateEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LastRejectedDateEnricher.java new file mode 100644 index 0000000..6406a1c --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LastRejectedDateEnricher.java @@ -0,0 +1,74 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.LegalsMap; +import ru.nbch.credit_tracker.model.FlagData; +import ru.nbch.credit_tracker.service.indic.FlagsService; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; + +import java.time.LocalDate; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +@Slf4j +@Service +public class LastRejectedDateEnricher { + + private final FlagsService flagsService; + private final PackageRecordsService packageRecordsService; + private final ExecutorService ioPool; + + public LastRejectedDateEnricher(FlagsService flagsService, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { + this.flagsService = flagsService; + this.packageRecordsService = packageRecordsService; + this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getRejectedDateUpdater().concurrency(), + creditTrackerSettings.getRejectedDateUpdater().concurrency(), + 0L, TimeUnit.MINUTES, + new LinkedBlockingQueue<>(creditTrackerSettings.getRejectedDateUpdater().queueCapacity()), + r -> { + Thread t = new Thread(r); + t.setName("last-rejected-fate-enricher-%d".formatted(t.threadId())); + return t; + }, + new ThreadPoolExecutor.CallerRunsPolicy() + ); + } + + private void process(List legalsMaps) { + log.debug("FlagsEnricher handle: lp size: {}", legalsMaps.size()); + + List ligalsNoRejecedDateFids = legalsMaps.stream() + .map(LegalsMap::getFid) + .toList(); + + Map fidDatesMap = flagsService.findLastRejecteDates(ligalsNoRejecedDateFids); + List lpFlagCacheOnUpdate = legalsMaps.stream() + .flatMap(legalMap -> { + Integer fid = legalMap.getFid(); + LocalDate lastRejectedDate = fidDatesMap.get(fid); + FlagData flagData = FlagData.builder() + .lastRejectedDate(lastRejectedDate) + .inn(legalMap.getInn()) + .regNum(legalMap.getRegNum()) + .build(); + return Stream.of(flagData); + }).toList(); + + packageRecordsService.updateLpLastRejectedDateCache(lpFlagCacheOnUpdate); + } + + public void accept(List legalsMaps) { + ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool; + log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size()); + this.ioPool.submit(() -> process(legalsMaps)); + } +} diff --git a/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml b/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml index 1c1a021..11a945c 100644 --- a/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml +++ b/src/main/resources/mybatis/mapper/indic/FlagsMapper.xml @@ -10,6 +10,11 @@ + + + + + + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml index cfd28b3..14200fd 100644 --- a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml @@ -97,8 +97,18 @@ limit #{batchSize} + + - WITH src AS (SELECT DISTINCT ON (inn, reg_num) inn, + WITH src AS (SELECT DISTINCT ON (inn, reg_num) + inn, reg_num, flag_90_12 FROM json_to_recordset(#{json}::json) AS src( @@ -126,6 +136,35 @@ AND dst.id IN (SELECT id FROM to_lock); + + WITH src AS (SELECT DISTINCT ON (inn, reg_num) + inn, + reg_num, + last_rejected_date + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num text, + last_rejected_date timestamp + ) + ORDER BY inn, reg_num), + to_lock AS (SELECT dst.id + FROM signals_legals_map dst + JOIN src ON src.inn = dst.inn AND src.reg_num = dst.reg_num + ORDER BY dst.inn, dst.reg_num + FOR UPDATE) + + UPDATE signals_legals_map dst + SET last_rejected_date = src.last_rejected_date + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num text, + last_rejected_date timestamp + ) + WHERE dst.inn = src.inn + AND dst.reg_num = src.reg_num + AND dst.id IN (SELECT id FROM to_lock); + +