From 6b4d6908d147116f5a06f53951b77a5116e15f5d Mon Sep 17 00:00:00 2001 From: etreschenkov Date: Wed, 14 Jan 2026 15:55:09 +0300 Subject: [PATCH] http://jira.mfd.msk:8088/browse/BKI-3241 [recalc person data] --- .../config/task/PipelineStarterConfig.java | 9 ++ .../config/task/StageConfig.java | 16 +++ .../entities/app/UserPackage.java | 1 + .../mapper/signal/PhonesMapMapper.java | 1 + .../model/impl/EmptyTerminatePayload.java | 6 + .../model/impl/VoidPayload.java | 24 ++++ .../service/UpdatingFlagsService.java | 30 ++++- .../service/task/PipelinePackageFabric.java | 39 +++++- .../task/PipelineRecalcPDOrchestrator.java | 76 ++++++++++++ .../service/task/impl/FidIdUpdater.java | 80 ++++++++++++ .../service/task/impl/FidNameUpdater.java | 67 ++++++++++ .../service/task/impl/KickRecalcPipeLine.java | 115 ++++++++++++++++++ .../service/task/impl/NoOpTerminator.java | 54 ++++++++ .../service/task/impl/PipelineMode.java | 3 +- .../mybatis/mapper/signal/PhonesMapMapper.xml | 19 +++ .../mapper/signal/UserPackageMapper.xml | 5 +- src/main/resources/sql/v2.21.ddl | 2 + 17 files changed, 538 insertions(+), 9 deletions(-) create mode 100644 src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java create mode 100644 src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java index 0cbde3a..fa434e0 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java @@ -6,13 +6,16 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; import ru.nbch.credit_tracker.service.signal.PackageRecordsService; import ru.nbch.credit_tracker.service.task.SetupParam; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl; import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRepeatedRestoreImpl; +import ru.nbch.credit_tracker.service.task.impl.NoOpTerminator; import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; @@ -30,6 +33,7 @@ public class PipelineStarterConfig { new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); case REPEATED_RESTORE -> new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings); + case RECALC_PERSONAL_DATA -> null; }; } @@ -37,4 +41,9 @@ public class PipelineStarterConfig { public Supplier> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) { return () -> new XmlMergeCoordinator(creditTrackerSettings); } + + @Bean + public Supplier> noOpTerminatorFactory() { + return NoOpTerminator::new; + } } diff --git a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java index 45db916..c78c000 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java @@ -5,6 +5,7 @@ import org.springframework.context.annotation.Configuration; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; @Configuration public class StageConfig { @@ -14,11 +15,26 @@ public class StageConfig { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); } + @Bean("nameUpdaterQueueFactory") + public StageQueueFactory nameUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); + } + @Bean("idEnricherQueueFactory") public StageQueueFactory idEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); } + @Bean("idUpdaterQueueFactory") + public StageQueueFactory idUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity()); + } + + @Bean("noOpQueueFactory") + public StageQueueFactory noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) { + return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); + } + @Bean("partFileQueueFactory") public StageQueueFactory partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) { return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity()); diff --git a/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java b/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java index fbc6765..f6397e6 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/app/UserPackage.java @@ -24,4 +24,5 @@ public class UserPackage { private LocalDateTime monitoringAt; private LocalDateTime lastSearchedAt; private String rejectStatus; + private String externalPackageName; } diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index daf359d..b741734 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -8,6 +8,7 @@ import ru.nbch.credit_tracker.entities.app.PhonesMap; @Mapper public interface PhonesMapMapper { List getCachedData(@Param("phones") List phones); + List searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("limit") int limit, @Param("offset") int offset); PhonesMap findByPhone(@Param("phone") Long phone); void updateIdCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json); diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java new file mode 100644 index 0000000..5db5767 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/EmptyTerminatePayload.java @@ -0,0 +1,6 @@ +package ru.nbch.credit_tracker.model.impl; + +import ru.nbch.credit_tracker.model.TerminatePayload; + +public class EmptyTerminatePayload implements TerminatePayload { +} diff --git a/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java b/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java new file mode 100644 index 0000000..2eb4e8b --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/model/impl/VoidPayload.java @@ -0,0 +1,24 @@ +package ru.nbch.credit_tracker.model.impl; + +import lombok.Getter; +import ru.nbch.credit_tracker.enums.PoisonType; +import ru.nbch.credit_tracker.model.StagePayload; + +@Getter +public class VoidPayload implements StagePayload { + private final PoisonType poison; + public static final VoidPayload POISON = new VoidPayload(PoisonType.NORMALLY); + + public VoidPayload(PoisonType poison) { + this.poison = poison; + } + + public VoidPayload() { + this.poison = PoisonType.NONE; + } + + @Override + public boolean isPoison() { + return poison == PoisonType.NORMALLY; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java index 3aa67b9..9365d12 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFlagsService.java @@ -4,15 +4,16 @@ import java.util.List; import lombok.extern.slf4j.Slf4j; import org.slf4j.MDC; import org.springframework.scheduling.annotation.EnableScheduling; -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.enums.Status; +import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; import ru.nbch.credit_tracker.service.task.Pipeline; import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; 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.impl.PipelineMode; @@ -26,19 +27,27 @@ import ru.nbch.credit_tracker.service.task.impl.PipelineMode; public class UpdatingFlagsService { private final Long daysToUpdateFlags; private final UserPackageMapper userPackageMapper; + private final PhonesMapMapper phonesMapMapper; private final PipelinePackageFabric pipelinePackageManager; private final PipelinePackageOrchestrator pipelineOrchestrator; + private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; - public UpdatingFlagsService(CreditTrackerSettings config, UserPackageMapper userPackageMapper, - PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator) { + public UpdatingFlagsService(CreditTrackerSettings config, + UserPackageMapper userPackageMapper, + PhonesMapMapper phonesMapMapper, + PipelinePackageFabric pipelinePackageManager, + PipelinePackageOrchestrator pipelineOrchestrator, + PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator) { this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1; this.userPackageMapper = userPackageMapper; + this.phonesMapMapper = phonesMapMapper; this.pipelinePackageManager = pipelinePackageManager; this.pipelineOrchestrator = pipelineOrchestrator; + this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; } - @Scheduled(cron = "${ct.update-flags-cron-expression}") - private void update() { +// @Scheduled(cron = "${ct.update-flags-cron-expression}") + private void updateOld() { log.info("Start updating flags"); List packages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name()); for (UserPackage userPackage : packages) { @@ -59,4 +68,15 @@ public class UpdatingFlagsService { } log.info("Finish updating flags"); } + +// @Scheduled(cron = "${ct.update-flags-cron-expression}") + private void update() { + log.info("Start updating flags"); + try { + Pipeline pipeline = pipelinePackageManager.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); + pipelineRecalcPDOrchestrator.startOrReject(pipeline); + } catch (Exception e) { + log.error("Failed to build pipeline", e); + } + } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index cd3d596..21ce561 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -8,12 +8,16 @@ import org.springframework.beans.factory.annotation.Qualifier; 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.EmptyTerminatePayload; import ru.nbch.credit_tracker.model.impl.PackageVariable; import ru.nbch.credit_tracker.model.impl.PartFilePayload; import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload; import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; +import ru.nbch.credit_tracker.service.task.impl.FidIdUpdater; import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; +import ru.nbch.credit_tracker.service.task.impl.FidNameUpdater; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; @@ -24,29 +28,48 @@ import ru.nbch.credit_tracker.service.task.setup.PipeLineBuilderV2; public class PipelinePackageFabric implements InitializingBean { private final FlagsEnricher flagsEnricherV2; private final StageQueueFactory idEnricherQueueFactory; + private final StageQueueFactory idUpdaterQueueFactory; private final StageQueueFactory nameEnricherQueueFactory; + private final StageQueueFactory nameUpdaterQueueFactory; private final StageQueueFactory sendSignalQueueFactory; + private final StageQueueFactory noOpQueueFactory; private final FidIdEnricher fidIdEnricherHandler; + private final FidIdUpdater fidIdUpdater; private final FidNameEnricher fidNameEnricher; + private final FidNameUpdater fidNameUpdater; private final Supplier> pipelineTerminatorFactory; + private final Supplier> noOpTerminatorFactory; private final Function> pipelineStarterFactory; private final CreditTrackerSettings creditTrackerSettings; private final PipelineBackoff pipelineBackoff; public PipelinePackageFabric(@Qualifier("idEnricherQueueFactory") StageQueueFactory idEnricherQueueFactory, + @Qualifier("idUpdaterQueueFactory") StageQueueFactory idUpdaterQueueFactory, @Qualifier("nameEnricherQueueFactory") StageQueueFactory nameEnricherQueueFactory, + @Qualifier("nameUpdaterQueueFactory") StageQueueFactory nameUpdaterQueueFactory, @Qualifier("partFileQueueFactory") StageQueueFactory sendSignalQueueFactory, + @Qualifier("noOpQueueFactory") StageQueueFactory noOpQueueFactory, FidIdEnricher fidIdEnricherHandler, + FidIdUpdater fidIdUpdater, + FidNameUpdater fidNameUpdater, Supplier> pipelineTerminatorFactory, + Supplier> noOpTerminatorFactory, Function> pipelineStarterFactory, - FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher, + FlagsEnricher flagsEnricherV2, + FidNameEnricher fidNameEnricher, CreditTrackerSettings creditTrackerSettings, PipelineBackoff pipelineBackoff) { this.idEnricherQueueFactory = idEnricherQueueFactory; + this.idUpdaterQueueFactory = idUpdaterQueueFactory; this.nameEnricherQueueFactory = nameEnricherQueueFactory; + this.nameUpdaterQueueFactory = nameUpdaterQueueFactory; this.sendSignalQueueFactory = sendSignalQueueFactory; + this.noOpQueueFactory = noOpQueueFactory; this.fidIdEnricherHandler = fidIdEnricherHandler; + this.fidIdUpdater = fidIdUpdater; + this.fidNameUpdater = fidNameUpdater; this.pipelineTerminatorFactory = pipelineTerminatorFactory; + this.noOpTerminatorFactory = noOpTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; this.flagsEnricherV2 = flagsEnricherV2; this.fidNameEnricher = fidNameEnricher; @@ -69,6 +92,20 @@ public class PipelinePackageFabric implements InitializingBean { .build(); } + public Pipeline buildRecalculationPD(SetupParam setupParam) throws Exception { + return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) + .handler(fidNameUpdater) + .andThen() + .stage(noOpQueueFactory.getObject()) + .handler(fidIdUpdater) + .andThen() + .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) + .pollFinishingQueue(noOpTerminatorFactory.get()) + .backoff(pipelineBackoff) + .context(PackageVariable.class) + .build(); + } + @Override public void afterPropertiesSet() throws Exception { log.info("pipeline settings: {}", creditTrackerSettings.getPipeline()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java new file mode 100644 index 0000000..5defcc9 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java @@ -0,0 +1,76 @@ +package ru.nbch.credit_tracker.service.task; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +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.model.impl.EmptyTerminatePayload; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; + +@Service +public class PipelineRecalcPDOrchestrator { + 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; + this.pipelineExecutorService = pipelineExecutorService; + this.semaphore = new Semaphore(1); + } + + public void startOrReject(Pipeline pipeline) { + synchronized (lock) { + if (!semaphore.tryAcquire()) { + log.debug("couldn't acquire permit from semaphore. task rejected..."); + return; + } + log.info("throttler permits after acquire: {}", semaphore.availablePermits()); + } + innerStart(pipeline); + } + + private void innerStart(Pipeline pipeline) { + try { + pipeline.start(pipelineExecutorService); + addCallback(pipeline); + } catch (Exception e) { + log.error(ExceptionUtils.getStackTrace(e)); + pipeline.close(); + release(); + } + } + + private void addCallback(Pipeline pipeline) { + 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()); + EmptyTerminatePayload res = (EmptyTerminatePayload) v; + } catch (Exception ex) { + log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex)); + } finally { + log.info("pipeline finished"); + pipeline.close(); + release(); + log.info("semaphore amount: {}", semaphore.availablePermits()); + } + }); + } + + private void release() { + log.debug("checking for delayed pipelines..."); + synchronized (lock) { + semaphore.release(); + } + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java new file mode 100644 index 0000000..6b9a5aa --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java @@ -0,0 +1,80 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.indic.Id; +import ru.nbch.credit_tracker.model.IdData; +import ru.nbch.credit_tracker.model.SubjectProfile; +import ru.nbch.credit_tracker.model.impl.SubjectPayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; +import ru.nbch.credit_tracker.service.indic.PersonServiceV2; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.service.task.Handler; +import ru.nbch.credit_tracker.utils.CTUtil; +import static ru.nbch.credit_tracker.utils.CTUtil.defineIdType; + +@Service +@Slf4j +public class FidIdUpdater implements Handler { + private final PersonServiceV2 personService; + private final PackageRecordsService packageRecordsService; + private final CreditTrackerSettings creditTrackerSettings; + + public FidIdUpdater(PersonServiceV2 personService, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { + this.personService = personService; + this.packageRecordsService = packageRecordsService; + this.creditTrackerSettings = creditTrackerSettings; + } + + @Override + public VoidPayload handle(SubjectPayload taskPayload) { + List subjectsProfile = taskPayload.getSubjects(); + log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); + + List noIdDataFids = subjectsProfile.stream() + .map(SubjectProfile::getFid) + .toList(); + Map idByFid = personService.filledPassportsData(noIdDataFids); + + List idCacheOnUpdate = new ArrayList<>(); + List personOnProcessing = subjectsProfile.stream() + .flatMap(subjectProfile -> { + Id id = idByFid.get(subjectProfile.getFid()); + if (id == null) { + log.warn("Id not found for fid: {}", subjectProfile.getFid()); + return Stream.empty(); + } + String serNum = CTUtil.normalizePassport(id); + IdData idData = IdData.builder() + .serNum(serNum) + .idType(defineIdType(id.getIdType(), id.getIdType2())) + .fid(subjectProfile.getFid()) + .phone(subjectProfile.getPhone()) + .build(); + return Stream.of(idData); + }) + .toList(); + + packageRecordsService.updateIdCache(idCacheOnUpdate); + log.debug("personOnProcessing size: {}", personOnProcessing.size()); + + return new VoidPayload(); + } + + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getPipeline().idEnricherStageConcurrency(); + } + + @Override + public VoidPayload cookPoison() { + return VoidPayload.POISON; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java new file mode 100644 index 0000000..88216cf --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -0,0 +1,67 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.indic.Name; +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.service.indic.PersonServiceV2; +import ru.nbch.credit_tracker.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.service.task.Handler; +import ru.nbch.credit_tracker.utils.CTUtil; + +@Service +@Slf4j +public class FidNameUpdater implements Handler { + private final PersonServiceV2 personService; + private final CreditTrackerSettings creditTrackerSettings; + private final PackageRecordsService packageRecordsService; + + public FidNameUpdater(PersonServiceV2 personService, + CreditTrackerSettings creditTrackerSettings, + PackageRecordsService packageRecordsService) { + this.personService = personService; + this.creditTrackerSettings = creditTrackerSettings; + this.packageRecordsService = packageRecordsService; + } + + @Override + public SubjectPayload handle(SubjectPayload taskPayload) { + List subjects = taskPayload.getSubjects(); + var ctx = obtainContext(PackageVariable.class, taskPayload); + log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size()); + Map nameByPhone = personService.findNameByNumber(subjects); + + List changingFidPhones = new ArrayList<>(); + List notFoundFidPhones = new ArrayList<>(); + subjects.forEach(subjectProfile -> { + Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone().toString())); + if (name == null) { + notFoundFidPhones.add(subjectProfile.getPhone()); + } else if (subjectProfile.getFid().equals(name.getFid())) { + changingFidPhones.add(subjectProfile); + } + }); + //1.set to null fid name and id data in signals_phones_map + //1.2 set FID_CHANGED = 1 for this package(select by phone) + + //2.set new fid name and id data in signals_phones_map + //2.2 set FID_CHANGED = 1 for this package(select by phone) + return new SubjectPayload(null, null, null, changingFidPhones); + } + + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); + } + + @Override + public SubjectPayload cookPoison() { + return SubjectPayload.POISON; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java new file mode 100644 index 0000000..e586504 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java @@ -0,0 +1,115 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.entities.app.PhonesMap; +import ru.nbch.credit_tracker.entities.app.UserPackage; +import ru.nbch.credit_tracker.enums.PoisonType; +import ru.nbch.credit_tracker.exceptions.SignalClientException; +import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.model.PipelineContext; +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.service.signal.PackageRecordsService; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; +import ru.nbch.credit_tracker.utils.error.ExceptionUtils; + +public abstract class KickRecalcPipeLine implements IPipeLineStarter { + private final Logger log = LoggerFactory.getLogger(getClass()); + protected final Long packageId; + protected final UserPackageMapper userPackageMapper; + protected final PhonesMapMapper phonesMapMapper; + protected final PackageRecordsService packageRecordsService; + protected final CreditTrackerSettings creditTrackerSettings; + private final int BATCH_SIZE; + private final Long daysToUpdateFlags; + private boolean isInterrupted = false; + protected int offset = 0; + private PipelineContext context; + + + public KickRecalcPipeLine(SetupParam setupParam, + UserPackageMapper userPackageMapper, + PhonesMapMapper phonesMapMapper, + PackageRecordsService packageRecordsService, + CreditTrackerSettings creditTrackerSettings) { + this.packageId = setupParam.packageId(); + this.userPackageMapper = userPackageMapper; + this.phonesMapMapper = phonesMapMapper; + this.packageRecordsService = packageRecordsService; + this.creditTrackerSettings = creditTrackerSettings; + this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize(); + this.daysToUpdateFlags = creditTrackerSettings.getDaysToUpdateFlags() != null ? creditTrackerSettings.getDaysToUpdateFlags() : 1; + } + + private static class PipelineStarterException extends SignalClientException { + public PipelineStarterException() { + super(null, null, null); + } + } + + @SuppressWarnings("unchecked") + @Override + public void setContext(PipelineContext context) { + this.context = (PipelineContext) context; + } + + @Override + public void startFillingQueue(BlockingQueue firstQueue) { + try { + log.info("Start kicker: mode:{}, packageId: {};", mode(), packageId); + context.store(PackageVariable.packageId, packageId); + context.store(PackageVariable.cacheMisses, new AtomicInteger()); + context.store(PackageVariable.cacheHits, new AtomicInteger()); + int limit = BATCH_SIZE; + UserPackage userPackage = userPackageMapper.findPackage(packageId); + List page = nextPage(limit); + while (!page.isEmpty()) { + log.debug("Get page {}", page.size()); + List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() + .mapId(phoneFidMap.getId()) + .phone(phoneFidMap.getPhone()) + .build()).toList(); + try { + SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles); + payload.setContext(context); + firstQueue.put(payload); + } catch (InterruptedException e) { + isInterrupted = true; + log.error("Pipeline starter was interrupted!"); + throw new PipelineStarterException(); + } + offset += limit; + page = nextPage(limit); + } + } catch (PipelineStarterException e) { + log.info("Pipeline starter shutting down..."); + } finally { + if (isInterrupted) { + //noinspection ReturnInsideFinallyBlock + return; + } + log.debug("Send poison pill"); + try { + SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY); + firstQueue.put(poison); + } catch (InterruptedException e) { + log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e)); + } + } + } + + protected PipelineMode mode() { + return PipelineMode.REPEATED_RESTORE; + } + protected List nextPage(Integer limit) { + return phonesMapMapper.searchedAtLessThan(daysToUpdateFlags.intValue(), limit, BATCH_SIZE); + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java new file mode 100644 index 0000000..f736916 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/NoOpTerminator.java @@ -0,0 +1,54 @@ +package ru.nbch.credit_tracker.service.task.impl; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.exception.ExceptionUtils; +import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload; +import ru.nbch.credit_tracker.model.impl.VoidPayload; +import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; + +@Slf4j +public final class NoOpTerminator implements IPipeLineTerminator { + private final CompletableFuture done = new CompletableFuture<>(); + private BlockingQueue inQueue; + + public NoOpTerminator() {} + + @Override + public void run() { + while (!Thread.currentThread().isInterrupted()) { + try { + VoidPayload payload = inQueue.poll(5, TimeUnit.SECONDS); + if (payload == null) { + continue; + } + if (payload.isPoison()) { + log.debug("Got poison"); + done.complete(new EmptyTerminatePayload()); + return; + } + } catch (InterruptedException ie) { + log.info("pipeline terminator was interrupted!"); + done.completeExceptionally(new PipelineTerminatorException()); + } catch (Exception e) { + log.error("Error: {}", ExceptionUtils.getStackTrace(e)); + } + } + done.completeExceptionally(new PipelineTerminatorException()); + } + + public static class PipelineTerminatorException extends RuntimeException { + } + + @Override + public void setPollingQueue(BlockingQueue mergerQueue) { + this.inQueue = mergerQueue; + } + + @Override + public CompletableFuture doneFuture() { + return done; + } +} diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java index bab15ac..3aa221e 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/PipelineMode.java @@ -8,5 +8,6 @@ package ru.nbch.credit_tracker.service.task.impl; public enum PipelineMode { DEFAULT, RESTORE, - REPEATED_RESTORE + REPEATED_RESTORE, + RECALC_PERSONAL_DATA } diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index 224db0a..425abf6 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -37,6 +37,25 @@ WHERE phone = #{phone} + + @@ -63,8 +64,8 @@ diff --git a/src/main/resources/sql/v2.21.ddl b/src/main/resources/sql/v2.21.ddl index 1f1a067..79cc789 100644 --- a/src/main/resources/sql/v2.21.ddl +++ b/src/main/resources/sql/v2.21.ddl @@ -9,6 +9,8 @@ alter table signals_user_packages add signal_types varchar(255); alter table signals_user_packages add REJECT_STATUS varchar(255); +alter table signals_user_packages + add EXTERNAL_PACKAGE_NAME varchar(255); alter table signals_hits