LP: aAdded updating fids for lp; Added abstraction for PipelineOrchestrators

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.update-np-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *}
-ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30}
+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}

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

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.*;
-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.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("^(?<member>[A-Za-z0-9]{6})_(?<package>[A-Za-z0-9_]+)_(?<ts>\\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);

src/main/java/ru/nbch/credit_tracker/mapper/signal/LegalsMapMapper.java

@@ -32,4 +32,12 @@ public interface LegalsMapMapper {
                                        @Param("lastId") Long lastId);
 
     List<LegalsMap> 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();
 }

src/main/java/ru/nbch/credit_tracker/mapper/signal/PackagesRecordsMapper.java

@@ -19,6 +19,8 @@ public interface PackagesRecordsMapper {
 
     List<Long> uniquePackageIdByPhones(@Param("json") String json);
 
+    List<Long> uniquePackageIdByInnAndRegNum(@Param("json") String json);
+
     List<UserPackagePhoneData> selectUserPackagePhoneData(@Param("json") String json);
 
     List<UserPackageLegalData> selectUserPackageLegalData(@Param("json") String json);

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<LegalsMap> subjects) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(subjects.stream()
+                    .map(phoneFidMap -> new HashMap<String, Object>() {{
+                        put("inn", phoneFidMap.getInn());
+                        put("reg_num", phoneFidMap.getRegNum());
+                    }}).toList());
+
+            legalsMapMapper.updateSearchedAt(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void resetFidAndFlagsByInnAndRegNum(List<LegalsMap> subjects) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(subjects.stream()
+                    .map(phoneFidMap -> new HashMap<String, Object>() {{
+                        put("inn", phoneFidMap.getInn());
+                        put("reg_num", phoneFidMap.getRegNum());
+                    }}).toList());
+
+            legalsMapMapper.resetFidAndFlagsByInnAndRegNum(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void updateFidCachedData(List<LegalsMap> subjects) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(subjects.stream()
+                    .map(phoneFidMap -> new HashMap<String, Object>() {{
+                        put("inn", phoneFidMap.getInn());
+                        put("reg_num", phoneFidMap.getRegNum());
+                        put("fid", phoneFidMap.getFid());
+                    }}).toList());
+
+            legalsMapMapper.updateFidCachedData(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

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() {

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.*;
-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;
 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 LpPipelinePackageFabric lpPipelinePackageFabric;
-    private final PipelinePackageOrchestrator pipelinePackageOrchestrator;
+    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,
+                              LpPipelinePackageFabric lpPipelinePackageFabric,
-                              PipelinePackageOrchestrator pipelinePackageOrchestrator,
+                              NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator,
+                              LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator,
+                              NpPipelinePackageOrchestrator npPipelinePackageOrchestrator,
+                              LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator,
                               UserPackageMapper userPackageMapper,
-                              PhonesMapService phonesMapService) {
+                              PhonesMapService phonesMapService,
+                              LegalsMapMapper legalsMapMapper) {
         this.npPipelinePackageFabric = npPipelinePackageFabric;
-        this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator;
+        this.lpPipelinePackageFabric = lpPipelinePackageFabric;
-        this.pipelinePackageOrchestrator = pipelinePackageOrchestrator;
+        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<UserPackage> 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);
+        }
     }
 }

src/main/java/ru/nbch/credit_tracker/service/signal/PackageRecordsService.java

@@ -113,7 +113,7 @@ public class PackageRecordsService {
         }
     }
 
