NP: get rid of unused kick pipelines;

LP: added lp kick pipelines with data loading

src/main/java/ru/nbch/credit_tracker/config/task/PipelineStarterConfig.java

@@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.config.task;
 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.LegalsMapMapper;
 import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 import ru.nbch.credit_tracker.model.impl.EmptyTerminatePayload;
@@ -25,15 +26,24 @@ public class PipelineStarterConfig {
     public Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(CreditTrackerSettings creditTrackerSettings,
                                                                                          UserPackageMapper userPackageMapper,
                                                                                          PackageRecordsService packageRecordsService,
-                                                                                         PhonesMapMapper phonesMapMapper) {
+                                                                                         PhonesMapMapper phonesMapMapper,
-        // TODO separate for subject type
+                                                                                         LegalsMapMapper legalsMapMapper) {
-        return (setupParam) -> switch (setupParam.mode()) {
+        return (setupParam) -> {
-            case DEFAULT, RESEND, RESTORE ->
+            if (setupParam.subjectType() == 1) {
-                new KickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
+                return switch (setupParam.mode()) {
-            case REPEATED_RESTORE, MAKE_XML_BY_REPEATED ->
+                    case DEFAULT, RESEND ->
-                new KickPipeLineRepeatedRestoreImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
+                            new NpKickPipeLineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
-            case RECALC_PERSONAL_DATA ->
+                    case RECALC_FID ->
-                new KickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings);
+                            new NpKickRecalcPipeLine(setupParam, userPackageMapper, phonesMapMapper, packageRecordsService, creditTrackerSettings);
+                };
+            } else {
+                return switch (setupParam.mode()) {
+                    case DEFAULT, RESEND ->
+                            new LpKickPipelineDefaultImpl(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
+                    case RECALC_FID ->
+                            new LpKickRecalcPipeLine(setupParam, userPackageMapper, legalsMapMapper, packageRecordsService, creditTrackerSettings);
+                };
+            }
         };
     }
 

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

@@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.mapper.signal;
 
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
+import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap;
 
 import java.util.List;
 
@@ -10,4 +11,7 @@ public interface LegalsFidMapMapper {
 
     List<Long> registerJson(@Param("json") String json);
 
+    List<SignalLegalFidMap> findFullLegalFidMap(@Param("packageId") Long packageId,
+                                                @Param("limit") int limit,
+                                                @Param("offset") int offset);
 }

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

@@ -22,5 +22,9 @@ public interface LegalsMapMapper {
     List<LegalsMap> findEligibleForFlagsInsert(@Param("lastId") Long lastId,
                                                @Param("batchSize") Integer batchSize);
 
+    List<LegalsMap> searchedAtLessThan(@Param("daysCount") Integer daysCount,
+                                       @Param("batchSize") int batchSize,
+                                       @Param("lastId") Long lastId);
+
     List<LegalsMap> findCachedByInnAndRegNUm(@Param("json") String json);
 }

src/main/java/ru/nbch/credit_tracker/service/mode/common/StartupRestoreService.java

@@ -64,7 +64,7 @@ public class StartupRestoreService {
                 log.trace("Deleted {} rows", count);
                 Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage);
                 log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
-                Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT));
+                Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT));
                 pipelineOrchestrator.startOrDelay(
                     () -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
                     pipeline, userPackage.getId()
@@ -81,7 +81,7 @@ public class StartupRestoreService {
         for (UserPackage userPackage : packages) {
             try {
                 log.debug("Build pipeline for packageId {}", userPackage.getId());
-                Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT));
+                Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT));
                 pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
             } catch (Exception e) {
                 log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));

src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingFidService.java

