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.200.200.133:5432/credit_tracker?currentSchema=test_3}
+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.username=${USERNAME_SIGNAL:credit_tracker_user}
-ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-}
+ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:password}
 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,10 +1,9 @@
 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 {
 
@@ -16,7 +15,7 @@ public class DeletePackageStatusComparator {
         try {
             Status status = Status.valueOf(str);
             return switch (status) {
-                case ON_MONITORING -> 1;
+                case ON_MONITORING, PROCESSING_REPEATED, NO_FID -> 1;
                 case DELETED -> 0;
                 default -> -1;
             };

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

@@ -1,16 +1,15 @@
 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.ON_MONITORING, Status.PROCESSING_REPEATED
     );
 
     public static Comparator<UserPackage> INSTANCE = Comparator
@@ -25,7 +24,7 @@ public class ReportListPackageStatusComparator {
         try {
             Status status = Status.valueOf(str);
             return switch (status) {
-                case ON_MONITORING -> 1;
+                case ON_MONITORING, PROCESSING_REPEATED -> 1;
                 default -> -1;
             };
         } catch (Throwable e) {

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

@@ -1,13 +1,14 @@
 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.ConsumerFactory;
+import org.springframework.kafka.core.*;
-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;
@@ -26,6 +27,7 @@ public class KafkaConfig {
 
     @Bean("signalKafkaEventQueue")
     public BlockingQueue<SignalKafkaEvent> signalKafkaEventQueue(CreditTrackerSettings settings) {
+        // TODO add capacity
         return new LinkedBlockingQueue<>();
     }
 
@@ -64,4 +66,19 @@ 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,40 +10,20 @@ import java.util.concurrent.Executors;
 @Configuration
 public class ExecutorsConfig {
 
-    @Bean(name = "npPipelineExecutorService",
+    @Bean(destroyMethod = "shutdown")
-            destroyMethod = "shutdown")
+    public ExecutorService pipelineExecutorService() {
-    public ExecutorService npPipelineExecutorService() {
         return Executors.newFixedThreadPool(15, r -> {
             Thread t = new Thread(r);
-            t.setName("np-pipe-starter-%d".formatted(t.threadId()));
+            t.setName("pipe-starter-%d".formatted(t.threadId()));
             return t;
         });
     }
 
-    @Bean(name = "lpPipelineExecutorService",
+    @Bean
-            destroyMethod = "shutdown")
+    public ExecutorService saverExecutorService(CreditTrackerSettings creditTrackerSettings) {
-    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("np-subject-saver-%d".formatted(t.threadId()));
+            t.setName("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,8 +2,10 @@ 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,7 +223,8 @@ 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 : packageStatus;
+            Status statusForClient = packageStatus == Status.ACCEPTED ? Status.PROCESSING :
+                Arrays.asList(Status.PROCESSING_REPEATED, Status.NO_FID).contains(packageStatus) ? Status.ON_MONITORING : packageStatus;
             assert statusForClient != null;
             pack.setStatus(statusForClient.name());
             if (packageStatus == Status.DELETED) {
@@ -243,7 +244,7 @@ public class SignalFacade {
                 -> clientExcp(SignalError.ERROR_016, XmlErrorFormat.MonitoringPackageDeleteError, packageName));
 
         Status status = EnumUtil.getEnum(Status.class, packageToDelete.getStatus());
-        if (Status.ON_MONITORING != status) {
+        if (!Arrays.asList(Status.ON_MONITORING, Status.NO_FID).contains(status)) {
             throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName);
         }
 
@@ -265,6 +266,7 @@ 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,6 +14,4 @@ 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,6 +17,10 @@ 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,8 +25,14 @@ 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);
 
@@ -41,5 +47,7 @@ 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,7 +5,6 @@ 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
@@ -14,7 +13,6 @@ import java.util.Map;
 @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,8 +17,6 @@ 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;
@@ -61,14 +59,6 @@ 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,13 +59,17 @@ public class UserPackageService {
     }
 
     public List<UserPackage> findActivePackages(String userId) {
-        return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name()));
+        return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name(), Status.NO_FID.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,10 +6,11 @@ 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.*;
+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.impl.PipelineMode;
 import ru.nbch.credit_tracker.service.task.impl.Saver;
 import ru.nbch.credit_tracker.utils.CTUtil;
@@ -30,30 +31,21 @@ public class StartupRestoreService {
 
     private final UserPackageMapper userPackageMapper;
     private final PackagesRecordsMapper phoneFidMapMapper;
-    private final LegalsFidMapMapper legalsFidMapMapper;
     private final NpPipelinePackageFabric npPipelinePackageManager;
-    private final NpPipelinePackageOrchestrator npPipelineOrchestrator;
+    private final NpPipelinePackageOrchestrator pipelineOrchestrator;
-    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 npPipelineOrchestrator,
+                                 NpPipelinePackageOrchestrator pipelineOrchestrator,
-                                 LpPipelinePackageFabric lpPipelinePackageManager,
-                                 LpPipelinePackageOrchestrator lpPipelineOrchestrator,
                                  CreditTrackerSettings creditTrackerSettings,
                                  Saver saver) {
         this.userPackageMapper = userPackageMapper;
         this.phoneFidMapMapper = phoneFidMapMapper;
-        this.legalsFidMapMapper = legalsFidMapMapper;
         this.npPipelinePackageManager = npPipelinePackageManager;
-        this.npPipelineOrchestrator = npPipelineOrchestrator;
+        this.pipelineOrchestrator = pipelineOrchestrator;
-        this.lpPipelinePackageManager = lpPipelinePackageManager;
-        this.lpPipelineOrchestrator = lpPipelineOrchestrator;
         this.creditTrackerSettings = creditTrackerSettings;
         this.saver = saver;
     }
@@ -61,64 +53,39 @@ 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) {
-            if (userPackage.getSubjectType() == 1) {
             try {
                 Long count = phoneFidMapMapper.deleteByPackageId(userPackage.getId());
-                    log.trace("Deleted {} np rows", count);
+                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));
-                    npPipelineOrchestrator.startOrDelay(
+                pipelineOrchestrator.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));
+                log.error("Error build default 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());
+        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name(), Status.PROCESSING_REPEATED.name());
         log.debug("Found {} packages in status PROCESSING", packages.size());
         for (UserPackage userPackage : packages) {
-            if (userPackage.getSubjectType() == 1) {
             try {
-                    log.debug("Build np pipeline for packageId {}", userPackage.getId());
+                log.debug("Build pipeline for packageId {}", userPackage.getId());
                 Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT));
-                    npPipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
+                pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
             } catch (Exception e) {
-                    log.error("Error build restore np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+                log.error("Error build restore 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,12 +5,15 @@ 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
@@ -21,6 +24,9 @@ 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;
 
@@ -29,12 +35,18 @@ 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;
     }
@@ -47,7 +59,7 @@ public class UpdatingFidService {
         }
         log.info("Delete unused phones");
         phonesMapService.deleteUnusedPhones();
-        log.info("Start updating fids");
+        log.info("Start updating fids and PD");
         try {
             Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, 1, PipelineMode.RECALC_FID));
             CompletableFuture<?> future = npPipelineRecalcOrchestrator.startOrReject(pipeline);
@@ -58,7 +70,17 @@ public class UpdatingFidService {
                         return;
                     }
 
-                    log.info("Updating NP fids successfully completed.");
+                    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);
+                        }
+                    });
                 } finally {
                     log.info("pipeline finished");
                     pipeline.close();
@@ -76,7 +98,7 @@ public class UpdatingFidService {
             log.info("The previous task has not yet been completed.");
             return;
         }
-        log.info("Delete unused lp maps");
+        log.info("Delete unused phones");
         legalsMapMapper.deleteUnusedMaps();
         log.info("Start updating fids");
         try {
@@ -89,7 +111,17 @@ public class UpdatingFidService {
                         return;
                     }
 
-                    log.info("Updating LP fids successfully completed.");
+                    List<UserPackage> packagesRepeated = userPackageMapper.findForRepeatedPackages();
+                    packagesRepeated.forEach(userPackage -> {
+                        Long packageId = userPackage.getId();
+                        SetupParam setupParam = new SetupParam(packageId, 2, PipelineMode.RESEND);
+                        try {
+                            Pipeline setupPipeline = lpPipelinePackageFabric.build(setupParam);
+                            lpPipelinePackageOrchestrator.startOrDelay(setupPipeline, packageId);
+                        } catch (Exception e) {
+                            log.error("Failed to build pipeline", e);
+                        }
+                    });
                 } finally {
                     log.info("pipeline finished");
                     pipeline.close();

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

@@ -8,7 +8,6 @@ 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;
@@ -24,21 +23,20 @@ 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.*;
+import java.util.ArrayList;
-import java.util.concurrent.*;
+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.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;
@@ -49,32 +47,20 @@ 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(
@@ -104,6 +90,8 @@ 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()));
@@ -113,43 +101,17 @@ public class KafkaMonitoringService {
 
         Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
 
-        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
         if (npMessages != null && !npMessages.isEmpty()) {
-                return processNpMessages(npMessages, segmentMap);
+            List<Hit> npHits = processNpMessages(npMessages, segmentMap);
+            hits.addAll(npHits);
         }
-            return Collections.emptyList();
-        }, processingExecutor);
 
-        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
         if (lpMessages != null && !lpMessages.isEmpty()) {
-                return processLpMessages(lpMessages, segmentMap);
+            List<Hit> lpHits = processLpMessages(lpMessages, segmentMap);
+            hits.addAll(lpHits);
         }
-            return Collections.emptyList();
-        }, processingExecutor);
 
