http://jira.mfd.msk:8088/browse/BKI-3241 [recalc person data]

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<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory(CreditTrackerSettings creditTrackerSettings) {
         return () -> new XmlMergeCoordinator(creditTrackerSettings);
     }
+
+    @Bean
+    public Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory() {
+        return NoOpTerminator::new;
+    }
 }

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<SubjectPayload> nameUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
+    }
+
     @Bean("idEnricherQueueFactory")
     public StageQueueFactory<SubjectPayload> idEnricherQueueFactory(CreditTrackerSettings creditTrackerSettings) {
         return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
     }
 
+    @Bean("idUpdaterQueueFactory")
+    public StageQueueFactory<SubjectPayload> idUpdaterQueueFactory(CreditTrackerSettings creditTrackerSettings) {
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().nameEnricherQueueCapacity());
+    }
+
+    @Bean("noOpQueueFactory")
+    public StageQueueFactory<VoidPayload> noOpQueueFactory(CreditTrackerSettings creditTrackerSettings) {
+        return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());
+    }
+
     @Bean("partFileQueueFactory")
     public StageQueueFactory<PartFilePayload> partFileQueueFactory(CreditTrackerSettings creditTrackerSettings) {
         return new StageQueueFactory<>(creditTrackerSettings.getPipeline().mergerQueueCapacity());

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

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<PhonesMap> getCachedData(@Param("phones") List<Long> phones);
+    List<PhonesMap> 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);

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 {
+}

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

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,
+    public UpdatingFlagsService(CreditTrackerSettings config,
-                                PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator) {
+                                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}")
+//    @Scheduled(cron = "${ct.update-flags-cron-expression}")
-    private void update() {
+    private void updateOld() {
         log.info("Start updating flags");
         List<UserPackage> 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);
+        }
+    }
 }

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<SubjectPayload> idEnricherQueueFactory;
+    private final StageQueueFactory<SubjectPayload> idUpdaterQueueFactory;
     private final StageQueueFactory<SubjectPayload> nameEnricherQueueFactory;
+    private final StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory;
     private final StageQueueFactory<PartFilePayload> sendSignalQueueFactory;
+    private final StageQueueFactory<VoidPayload> noOpQueueFactory;
     private final FidIdEnricher fidIdEnricherHandler;
+    private final FidIdUpdater fidIdUpdater;
     private final FidNameEnricher fidNameEnricher;
+    private final FidNameUpdater fidNameUpdater;
     private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
+    private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory;
     private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
     private final CreditTrackerSettings creditTrackerSettings;
     private final PipelineBackoff pipelineBackoff;
 
     public PipelinePackageFabric(@Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
+                                 @Qualifier("idUpdaterQueueFactory") StageQueueFactory<SubjectPayload> idUpdaterQueueFactory,
                                  @Qualifier("nameEnricherQueueFactory") StageQueueFactory<SubjectPayload> nameEnricherQueueFactory,
+                                 @Qualifier("nameUpdaterQueueFactory") StageQueueFactory<SubjectPayload> nameUpdaterQueueFactory,
                                  @Qualifier("partFileQueueFactory") StageQueueFactory<PartFilePayload> sendSignalQueueFactory,
+                                 @Qualifier("noOpQueueFactory") StageQueueFactory<VoidPayload> noOpQueueFactory,
                                  FidIdEnricher fidIdEnricherHandler,
+                                 FidIdUpdater fidIdUpdater,
+                                 FidNameUpdater fidNameUpdater,
                                  Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
+                                 Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
                                  Function<SetupParam, IPipeLineStarter<SubjectPayload>> 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());

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();
+        }
+    }
+}

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<SubjectPayload, VoidPayload> {
+    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<SubjectProfile> subjectsProfile = taskPayload.getSubjects();
+        log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
+
+        List<Integer> noIdDataFids = subjectsProfile.stream()
+            .map(SubjectProfile::getFid)
+            .toList();
+        Map<Integer, Id> idByFid = personService.filledPassportsData(noIdDataFids);
+
+        List<IdData> idCacheOnUpdate = new ArrayList<>();
+        List<IdData> 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;
+    }
+}

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<SubjectPayload, SubjectPayload> {
+    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<SubjectProfile> subjects = taskPayload.getSubjects();
+        var ctx = obtainContext(PackageVariable.class, taskPayload);
+        log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size());
+        Map<String, Name> nameByPhone = personService.findNameByNumber(subjects);
+
+        List<SubjectProfile> changingFidPhones = new ArrayList<>();
+        List<Long> 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;
+    }
+}

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<SubjectPayload> {
+    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<PackageVariable> 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<PackageVariable>) context;
+    }
+
+    @Override
+    public void startFillingQueue(BlockingQueue<SubjectPayload> 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<PhonesMap> page = nextPage(limit);
+            while (!page.isEmpty()) {
+                log.debug("Get page {}", page.size());
+                List<SubjectProfile> 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<PhonesMap> nextPage(Integer limit) {
+        return phonesMapMapper.searchedAtLessThan(daysToUpdateFlags.intValue(), limit, BATCH_SIZE);
+    }
+}

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<VoidPayload, EmptyTerminatePayload> {
+    private final CompletableFuture<EmptyTerminatePayload> done = new CompletableFuture<>();
+    private BlockingQueue<VoidPayload> 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<VoidPayload> mergerQueue) {
+        this.inQueue = mergerQueue;
+    }
+
+    @Override
+    public CompletableFuture<EmptyTerminatePayload> doneFuture() {
+        return done;
+    }
+}

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
 }

src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml

@@ -37,6 +37,25 @@
         WHERE phone = #{phone}
     </select>
 
+    <select id="searchedAtLessThan" resultMap="PhoneMap">
+        select fid,
+               phone,
+               name_1,
+               first,
+               middle,
+               birth_dt,
+               ser_num,
+               id_type,
+               flag_debt5000,
+               flag_90_12,
+               searched_at,
+               calculated_at
+        FROM signals_phones_map
+        WHERE searched_at &gt; now() - make_interval(days => #{daysCount})
+        ORDER BY id
+        LIMIT #{limit} OFFSET #{offset}
+    </select>
+
     <select id="getCachedData" resultMap="PhoneMap" resultType="java.util.List">
         SELECT fid,
         phone,

src/main/resources/mybatis/mapper/signal/UserPackageMapper.xml

@@ -17,6 +17,7 @@
         <result property="lastSearchedAt" column="last_searched_at"/>
         <result property="rejectStatus" column="reject_status"/>
         <result property="signalTypes" column="signal_types"/>
+        <result property="externalPackageName" column="external_package_name"/>
     </resultMap>
 
     <select id="isPackageNotUnique" resultType="boolean">
@@ -63,8 +64,8 @@
 
     <select id="insertPackage" resultType="java.lang.Long"
             parameterType="ru.nbch.credit_tracker.entities.app.UserPackage">
-        INSERT INTO signals_user_packages (membercode, package_name, signal_types,  created_at, status)
+        INSERT INTO signals_user_packages (membercode, package_name, signal_types,  created_at, status, external_package_name)
-        values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}) RETURNING id
+        values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}, concat(substring(#{serPackage.userId}, 1, 6), , '_', #{userPackage.packageName})) RETURNING id
     </select>
 
     <update id="updatePackageStatus">

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