@@ -51,7 +51,7 @@ public class UpdatingFidService {
         phonesMapService.deleteUnusedPhones();
         log.info("Start updating fids and PD");
         try {
-            Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA));
+            Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, 1, PipelineMode.RECALC_FID));
             CompletableFuture<?> future = pipelineRecalcPDOrchestrator.startOrReject(pipeline);
             future.whenComplete((o, throwable) -> {
                 try {
@@ -63,7 +63,7 @@ public class UpdatingFidService {
                     List<UserPackage> packagesRepeated = userPackageMapper.findForRepeatedPackages();
                     packagesRepeated.forEach(userPackage -> {
                         Long packageId = userPackage.getId();
-                        SetupParam setupParam = new SetupParam(packageId, null, PipelineMode.RESEND);
+                        SetupParam setupParam = new SetupParam(packageId, 1, PipelineMode.RESEND);
                         try {
                             Pipeline makeXmlByCache = npPipelinePackageFabric.build(setupParam);
                             pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId);
@@ -82,7 +82,7 @@ public class UpdatingFidService {
         }
     }
 
-    @Scheduled(cron = "${ct.update-fid-cron-expression}")
+    @Scheduled(cron = "${ct.update-fid-cron-expression}") // TODO another property
     public void updateLp() {
         // TODO
     }

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

@@ -8,6 +8,8 @@ import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.app.LegalsMap;
 import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap;
 import ru.nbch.credit_tracker.entities.app.PhonesMap;
+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 ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
 import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
@@ -33,15 +35,18 @@ public class PackageRecordsService {
     private final PhonesMapMapper phonesMapMapper;
     private final LegalsMapMapper legalsMapMapper;
     private final PackagesRecordsMapper packagesRecordsMapper;
+    private final LegalsFidMapMapper legalsFidMapMapper;
     private final ObjectMapper entityToJsonMapper;
 
     public PackageRecordsService(PhonesMapMapper phonesMapMapper,
                                  LegalsMapMapper legalsMapMapper,
                                  PackagesRecordsMapper packagesRecordsMapper,
+                                 LegalsFidMapMapper legalsFidMapMapper,
                                  ObjectMapper entityToJsonMapper) {
         this.phonesMapMapper = phonesMapMapper;
         this.legalsMapMapper = legalsMapMapper;
         this.packagesRecordsMapper = packagesRecordsMapper;
+        this.legalsFidMapMapper = legalsFidMapMapper;
         this.entityToJsonMapper = entityToJsonMapper;
     }
 
@@ -75,6 +80,10 @@ public class PackageRecordsService {
         return packagesRecordsMapper.findFullPhoneFidMap(packageId, limit, offset);
     }
 
+    public List<SignalLegalFidMap> findFullLegalFidMap(Long packageId, Integer limit, int offset) {
+        return legalsFidMapMapper.findFullLegalFidMap(packageId, limit, offset);
+    }
+
     public void saveToCache(List<PhonesMap> phonesMaps) {
         try {
             String json = entityToJsonMapper.writeValueAsString(phonesMaps.stream()

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

@@ -1,42 +0,0 @@
-package ru.nbch.credit_tracker.service.task.impl;
-
-import java.util.ArrayList;
-import java.util.List;
-import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap;
-import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
-import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
-import ru.nbch.credit_tracker.service.task.SetupParam;
-
-public class KickPipeLineRepeatedRestoreImpl extends KickPipeLine {
-
-    private boolean oldRowsWereFound = false;
-    private final Integer daysToUpdateFlags;
-    public KickPipeLineRepeatedRestoreImpl(SetupParam setupParam,
-                                           UserPackageMapper userPackageMapper,
-                                           PackageRecordsService packageRecordsService,
-                                           CreditTrackerSettings creditTrackerSettings) {
-        super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
-        this.daysToUpdateFlags = creditTrackerSettings.getDaysToUpdateFlags();
-    }
-
-    @Override
-    protected PipelineMode mode() {
-        return PipelineMode.RESTORE;
-    }
-
-    @Override
-    protected List<PackagesRecordsMap> nextPage(Integer limit) {
-        List<PackagesRecordsMap> pageWithNullFids = new ArrayList<>();
-        //fixme здесь должен быть запрос к таблице signals_phones_map, см. xml маппер
-//        if (!oldRowsWereFound) {
-//            pageWithNullFids = phoneFidMapMapper.pageByPackageIdAndOlderThanN(packageId, limit, offset, daysToUpdateFlags);
-//        }
-//        if (pageWithNullFids.isEmpty()) {
-//            oldRowsWereFound = true;
-//            offset = 0;
-//            return phoneFidMapMapper.pageByPackageIdAndNewerThanN(packageId, limit, offset, daysToUpdateFlags);
-//        }
-        return pageWithNullFids;
-    }
-}

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

@@ -1,42 +0,0 @@
-package ru.nbch.credit_tracker.service.task.impl;
-
-import java.util.ArrayList;
-import java.util.List;
-import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap;
-import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
-import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
-import ru.nbch.credit_tracker.service.task.SetupParam;
-
-public class KickPipeLineRestoreImpl extends KickPipeLine {
-
-    private boolean nullFidsWereFound = false;
-    public KickPipeLineRestoreImpl(SetupParam setupParam,
-                                   UserPackageMapper userPackageMapper,
-                                   PackageRecordsService packageRecordsService,
-                                   CreditTrackerSettings creditTrackerSettings) {
-        super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
-    }
-
-    @Override
-    protected PipelineMode mode() {
-        return PipelineMode.REPEATED_RESTORE;
-    }
-
-    @Override
-    protected List<PackagesRecordsMap> nextPage(Integer limit) {
-        //todo рассмотреть к удалению
-        List<PackagesRecordsMap> page = new ArrayList<>();
-//        if (!nullFidsWereFound) {
-//            page = phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset);
-//        }
-//        if (page.isEmpty()) {
-//            if (!nullFidsWereFound) {
-//                nullFidsWereFound = true;
-//                offset = 0;
-//            }
-//            return phoneFidMapMapper.findNotNullFidPhoneFidMap(packageId, limit, offset);
-//        }
-        return page;
-    }
-}

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

@@ -0,0 +1,111 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap;
+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.UserPackageMapper;
+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.utils.error.ExceptionUtils;
+import ru.nbch.pipeline.engine.core.PipelineContext;
+import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
+
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public abstract class LpKickPipeLine implements IPipeLineStarter<SubjectPayload> {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    protected final Long packageId;
+    protected final PipelineMode mode;
+    protected final UserPackageMapper userPackageMapper;
+    protected final PackageRecordsService packageRecordsService;
+    protected final CreditTrackerSettings creditTrackerSettings;
+    private final int BATCH_SIZE;
+    private boolean isInterrupted = false;
+    protected int offset = 0;
+    private PipelineContext<PackageVariable> context;
+
+
+    public LpKickPipeLine(SetupParam setupParam,
+                          UserPackageMapper userPackageMapper,
+                          PackageRecordsService packageRecordsService,
+                          CreditTrackerSettings creditTrackerSettings) {
+        this.packageId = setupParam.packageId();
+        this.userPackageMapper = userPackageMapper;
+        this.packageRecordsService = packageRecordsService;
+        this.creditTrackerSettings = creditTrackerSettings;
+        this.BATCH_SIZE = creditTrackerSettings.getLpPipeline().subjectBatchSize();
+        this.mode = setupParam.mode();
+    }
+
+    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());
+            context.store(PackageVariable.mode, mode);
+            int limit = BATCH_SIZE;
+            UserPackage userPackage = userPackageMapper.findPackage(packageId);
+            List<SignalLegalFidMap> page = nextPage(limit);
+            while (!page.isEmpty()) {
+                log.debug("Get page {}", page.size());
+                List<SubjectProfile> subjectProfiles = page.stream().map(legalFidMap -> SubjectProfile.builder()
+                    .mapId(legalFidMap.getId())
+                    .identifier(legalFidMap.getSubjectId())
+                    .inn(legalFidMap.getInn())
+                    .regNum(legalFidMap.getRegNum())
+                    .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 abstract PipelineMode mode();
+    protected abstract  List<SignalLegalFidMap> nextPage(Integer limit);
+}

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

@@ -0,0 +1,26 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.SignalLegalFidMap;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
+import ru.nbch.credit_tracker.service.task.SetupParam;
+
+import java.util.List;
+
+public class LpKickPipelineDefaultImpl extends LpKickPipeLine{
+
+    public LpKickPipelineDefaultImpl(SetupParam setupParam, UserPackageMapper userPackageMapper, PackageRecordsService packageRecordsService, CreditTrackerSettings creditTrackerSettings) {
+        super(setupParam, userPackageMapper, packageRecordsService, creditTrackerSettings);
+    }
+
+    @Override
+    protected PipelineMode mode() {
+        return PipelineMode.DEFAULT;
+    }
+
+    @Override
+    protected List<SignalLegalFidMap> nextPage(Integer limit) {
+        return packageRecordsService.findFullLegalFidMap(packageId, limit, offset);
+    }
+}

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

@@ -0,0 +1,112 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.LegalsMap;
+import ru.nbch.credit_tracker.enums.PoisonType;
+import ru.nbch.credit_tracker.exceptions.SignalClientException;
+import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+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.utils.error.ExceptionUtils;
+import ru.nbch.pipeline.engine.core.PipelineContext;
+import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
+
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+
+public class LpKickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    protected final Long packageId;
+    protected final UserPackageMapper userPackageMapper;
+    protected final LegalsMapMapper legalsMapMapper;
+    protected final PackageRecordsService packageRecordsService;
+    protected final CreditTrackerSettings creditTrackerSettings;
+    private final int BATCH_SIZE;
+    private final Long daysToUpdateFids;
+    private boolean isInterrupted = false;
+    private PipelineContext<PackageVariable> context;
+
+
+    public LpKickRecalcPipeLine(SetupParam setupParam,
+                                UserPackageMapper userPackageMapper,
+                                LegalsMapMapper legalsMapMapper,
+                                PackageRecordsService packageRecordsService,
+                                CreditTrackerSettings creditTrackerSettings) {
+        this.packageId = setupParam.packageId();
+        this.userPackageMapper = userPackageMapper;
+        this.legalsMapMapper = legalsMapMapper;
+        this.packageRecordsService = packageRecordsService;
+        this.creditTrackerSettings = creditTrackerSettings;
+        this.BATCH_SIZE = creditTrackerSettings.getLpPipeline().subjectBatchSize();
+        this.daysToUpdateFids = creditTrackerSettings.getDaysToUpdateFids() != null ? creditTrackerSettings.getDaysToUpdateFids() : 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:{};", mode());
+            long lastId = 0;
+            List<LegalsMap> page = nextPage(lastId);
+            while (!page.isEmpty()) {
+                log.debug("Get page {}", page.size());
+                List<SubjectProfile> subjectProfiles = page.stream().map(legalsMap -> SubjectProfile.builder()
+                        .mapId(legalsMap.getId())
+                        .inn(legalsMap.getInn())
+                        .regNum(legalsMap.getRegNum())
+                        .fid(legalsMap.getFid())
+                        .build()).toList();
+                try {
+                    SubjectPayload payload = new SubjectPayload(null, null, null, subjectProfiles);
+                    payload.setContext(context);
+                    firstQueue.put(payload);
+                } catch (InterruptedException e) {
+                    isInterrupted = true;
+                    log.error("Pipeline starter was interrupted!");
+                    throw new PipelineStarterException();
+                }
+                lastId = page.getLast().getId();
+                page = nextPage(lastId);
+            }
+        } 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.RECALC_FID;
+    }
+
+    protected List<LegalsMap> nextPage(long lastId) {
+        return legalsMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId);
+    }
+}

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

@@ -21,7 +21,7 @@ import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 
-public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
+public abstract class NpKickPipeLine implements IPipeLineStarter<SubjectPayload> {
     private final Logger log = LoggerFactory.getLogger(getClass());
     protected final Long packageId;
     protected final PipelineMode mode;
@@ -34,7 +34,7 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
     private PipelineContext<PackageVariable> context;
 
 
-    public KickPipeLine(SetupParam setupParam,
+    public NpKickPipeLine(SetupParam setupParam,
                           UserPackageMapper userPackageMapper,
                           PackageRecordsService packageRecordsService,
                           CreditTrackerSettings creditTrackerSettings) {

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

@@ -1,16 +1,16 @@
 package ru.nbch.credit_tracker.service.task.impl;
 
-import java.util.List;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.PackagesRecordsMap;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.task.SetupParam;
 
-public class KickPipeLineDefaultImpl extends KickPipeLine {
+import java.util.List;
-    private boolean cachedPhonesWereFound = false;
 
-    public KickPipeLineDefaultImpl(SetupParam setupParam,
+public class NpKickPipeLineDefaultImpl extends NpKickPipeLine {
+
+    public NpKickPipeLineDefaultImpl(SetupParam setupParam,
                                      UserPackageMapper userPackageMapper,
                                      PackageRecordsService packageRecordsService,
                                      CreditTrackerSettings creditTrackerSettings) {

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

@@ -20,7 +20,7 @@ import ru.nbch.pipeline.engine.core.setup.IPipeLineStarter;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 
-public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
+public class NpKickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
     private final Logger log = LoggerFactory.getLogger(getClass());
     protected final Long packageId;
     protected final UserPackageMapper userPackageMapper;
@@ -33,7 +33,7 @@ public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
     private PipelineContext<PackageVariable> context;
 
 
-    public KickRecalcPipeLine(SetupParam setupParam,
+    public NpKickRecalcPipeLine(SetupParam setupParam,
                                 UserPackageMapper userPackageMapper,
                                 PhonesMapMapper phonesMapMapper,
                                 PackageRecordsService packageRecordsService,
@@ -102,7 +102,7 @@ public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
     }
 
     protected PipelineMode mode() {
-        return PipelineMode.REPEATED_RESTORE;
+        return PipelineMode.RECALC_FID;
     }
     protected List<PhonesMap> nextPage(long lastId) {
         return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId);

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

@@ -8,8 +8,5 @@ package ru.nbch.credit_tracker.service.task.impl;
 public enum PipelineMode {
     DEFAULT,
     RESEND,
-    RESTORE,
+    RECALC_FID
-    REPEATED_RESTORE,
-    RECALC_PERSONAL_DATA,
-    MAKE_XML_BY_REPEATED
 }

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

@@ -5,6 +5,14 @@
 
 <mapper namespace="ru.nbch.credit_tracker.mapper.signal.LegalsFidMapMapper">
 
+    <resultMap id="LegalFidMapResultMap" type="ru.nbch.credit_tracker.entities.app.SignalLegalFidMap">
+        <result property="id" column="id"/>
+        <result property="packageId" column="package_id"/>
+        <result property="subjectId" column="subject_id"/>
+        <result property="inn" column="inn"/>
+        <result property="regNum" column="reg_num"/>
+    </resultMap>
+
     <select id="registerJson" parameterType="string" resultType="long">
         WITH src AS (SELECT row_number() OVER () AS rn,
                             subject_id,
@@ -29,4 +37,16 @@
             RETURNING id;
     </select>
 
+    <select id="findFullLegalFidMap" resultMap="LegalFidMapResultMap" resultType="java.util.List">
+        SELECT id,
+               package_id,
+               subject_id,
+               inn,
+               reg_num
+        FROM signals_legal_fid_map
+        WHERE package_id = #{packageId}
+        ORDER BY id
+        LIMIT #{limit} OFFSET #{offset}
+    </select>
+
 </mapper>

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

@@ -126,4 +126,19 @@
           AND dst.id IN (SELECT id FROM to_lock);
     </update>
 
+
+    <select id="searchedAtLessThan" resultMap="LegalsMap">
+        select id,
+               inn,
+               reg_num,
+               fid,
+               searched_at,
+               flag_90_12,
+               calculated_at,
+               last_rejected_date
+        FROM signals_legals_map
+        WHERE searched_at &lt; now() - make_interval(days => #{daysCount})  and id > #{lastId}
+        order by id
+        limit #{batchSize}
+    </select>
 </mapper>

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

@@ -10,15 +10,13 @@
         <result property="packageId" column="package_id"/>
         <result property="subjectId" column="subject_id"/>
         <result property="phone_l" column="phone_l"/>
-        <result property="errorText" column="error_text"/>
     </resultMap>
 
     <select id="findById" parameterType="long" resultMap="PhoneFidMapResultMap">
         SELECT id,
                package_id,
                subject_id,
-               phone_l,
+               phone_l
-               error_text
         FROM signals_phone_fid_map
         WHERE id = #{id}
     </select>
@@ -27,8 +25,7 @@
         SELECT id,
                package_id,
                subject_id,
-               phone_l,
+               phone_l
-               error_text
         FROM signals_phone_fid_map
         WHERE phone_l = #{phone}
     </select>
@@ -37,8 +34,7 @@
         SELECT id,
                package_id,
                subject_id,
-               phone_l,
+               phone_l
-               error_text
         FROM signals_phone_fid_map
         WHERE phone_l IN
         <foreach item='phone' collection='phones' open='(' separator=',' close=')'>
@@ -114,8 +110,7 @@
         SELECT id,
                package_id,
                subject_id,
-               phone_l,
+               phone_l
-               error_text
         FROM signals_phone_fid_map
         WHERE package_id = #{packageId}
         ORDER BY id