-        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) {
@@ -228,17 +190,12 @@ public class KafkaMonitoringService {
                         Integer segment = segmentMap.get(memberCode.substring(0, 6));
                         event.setSegment(segment != null ? segment : -1);
 
-                        String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
+                        putToQueue(event);
-                                + "_" + userPackagePhoneData.getPackageName();
-                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
-                        if (reportFileName == null) {
-                            continue; // событие не встало в очередь — hit не пишем
-                        }
                         Hit hit = Hit.builder()
                                 .packageId(userPackagePhoneData.getPackageId())
                                 .subjectId(subjectId)
                                 .signalType(signalType)
-                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                .reportFileName("") // TODO
                                 .inqT(message.getInquiryType())
                                 .busCategory(busCategory)
                                 .inqDt(message.getInquiryDate())
@@ -333,17 +290,12 @@ public class KafkaMonitoringService {
                         Integer segment = segmentMap.get(memberCode.substring(0, 6));
                         event.setSegment(segment != null ? segment : -1);
 
-                        String reportKey = userPackageLegalData.getUserId().substring(0, 6)
+                        putToQueue(event);
-                                + "_" + userPackageLegalData.getPackageName();
-                        String reportFileName = resolveNameAndEnqueue(reportKey, event);
-                        if (reportFileName == null) {
-                            continue; // событие не встало в очередь — hit не пишем
-                        }
                         Hit hit = Hit.builder()
                                 .packageId(userPackageLegalData.getPackageId())
                                 .subjectId(subjectId)
                                 .signalType(signalType)
-                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                .reportFileName("") // TODO
                                 .inqT(message.getInquiryType())
                                 .busCategory(busCategory)
                                 .inqDt(message.getInquiryDate())
@@ -370,18 +322,12 @@ public class KafkaMonitoringService {
                 (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
     }
 
-    /**
+    private void putToQueue(SignalKafkaEvent event) {
-     * Резолвит имя результирующего файла и ставит событие в очередь.
-     *
-     * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
-     */
-    private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
         try {
-            return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+            signalEventsQueue.put(event);
         } catch (InterruptedException e) {
             log.error("Error while putting signal event", e);
             Thread.currentThread().interrupt();
-            return null;
         }
     }
 
@@ -393,7 +339,9 @@ public class KafkaMonitoringService {
         processingPermits.acquire(permits);
 
         try {
-            reportNameRegistry.enqueueRotate();
+            SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
+            rotateEvent.setEventType(EventType.ROTATE);
+            signalEventsQueue.put(rotateEvent);
             log.info("ROTATE event sent to queue");
 
             SignalKafkaEvent stopEvent = new SignalKafkaEvent();

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

@@ -1,72 +0,0 @@
-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,10 +3,14 @@ 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;
@@ -19,12 +23,12 @@ public class RotateSheduler {
     );
     private volatile boolean isShuttingdDown = false;
 
-    private final ReportNameRegistry reportNameRegistry;
+    private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
     private final CreditTrackerSettings config;
 
-    public RotateSheduler(ReportNameRegistry reportNameRegistry,
+    public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
                           CreditTrackerSettings config) {
-        this.reportNameRegistry = reportNameRegistry;
+        this.signalEventsQueue = signalEventsQueue;
         this.config = config;
     }
 
@@ -40,7 +44,9 @@ public class RotateSheduler {
                 return;
             }
             try {
-                reportNameRegistry.enqueueRotate();
+                SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
+                rotateEvent.setEventType(EventType.ROTATE);
+                signalEventsQueue.put(rotateEvent);
             } 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,7 +34,6 @@ 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;
@@ -86,7 +85,7 @@ public class SignalKafkaEventWriterService {
                 }
 
                 if (event.getEventType() == EventType.ROTATE) {
-                    rotate(event.getReportNames());
+                    rotate();
                 }
             }
         }
@@ -135,16 +134,14 @@ public class SignalKafkaEventWriterService {
     }
 
     /**
-     * Упаковывает все результирующие xml в архив.
+     * Упаковывает все результирующие xml в архив
-     *
-     * @param reportNames снапшот имён файлов закрываемой эпохи (ключ: membercode_packageName)
      */
-    private void rotate(Map<String, String> reportNames) {
+    private void rotate() {
         List<UserPackage> packagesToUpdate = new ArrayList<>();
         List<UserPackage> userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
         for (UserPackage userPackage : userPackages) {
             try {
-                String reportName = movedAndZip(userPackage, reportNames);
+                String reportName = movedAndZip(userPackage);
                 userPackage.setLastDownloadedReport(reportName);
                 packagesToUpdate.add(userPackage);
             } catch (IOException e) {
@@ -170,7 +167,7 @@ public class SignalKafkaEventWriterService {
         }
     }
 
-    private String movedAndZip(UserPackage userPackage, Map<String, String> reportNames) throws IOException {
+    private String movedAndZip(UserPackage userPackage) throws IOException {
         String finalName = "";
         String commonMembercode = userPackage.getUserId().substring(0, 6);
         Path tmpFilePath = Path.of(config.getLoadDir(),
@@ -187,17 +184,11 @@ public class SignalKafkaEventWriterService {
             if (!Files.exists(rootReportPath)) {
                 Files.createDirectories(rootReportPath);
             }
-
+            LocalDateTime newReportDate = LocalDateTime.now();
-            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));
+                    newReportDate.format(CTUtil.responseReportDateTimeFormatter));
-            }
 
             Path responsePath = rootReportPath.resolve(finalName);
 

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

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

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

@@ -47,12 +47,11 @@ 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);
-        boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
+        processByNumber(subjects, cachedRecords, ctx);
-        taskPayload.setHasFids(hasFids);
         return taskPayload;
     }
 
-    private boolean processByNumber(List<SubjectProfile> subjects,
+    private void processByNumber(List<SubjectProfile> subjects,
                                  Map<Pair<Long, Long>, LegalsMap> cachedRecords,
                                  PipelineContext<PackageVariable> context) {
         List<SubjectProfile> noFidsData = subjects.stream()
@@ -126,10 +125,7 @@ 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,6 +1,5 @@
 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;
@@ -10,12 +9,14 @@ import java.util.concurrent.ExecutorService;
 
 @Service
 public class LpPipelinePackageOrchestrator extends PackageOrchestrator {
+    private final CreditTrackerSettings creditTrackerSettings;
 
     public LpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
-                                         UserPackageMapper userPackageMapper,
+                                         UserPackageMapper userPackageMapper, // TODO
-                                         @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService,
+                                         ExecutorService pipelineExecutorService, // TODO
-                                         @Qualifier("lpSaverExecutorService") ExecutorService lpSaverExecutorService) {
+                                         ExecutorService saverExecutorService) { // TODO
-        super(creditTrackerSettings, userPackageMapper, lpPipelineExecutorService, lpSaverExecutorService);
+        super(userPackageMapper, pipelineExecutorService, saverExecutorService);
+        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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

@@ -1,6 +1,5 @@
 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;
@@ -10,9 +9,12 @@ import java.util.concurrent.ExecutorService;
 @Service
 public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator{
 
+    private final CreditTrackerSettings creditTrackerSettings;
+
     public LpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
-                                        @Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService) {
+                                        ExecutorService pipelineExecutorService) { // TODO another executor
-        super(creditTrackerSettings, lpPipelineExecutorService);
+        super(pipelineExecutorService);
+        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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

@@ -1,6 +1,5 @@
 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;
@@ -10,12 +9,14 @@ import java.util.concurrent.ExecutorService;
 
 @Service
 public class NpPipelinePackageOrchestrator extends PackageOrchestrator {
+    private final CreditTrackerSettings creditTrackerSettings;
 
     public NpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
                                          UserPackageMapper userPackageMapper,
-                                         @Qualifier("npPipelineExecutorService") ExecutorService npPipelineExecutorService,
+                                         ExecutorService pipelineExecutorService,
-                                         @Qualifier("npSaverExecutorService") ExecutorService npSaverExecutorService) {
+                                         ExecutorService saverExecutorService) {
-        super(creditTrackerSettings, userPackageMapper, npPipelineExecutorService, npSaverExecutorService);
+        super(userPackageMapper, pipelineExecutorService, saverExecutorService);
+        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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

@@ -1,6 +1,5 @@
 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;
@@ -10,9 +9,11 @@ import java.util.concurrent.ExecutorService;
 @Service
 public class NpPipelineRecalcOrchestrator extends RecalcOrchestrator {
 
+    private final CreditTrackerSettings creditTrackerSettings;
     public NpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
-                                        @Qualifier("npPipelineExecutorService")ExecutorService npPipelineExecutorService) {
+                                        ExecutorService pipelineExecutorService) {
-        super(creditTrackerSettings, npPipelineExecutorService);
+        super(pipelineExecutorService);
+        this.creditTrackerSettings = creditTrackerSettings;
     }
 
     @Override

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

@@ -2,7 +2,6 @@ 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;
@@ -21,7 +20,6 @@ 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;
@@ -29,11 +27,9 @@ public abstract class PackageOrchestrator {
     private final java.util.Deque<PackageOrchestrator.Launch> pending = new java.util.ArrayDeque<>();
     private final Semaphore semaphore;
 
-    public PackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
+    public PackageOrchestrator(UserPackageMapper userPackageMapper,
-                               UserPackageMapper userPackageMapper,
                                ExecutorService pipelineExecutorService,
                                ExecutorService saverExecutorService) {
-        this.creditTrackerSettings = creditTrackerSettings;
         this.userPackageMapper = userPackageMapper;
         this.pipelineExecutorService = pipelineExecutorService;
         this.saverExecutorService = saverExecutorService;
@@ -112,6 +108,8 @@ 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,7 +2,6 @@ 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;
@@ -14,14 +13,11 @@ 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(CreditTrackerSettings creditTrackerSettings,
+    public RecalcOrchestrator(ExecutorService pipelineExecutorService) {
-                              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,6 +107,12 @@ 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,12 +46,11 @@ 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);
-        boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
+        processByNumber(subjects, cachedRecords, ctx);
-        taskPayload.setHasFids(hasFids);
         return taskPayload;
     }
 
-    private boolean processByNumber(List<SubjectProfile> subjects,
+    private void processByNumber(List<SubjectProfile> subjects,
                                  Map<Long, PhonesMap> cachedRecords,
                                  PipelineContext<PackageVariable> context) {
         List<SubjectProfile> noFidsData = subjects.stream()
@@ -98,8 +97,6 @@ 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,7 +8,9 @@ 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;
@@ -23,14 +25,20 @@ 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,
-                        PhonesMapService phonesMapService) {
+                        PackageRecordsService packageRecordsService,
+                        PhonesMapService phonesMapService,
+                        UserPackageService userPackageService) {
         this.personService = personService;
         this.creditTrackerSettings = creditTrackerSettings;
+        this.packageRecordsService = packageRecordsService;
         this.phonesMapService = phonesMapService;
+        this.userPackageService = userPackageService;
     }
 
     @Override
@@ -71,6 +79,12 @@ 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.isHasFids());
+                    hasFids.set(part.getSubjects().stream().anyMatch(s -> s.getFid() != null));
                 }
                 if (part.isPoison()) {
                     log.debug("Got poison");

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

@@ -49,10 +49,4 @@
         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">
+    <select id="findCachedByInnAndRegNUm" resultMap="LegalsMap">
-        WITH src AS (SELECT inn, reg_num
+        WITH src AS (SELECT inn, regNum
                      FROM json_to_recordset(#{json}::json) AS t(
                                                                 inn bigint,
-                                                                reg_num bigint
+                                                                regNum 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, reg_num FROM src)
+        WHERE (slm.inn, slm.reg_num) IN (SELECT inn, regNum FROM src)
     </select>
 
     <select id="findEligibleForFlagsUpdate" resultMap="LegalsMap">
@@ -231,7 +231,8 @@
                             AND sup.status IN
                                 (
                                  'ON_MONITORING',
-                                 'PROCESSING'
+                                 'PROCESSING',
+                                 'PROCESSING_REPEATED'
                                     ))
     </delete>
 

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

@@ -42,6 +42,30 @@
         </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(
@@ -54,17 +78,17 @@
     </select>
 
     <select id="selectUserPackageLegalData" resultType="ru.nbch.credit_tracker.model.UserPackageLegalData">
-        WITH src AS (SELECT inn, reg_num
+        WITH src AS (SELECT inn, regNum
                      FROM json_to_recordset(#{json}::json) AS t(
                                                                 inn bigint,
-                                                                reg_num bigint
+                                                                regNum 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, reg_num FROM src)
+        WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, regNum FROM src)
           AND sup.status = 'ON_MONITORING'
     </select>
 

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

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

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

@@ -14,7 +14,9 @@
         <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">
@@ -61,8 +63,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)
+        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}) RETURNING id
+        values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.subjectType}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}, concat(substring(#{userPackage.userId}, 1, 6), '_', #{userPackage.packageName})) RETURNING id
     </select>
 
     <update id="updatePackageStatus">
@@ -71,6 +73,16 @@
         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
@@ -93,7 +105,20 @@
             <if test="monitoringAt != null">
                 monitoring_at = #{monitoringAt},
             </if>
-            status = 'ON_MONITORING'
+            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}
         WHERE id = #{id}
     </update>
 
@@ -106,6 +131,7 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
+               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE id = #{id}
@@ -121,12 +147,28 @@
                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,
@@ -136,9 +178,10 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
+               reject_status,
                signal_types
         FROM signals_user_packages
-        WHERE status = 'ON_MONITORING'
+        WHERE status = 'ON_MONITORING' and reject_status is null
     </select>
 
 
@@ -151,6 +194,7 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
+               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE membercode IS NOT NULL
@@ -170,6 +214,7 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
+               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE membercode IS NOT NULL
@@ -185,6 +230,7 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
+               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE status in
@@ -202,6 +248,7 @@
                deactivated_at,
                last_downloaded_report,
                monitoring_at,
+               reject_status,
                signal_types
         FROM signals_user_packages
         WHERE LEFT(membercode, 6) = #{memberCode} AND package_name = #{packageName}