diff --git a/credit-tracker.properties b/credit-tracker.properties index eacf555..2489b89 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -82,8 +82,10 @@ ct.monitoring-interval=${MONITORING_INTERVAL:5} ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *} ct.new-flags-cron-expression=${NEW_FLAGS_CRON_EXPRESSION:0 25 18 * * *} ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30} -ct.update-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *} -ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30} +ct.update-np-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *} +ct.update-lp-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *} +ct.days-to-update-np-fids=${DAYS_TO_UPDATE_FIDS:30} +ct.days-to-update-lp-fids=${DAYS_TO_UPDATE_FIDS:30} #kafka ct.kafka-consumer-settings.topic=${CONSUMER_TOPIC:my-topic} 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 e3fa7dc..1c91979 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 @@ -39,7 +39,8 @@ public class CreditTrackerSettings { private Long monitoringInterval; private String updateFlagsCronExpression; private Integer daysToUpdateFlags; - private Long daysToUpdateFids; + private Long daysToUpdateNpFids; + private Long daysToUpdateLpFids; private String encoding; private Integer saverSubjectBatchSize; } 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 d4e7e1e..29df35e 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java @@ -29,10 +29,7 @@ import ru.nbch.credit_tracker.model.MonitoringValidationError; import ru.nbch.credit_tracker.model.ValidationResult; import ru.nbch.credit_tracker.service.UserPackageService; import ru.nbch.credit_tracker.service.scoring.AuthorizationService; -import ru.nbch.credit_tracker.service.task.LpPipelinePackageFabric; -import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; -import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; -import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.task.*; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; import ru.nbch.credit_tracker.service.task.impl.Saver; import ru.nbch.credit_tracker.utils.CtFileUtils; @@ -67,7 +64,8 @@ public class SignalFacade { private final NpPipelinePackageFabric npPipelinePackageFabric; private final LpPipelinePackageFabric lpPipelinePackageFabric; private final Saver saver; - private final PipelinePackageOrchestrator pipelineOrchestrator; + private final NpPipelinePackageOrchestrator pipelineOrchestrator; + private final LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator; private final Pattern DOWNLOAD_FILE_PATTERN = Pattern.compile("^(?[A-Za-z0-9]{6})_(?[A-Za-z0-9_]+)_(?\\d{4}\\d{2}\\d{2}_\\d{2}\\d{2}\\d{2})(?:\\.xml)?\\.zip$"); public SignalFacade(MonitoringRequestValidator monitoringRequestValidator, @@ -76,7 +74,8 @@ public class SignalFacade { AuthorizationService authorizationService, NpPipelinePackageFabric npPipelinePackageManager, LpPipelinePackageFabric lpPipelinePackageFabric, - Saver saver, PipelinePackageOrchestrator pipelineOrchestrator) { + Saver saver, NpPipelinePackageOrchestrator pipelineOrchestrator, + LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator) { this.monitoringRequestValidator = monitoringRequestValidator; this.config = config; this.userPackageService = userPackageService; @@ -85,6 +84,7 @@ public class SignalFacade { this.lpPipelinePackageFabric = lpPipelinePackageFabric; this.saver = saver; this.pipelineOrchestrator = pipelineOrchestrator; + this.lpPipelinePackageOrchestrator = lpPipelinePackageOrchestrator; } public void setupMonitoringRequest(MultipartFile file) { @@ -160,7 +160,7 @@ public class SignalFacade { }, pipeline, packageId); } else { Pipeline pipeline = lpPipelinePackageFabric.build(new SetupParam(packageId, 2, PipelineMode.DEFAULT)); - pipelineOrchestrator.startOrDelay(() -> { + lpPipelinePackageOrchestrator.startOrDelay(() -> { saver.readAndSaveLpPackage(packageId, parsedFilePath); CtFileUtils.deleteQuietly(parsedFilePath); }, pipeline, packageId); 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 9a22a05..e930a2d 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 @@ -32,4 +32,12 @@ public interface LegalsMapMapper { @Param("lastId") Long lastId); List findCachedByInnAndRegNUm(@Param("json") String json); + + void updateSearchedAt(@Param("json") String json); + + void resetFidAndFlagsByInnAndRegNum(@Param("json") String json); + + void updateFidCachedData(@Param("json") String json); + + void deleteUnusedMaps(); } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java index 76959a4..04dbe2e 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java @@ -19,6 +19,8 @@ public interface PackagesRecordsMapper { List uniquePackageIdByPhones(@Param("json") String json); + List uniquePackageIdByInnAndRegNum(@Param("json") String json); + List selectUserPackagePhoneData(@Param("json") String json); List selectUserPackageLegalData(@Param("json") String json); diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/LegalsMapService.java b/src/main/java/ru/nbch/credit_tracker/service/indic/LegalsMapService.java index dd3a4f2..5849a01 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/LegalsMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/LegalsMapService.java @@ -4,8 +4,10 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.entities.app.LegalsMap; import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap; import ru.nbch.credit_tracker.mapper.signal.LegalsFidMapMapper; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import java.util.HashMap; import java.util.List; @@ -15,11 +17,14 @@ import java.util.List; public class LegalsMapService { private final LegalsFidMapMapper legalsFidMapMapper; + private final LegalsMapMapper legalsMapMapper; private final ObjectMapper entityToJsonMapper; public LegalsMapService(LegalsFidMapMapper legalsFidMapMapper, + LegalsMapMapper legalsMapMapper, ObjectMapper entityToJsonMapper) { this.legalsFidMapMapper = legalsFidMapMapper; + this.legalsMapMapper = legalsMapMapper; this.entityToJsonMapper = entityToJsonMapper; } @@ -37,4 +42,47 @@ public class LegalsMapService { throw new RuntimeException(e); } } + + public void updateSearchedAt(List subjects) { + try { + String json = entityToJsonMapper.writeValueAsString(subjects.stream() + .map(phoneFidMap -> new HashMap() {{ + put("inn", phoneFidMap.getInn()); + put("reg_num", phoneFidMap.getRegNum()); + }}).toList()); + + legalsMapMapper.updateSearchedAt(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + + public void resetFidAndFlagsByInnAndRegNum(List subjects) { + try { + String json = entityToJsonMapper.writeValueAsString(subjects.stream() + .map(phoneFidMap -> new HashMap() {{ + put("inn", phoneFidMap.getInn()); + put("reg_num", phoneFidMap.getRegNum()); + }}).toList()); + + legalsMapMapper.resetFidAndFlagsByInnAndRegNum(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + + public void updateFidCachedData(List subjects) { + try { + String json = entityToJsonMapper.writeValueAsString(subjects.stream() + .map(phoneFidMap -> new HashMap() {{ + put("inn", phoneFidMap.getInn()); + put("reg_num", phoneFidMap.getRegNum()); + put("fid", phoneFidMap.getFid()); + }}).toList()); + + legalsMapMapper.updateFidCachedData(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java index 31583f9..a946f59 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java @@ -9,7 +9,7 @@ import ru.nbch.credit_tracker.enums.Status; import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; -import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; +import ru.nbch.credit_tracker.service.task.NpPipelinePackageOrchestrator; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; import ru.nbch.credit_tracker.service.task.impl.Saver; @@ -32,14 +32,14 @@ public class StartupRestoreService { private final UserPackageMapper userPackageMapper; private final PackagesRecordsMapper phoneFidMapMapper; private final NpPipelinePackageFabric npPipelinePackageManager; - private final PipelinePackageOrchestrator pipelineOrchestrator; + private final NpPipelinePackageOrchestrator pipelineOrchestrator; private final CreditTrackerSettings creditTrackerSettings; private final Saver saver; public StartupRestoreService(UserPackageMapper userPackageMapper, PackagesRecordsMapper phoneFidMapMapper, NpPipelinePackageFabric npPipelinePackageManager, - PipelinePackageOrchestrator pipelineOrchestrator, + NpPipelinePackageOrchestrator pipelineOrchestrator, CreditTrackerSettings creditTrackerSettings, Saver saver) { this.userPackageMapper = userPackageMapper; @@ -53,6 +53,7 @@ public class StartupRestoreService { public void run() { acceptedRestore(); processingRestore(); + // TODO LP } private void acceptedRestore() { diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java index 6986743..69cfb96 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java @@ -6,12 +6,10 @@ import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.entities.app.UserPackage; +import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.signal.PhonesMapService; -import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric; -import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; -import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator; -import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.task.*; import ru.nbch.credit_tracker.service.task.impl.PipelineMode; import ru.nbch.pipeline.engine.core.Pipeline; @@ -23,27 +21,39 @@ import java.util.concurrent.CompletableFuture; @EnableScheduling public class UpdatingFidService { private final NpPipelinePackageFabric npPipelinePackageFabric; - private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; - private final PipelinePackageOrchestrator pipelinePackageOrchestrator; + private final LpPipelinePackageFabric lpPipelinePackageFabric; + private final NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator; + private final LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator; + private final NpPipelinePackageOrchestrator npPipelinePackageOrchestrator; + private final LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator; private final UserPackageMapper userPackageMapper; private final PhonesMapService phonesMapService; + private final LegalsMapMapper legalsMapMapper; public UpdatingFidService(CreditTrackerSettings config, NpPipelinePackageFabric npPipelinePackageFabric, - PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, - PipelinePackageOrchestrator pipelinePackageOrchestrator, + LpPipelinePackageFabric lpPipelinePackageFabric, + NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator, + LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator, + NpPipelinePackageOrchestrator npPipelinePackageOrchestrator, + LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator, UserPackageMapper userPackageMapper, - PhonesMapService phonesMapService) { + PhonesMapService phonesMapService, + LegalsMapMapper legalsMapMapper) { this.npPipelinePackageFabric = npPipelinePackageFabric; - this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; - this.pipelinePackageOrchestrator = pipelinePackageOrchestrator; + this.lpPipelinePackageFabric = lpPipelinePackageFabric; + this.npPipelineRecalcOrchestrator = npPipelineRecalcOrchestrator; + this.lpPipelineRecalcOrchestrator = lpPipelineRecalcOrchestrator; + this.npPipelinePackageOrchestrator = npPipelinePackageOrchestrator; + this.lpPipelinePackageOrchestrator = lpPipelinePackageOrchestrator; this.userPackageMapper = userPackageMapper; this.phonesMapService = phonesMapService; + this.legalsMapMapper = legalsMapMapper; } - @Scheduled(cron = "${ct.update-fid-cron-expression}") + @Scheduled(cron = "${ct.update-np-fid-cron-expression}") public void updateNp() { - if (!pipelineRecalcPDOrchestrator.checkPermits()) { + if (!npPipelineRecalcOrchestrator.checkPermits()) { log.info("The previous task has not yet been completed."); return; } @@ -52,7 +62,7 @@ public class UpdatingFidService { log.info("Start updating fids and PD"); try { Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, 1, PipelineMode.RECALC_FID)); - CompletableFuture future = pipelineRecalcPDOrchestrator.startOrReject(pipeline); + CompletableFuture future = npPipelineRecalcOrchestrator.startOrReject(pipeline); future.whenComplete((o, throwable) -> { try { if (throwable != null) { @@ -66,7 +76,7 @@ public class UpdatingFidService { SetupParam setupParam = new SetupParam(packageId, 1, PipelineMode.RESEND); try { Pipeline makeXmlByCache = npPipelinePackageFabric.build(setupParam); - pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId); + npPipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId); } catch (Exception e) { log.error("Failed to build pipeline", e); } @@ -74,7 +84,7 @@ public class UpdatingFidService { } finally { log.info("pipeline finished"); pipeline.close(); - pipelineRecalcPDOrchestrator.release(); + npPipelineRecalcOrchestrator.release(); } }); } catch (Exception e) { @@ -82,8 +92,44 @@ public class UpdatingFidService { } } - @Scheduled(cron = "${ct.update-fid-cron-expression}") // TODO another property + @Scheduled(cron = "${ct.update-lp-fid-cron-expression}") public void updateLp() { - // TODO + if (!lpPipelineRecalcOrchestrator.checkPermits()) { + log.info("The previous task has not yet been completed."); + return; + } + log.info("Delete unused phones"); + legalsMapMapper.deleteUnusedMaps(); + log.info("Start updating fids"); + try { + Pipeline pipeline = lpPipelinePackageFabric.buildRecalculation(new SetupParam(null, 2, PipelineMode.RECALC_FID)); + CompletableFuture future = lpPipelineRecalcOrchestrator.startOrReject(pipeline); + future.whenComplete((o, throwable) -> { + try { + if (throwable != null) { + log.error(throwable.getMessage(), throwable); + return; + } + + List packagesRepeated = userPackageMapper.findForRepeatedPackages(); + packagesRepeated.forEach(userPackage -> { + Long packageId = userPackage.getId(); + SetupParam setupParam = new SetupParam(packageId, 2, PipelineMode.RESEND); + try { + Pipeline setupPipeline = lpPipelinePackageFabric.build(setupParam); + lpPipelinePackageOrchestrator.startOrDelay(setupPipeline, packageId); + } catch (Exception e) { + log.error("Failed to build pipeline", e); + } + }); + } finally { + log.info("pipeline finished"); + pipeline.close(); + lpPipelineRecalcOrchestrator.release(); + } + }); + } catch (Exception e) { + log.error("Failed to build pipeline", e); + } } } 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 3c558b8..ca91930 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 @@ -113,7 +113,7 @@ public class PackageRecordsService { } } - public List selectUniquePackageIds(List phones) { + public List selectNpUniquePackageIds(List phones) { try { String json = entityToJsonMapper.writeValueAsString(phones.stream() .filter(Objects::nonNull) @@ -126,6 +126,20 @@ public class PackageRecordsService { } } + public List selectLpUniquePackageIds(List legalsMaps) { + try { + List pairs = legalsMaps.stream() + .filter(Objects::nonNull) + .map(legalMap -> new InnRegnumPair(legalMap.getInn(), legalMap.getRegNum())) + .toList(); + + String json = entityToJsonMapper.writeValueAsString(pairs); + return packagesRecordsMapper.uniquePackageIdByInnAndRegNum(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public List selectUserPackagePhoneData(Set phones) { try { String json = entityToJsonMapper.writeValueAsString(phones.stream() diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java index cf01db8..1992010 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java @@ -144,7 +144,7 @@ public class LpFidEnricher implements Handler { @Override public int preferredConcurrency() { - return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); + return creditTrackerSettings.getLpPipeline().nameEnricherStageConcurrency(); } @Override diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java index 6ddbbcb..4acf644 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageFabric.java @@ -7,6 +7,7 @@ import org.springframework.stereotype.Component; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.config.task.StageQueueFactory; import ru.nbch.credit_tracker.model.impl.*; +import ru.nbch.credit_tracker.service.task.impl.LpFidUpdater; import ru.nbch.pipeline.engine.core.Pipeline; import ru.nbch.pipeline.engine.core.PipelineBackoff; import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter; @@ -25,6 +26,7 @@ public class LpPipelinePackageFabric implements InitializingBean { private final StageQueueFactory lpFinishSetupQueueFactory; private final StageQueueFactory noOpQueueFactory; private final LpFidEnricher lpFidEnricher; + private final LpFidUpdater lpFidUpdater; private final Supplier> pipelineTerminatorFactory; private final Supplier> noOpTerminatorFactory; private final Function> pipelineStarterFactory; @@ -37,6 +39,7 @@ public class LpPipelinePackageFabric implements InitializingBean { @Qualifier("lpFinishSetupQueueFactory") StageQueueFactory lpFinishSetupQueueFactory, @Qualifier("lpNoOpQueueFactory") StageQueueFactory lpNoOpQueueFactory, LpFidEnricher lpFidEnricher, + LpFidUpdater lpFidUpdater, Supplier> pipelineTerminatorFactory, Supplier> noOpTerminatorFactory, Function> pipelineStarterFactory, @@ -47,6 +50,7 @@ public class LpPipelinePackageFabric implements InitializingBean { this.lpFinishSetupQueueFactory = lpFinishSetupQueueFactory; this.noOpQueueFactory = lpNoOpQueueFactory; this.lpFidEnricher = lpFidEnricher; + this.lpFidUpdater = lpFidUpdater; this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.noOpTerminatorFactory = noOpTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; @@ -69,4 +73,15 @@ public class LpPipelinePackageFabric implements InitializingBean { public void afterPropertiesSet() throws Exception { log.info("pipeline settings: {}", creditTrackerSettings.getLpPipeline()); } + + public Pipeline buildRecalculation(SetupParam setupParam) throws Exception { + return PipeLineBuilderV2.first(lpFidUpdaterQueueFactory.getObject(), noOpQueueFactory.getObject()) + .handler(lpFidUpdater) + .andThen() + .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) + .pollFinishingQueue(noOpTerminatorFactory.get()) + .backoff(pipelineBackoff) + .context(PackageVariable.class) + .build(); + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java new file mode 100644 index 0000000..1285a76 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelinePackageOrchestrator.java @@ -0,0 +1,26 @@ +package ru.nbch.credit_tracker.service.task; + +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.settings.PipelineSettings; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; + +import java.util.concurrent.ExecutorService; + +@Service +public class LpPipelinePackageOrchestrator extends PackageOrchestrator { + private final CreditTrackerSettings creditTrackerSettings; + + public LpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, + UserPackageMapper userPackageMapper, // TODO + ExecutorService pipelineExecutorService, // TODO + ExecutorService saverExecutorService) { // TODO + super(userPackageMapper, pipelineExecutorService, saverExecutorService); + this.creditTrackerSettings = creditTrackerSettings; + } + + @Override + protected PipelineSettings getPipeline() { + return creditTrackerSettings.getPipeline(); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java new file mode 100644 index 0000000..6c70099 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/LpPipelineRecalcOrchestrator.java @@ -0,0 +1,24 @@ +package ru.nbch.credit_tracker.service.task; + +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.settings.PipelineSettings; + +import java.util.concurrent.ExecutorService; + +@Service +public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator{ + + private final CreditTrackerSettings creditTrackerSettings; + + public LpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings, + ExecutorService pipelineExecutorService) { // TODO another executor + super(pipelineExecutorService); + this.creditTrackerSettings = creditTrackerSettings; + } + + @Override + protected PipelineSettings getPipelineSettings() { + return creditTrackerSettings.getLpPipeline(); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java new file mode 100644 index 0000000..fe6f59a --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelinePackageOrchestrator.java @@ -0,0 +1,26 @@ +package ru.nbch.credit_tracker.service.task; + +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.settings.PipelineSettings; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; + +import java.util.concurrent.ExecutorService; + +@Service +public class NpPipelinePackageOrchestrator extends PackageOrchestrator { + private final CreditTrackerSettings creditTrackerSettings; + + public NpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, + UserPackageMapper userPackageMapper, + ExecutorService pipelineExecutorService, + ExecutorService saverExecutorService) { + super(userPackageMapper, pipelineExecutorService, saverExecutorService); + this.creditTrackerSettings = creditTrackerSettings; + } + + @Override + protected PipelineSettings getPipeline() { + return creditTrackerSettings.getPipeline(); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java new file mode 100644 index 0000000..7273ec4 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/NpPipelineRecalcOrchestrator.java @@ -0,0 +1,23 @@ +package ru.nbch.credit_tracker.service.task; + +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.settings.PipelineSettings; + +import java.util.concurrent.ExecutorService; + +@Service +public class NpPipelineRecalcOrchestrator extends RecalcOrchestrator { + + private final CreditTrackerSettings creditTrackerSettings; + public NpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings, + ExecutorService pipelineExecutorService) { + super(pipelineExecutorService); + this.creditTrackerSettings = creditTrackerSettings; + } + + @Override + protected PipelineSettings getPipelineSettings() { + return creditTrackerSettings.getPipeline(); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java new file mode 100644 index 0000000..da16fb5 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PackageOrchestrator.java @@ -0,0 +1,148 @@ +package ru.nbch.credit_tracker.service.task; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.PipelineSettings; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.impl.PackageVariable; +import ru.nbch.credit_tracker.model.impl.SignalPayload; +import ru.nbch.credit_tracker.service.task.impl.PipelineMode; +import ru.nbch.credit_tracker.service.task.impl.SetupMonitoringPipelineTerminator; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; +import ru.nbch.pipeline.engine.core.Pipeline; + +import java.time.LocalDateTime; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +public abstract class PackageOrchestrator { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final UserPackageMapper userPackageMapper; + private final ExecutorService pipelineExecutorService; + private final ExecutorService saverExecutorService; + private final Object lock = new Object(); + private final java.util.Deque pending = new java.util.ArrayDeque<>(); + private final Semaphore semaphore; + + public PackageOrchestrator(UserPackageMapper userPackageMapper, + ExecutorService pipelineExecutorService, + ExecutorService saverExecutorService) { + this.userPackageMapper = userPackageMapper; + this.pipelineExecutorService = pipelineExecutorService; + this.saverExecutorService = saverExecutorService; + this.semaphore = new Semaphore(getPipeline().parallelRequests()); + } + + private record Launch(Runnable before, Pipeline pipeline, Long packageId) { + } + + public void startOrDelay(Pipeline pipeline, Long packageId) { + startOrDelay(null, pipeline, packageId); + } + + public void startOrDelay(Runnable before, Pipeline pipeline, Long packageId) { + synchronized (lock) { + if (!semaphore.tryAcquire()) { + log.debug("couldn't acquire permit from semaphore. delaying task for {}...", packageId); + pending.addLast(new PackageOrchestrator.Launch(before, pipeline, packageId)); + return; + } + log.info("throttler permits after acquire: {}", semaphore.availablePermits()); + } + innerStartOrDelay(before, pipeline, packageId); + } + + private void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) { + try { + if (before != null) { + CompletableFuture + .runAsync(before, saverExecutorService) + .handle((v, t) -> { + if (t != null) { + log.error("`before` failed: {}", ExceptionUtils.getStackTrace(t)); + pipeline.close(); + checkForDelayedTasks(); // передать слот дальше или освободить + } else { + pipeline.start(pipelineExecutorService); + addCallback(pipeline, packageId); + } + return null; + }); + } else { + pipeline.start(pipelineExecutorService); + addCallback(pipeline, packageId); + } + } catch (Exception e) { + log.error(ExceptionUtils.getStackTrace(e)); + pipeline.close(); + checkForDelayedTasks(); + } + } + + private void addCallback(Pipeline pipeline, Long packageId) { + pipeline.getTerminator() + .doneFuture() + .orTimeout(getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS) + .whenComplete((v, e) -> { + try { + long finishTime = System.currentTimeMillis(); + log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime()); + SignalPayload res = (SignalPayload) v; + var ctx = pipeline.obtainContext(PackageVariable.class); + AtomicInteger cacheHits = ctx.get(PackageVariable.cacheHits); + AtomicInteger cacheMisses = ctx.get(PackageVariable.cacheMisses); + log.info( + "Cache stats: hits={}, misses={}, total={}", + cacheHits.get(), + cacheMisses.get(), + cacheHits.get() + cacheMisses.get()); + if (e instanceof SetupMonitoringPipelineTerminator.PipelineTerminatorException) { + log.info("when complete: terminator was interrupted."); + return; + } else if (e != null) { + log.error(ExceptionUtils.getStackTrace(e)); + return; + } + if (!res.hasFids()) { + log.info("No one fid was found by packageId : {}.", packageId); + userPackageMapper.updateNoFidMonitoring(packageId); + return; + } + + PipelineMode mode = ctx.get(PackageVariable.mode); + if (mode != null && mode.equals(PipelineMode.DEFAULT)) { + userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); + } else { + userPackageMapper.updatePackageOnMonitoringSuccess(packageId, null); + } + } catch (Exception ex) { + log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); + } finally { + log.info("packageId={}: pipeline finished", packageId); + pipeline.close(); + checkForDelayedTasks(); + log.info("semaphore amount: {}", semaphore.availablePermits()); + } + }); + } + + private void checkForDelayedTasks() { + log.debug("checking for delayed pipelines..."); + PackageOrchestrator.Launch delayedLaunch; + synchronized (lock) { + delayedLaunch = pending.pollFirst(); + if (delayedLaunch == null) { + log.debug("no delayed pipelines found; releasing semaphore..."); + semaphore.release(); + return; + } + } + log.info("delayed pipeline found: packageId={}. continue work...", delayedLaunch.packageId); + innerStartOrDelay(delayedLaunch.before, delayedLaunch.pipeline, delayedLaunch.packageId); + } + + protected abstract PipelineSettings getPipeline(); +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java deleted file mode 100644 index 61a9d31..0000000 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java +++ /dev/null @@ -1,151 +0,0 @@ -package ru.nbch.credit_tracker.service.task; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; -import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; -import ru.nbch.credit_tracker.model.impl.PackageVariable; -import ru.nbch.credit_tracker.model.impl.SignalPayload; -import ru.nbch.credit_tracker.service.task.impl.PipelineMode; -import ru.nbch.credit_tracker.service.task.impl.SetupMonitoringPipelineTerminator; -import ru.nbch.credit_tracker.utils.error.ExceptionUtils; -import ru.nbch.pipeline.engine.core.Pipeline; - -import java.time.LocalDateTime; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -@Service -public class PipelinePackageOrchestrator { - private final Logger log = LoggerFactory.getLogger(getClass()); - private final CreditTrackerSettings creditTrackerSettings; - private final UserPackageMapper userPackageMapper; - private final ExecutorService pipelineExecutorService; - private final ExecutorService saverExecutorService; - private final Object lock = new Object(); - private final java.util.Deque pending = new java.util.ArrayDeque<>(); - private final Semaphore semaphore; - - public PipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings, - UserPackageMapper userPackageMapper, - ExecutorService pipelineExecutorService, - ExecutorService saverExecutorService) { - this.creditTrackerSettings = creditTrackerSettings; - this.userPackageMapper = userPackageMapper; - this.pipelineExecutorService = pipelineExecutorService; - this.saverExecutorService = saverExecutorService; - this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests()); - } - - private record Launch(Runnable before, Pipeline pipeline, Long packageId) { - } - - public void startOrDelay(Pipeline pipeline, Long packageId) { - startOrDelay(null, pipeline, packageId); - } - - public void startOrDelay(Runnable before, Pipeline pipeline, Long packageId) { - synchronized (lock) { - if (!semaphore.tryAcquire()) { - log.debug("couldn't acquire permit from semaphore. delaying task for {}...", packageId); - pending.addLast(new Launch(before, pipeline, packageId)); - return; - } - log.info("throttler permits after acquire: {}", semaphore.availablePermits()); - } - innerStartOrDelay(before, pipeline, packageId); - } - - private void innerStartOrDelay(Runnable before, Pipeline pipeline, Long packageId) { - try { - if (before != null) { - CompletableFuture - .runAsync(before, saverExecutorService) - .handle((v, t) -> { - if (t != null) { - log.error("`before` failed: {}", ExceptionUtils.getStackTrace(t)); - pipeline.close(); - checkForDelayedTasks(); // передать слот дальше или освободить - } else { - pipeline.start(pipelineExecutorService); - addCallback(pipeline, packageId); - } - return null; - }); - } else { - pipeline.start(pipelineExecutorService); - addCallback(pipeline, packageId); - } - } catch (Exception e) { - log.error(ExceptionUtils.getStackTrace(e)); - pipeline.close(); - checkForDelayedTasks(); - } - } - - private void addCallback(Pipeline pipeline, Long packageId) { - pipeline.getTerminator() - .doneFuture() - .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS) - .whenComplete((v, e) -> { - try { - long finishTime = System.currentTimeMillis(); - log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime()); - SignalPayload res = (SignalPayload) v; - var ctx = pipeline.obtainContext(PackageVariable.class); - AtomicInteger cacheHits = ctx.get(PackageVariable.cacheHits); - AtomicInteger cacheMisses = ctx.get(PackageVariable.cacheMisses); - log.info( - "Cache stats: hits={}, misses={}, total={}", - cacheHits.get(), - cacheMisses.get(), - cacheHits.get() + cacheMisses.get()); - if (e instanceof SetupMonitoringPipelineTerminator.PipelineTerminatorException) { - log.info("when complete: terminator was interrupted."); - return; - } else if (e != null) { - log.error(ExceptionUtils.getStackTrace(e)); - return; - } - if (!res.hasFids()) { - log.info("No one fid was found by packageId : {}.", packageId); - userPackageMapper.updateNoFidMonitoring(packageId); - return; - } - - PipelineMode mode = ctx.get(PackageVariable.mode); - if (mode != null && mode.equals(PipelineMode.DEFAULT)) { - userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); - } else { - userPackageMapper.updatePackageOnMonitoringSuccess(packageId, null); - } - } catch (Exception ex) { - log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); - } finally { - log.info("packageId={}: pipeline finished", packageId); - pipeline.close(); - checkForDelayedTasks(); - log.info("semaphore amount: {}", semaphore.availablePermits()); - } - }); - } - - private void checkForDelayedTasks() { - log.debug("checking for delayed pipelines..."); - Launch delayedLaunch; - synchronized (lock) { - delayedLaunch = pending.pollFirst(); - if (delayedLaunch == null) { - log.debug("no delayed pipelines found; releasing semaphore..."); - semaphore.release(); - return; - } - } - log.info("delayed pipeline found: packageId={}. continue work...", delayedLaunch.packageId); - innerStartOrDelay(delayedLaunch.before, delayedLaunch.pipeline, delayedLaunch.packageId); - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java similarity index 77% rename from src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java rename to src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java index 1c49013..8881082 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/RecalcOrchestrator.java @@ -2,8 +2,7 @@ package ru.nbch.credit_tracker.service.task; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.stereotype.Service; -import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.config.settings.PipelineSettings; import ru.nbch.credit_tracker.utils.error.ExceptionUtils; import ru.nbch.pipeline.engine.core.Pipeline; @@ -12,17 +11,13 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; -@Service -public class PipelineRecalcPDOrchestrator { +public abstract class RecalcOrchestrator { private final Logger log = LoggerFactory.getLogger(getClass()); - private final CreditTrackerSettings creditTrackerSettings; private final ExecutorService pipelineExecutorService; private final Object lock = new Object(); private final Semaphore semaphore; - public PipelineRecalcPDOrchestrator(CreditTrackerSettings creditTrackerSettings, - ExecutorService pipelineExecutorService) { - this.creditTrackerSettings = creditTrackerSettings; + public RecalcOrchestrator(ExecutorService pipelineExecutorService) { this.pipelineExecutorService = pipelineExecutorService; this.semaphore = new Semaphore(1); } @@ -43,6 +38,7 @@ public class PipelineRecalcPDOrchestrator { return semaphore.availablePermits() > 0; } } + private CompletableFuture innerStart(Pipeline pipeline) { try { pipeline.start(pipelineExecutorService); @@ -57,8 +53,8 @@ public class PipelineRecalcPDOrchestrator { private CompletableFuture addCallback(Pipeline pipeline) { return pipeline.getTerminator() - .doneFuture() - .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS); + .doneFuture() + .orTimeout(getPipelineSettings().globalTimeout().toMillis(), TimeUnit.MILLISECONDS); } public void release() { @@ -68,4 +64,6 @@ public class PipelineRecalcPDOrchestrator { log.info("semaphore amount: {}", semaphore.availablePermits()); } } + + protected abstract PipelineSettings getPipelineSettings(); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpFidUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpFidUpdater.java new file mode 100644 index 0000000..96bcb6e --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpFidUpdater.java @@ -0,0 +1,136 @@ +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.FidIdData; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.PackageVariable; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; +import ru.nbch.credit_tracker.service.UserPackageService; +import ru.nbch.credit_tracker.service.indic.LegalsMapService; +import ru.nbch.credit_tracker.service.indic.PersonService; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.utils.CTUtil; +import ru.nbch.pipeline.engine.core.Handler; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +@Service +@Slf4j +public class LpFidUpdater implements Handler { + private final PersonService personService; + private final CreditTrackerSettings creditTrackerSettings; + private final PackageRecordsService packageRecordsService; + private final LegalsMapService legalsMapService; + private final UserPackageService userPackageService; + + public LpFidUpdater(PersonService personService, + CreditTrackerSettings creditTrackerSettings, + PackageRecordsService packageRecordsService, + LegalsMapService legalsMapService, + UserPackageService userPackageService) { + this.personService = personService; + this.creditTrackerSettings = creditTrackerSettings; + this.packageRecordsService = packageRecordsService; + this.legalsMapService = legalsMapService; + this.userPackageService = userPackageService; + } + + @Override + public VoidPayload handle(SubjectPayload taskPayload) { + List subjects = taskPayload.getSubjects(); + var ctx = obtainContext(PackageVariable.class, taskPayload); + log.debug("LpFidUpdaterStage handle, size of bucket for save: {}", subjects.size()); + + Map> fidsByInnMap = personService.findFidsByInn(subjects); + Map> fidsByRegNumMap = personService.findFidsByRegNum(subjects); + + List changingFidLegalMaps = new ArrayList<>(); + List notFoundFidLegalMaps = new ArrayList<>(); + List notChangingFidLegalMaps = new ArrayList<>(); + + for (SubjectProfile subject : subjects) { + Long inn = subject.getInn(); + Long regNum = subject.getRegNum(); + + List fidByInn = fidsByInnMap.get(CTUtil.normalizeInnZeroesOnly(inn)); + List fidByRegNum = fidsByRegNumMap.get(CTUtil.normalizeRegNumZeroesOnly(regNum)); + + if (fidByInn == null || fidByInn.isEmpty()) { + // fid не найден + notFoundFidLegalMaps.add(fillLegalMap(subject, null)); + continue; + } + + List fidsByInn = fidByInn.stream().map(FidIdData::getFid).toList(); + + if (fidByRegNum != null && !fidByRegNum.isEmpty()) { + List fidsByRegNum = fidByRegNum.stream().map(FidIdData::getFid).toList(); + + // находим все пересечения по фидам + List intersection = new ArrayList<>(fidsByInn); + intersection.retainAll(fidsByRegNum); + + if (!intersection.isEmpty()) { + // берем максимальное пересечение + Integer maxFid = Collections.max(intersection); + + if (maxFid.equals(subject.getFid())) { + notChangingFidLegalMaps.add(fillLegalMap(subject, maxFid)); + } else { + changingFidLegalMaps.add(fillLegalMap(subject, maxFid)); + } + continue; + } + } + notFoundFidLegalMaps.add(fillLegalMap(subject, null)); + } + + List changingFids = new ArrayList<>(); + if (!notFoundFidLegalMaps.isEmpty()) { + legalsMapService.resetFidAndFlagsByInnAndRegNum(notFoundFidLegalMaps); + changingFids.addAll(notFoundFidLegalMaps); + } + + if (!changingFidLegalMaps.isEmpty()) { + legalsMapService.updateFidCachedData(changingFidLegalMaps); + changingFids.addAll(changingFidLegalMaps); + } + + if (!notChangingFidLegalMaps.isEmpty()) { + legalsMapService.updateSearchedAt(notChangingFidLegalMaps); + } + + if (!changingFids.isEmpty()) { + List packageIds = packageRecordsService.selectLpUniquePackageIds(changingFids); + if (!packageIds.isEmpty()) { + userPackageService.setFidChanged(packageIds); + } + } + return new VoidPayload(); + } + + private LegalsMap fillLegalMap(SubjectProfile subject, Integer fid) { + return LegalsMap.builder() + .inn(subject.getInn()) + .regNum(subject.getRegNum()) + .fid(fid) + .build(); + } + + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getLpPipeline().nameUpdaterStageConcurrency(); + } + + @Override + public VoidPayload cookPoison() { + return VoidPayload.POISON; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java index 494c2bd..93fe289 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java @@ -44,7 +44,7 @@ public class LpKickRecalcPipeLine implements IPipeLineStarter { this.packageRecordsService = packageRecordsService; this.creditTrackerSettings = creditTrackerSettings; this.BATCH_SIZE = creditTrackerSettings.getLpPipeline().subjectBatchSize(); - this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateFids() != null ? creditTrackerSettings.getDaysToUpdateFids() : 1; + this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateLpFids() != null ? creditTrackerSettings.getDaysToUpdateLpFids() : 1; } private static class PipelineStarterException extends SignalClientException { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java index 81b61e0..caedd46 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java @@ -80,7 +80,7 @@ public class NpFidUpdater implements Handler { } if (!changingFids.isEmpty()) { - List packageIds = packageRecordsService.selectUniquePackageIds(changingFids); + List packageIds = packageRecordsService.selectNpUniquePackageIds(changingFids); if (!packageIds.isEmpty()) { userPackageService.setFidChanged(packageIds); } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java index 6c35b9d..44e2adb 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java @@ -44,7 +44,7 @@ public class NpKickRecalcPipeLine implements IPipeLineStarter { this.packageRecordsService = packageRecordsService; this.creditTrackerSettings = creditTrackerSettings; this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); - this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateFids() != null ? creditTrackerSettings.getDaysToUpdateFids() : 1; + this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateNpFids() != null ? creditTrackerSettings.getDaysToUpdateNpFids() : 1; } private static class PipelineStarterException extends SignalClientException { diff --git a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml index 14200fd..deffe91 100644 --- a/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/LegalsMapMapper.xml @@ -113,7 +113,7 @@ flag_90_12 FROM json_to_recordset(#{json}::json) AS src( inn bigint, - reg_num text, + reg_num bigint, flag_90_12 smallint ) ORDER BY inn, reg_num), @@ -128,7 +128,7 @@ calculated_at = now() FROM json_to_recordset(#{json}::json) AS src( inn bigint, - reg_num text, + reg_num bigint, flag_90_12 smallint ) WHERE dst.inn = src.inn @@ -143,7 +143,7 @@ last_rejected_date FROM json_to_recordset(#{json}::json) AS src( inn bigint, - reg_num text, + reg_num bigint, last_rejected_date timestamp ) ORDER BY inn, reg_num), @@ -157,7 +157,7 @@ SET last_rejected_date = src.last_rejected_date FROM json_to_recordset(#{json}::json) AS src( inn bigint, - reg_num text, + reg_num bigint, last_rejected_date timestamp ) WHERE dst.inn = src.inn @@ -180,4 +180,60 @@ order by id limit #{batchSize} + + + update signals_legals_map dst + set searched_at = now() + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num bigint) + where dst.inn = src.inn AND dst.reg_num = src.reg_num + + + + update signals_legals_map dst + set searched_at = now(), + fid = null, + flag_90_12 = null, + calculated_at = null, + last_rejected_date = null + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num bigint) + where dst.inn = src.inn AND dst.reg_num = src.reg_num + + + + + update signals_legals_map dst + set searched_at = now(), + fid = src.fid, + flag_90_12 = null, + calculated_at = null, + last_rejected_date = null + FROM json_to_recordset(#{json}::json) AS src( + inn bigint, + reg_num bigint, + fid int) + where dst.inn = src.inn AND dst.reg_num = src.reg_num + + + + + DELETE + FROM signals_legals_map slm + WHERE NOT EXISTS (SELECT 1 + FROM signals_legal_fid_map slr + JOIN signals_user_packages sup + ON sup.id = slr.package_id + WHERE slr.inn = slm.inn + AND slr.reg_num = slm.reg_num + AND sup.status IN + ( + 'ON_MONITORING', + 'PROCESSING', + 'PROCESSING_REPEATED' + )) + + \ No newline at end of file diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml index 3f2c880..75257f0 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml @@ -53,6 +53,19 @@ where spr.phone_l in (select * from src) and sup.status = 'ON_MONITORING' + +