get rid of setting reject status

credit-tracker.properties

@@ -22,9 +22,9 @@ ct.indic-data-source-settings.leak-detection-threshold=${LEAK_DETECTION_THRESHOL
 
 # signal
 ct.signal-data-source-settings.driver=org.postgresql.Driver
-ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.230.227.125:5432/credit_tracker?currentSchema=test}
+ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.200.200.133:5432/credit_tracker?currentSchema=test_3}
 ct.signal-data-source-settings.username=${USERNAME_SIGNAL:credit_tracker_user}
-ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:password}
+ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-}
 ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:40}
 ct.signal-data-source-settings.minimum-idle=${MINIMUM_IDLE_SIGNAL:10}
 ct.signal-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SIGNAL:10000}

src/main/java/ru/nbch/credit_tracker/component/comparator/DeletePackageStatusComparator.java

@@ -1,9 +1,10 @@
 package ru.nbch.credit_tracker.component.comparator;
 
+import ru.nbch.credit_tracker.entities.app.UserPackage;
+import ru.nbch.credit_tracker.enums.Status;
+
 import java.util.Comparator;
 import java.util.function.ToIntFunction;
-import ru.nbch.credit_tracker.entities.app.UserPackage;
-import ru.nbch.credit_tracker.enums.Status;
 
 public class DeletePackageStatusComparator {
 
@@ -15,7 +16,7 @@ public class DeletePackageStatusComparator {
         try {
             Status status = Status.valueOf(str);
             return switch (status) {
-                case ON_MONITORING, PROCESSING_REPEATED, NO_FID -> 1;
+                case ON_MONITORING -> 1;
                 case DELETED -> 0;
                 default -> -1;
             };

src/main/java/ru/nbch/credit_tracker/component/comparator/ReportListPackageStatusComparator.java

@@ -1,15 +1,16 @@
 package ru.nbch.credit_tracker.component.comparator;
 
+import ru.nbch.credit_tracker.entities.app.UserPackage;
+import ru.nbch.credit_tracker.enums.Status;
+
 import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.function.ToIntFunction;
-import ru.nbch.credit_tracker.entities.app.UserPackage;
-import ru.nbch.credit_tracker.enums.Status;
 
 public class ReportListPackageStatusComparator {
 
     public static final EnumSet<Status> reportedStatuses = EnumSet.of(
-        Status.ON_MONITORING, Status.PROCESSING_REPEATED
+        Status.ON_MONITORING
     );
 
     public static Comparator<UserPackage> INSTANCE = Comparator
@@ -24,7 +25,7 @@ public class ReportListPackageStatusComparator {
         try {
             Status status = Status.valueOf(str);
             return switch (status) {
-                case ON_MONITORING, PROCESSING_REPEATED -> 1;
+                case ON_MONITORING -> 1;
                 default -> -1;
             };
         } catch (Throwable e) {

src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java

@@ -1,14 +1,13 @@
 package ru.nbch.credit_tracker.config.kafka;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
-import org.apache.kafka.common.serialization.StringSerializer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
-import org.springframework.kafka.core.*;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.support.serializer.JsonDeserializer;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@@ -27,7 +26,6 @@ public class KafkaConfig {
 
     @Bean("signalKafkaEventQueue")
     public BlockingQueue<SignalKafkaEvent> signalKafkaEventQueue(CreditTrackerSettings settings) {
-        // TODO add capacity
         return new LinkedBlockingQueue<>();
     }
 
@@ -66,19 +64,4 @@ public class KafkaConfig {
                 );
     }
 
-    @Bean
-    public ProducerFactory<String, String> producerFactory() {
-        Map<String, Object> props = new HashMap<>();
-        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
-        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
-        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        return new DefaultKafkaProducerFactory<>(props);
-    }
-
-    @Bean
-    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
-        return new KafkaTemplate<>(producerFactory);
-    }
-
 }

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

@@ -10,20 +10,40 @@ import java.util.concurrent.Executors;
 @Configuration
 public class ExecutorsConfig {
 
-    @Bean(destroyMethod = "shutdown")
-    public ExecutorService pipelineExecutorService() {
+    @Bean(name = "npPipelineExecutorService",
+            destroyMethod = "shutdown")
+    public ExecutorService npPipelineExecutorService() {
         return Executors.newFixedThreadPool(15, r -> {
             Thread t = new Thread(r);
-            t.setName("pipe-starter-%d".formatted(t.threadId()));
+            t.setName("np-pipe-starter-%d".formatted(t.threadId()));
             return t;
         });
     }
 
-    @Bean
-    public ExecutorService saverExecutorService(CreditTrackerSettings creditTrackerSettings) {
+    @Bean(name = "lpPipelineExecutorService",
+            destroyMethod = "shutdown")
+    public ExecutorService lpPipelineExecutorService() {
+        return Executors.newFixedThreadPool(15, r -> {
+            Thread t = new Thread(r);
+            t.setName("lp-pipe-starter-%d".formatted(t.threadId()));
+            return t;
+        });
+    }
+
+    @Bean("npSaverExecutorService")
+    public ExecutorService npSaverExecutorService(CreditTrackerSettings creditTrackerSettings) {
         return Executors.newFixedThreadPool(creditTrackerSettings.getNpPipeline().parallelRequests(), r -> {
             Thread t = new Thread(r);
-            t.setName("subject-saver-%d".formatted(t.threadId()));
+            t.setName("np-subject-saver-%d".formatted(t.threadId()));
+            return t;
+        });
+    }
+
+    @Bean("lpSaverExecutorService")
+    public ExecutorService lpSaverExecutorService(CreditTrackerSettings creditTrackerSettings) {
+        return Executors.newFixedThreadPool(creditTrackerSettings.getNpPipeline().parallelRequests(), r -> {
+            Thread t = new Thread(r);
+            t.setName("lp-subject-saver-%d".formatted(t.threadId()));
             return t;
         });
     }

src/main/java/ru/nbch/credit_tracker/enums/Status.java

@@ -2,10 +2,8 @@ package ru.nbch.credit_tracker.enums;
 
 public enum Status {
     ON_MONITORING, // пакет стоит на мониторинге в оф. сервисе
-    NO_FID,
     REJECTED, // отклонен на стороне НБКИ
     PROCESSING, // идет поиск телефона и расчет флагов по пакету
-    PROCESSING_REPEATED,
     ACCEPTED,
     DELETED, // пакет удален
     ERROR; // произошла ошибка при постановке пакета на мониторинг в оф. сервис

src/main/java/ru/nbch/credit_tracker/facade/SignalFacade.java

@@ -223,8 +223,7 @@ public class SignalFacade {
             pack.setPackageName(data.getPackageName());
             pack.setCreatedAt(data.getCreatedAt() != null ? data.getCreatedAt().toString() : null);
             Status packageStatus = EnumUtil.getEnum(Status.class, data.getStatus());
-            Status statusForClient = packageStatus == Status.ACCEPTED ? Status.PROCESSING :
-                Arrays.asList(Status.PROCESSING_REPEATED, Status.NO_FID).contains(packageStatus) ? Status.ON_MONITORING : packageStatus;
+            Status statusForClient = packageStatus == Status.ACCEPTED ? Status.PROCESSING : packageStatus;
             assert statusForClient != null;
             pack.setStatus(statusForClient.name());
             if (packageStatus == Status.DELETED) {
@@ -244,7 +243,7 @@ public class SignalFacade {
                 -> clientExcp(SignalError.ERROR_016, XmlErrorFormat.MonitoringPackageDeleteError, packageName));
 
         Status status = EnumUtil.getEnum(Status.class, packageToDelete.getStatus());
-        if (!Arrays.asList(Status.ON_MONITORING, Status.NO_FID).contains(status)) {
+        if (Status.ON_MONITORING != status) {
             throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName);
         }
 
@@ -266,7 +265,6 @@ public class SignalFacade {
             case PROCESSING, ACCEPTED -> SignalError.ERROR_023;
             case REJECTED, ERROR -> SignalError.ERROR_024;
             case DELETED -> SignalError.ERROR_025;
-            case PROCESSING_REPEATED -> SignalError.ERROR_026;
             default -> throw new IllegalStateException("Unexpected value: " + status);
         };
     }

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

@@ -14,4 +14,6 @@ public interface LegalsFidMapMapper {
     List<SignalLegalFidMap> findFullLegalFidMap(@Param("packageId") Long packageId,
                                                 @Param("limit") int limit,
                                                 @Param("offset") int offset);
+
+    Long deleteByPackageId(@Param("packageId") Long id);
 }

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

@@ -31,7 +31,7 @@ public interface LegalsMapMapper {
                                        @Param("batchSize") int batchSize,
                                        @Param("lastId") Long lastId);
 
-    List<LegalsMap> findCachedByInnAndRegNUm(@Param("json") String json);
+    List<LegalsMap> findCachedByInnAndRegNum(@Param("json") String json);
 
     void updateSearchedAt(@Param("json") String json);
 

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

@@ -17,10 +17,6 @@ public interface PackagesRecordsMapper {
                                                  @Param("limit") int limit,
                                                  @Param("offset") int offset);
 
-    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/mapper/signal/UserPackageMapper.java

@@ -25,14 +25,8 @@ public interface UserPackageMapper {
     void updatePackageOnMonitoringSuccess(@Param("id") Long id,
                                           @Param("monitoringAt") LocalDateTime monitoringAt);
 
-    void updateNoFidMonitoring(@Param("id") Long id);
-
-    void setFidChanged(@Param("packageIds") List<Long> id);
-
     UserPackage findPackage(@Param("id") Long id);
 
-    List<UserPackage> findForRepeatedPackages();
-
     List<UserPackage> findUserPackages(@Param("userId") String userId,
                                        @Param("statuses") List<String> status);
 
@@ -47,7 +41,5 @@ public interface UserPackageMapper {
     List<UserPackage> findByMemberCodeAndPackageName(@Param("memberCode") String memberCode,
                                                      @Param("packageName") String packageName);
 
-    void setRejectedStatus(@Param("id") Long id, @Param("status") String status);
-
     void updatePackageLastDownloadedReport(@Param("json") String json);
 }

src/main/java/ru/nbch/credit_tracker/model/SignalKafkaEvent.java

@@ -5,6 +5,7 @@ import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 import ru.nbch.credit_tracker.enums.EventType;
 
 import java.time.LocalDateTime;
+import java.util.Map;
 
 @Getter
 @Setter
@@ -13,6 +14,7 @@ import java.time.LocalDateTime;
 @AllArgsConstructor
 public class SignalKafkaEvent {
     private EventType eventType;
+    private Map<String, String> reportNames;
 
     private UserPackagePhoneData userPackagePhoneData;
     private UserPackageLegalData userPackageLegalData;

src/main/java/ru/nbch/credit_tracker/model/impl/SubjectPayload.java

@@ -17,6 +17,8 @@ public class SubjectPayload implements StagePayload {
     public static final SubjectPayload POISON = new SubjectPayload(null, PoisonType.NORMALLY);
     private final PoisonType poison;
 
+    private boolean hasFids = false;
+
     public SubjectPayload(Long packageId, PoisonType poison) {
         this.memberCode = null;
         this.packageName = null;
@@ -59,6 +61,14 @@ public class SubjectPayload implements StagePayload {
         return packageName;
     }
 
+    public boolean isHasFids() {
+        return hasFids;
+    }
+
+    public void setHasFids(boolean hasFids) {
+        this.hasFids = hasFids;
+    }
+
     @Override
     public boolean isPoison() {
         return poison == PoisonType.NORMALLY;

src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java

@@ -59,17 +59,13 @@ public class UserPackageService {
     }
 
     public List<UserPackage> findActivePackages(String userId) {
-        return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name(), Status.NO_FID.name()));
+        return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name()));
     }
 
     public UserPackage findByPackageName(String packageName) {
         return userPackageMapper.findByPackageName(packageName);
     }
 
-    public void setFidChanged(List<Long> id) {
-        userPackageMapper.setFidChanged(id);
-    }
-
     public List<UserPackage> findAllPackages(String userId) {
         return userPackageMapper.findAllPackagesByUserId(userId.substring(0, 6));
     }

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

@@ -6,11 +6,10 @@ 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.LegalsFidMapMapper;
 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.NpPipelinePackageOrchestrator;
-import ru.nbch.credit_tracker.service.task.SetupParam;
+import ru.nbch.credit_tracker.service.task.*;
 import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
 import ru.nbch.credit_tracker.service.task.impl.Saver;
 import ru.nbch.credit_tracker.utils.CTUtil;
@@ -31,21 +30,30 @@ public class StartupRestoreService {
 
     private final UserPackageMapper userPackageMapper;
     private final PackagesRecordsMapper phoneFidMapMapper;
+    private final LegalsFidMapMapper legalsFidMapMapper;
     private final NpPipelinePackageFabric npPipelinePackageManager;
-    private final NpPipelinePackageOrchestrator pipelineOrchestrator;
+    private final NpPipelinePackageOrchestrator npPipelineOrchestrator;
+    private final LpPipelinePackageFabric lpPipelinePackageManager;
+    private final LpPipelinePackageOrchestrator lpPipelineOrchestrator;
     private final CreditTrackerSettings creditTrackerSettings;
     private final Saver saver;
 
     public StartupRestoreService(UserPackageMapper userPackageMapper,
                                  PackagesRecordsMapper phoneFidMapMapper,
+                                 LegalsFidMapMapper legalsFidMapMapper,
                                  NpPipelinePackageFabric npPipelinePackageManager,
-                                 NpPipelinePackageOrchestrator pipelineOrchestrator,
+                                 NpPipelinePackageOrchestrator npPipelineOrchestrator,
+                                 LpPipelinePackageFabric lpPipelinePackageManager,
+                                 LpPipelinePackageOrchestrator lpPipelineOrchestrator,
                                  CreditTrackerSettings creditTrackerSettings,
                                  Saver saver) {
         this.userPackageMapper = userPackageMapper;
         this.phoneFidMapMapper = phoneFidMapMapper;
+        this.legalsFidMapMapper = legalsFidMapMapper;
         this.npPipelinePackageManager = npPipelinePackageManager;
-        this.pipelineOrchestrator = pipelineOrchestrator;
+        this.npPipelineOrchestrator = npPipelineOrchestrator;
+        this.lpPipelinePackageManager = lpPipelinePackageManager;
+        this.lpPipelineOrchestrator = lpPipelineOrchestrator;
         this.creditTrackerSettings = creditTrackerSettings;
         this.saver = saver;
     }
@@ -53,39 +61,64 @@ public class StartupRestoreService {
     public void run() {
         acceptedRestore();
         processingRestore();
-        // TODO LP
     }
 
     private void acceptedRestore() {
         List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name());
         log.debug("Found {} packages in status ACCEPTED", packages.size());
         for (UserPackage userPackage : packages) {
-            try {
-                Long count = phoneFidMapMapper.deleteByPackageId(userPackage.getId());
-                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(), 1, PipelineMode.DEFAULT));
-                pipelineOrchestrator.startOrDelay(
-                    () -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
-                    pipeline, userPackage.getId()
-                );
-            } catch (Exception e) {
-                log.error("Error build default pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+            if (userPackage.getSubjectType() == 1) {
+                try {
+                    Long count = phoneFidMapMapper.deleteByPackageId(userPackage.getId());
+                    log.trace("Deleted {} np 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(), 1, PipelineMode.DEFAULT));
+                    npPipelineOrchestrator.startOrDelay(
+                            () -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
+                            pipeline, userPackage.getId()
+                    );
+                } catch (Exception e) {
+                    log.error("Error build default np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+                }
+            } else {
+                try {
+                    Long count = legalsFidMapMapper.deleteByPackageId(userPackage.getId());
+                    log.trace("Deleted {} lp rows", count);
+                    Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage);
+                    log.debug("Try to read and build lp pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
+                    Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));
+                    lpPipelineOrchestrator.startOrDelay(
+                            () -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
+                            pipeline, userPackage.getId()
+                    );
+                } catch (Exception e) {
+                    log.error("Error build default lp pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+                }
             }
         }
     }
 
     private void processingRestore() {
-        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name(), Status.PROCESSING_REPEATED.name());
+        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name());
         log.debug("Found {} packages in status PROCESSING", packages.size());
         for (UserPackage userPackage : packages) {
-            try {
-                log.debug("Build pipeline for packageId {}", userPackage.getId());
-                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));
+            if (userPackage.getSubjectType() == 1) {
+                try {
+                    log.debug("Build np pipeline for packageId {}", userPackage.getId());
+                    Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT));
+                    npPipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
+                } catch (Exception e) {
+                    log.error("Error build restore np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+                }
+            } else {
+                try {
+                    log.debug("Build lp pipeline for packageId {}", userPackage.getId());
+                    Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));
+                    lpPipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
+                } catch (Exception e) {
+                    log.error("Error build restore lp pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+                }
             }
         }
     }

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

@@ -5,15 +5,12 @@ 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.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.*;
 import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
 import ru.nbch.pipeline.engine.core.Pipeline;
 
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 @Slf4j
@@ -24,9 +21,6 @@ public class UpdatingFidService {
     private final LpPipelinePackageFabric lpPipelinePackageFabric;
     private final NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator;
     private final LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator;
-    private final NpPipelinePackageOrchestrator npPipelinePackageOrchestrator;
-    private final LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator;
-    private final UserPackageMapper userPackageMapper;
     private final PhonesMapService phonesMapService;
     private final LegalsMapMapper legalsMapMapper;
 
@@ -35,18 +29,12 @@ public class UpdatingFidService {
                               LpPipelinePackageFabric lpPipelinePackageFabric,
                               NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator,
                               LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator,
-                              NpPipelinePackageOrchestrator npPipelinePackageOrchestrator,
-                              LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator,
-                              UserPackageMapper userPackageMapper,
                               PhonesMapService phonesMapService,
                               LegalsMapMapper legalsMapMapper) {
         this.npPipelinePackageFabric = npPipelinePackageFabric;
         this.lpPipelinePackageFabric = lpPipelinePackageFabric;
         this.npPipelineRecalcOrchestrator = npPipelineRecalcOrchestrator;
         this.lpPipelineRecalcOrchestrator = lpPipelineRecalcOrchestrator;
-        this.npPipelinePackageOrchestrator = npPipelinePackageOrchestrator;
-        this.lpPipelinePackageOrchestrator = lpPipelinePackageOrchestrator;
-        this.userPackageMapper = userPackageMapper;
         this.phonesMapService = phonesMapService;
         this.legalsMapMapper = legalsMapMapper;
     }
@@ -59,7 +47,7 @@ public class UpdatingFidService {
         }
         log.info("Delete unused phones");
         phonesMapService.deleteUnusedPhones();
-        log.info("Start updating fids and PD");
+        log.info("Start updating fids");
         try {
             Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, 1, PipelineMode.RECALC_FID));
             CompletableFuture<?> future = npPipelineRecalcOrchestrator.startOrReject(pipeline);
@@ -70,17 +58,7 @@ public class UpdatingFidService {
                         return;
                     }
 
-                    List<UserPackage> packagesRepeated = userPackageMapper.findForRepeatedPackages();
-                    packagesRepeated.forEach(userPackage -> {
-                        Long packageId = userPackage.getId();
-                        SetupParam setupParam = new SetupParam(packageId, 1, PipelineMode.RESEND);
-                        try {
-                            Pipeline makeXmlByCache = npPipelinePackageFabric.build(setupParam);
-                            npPipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId);
-                        } catch (Exception e) {
-                            log.error("Failed to build pipeline", e);
-                        }
-                    });
+                    log.info("Updating NP fids successfully completed.");
                 } finally {
                     log.info("pipeline finished");
                     pipeline.close();
@@ -98,7 +76,7 @@ public class UpdatingFidService {
             log.info("The previous task has not yet been completed.");
             return;
         }
-        log.info("Delete unused phones");
+        log.info("Delete unused lp maps");
         legalsMapMapper.deleteUnusedMaps();
         log.info("Start updating fids");
         try {
@@ -111,17 +89,7 @@ public class UpdatingFidService {
                         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);
-                        }
-                    });
+                    log.info("Updating LP fids successfully completed.");
                 } finally {
                     log.info("pipeline finished");
                     pipeline.close();

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java

@@ -8,6 +8,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.Hit;
@@ -23,20 +24,21 @@ import ru.nbch.credit_tracker.service.signal.HitsService;
 import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.signal.SegmentMapService;
 import ru.nbch.credit_tracker.utils.CTUtil;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Semaphore;
+import java.util.*;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
 @Slf4j
 @Service
 @DependsOn(value = "signalKafkaEventWriterService")
 public class KafkaMonitoringService {
+
+    private final ExecutorService processingExecutor;
+    private final ExecutorService dbExecutor;
+
     private volatile boolean isShuttingdDown = false;
     private final int permits;
     private final Semaphore processingPermits;
@@ -47,20 +49,32 @@ public class KafkaMonitoringService {
     private final SourceService sourceService;
     private final SegmentMapService segmentMapService;
     private final HitsService hitsService;
+    private final ReportNameRegistry reportNameRegistry;
 
     public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
                                   PackageRecordsService packageRecordsService,
                                   SourceService sourceService,
                                   SegmentMapService segmentMapService,
                                   CreditTrackerSettings settings,
-                                  HitsService hitsService) {
+                                  HitsService hitsService,
+                                  ReportNameRegistry reportNameRegistry) {
         this.signalEventsQueue = signalEventsQueue;
         this.packageRecordsService = packageRecordsService;
         this.sourceService = sourceService;
         this.segmentMapService = segmentMapService;
+        this.reportNameRegistry = reportNameRegistry;
         this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
         this.hitsService = hitsService;
         this.processingPermits = new Semaphore(permits);
+
+        this.processingExecutor = Executors.newFixedThreadPool(
+                permits * 2,
+                new CustomizableThreadFactory("kafka-processor-")
+        );
+        this.dbExecutor = Executors.newFixedThreadPool(
+                permits,
+                new CustomizableThreadFactory("kafka-db-writer-")
+        );
     }
 
     @KafkaListener(
@@ -90,8 +104,6 @@ public class KafkaMonitoringService {
     }
 
     private void process(List<PackageMessage> messages) {
-        List<Hit> hits = new ArrayList<>();
-
         Map<Boolean, List<PackageMessage>> partition = messages.stream()
                 .filter(this::filterMessages)
                 .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));
@@ -101,17 +113,43 @@ public class KafkaMonitoringService {
 
         Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
 
-        if (npMessages != null && !npMessages.isEmpty()) {
-            List<Hit> npHits = processNpMessages(npMessages, segmentMap);
-            hits.addAll(npHits);
-        }
+        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
+            if (npMessages != null && !npMessages.isEmpty()) {
+                return processNpMessages(npMessages, segmentMap);
+            }
+            return Collections.emptyList();
+        }, processingExecutor);
 
-        if (lpMessages != null && !lpMessages.isEmpty()) {
-            List<Hit> lpHits = processLpMessages(lpMessages, segmentMap);
-            hits.addAll(lpHits);
-        }
+        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
+            if (lpMessages != null && !lpMessages.isEmpty()) {
+                return processLpMessages(lpMessages, segmentMap);
+            }
+            return Collections.emptyList();
+        }, processingExecutor);
 
-        hitsService.insertHits(hits);
+        CompletableFuture<List<Hit>> combinedFuture = npFuture
+                .thenCombine(lpFuture, (npHits, lpHits) -> {
+                    List<Hit> allHits = new ArrayList<>();
+                    if (npHits != null) allHits.addAll(npHits);
+                    if (lpHits != null) allHits.addAll(lpHits);
+                    return allHits;
+                });
+
+        combinedFuture.thenAcceptAsync(hits -> {
+            if (!hits.isEmpty()) {
+                try {
+                    hitsService.insertHits(hits);
+                } catch (Exception e) {
+                    log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
+                }
+            }
+        }, dbExecutor);
+
+        try {
+            combinedFuture.get(1, TimeUnit.MINUTES);
+        } catch (Exception e) {
+            log.error("Error processing messages", e);
+        }
     }
 
     private List<Hit> processNpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
@@ -190,12 +228,17 @@ public class KafkaMonitoringService {
                         Integer segment = segmentMap.get(memberCode.substring(0, 6));
                         event.setSegment(segment != null ? segment : -1);
 
-                        putToQueue(event);
+                        String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
+                                + "_" + userPackagePhoneData.getPackageName();
+                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
+                        if (reportFileName == null) {
+                            continue; // событие не встало в очередь — hit не пишем
+                        }
                         Hit hit = Hit.builder()
                                 .packageId(userPackagePhoneData.getPackageId())
                                 .subjectId(subjectId)
                                 .signalType(signalType)
-                                .reportFileName("") // TODO
+                                .reportFileName(reportFileName.replace(".xml", ".zip"))
                                 .inqT(message.getInquiryType())
                                 .busCategory(busCategory)
                                 .inqDt(message.getInquiryDate())
@@ -290,12 +333,17 @@ public class KafkaMonitoringService {
                         Integer segment = segmentMap.get(memberCode.substring(0, 6));
                         event.setSegment(segment != null ? segment : -1);
 
-                        putToQueue(event);
+                        String reportKey = userPackageLegalData.getUserId().substring(0, 6)
+                                + "_" + userPackageLegalData.getPackageName();
+                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
+                        if (reportFileName == null) {
+                            continue; // событие не встало в очередь — hit не пишем
+                        }
                         Hit hit = Hit.builder()
                                 .packageId(userPackageLegalData.getPackageId())
                                 .subjectId(subjectId)
                                 .signalType(signalType)
-                                .reportFileName("") // TODO
+                                .reportFileName(reportFileName.replace(".xml", ".zip"))
                                 .inqT(message.getInquiryType())
                                 .busCategory(busCategory)
                                 .inqDt(message.getInquiryDate())
@@ -322,12 +370,18 @@ public class KafkaMonitoringService {
                 (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
     }
 
-    private void putToQueue(SignalKafkaEvent event) {
+    /**
+     * Резолвит имя результирующего файла и ставит событие в очередь.
+     *
+     * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
+     */
+    private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
         try {
-            signalEventsQueue.put(event);
+            return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
         } catch (InterruptedException e) {
             log.error("Error while putting signal event", e);
             Thread.currentThread().interrupt();
+            return null;
         }
     }
 
@@ -339,9 +393,7 @@ public class KafkaMonitoringService {
         processingPermits.acquire(permits);
 
         try {
-            SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
-            rotateEvent.setEventType(EventType.ROTATE);
-            signalEventsQueue.put(rotateEvent);
+            reportNameRegistry.enqueueRotate();
             log.info("ROTATE event sent to queue");
 
             SignalKafkaEvent stopEvent = new SignalKafkaEvent();

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/ReportNameRegistry.java

@@ -0,0 +1,72 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.enums.EventType;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+import ru.nbch.credit_tracker.utils.CTUtil;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * Реестр имён результирующих файлов текущего цикла ротации.
+ * Имя фиксируется при первом сигнале цикла для пакета (timestamp — момент первого сигнала).
+ * <p>
+ * событие, получившее имя эпохи N, встаёт в очередь раньше, чем ROTATE,
+ * закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через
+ * {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}.
+ */
+@Component
+public class ReportNameRegistry {
+
+    // ключ: membercode_packageName; значение: финальное имя xml-файла
+    private final ConcurrentHashMap<String, String> names = new ConcurrentHashMap<>();
+    private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+    private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
+
+    public ReportNameRegistry(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue) {
+        this.signalEventsQueue = signalEventsQueue;
+    }
+
+    /**
+     * Резолвит имя файла для пакета и атомарно с этим ставит WRITE-событие в очередь.
+     *
+     * @param key membercode_packageName
+     * @return имя xml-файла, в который попадёт сигнал
+     */
+    public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException {
+        lock.readLock().lock();
+        try {
+            String name = names.computeIfAbsent(key, k ->
+                    String.format("%s_%s.xml", k,
+                            LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter)));
+            signalEventsQueue.put(event);
+            return name;
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    /**
+     * Закрывает текущую эпоху: снапшот имён уезжает внутри ROTATE-события,
+     * реестр очищается атомарно с постановкой события в очередь.
+     */
+    public void enqueueRotate() throws InterruptedException {
+        lock.writeLock().lock();
+        try {
+            SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
+            rotateEvent.setEventType(EventType.ROTATE);
+            rotateEvent.setReportNames(new HashMap<>(names));
+            names.clear();
+            signalEventsQueue.put(rotateEvent);
+        } finally {
+            lock.writeLock().unlock();
+        }
+    }
+}

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/RotateSheduler.java

@@ -3,14 +3,10 @@ package ru.nbch.credit_tracker.service.mode.monitoring;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Component;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.enums.EventType;
-import ru.nbch.credit_tracker.model.SignalKafkaEvent;
 
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -23,12 +19,12 @@ public class RotateSheduler {
     );
     private volatile boolean isShuttingdDown = false;
 
-    private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
+    private final ReportNameRegistry reportNameRegistry;
     private final CreditTrackerSettings config;
 
-    public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
+    public RotateSheduler(ReportNameRegistry reportNameRegistry,
                           CreditTrackerSettings config) {
-        this.signalEventsQueue = signalEventsQueue;
+        this.reportNameRegistry = reportNameRegistry;
         this.config = config;
     }
 
@@ -44,9 +40,7 @@ public class RotateSheduler {
                 return;
             }
             try {
-                SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
-                rotateEvent.setEventType(EventType.ROTATE);
-                signalEventsQueue.put(rotateEvent);
+                reportNameRegistry.enqueueRotate();
             } catch (InterruptedException e) {
                 log.error("Error while putting signal event", e);
                 Thread.currentThread().interrupt();

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java

@@ -34,6 +34,7 @@ import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -85,7 +86,7 @@ public class SignalKafkaEventWriterService {
                 }
 
                 if (event.getEventType() == EventType.ROTATE) {
-                    rotate();
+                    rotate(event.getReportNames());
                 }
             }
         }
@@ -134,14 +135,16 @@ public class SignalKafkaEventWriterService {
     }
 
     /**
-     * Упаковывает все результирующие xml в архив
+     * Упаковывает все результирующие xml в архив.
+     *
+     * @param reportNames снапшот имён файлов закрываемой эпохи (ключ: membercode_packageName)
      */
-    private void rotate() {
+    private void rotate(Map<String, String> reportNames) {
         List<UserPackage> packagesToUpdate = new ArrayList<>();
         List<UserPackage> userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
         for (UserPackage userPackage : userPackages) {
             try {
-                String reportName = movedAndZip(userPackage);
+                String reportName = movedAndZip(userPackage, reportNames);
                 userPackage.setLastDownloadedReport(reportName);
                 packagesToUpdate.add(userPackage);
             } catch (IOException e) {
@@ -167,7 +170,7 @@ public class SignalKafkaEventWriterService {
         }
     }
 
-    private String movedAndZip(UserPackage userPackage) throws IOException {
+    private String movedAndZip(UserPackage userPackage, Map<String, String> reportNames) throws IOException {
         String finalName = "";
         String commonMembercode = userPackage.getUserId().substring(0, 6);
         Path tmpFilePath = Path.of(config.getLoadDir(),
@@ -184,11 +187,17 @@ public class SignalKafkaEventWriterService {
             if (!Files.exists(rootReportPath)) {
                 Files.createDirectories(rootReportPath);
             }
-            LocalDateTime newReportDate = LocalDateTime.now();
-            finalName = String.format("%s_%s_%s.xml",
-                    commonMembercode,
-                    userPackage.getPackageName(),
-                    newReportDate.format(CTUtil.responseReportDateTimeFormatter));
+
+            String reportKey = commonMembercode + "_" + userPackage.getPackageName();
+            finalName = reportNames != null ? reportNames.get(reportKey) : null;
+            if (finalName == null) {
+                // tmp-файл есть, а имени в реестре нет —
+                // генерируем имя по времени ротации
+                finalName = String.format("%s_%s_%s.xml",
+                        commonMembercode,
+                        userPackage.getPackageName(),
+                        LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter));
+            }
 
             Path responsePath = rootReportPath.resolve(finalName);
 

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

@@ -56,12 +56,13 @@ public class PackageRecordsService {
 
     public Map<Pair<Long, Long>, LegalsMap> findCachedLegalsByInnAndRegNums(List<SubjectProfile> subjects) {
         try {
-            List<InnRegnumPair> innAndRegNums = subjects.stream()
-                    .map(s -> new InnRegnumPair(s.getInn(), s.getRegNum()))
-                    .toList();
-            String json = entityToJsonMapper.writeValueAsString(innAndRegNums);
-
-            return legalsMapMapper.findCachedByInnAndRegNUm(json)
+            String json = entityToJsonMapper.writeValueAsString(subjects.stream()
+                    .filter(Objects::nonNull)
+                    .map(s -> new HashMap<String, Object>() {{
+                        put("inn", s.getInn());
+                        put("reg_num", s.getRegNum());
+                    }}).toList());
+            return legalsMapMapper.findCachedByInnAndRegNum(json)
                     .stream()
                     .collect(Collectors.toMap(
                             lm -> Pair.of(lm.getInn(), lm.getRegNum()),
@@ -109,33 +110,6 @@ public class PackageRecordsService {
         }
     }
 
-    public List<Long> selectNpUniquePackageIds(List<Long> phones) {
-        try {
-            String json = entityToJsonMapper.writeValueAsString(phones.stream()
-                    .filter(Objects::nonNull)
-                    .map(phone -> new HashMap<String, Object>() {{
-                        put("phone", phone);
-                    }}).toList());
-            return packagesRecordsMapper.uniquePackageIdByPhones(json);
-        } catch (JsonProcessingException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    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()
@@ -224,11 +198,13 @@ public class PackageRecordsService {
 
     public List<UserPackageLegalData> selectUserPackageLegalData(Set<Pair<Long, Long>> uniqueInnRegnumPairs) {
         try {
-            List<InnRegnumPair> pairs = uniqueInnRegnumPairs.stream()
-                    .map(pair -> new InnRegnumPair(pair.getKey(), pair.getValue()))
-                    .toList();
+            String json = entityToJsonMapper.writeValueAsString(uniqueInnRegnumPairs.stream()
+                    .filter(Objects::nonNull)
+                    .map(s -> new HashMap<String, Object>() {{
+                        put("inn", s.getKey());
+                        put("reg_num", s.getValue());
+                    }}).toList());
 
-            String json = entityToJsonMapper.writeValueAsString(pairs);
             return packagesRecordsMapper.selectUserPackageLegalData(json);
         } catch (JsonProcessingException e) {
             throw new RuntimeException(e);
@@ -249,6 +225,4 @@ public class PackageRecordsService {
         }
     }
 
-    public record InnRegnumPair(Long inn, Long regNum) {
-    }
 }

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

@@ -47,11 +47,12 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
         log.debug("LpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
         //найдем inn + regNum в табличном кэше
         Map<Pair<Long, Long>, LegalsMap> cachedRecords = packageRecordsService.findCachedLegalsByInnAndRegNums(subjects);
-        processByNumber(subjects, cachedRecords, ctx);
+        boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
+        taskPayload.setHasFids(hasFids);
         return taskPayload;
     }
 
-    private void processByNumber(List<SubjectProfile> subjects,
+    private boolean processByNumber(List<SubjectProfile> subjects,
                                  Map<Pair<Long, Long>, LegalsMap> cachedRecords,
                                  PipelineContext<PackageVariable> context) {
         List<SubjectProfile> noFidsData = subjects.stream()
@@ -125,7 +126,10 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
             .filter(legalsMap -> seen.add(Pair.of(legalsMap.getInn(), legalsMap.getRegNum())))
             .toList();
 
+
         packageRecordsService.saveLegalsToCache(subjectForSave);
+
+        return subjectForSave.stream().anyMatch(s -> s.getFid() != null) || (noFidsData.isEmpty() && !cachedRecords.isEmpty());
     }
 
     private void fillByCache(SubjectProfile subjectProfile, LegalsMap cachedRecord) {

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.task;
 
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.config.settings.PipelineSettings;
@@ -9,14 +10,12 @@ 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;
+                                         UserPackageMapper userPackageMapper,
+                                         @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService,
+                                         @Qualifier("lpSaverExecutorService") ExecutorService lpSaverExecutorService) {
+        super(creditTrackerSettings, userPackageMapper, lpPipelineExecutorService, lpSaverExecutorService);
     }
 
     @Override

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.task;
 
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.config.settings.PipelineSettings;
@@ -7,14 +8,11 @@ import ru.nbch.credit_tracker.config.settings.PipelineSettings;
 import java.util.concurrent.ExecutorService;
 
 @Service
-public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator{
-
-    private final CreditTrackerSettings creditTrackerSettings;
+public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator {
 
     public LpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
-                                        ExecutorService pipelineExecutorService) { // TODO another executor
-        super(pipelineExecutorService);
-        this.creditTrackerSettings = creditTrackerSettings;
+                                        @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService) {
+        super(creditTrackerSettings, lpPipelineExecutorService);
     }
 
     @Override

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.task;
 
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.config.settings.PipelineSettings;
@@ -9,14 +10,12 @@ 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;
+                                         @Qualifier("npPipelineExecutorService") ExecutorService npPipelineExecutorService,
+                                         @Qualifier("npSaverExecutorService") ExecutorService npSaverExecutorService) {
+        super(creditTrackerSettings, userPackageMapper, npPipelineExecutorService, npSaverExecutorService);
     }
 
     @Override

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.task;
 
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.config.settings.PipelineSettings;
@@ -9,11 +10,9 @@ 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;
+                                        @Qualifier("npPipelineExecutorService")ExecutorService npPipelineExecutorService) {
+        super(creditTrackerSettings, npPipelineExecutorService);
     }
 
     @Override

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

@@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.service.task;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+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 ru.nbch.credit_tracker.model.impl.PackageVariable;
@@ -20,6 +21,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 public abstract class PackageOrchestrator {
     private final Logger log = LoggerFactory.getLogger(getClass());
+    protected final CreditTrackerSettings creditTrackerSettings;
     private final UserPackageMapper userPackageMapper;
     private final ExecutorService pipelineExecutorService;
     private final ExecutorService saverExecutorService;
@@ -27,9 +29,11 @@ public abstract class PackageOrchestrator {
     private final java.util.Deque<PackageOrchestrator.Launch> pending = new java.util.ArrayDeque<>();
     private final Semaphore semaphore;
 
-    public PackageOrchestrator(UserPackageMapper userPackageMapper,
+    public PackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
+                               UserPackageMapper userPackageMapper,
                                ExecutorService pipelineExecutorService,
                                ExecutorService saverExecutorService) {
+        this.creditTrackerSettings = creditTrackerSettings;
         this.userPackageMapper = userPackageMapper;
         this.pipelineExecutorService = pipelineExecutorService;
         this.saverExecutorService = saverExecutorService;
@@ -108,8 +112,6 @@ public abstract class PackageOrchestrator {
                         }
                         if (!res.hasFids()) {
                             log.info("No one fid was found by packageId : {}.", packageId);
-                            userPackageMapper.updateNoFidMonitoring(packageId);
-                            return;
                         }
 
                         PipelineMode mode = ctx.get(PackageVariable.mode);

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

@@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.service.task;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.config.settings.PipelineSettings;
 import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 import ru.nbch.pipeline.engine.core.Pipeline;
@@ -13,11 +14,14 @@ import java.util.concurrent.TimeUnit;
 
 public abstract class RecalcOrchestrator {
     private final Logger log = LoggerFactory.getLogger(getClass());
+    protected final CreditTrackerSettings creditTrackerSettings;
     private final ExecutorService pipelineExecutorService;
     private final Object lock = new Object();
     private final Semaphore semaphore;
 
-    public RecalcOrchestrator(ExecutorService pipelineExecutorService) {
+    public RecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
+                              ExecutorService pipelineExecutorService) {
+        this.creditTrackerSettings = creditTrackerSettings;
         this.pipelineExecutorService = pipelineExecutorService;
         this.semaphore = new Semaphore(1);
     }

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

@@ -107,12 +107,6 @@ public class LpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
             legalsMapService.updateSearchedAt(notChangingFidLegalMaps);
         }
 
-        if (!changingFids.isEmpty()) {
-            List<Long> packageIds = packageRecordsService.selectLpUniquePackageIds(changingFids);
-            if (!packageIds.isEmpty()) {
-                userPackageService.setFidChanged(packageIds);
-            }
-        }
         return new VoidPayload();
     }
 

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

@@ -46,11 +46,12 @@ public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
         log.debug("NpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
         //найдем телефоны в табличном кэше
         Map<Long, PhonesMap> cachedRecords = packageRecordsService.findCachedPhones(subjects);
-        processByNumber(subjects, cachedRecords, ctx);
+        boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
+        taskPayload.setHasFids(hasFids);
         return taskPayload;
     }
 
-    private void processByNumber(List<SubjectProfile> subjects,
+    private boolean processByNumber(List<SubjectProfile> subjects,
                                  Map<Long, PhonesMap> cachedRecords,
                                  PipelineContext<PackageVariable> context) {
         List<SubjectProfile> noFidsData = subjects.stream()
@@ -97,6 +98,8 @@ public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
             .toList();
 
         packageRecordsService.saveToCache(subjectForSave);
+
+        return subjectForSave.stream().anyMatch(s -> s.getFid() != null) || (noFidsData.isEmpty() && !cachedRecords.isEmpty());
     }
 
     private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) {

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

@@ -8,9 +8,7 @@ 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.PersonService;
-import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
 import ru.nbch.credit_tracker.service.signal.PhonesMapService;
 import ru.nbch.credit_tracker.utils.CTUtil;
 import ru.nbch.pipeline.engine.core.Handler;
@@ -25,20 +23,14 @@ import java.util.stream.Stream;
 public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
     private final PersonService personService;
     private final CreditTrackerSettings creditTrackerSettings;
-    private final PackageRecordsService packageRecordsService;
     private final PhonesMapService phonesMapService;
-    private final UserPackageService userPackageService;
 
     public NpFidUpdater(PersonService personService,
                         CreditTrackerSettings creditTrackerSettings,
-                        PackageRecordsService packageRecordsService,
-                        PhonesMapService phonesMapService,
-                        UserPackageService userPackageService) {
+                        PhonesMapService phonesMapService) {
         this.personService = personService;
         this.creditTrackerSettings = creditTrackerSettings;
-        this.packageRecordsService = packageRecordsService;
         this.phonesMapService = phonesMapService;
-        this.userPackageService = userPackageService;
     }
 
     @Override
@@ -79,12 +71,6 @@ public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
             phonesMapService.updateSearchedAt(notChangedPhones);
         }
 
-        if (!changingFids.isEmpty()) {
-            List<Long> packageIds = packageRecordsService.selectNpUniquePackageIds(changingFids);
-            if (!packageIds.isEmpty()) {
-                userPackageService.setFidChanged(packageIds);
-            }
-        }
         return new VoidPayload();
     }
 

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

@@ -34,7 +34,7 @@ public final class SetupMonitoringPipelineTerminator implements IPipeLineTermina
                     continue;
                 }
                 if (!part.isPoison() && !hasFids.get()) {
-                    hasFids.set(part.getSubjects().stream().anyMatch(s -> s.getFid() != null));
+                    hasFids.set(part.isHasFids());
                 }
                 if (part.isPoison()) {
                     log.debug("Got poison");

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

@@ -49,4 +49,10 @@
         LIMIT #{limit} OFFSET #{offset}
     </select>
 
+    <delete id="deleteByPackageId" parameterType="long">
+        DELETE
+        FROM signals_legal_fid_map
+        WHERE package_id = #{packageId}
+    </delete>
+
 </mapper>

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

@@ -8,7 +8,7 @@
     <resultMap id="LegalsMap" type="ru.nbch.credit_tracker.entities.app.LegalsMap">
         <result property="id" column="id"/>
         <result property="inn" column="inn"/>
-        <result property="regNUm" column="reg_num"/>
+        <result property="regNum" column="reg_num"/>
         <result property="fid" column="fid"/>
         <result property="searchedAt" column="searched_at"/>
         <result property="flag90days" column="flag_90_12"/>
@@ -58,11 +58,11 @@
         </foreach>
     </select>
 
-    <select id="findCachedByInnAndRegNUm" resultMap="LegalsMap">
-        WITH src AS (SELECT inn, regNum
+    <select id="findCachedByInnAndRegNum" resultMap="LegalsMap">
+        WITH src AS (SELECT inn, reg_num
                      FROM json_to_recordset(#{json}::json) AS t(
                                                                 inn bigint,
-                                                                regNum bigint
+                                                                reg_num bigint
                          ))
         SELECT slm.id AS id,
                slm.fid as fid,
@@ -73,7 +73,7 @@
                slm.calculated_at AS calculated_at,
                slm.last_rejected_date AS last_rejected_date
         FROM signals_legals_map slm
-        WHERE (slm.inn, slm.reg_num) IN (SELECT inn, regNum FROM src)
+        WHERE (slm.inn, slm.reg_num) IN (SELECT inn, reg_num FROM src)
     </select>
 
     <select id="findEligibleForFlagsUpdate" resultMap="LegalsMap">
@@ -231,8 +231,7 @@
                             AND sup.status IN
                                 (
                                  'ON_MONITORING',
-                                 'PROCESSING',
-                                 'PROCESSING_REPEATED'
+                                 'PROCESSING'
                                     ))
     </delete>
 

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

@@ -42,30 +42,6 @@
         </foreach>
     </select>
 
-    <select id="uniquePackageIdByPhones" resultType="java.lang.Long">
-        WITH src AS (SELECT phone
-                     FROM json_to_recordset(#{json}::json) AS t(
-                                                                phone bigint
-                         ))
-        select distinct (package_id)
-        FROM signals_phone_fid_map spr
-                 join signals_user_packages sup on sup.id = spr.package_id
-        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(
@@ -78,17 +54,17 @@
     </select>
 
     <select id="selectUserPackageLegalData" resultType="ru.nbch.credit_tracker.model.UserPackageLegalData">
-        WITH src AS (SELECT inn, regNum
+        WITH src AS (SELECT inn, reg_num
                      FROM json_to_recordset(#{json}::json) AS t(
                                                                 inn bigint,
-                                                                regNum bigint
+                                                                reg_num bigint
                          ))
         SELECT sup.id AS packageId, sup.signal_types AS signalTypes, sup.membercode AS userId,
                sup.package_name AS packageName, slfm.inn AS inn, slfm.reg_num AS regNum,
                slfm.subject_id AS subjectId
         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)
+        WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, reg_num FROM src)
           AND sup.status = 'ON_MONITORING'
     </select>
 

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

@@ -210,8 +210,7 @@
                             AND sup.status IN
                                 (
                                  'ON_MONITORING',
-                                 'PROCESSING',
-                                 'PROCESSING_REPEATED'
+                                 'PROCESSING'
                                     ))
     </delete>
 </mapper>

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

@@ -14,9 +14,7 @@
         <result property="deactivatedAt" column="deactivated_at"/>
         <result property="lastDownloadedReport" column="last_downloaded_report"/>
         <result property="monitoringAt" column="monitoring_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 +61,8 @@
 
     <select id="insertPackage" resultType="java.lang.Long"
             parameterType="ru.nbch.credit_tracker.entities.app.UserPackage">
-        INSERT INTO signals_user_packages (membercode, package_name, subject_type, signal_types,  created_at, status, external_package_name)
-        values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.subjectType}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}, concat(substring(#{userPackage.userId}, 1, 6), '_', #{userPackage.packageName})) RETURNING id
+        INSERT INTO signals_user_packages (membercode, package_name, subject_type, signal_types,  created_at, status)
+        values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.subjectType}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}) RETURNING id
     </select>
 
     <update id="updatePackageStatus">
@@ -73,16 +71,6 @@
         WHERE id = #{packageId}
     </update>
 
-    <update id="setFidChanged">
-        UPDATE signals_user_packages
-        SET fid_changed = 1,
-            status = 'PROCESSING_REPEATED'
-        WHERE id in
-        <foreach item="packageId" collection="packageIds" open="(" separator="," close=")">
-            #{packageId}
-        </foreach>
-    </update>
-
     <update id="updatePackageLastDownloadedReport" parameterType="string">
         WITH src AS (
             SELECT
@@ -105,20 +93,7 @@
             <if test="monitoringAt != null">
                 monitoring_at = #{monitoringAt},
             </if>
-            status = 'ON_MONITORING',
-            fid_changed = null
-        WHERE id = #{id}
-    </update>
-
-    <update id="updateNoFidMonitoring">
-        UPDATE signals_user_packages
-        SET monitoring_at = created_at, status = 'NO_FID', fid_changed = null
-        WHERE id = #{id}
-    </update>
-
-    <update id="setRejectedStatus">
-        UPDATE signals_user_packages
-        SET reject_status = #{status}
+            status = 'ON_MONITORING'
         WHERE id = #{id}
     </update>
 
@@ -131,7 +106,6 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE id = #{id}
@@ -147,28 +121,12 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE package_name = #{packageName}
         LIMIT 1
     </select>
 
-    <select id="findForRepeatedPackages" resultMap="packageDataResultMap">
-        SELECT id,
-               membercode,
-               package_name,
-               created_at,
-               status,
-               deactivated_at,
-               last_downloaded_report,
-               monitoring_at,
-               reject_status,
-               signal_types
-        FROM signals_user_packages
-        WHERE fid_changed = 1 and status = 'PROCESSING_REPEATED'
-    </select>
-
     <select id="findOnMonitoringPackages" resultMap="packageDataResultMap">
         SELECT id,
                membercode,
@@ -178,10 +136,9 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
-        WHERE status = 'ON_MONITORING' and reject_status is null
+        WHERE status = 'ON_MONITORING'
     </select>
 
 
@@ -194,7 +151,6 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE membercode IS NOT NULL
@@ -214,7 +170,6 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE membercode IS NOT NULL
@@ -230,7 +185,6 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE status in
@@ -248,7 +202,6 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
-               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE LEFT(membercode, 6) = #{memberCode} AND package_name = #{packageName}