-    public List<Long> selectUniquePackageIds(List<Long> phones) {
+    public List<Long> selectNpUniquePackageIds(List<Long> phones) {
         try {
             String json = entityToJsonMapper.writeValueAsString(phones.stream()
                     .filter(Objects::nonNull)
@@ -126,6 +126,20 @@ public class PackageRecordsService {
         }
     }
 
+    public List<Long> selectLpUniquePackageIds(List<LegalsMap> legalsMaps) {
+        try {
+            List<InnRegnumPair> 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<UserPackagePhoneData> selectUserPackagePhoneData(Set<Long> phones) {
         try {
             String json = entityToJsonMapper.writeValueAsString(phones.stream()

src/main/java/ru/nbch/credit_tracker/service/task/LpFidEnricher.java

@@ -144,7 +144,7 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
 
     @Override
     public int preferredConcurrency() {
-        return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency();
+        return creditTrackerSettings.getLpPipeline().nameEnricherStageConcurrency();
     }
 
     @Override

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<SubjectPayload> lpFinishSetupQueueFactory;
     private final StageQueueFactory<VoidPayload> noOpQueueFactory;
     private final LpFidEnricher lpFidEnricher;
+    private final LpFidUpdater lpFidUpdater;
     private final Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory;
     private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory;
     private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
@@ -37,6 +39,7 @@ public class LpPipelinePackageFabric implements InitializingBean {
                                    @Qualifier("lpFinishSetupQueueFactory") StageQueueFactory<SubjectPayload> lpFinishSetupQueueFactory,
                                    @Qualifier("lpNoOpQueueFactory") StageQueueFactory<VoidPayload> lpNoOpQueueFactory,
                                    LpFidEnricher lpFidEnricher,
+                                   LpFidUpdater lpFidUpdater,
                                    Supplier<IPipeLineTerminator<SubjectPayload, SignalPayload>> pipelineTerminatorFactory,
                                    Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
                                    Function<SetupParam, IPipeLineStarter<SubjectPayload>> 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();
+    }
 }

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

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

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

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

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

src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java

@@ -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<Launch> 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);
-    }
-}

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.PipelineSettings;
-import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 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 abstract class RecalcOrchestrator {
-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,
+    public RecalcOrchestrator(ExecutorService pipelineExecutorService) {
-                                        ExecutorService pipelineExecutorService) {
-        this.creditTrackerSettings = creditTrackerSettings;
         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);
@@ -58,7 +54,7 @@ public class PipelineRecalcPDOrchestrator {
     private CompletableFuture<?> addCallback(Pipeline pipeline) {
         return pipeline.getTerminator()
                 .doneFuture()
-            .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS);
+                .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();
 }

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<SubjectPayload, VoidPayload> {
+    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<SubjectProfile> subjects = taskPayload.getSubjects();
+        var ctx = obtainContext(PackageVariable.class, taskPayload);
+        log.debug("LpFidUpdaterStage handle, size of bucket for save: {}", subjects.size());
+
+        Map<String, List<FidIdData>> fidsByInnMap = personService.findFidsByInn(subjects);
+        Map<String, List<FidIdData>> fidsByRegNumMap = personService.findFidsByRegNum(subjects);
+
+        List<LegalsMap> changingFidLegalMaps = new ArrayList<>();
+        List<LegalsMap> notFoundFidLegalMaps = new ArrayList<>();
+        List<LegalsMap> notChangingFidLegalMaps = new ArrayList<>();
+
+        for (SubjectProfile subject : subjects) {
+            Long inn = subject.getInn();
+            Long regNum = subject.getRegNum();
+
+            List<FidIdData> fidByInn = fidsByInnMap.get(CTUtil.normalizeInnZeroesOnly(inn));
+            List<FidIdData> fidByRegNum = fidsByRegNumMap.get(CTUtil.normalizeRegNumZeroesOnly(regNum));
+
+            if (fidByInn == null || fidByInn.isEmpty()) {
+                // fid не найден
+                notFoundFidLegalMaps.add(fillLegalMap(subject, null));
+                continue;
+            }
+
+            List<Integer> fidsByInn = fidByInn.stream().map(FidIdData::getFid).toList();
+
+            if (fidByRegNum != null && !fidByRegNum.isEmpty()) {
+                List<Integer> fidsByRegNum = fidByRegNum.stream().map(FidIdData::getFid).toList();
+
+                // находим все пересечения по фидам
+                List<Integer> 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<LegalsMap> 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<Long> 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;
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/task/impl/LpKickRecalcPipeLine.java

@@ -44,7 +44,7 @@ public class LpKickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
         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 {

src/main/java/ru/nbch/credit_tracker/service/task/impl/NpFidUpdater.java

@@ -80,7 +80,7 @@ public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
         }
 
         if (!changingFids.isEmpty()) {
-            List<Long> packageIds = packageRecordsService.selectUniquePackageIds(changingFids);
+            List<Long> packageIds = packageRecordsService.selectNpUniquePackageIds(changingFids);
             if (!packageIds.isEmpty()) {
                 userPackageService.setFidChanged(packageIds);
             }

src/main/java/ru/nbch/credit_tracker/service/task/impl/NpKickRecalcPipeLine.java

@@ -44,7 +44,7 @@ public class NpKickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
         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 {

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}
     </select>
+
+    <update id="updateSearchedAt" parameterType="string">
+        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>
+
+    <update id="resetFidAndFlagsByInnAndRegNum" parameterType="string">
+        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>
+
+
+    <update id="updateFidCachedData" parameterType="string">
+        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
+    </update>
+
+
+    <delete id="deleteUnusedMaps">
+        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'
+                                    ))
+    </delete>
+
 </mapper>

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'
     </select>
 
+    <select id="uniquePackageIdByInnAndRegNum" resultType="java.lang.Long">
+        WITH src AS (SELECT inn, regNum
+                     FROM json_to_recordset(#{json}::json) AS t(
+                                                                inn bigint,
+                                                                regNum bigint
+                         ))
+        select distinct (package_id)
+        FROM signals_legal_fid_map slfm
+                 JOIN signals_user_packages sup ON sup.id = slfm.package_id
+        WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, regNum FROM src)
+          AND sup.status = 'ON_MONITORING'
+    </select>
+
     <select id="selectUserPackagePhoneData" resultType="ru.nbch.credit_tracker.model.UserPackagePhoneData">
         WITH src AS (SELECT phone
                      FROM json_to_recordset(#{json}::json) AS t(