restore package in ACCEPTER;PROCCESSING statuses

src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java

@@ -4,17 +4,18 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.ConfigurableApplicationContext;
 import ru.nbch.credit_tracker.service.MonitoringService;
+import ru.nbch.credit_tracker.service.UpdateProcessingPackagesService;
 
 @SpringBootApplication
 public class CreditTrackerApplication {
 
 	public static void main(String[] args) {
 		ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args);
-//		UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
+		UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
 		MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
 //		UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class);
-//		updateProcessingPackagesService.run();
+		updateProcessingPackagesService.run();
-		monitoringService.run();
+//		monitoringService.run();
 //		updatingFlagsService.run();
 	}
 

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

@@ -0,0 +1,29 @@
+package ru.nbch.credit_tracker.config.task;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+
+@Configuration
+public class ExecutorsConfig {
+
+    @Bean
+    public ExecutorService pipelineExecutorService() {
+        return Executors.newFixedThreadPool(10, r -> {
+            Thread t = new Thread(r);
+            t.setName("pipe-starter-%d".formatted(t.threadId()));
+            return t;
+        });
+    }
+
+    @Bean
+    public ExecutorService saverExecutorService(CreditTrackerSettings creditTrackerSettings) {
+        return Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> {
+            Thread t = new Thread(r);
+            t.setName("subject-saver-%d".formatted(t.threadId()));
+            return t;
+        });
+    }
+}

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

@@ -1,7 +1,6 @@
 package ru.nbch.credit_tracker.config.task;
 
-import java.nio.file.Path;
+import java.util.function.Function;
-import java.util.function.BiFunction;
 import java.util.function.Supplier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -11,7 +10,9 @@ import ru.nbch.credit_tracker.model.impl.PartFilePayload;
 import ru.nbch.credit_tracker.model.impl.SignalXmlFilePayload;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.indic.PhoneService2;
-import ru.nbch.credit_tracker.service.task.impl.KickPipeLineImpl;
+import ru.nbch.credit_tracker.service.task.SetupParam;
+import ru.nbch.credit_tracker.service.task.impl.KickPipeLineDefaultImpl;
+import ru.nbch.credit_tracker.service.task.impl.KickPipeLineRestoreImpl;
 import ru.nbch.credit_tracker.service.task.impl.XmlMergeCoordinator;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
 import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator;
@@ -20,11 +21,16 @@ import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 @Configuration
 public class PipelineStarterConfig {
     @Bean
-    public BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml,
+    public Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory(StaxXmlProcessor xml,
                                                                                          CreditTrackerSettings creditTrackerSettings,
                                                                                          UserPackageMapper userPackageMapper,
                                                                                          PhoneService2 phoneService2) {
-        return (packageId, packagePath) -> new KickPipeLineImpl(xml, packageId, packagePath, userPackageMapper, phoneService2, creditTrackerSettings);
+        return (setupParam) -> switch (setupParam.mode()) {
+            case DEFAULT ->
+                new KickPipeLineDefaultImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
+            case RESTORE ->
+                new KickPipeLineRestoreImpl(xml, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
+        };
     }
 
     @Bean

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

@@ -7,6 +7,8 @@ import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import javax.xml.stream.XMLStreamException;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.io.FileUrlResource;
@@ -38,11 +40,16 @@ import ru.nbch.credit_tracker.model.ValidationResult;
 import ru.nbch.credit_tracker.service.UserPackageService;
 import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
+import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
+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;
 import ru.nbch.credit_tracker.utils.EnumUtil;
 import ru.nbch.credit_tracker.utils.LogUtil;
 import ru.nbch.credit_tracker.utils.ZipUtils;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.clientExcp;
 import static ru.nbch.credit_tracker.utils.error.ExceptionUtils.serverExcp;
 
@@ -56,18 +63,28 @@ public class SignalFacadeV3 {
     private final AuthorizationService authorizationService;
     private final PipelinePackageManager pipelinePackageManager;
     private final SignalRestService signalRestService;
+    private final Saver saver;
+    private final ExecutorService pipelineExecutorService;
+    private final ExecutorService saverExecutorService;
+
     public SignalFacadeV3(MonitoringRequestValidator monitoringRequestValidator,
                           CreditTrackerSettings config,
                           UserPackageService userPackageService,
                           AuthorizationService authorizationService,
                           PipelinePackageManager pipelinePackageManager,
-                          SignalRestService signalRestService) {
+                          SignalRestService signalRestService,
+                          Saver saver,
+                          ExecutorService pipelineExecutorService,
+                          ExecutorService saverExecutorService) {
         this.monitoringRequestValidator = monitoringRequestValidator;
         this.config = config;
         this.userPackageService = userPackageService;
         this.authorizationService = authorizationService;
         this.pipelinePackageManager = pipelinePackageManager;
         this.signalRestService = signalRestService;
+        this.saverExecutorService = saverExecutorService;
+        this.pipelineExecutorService = pipelineExecutorService;
+        this.saver = saver;
     }
 
     public void setupMonitoringRequest(MultipartFile file) {
@@ -129,7 +146,17 @@ public class SignalFacadeV3 {
             Long packageId = userPackageService.registerPackage(userId, packageName);
             LogUtil.addMdcLogKey("%s %d".formatted(packageName, packageId));
             log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name());
-            pipelinePackageManager.buildAndRun(packageId, parsedFilePath);
+            Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT));
+            CompletableFuture
+                .runAsync(() -> {
+                    try {
+                        saver.readAndSave(packageId, parsedFilePath);
+                    } catch (Throwable t) {
+                        log.error(ExceptionUtils.getStackTrace(t));
+                        pipeline.close();
+                    }
+                }, saverExecutorService)
+                .thenRun(() -> pipeline.start(pipelineExecutorService));
         } catch (Exception e) {
             try {
                 Files.delete(parsedFilePath);

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

@@ -18,6 +18,7 @@ public interface PhoneFidMapMapper {
     void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
 
     List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset);
+    List<PhoneFidMap> findNullFidPhoneFidMap(@Param("packageId") Long packageId, @Param("limit") int limit, @Param("offset") int offset);
 
     void updatePhoneFidMaps(@Param("subjects") Collection<PhoneFidMap> subjects);
     void updateOne(@Param("subject") PhoneFidMap subject);
@@ -28,4 +29,6 @@ public interface PhoneFidMapMapper {
     PhoneFidMap findPhoneFidMapById(@Param("id") Long packageId);
 
     boolean hasNonNullFids(@Param("packageId") Long packageId);
+
+    Long deleteByPackageId(@Param("packageId") Long id);
 }

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

@@ -35,7 +35,7 @@ import ru.nbch.credit_tracker.service.indic.PhoneService;
 import ru.nbch.credit_tracker.service.indic.PhoneService2;
 import ru.nbch.credit_tracker.service.signal.HitsService;
 import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
-import ru.nbch.credit_tracker.service.signal.UserPackageService;
+import ru.nbch.credit_tracker.service.signal.UserPackageDao;
 import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 import ru.nbch.credit_tracker.utils.CTUtil;
@@ -48,7 +48,7 @@ public class SignalService {
 
     private final CreditTrackerSettings config;
     private final HitsService hitsService;
-    private final UserPackageService userPackageService;
+    private final UserPackageDao userPackageService;
     private final PhoneFidMapService phoneFidMapService;
     private final PhoneService2 phoneService2;
     private final PhoneService phoneService;
@@ -60,7 +60,7 @@ public class SignalService {
 
     public SignalService(CreditTrackerSettings config,
                          HitsService hitsService,
-                         UserPackageService userPackageService,
+                         UserPackageDao userPackageService,
                          PhoneFidMapService phoneFidMapService,
                          PhoneService2 phoneService2,
                          PhoneService phoneService,

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

@@ -1,15 +1,27 @@
 package ru.nbch.credit_tracker.service;
 
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.aop.framework.AopContext;
 import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.mdc.SetAndClearMdc;
+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.PhoneFidMapMapper;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
-
+import ru.nbch.credit_tracker.service.task.Pipeline;
-import java.util.List;
+import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
+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;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
 /**
  * Сервис для завершения обработки пакетов находящихся в статусе PROCESSING
@@ -23,25 +35,90 @@ public class UpdateProcessingPackagesService {
     private UpdateProcessingPackagesService self;
 
     private final SignalService signalService;
+    private final UserPackageMapper userPackageMapper;
+    private final PhoneFidMapMapper phoneFidMapMapper;
     private final SignalRestService signalRestService;
     private final MonitoringErrorProcesingService monitoringErrorProcesingService;
+    private final PipelinePackageManager pipelinePackageManager;
+    private final CreditTrackerSettings creditTrackerSettings;
+    private final Saver saver;
+    private final ExecutorService pipelineExecutorService;
+    private final ExecutorService saverExecutorService;
 
     public UpdateProcessingPackagesService(SignalService signalService,
+                                           UserPackageMapper userPackageMapper,
+                                           PhoneFidMapMapper phoneFidMapMapper,
                                            SignalRestService signalRestService,
-                                           MonitoringErrorProcesingService monitoringErrorProcesingService) {
+                                           MonitoringErrorProcesingService monitoringErrorProcesingService,
+                                           PipelinePackageManager pipelinePackageManager,
+                                           CreditTrackerSettings creditTrackerSettings,
+                                           Saver saver,
+                                           ExecutorService pipelineExecutorService,
+                                           ExecutorService saverExecutorService) {
         this.signalService = signalService;
+        this.userPackageMapper = userPackageMapper;
+        this.phoneFidMapMapper = phoneFidMapMapper;
         this.signalRestService = signalRestService;
         this.monitoringErrorProcesingService = monitoringErrorProcesingService;
+        this.pipelinePackageManager = pipelinePackageManager;
+        this.creditTrackerSettings = creditTrackerSettings;
+        this.saver = saver;
+        this.pipelineExecutorService = pipelineExecutorService;
+        this.saverExecutorService = saverExecutorService;
     }
 
     public void run() {
-        log.info("Start updating processing packages");
+        acceptedRestore();
-        List<UserPackage> packages = signalService.findPackagesOnProcessing();
+        processingRestore();
-        for (UserPackage userPackage : packages) {
+//        log.info("Start updating processing packages");
-            log.info("Updating flags for Package {}", userPackage.getPackageName());
+//        List<UserPackage> packages = signalService.findPackagesOnProcessing();
-            getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName());
+//        for (UserPackage userPackage : packages) {
+//            log.info("Updating flags for Package {}", userPackage.getPackageName());
+//            getCurrentProxy().process(userPackage.getId(), userPackage.getPackageName());
+//        }
+//        log.info("Finish updating processing packages");
+    }
+
+
+    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);
+                //todo ???запускает асинхронный процесс без ожидания, на старте мы должны восстановиться один раз и
+                // видимо дождаться выполнения, иначе может не хватить пула, памяти или чего еще
+                Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), xmlReqPath, PipelineMode.DEFAULT));
+                CompletableFuture
+                    .runAsync(() -> {
+                        try {
+                            saver.readAndSave(userPackage.getId(), xmlReqPath);
+                        } catch (Throwable t) {
+                            log.error(ExceptionUtils.getStackTrace(t));
+                            pipeline.close();
+                        }
+                    }, saverExecutorService)
+                    .thenRun(() -> pipeline.start(pipelineExecutorService));
+            } catch (Exception e) {
+                log.error("Error build default pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+            }
+        }
+    }
+
+    private void processingRestore() {
+        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name());
+        log.debug("Found {} packages in status PROCESSING", packages.size());
+        for (UserPackage userPackage : packages) {
+            try {
+                Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.RESTORE));
+                pipeline.start(pipelineExecutorService);
+            } catch (Exception e) {
+                log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
+            }
         }
-        log.info("Finish updating processing packages");
     }
 
     @SetAndClearMdc(argIndex = 1)

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

@@ -7,13 +7,14 @@ import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.app.UserPackage;
 import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
 import ru.nbch.credit_tracker.enums.Status;
+import ru.nbch.credit_tracker.service.signal.UserPackageDao;
 
 @Slf4j
 @Service
 public class UserPackageService {
-    private final ru.nbch.credit_tracker.service.signal.UserPackageService userPackageService;
+    private final UserPackageDao userPackageService;
 
-    public UserPackageService(ru.nbch.credit_tracker.service.signal.UserPackageService userPackageService) {
+    public UserPackageService(UserPackageDao userPackageService) {
         this.userPackageService = userPackageService;
     }
 

src/main/java/ru/nbch/credit_tracker/service/indic/PhoneService2.java

@@ -91,7 +91,12 @@ public class PhoneService2 {
         }
     }
 
-    public List<PhoneFidMap> nextPage(Long packageId, Integer limit, Integer offset) {
+    public List<PhoneFidMap> pageByPackageId(Long packageId, Integer limit, Integer offset) {
         return phoneFidMapMapper.findFullPhoneFidMap(packageId, limit, offset);
     }
+
+    public List<PhoneFidMap> pageByPackageIdWithNullFids(Long packageId, Integer limit, Integer offset) {
+        return phoneFidMapMapper.findNullFidPhoneFidMap(packageId, limit, offset);
+    }
+
 }

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

@@ -9,11 +9,11 @@ import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 
 @Service
-public class UserPackageService {
+public class UserPackageDao {
 
     private final UserPackageMapper mapper;
 
-    public UserPackageService(UserPackageMapper mapper) {
+    public UserPackageDao(UserPackageMapper mapper) {
         this.mapper = mapper;
     }
 

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

@@ -1,13 +1,9 @@
 package ru.nbch.credit_tracker.service.task;
 
-import java.nio.file.Path;
 import java.time.LocalDateTime;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
-import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.InitializingBean;
@@ -42,18 +38,10 @@ public class PipelinePackageManager implements InitializingBean {
     private final FidIdEnricher fidIdEnricherHandler;
     private final FidNameEnricher fidNameEnricher;
     private final Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
-    private final BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
+    private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
     private final CreditTrackerSettings creditTrackerSettings;
-    private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> {
-        Thread t = new Thread(r);
-        t.setName("pipe-starter-%d".formatted(t.threadId()));
-        return t;
-    });
-
-    private final ExecutorService subjectSaverExecutor;
     private final Semaphore semaphore;
     private final PipelineBackoff pipelineBackoff;
-    private final Saver saver;
 
     public PipelinePackageManager(SignalRestService signalRestService,
                                   @Qualifier("idEnricherQueueFactory") StageQueueFactory<SubjectPayload> idEnricherQueueFactory,
@@ -62,7 +50,7 @@ public class PipelinePackageManager implements InitializingBean {
                                   UserPackageMapper userPackageMapper,
                                   FidIdEnricher fidIdEnricherHandler,
                                   Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
-                                  BiFunction<Long, Path, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
+                                  Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory,
                                   FlagsEnricher flagsEnricherV2, FidNameEnricher fidNameEnricher,
                                   CreditTrackerSettings creditTrackerSettings,
                                   PipelineBackoff pipelineBackoff, Saver saver) {
@@ -79,19 +67,15 @@ public class PipelinePackageManager implements InitializingBean {
         this.creditTrackerSettings = creditTrackerSettings;
         this.semaphore = new Semaphore(creditTrackerSettings.getPipeline().parallelRequests());
         this.pipelineBackoff = pipelineBackoff;
-        this.saver = saver;
-        this.subjectSaverExecutor = Executors.newFixedThreadPool(creditTrackerSettings.getPipeline().parallelRequests(), r -> {
-            Thread t = new Thread(r);
-            t.setName("subject-saver-%d".formatted(t.threadId()));
-            return t;
-        });
     }
 
-    public void buildAndRun(Long packageId, Path pkgPath) throws Exception {
+    public Pipeline buildAndRun(SetupParam setupParam) throws Exception {
         if (!semaphore.tryAcquire()) {
             throw new RuntimeException("Too many concurrent requests");
         }
         log.info("throttler permits after acquire: {}", semaphore.availablePermits());
+
+        Long packageId = setupParam.packageId();
         long startTime = System.currentTimeMillis();
         try {
             Pipeline pipeline = PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
@@ -101,7 +85,7 @@ public class PipelinePackageManager implements InitializingBean {
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)
                     .andThen()
-                .fillInitialQueue(pipelineStarterFactory.apply(packageId, pkgPath))
+                .fillInitialQueue(pipelineStarterFactory.apply(setupParam))
                 .pollFinishingQueue(pipelineTerminatorFactory.get())
                 .semaphore(semaphore)
                 .backoff(pipelineBackoff)
@@ -136,17 +120,7 @@ public class PipelinePackageManager implements InitializingBean {
                         log.info("semaphore amount: {}", semaphore.availablePermits());
                     }
                 });
-
+            return pipeline;
-            CompletableFuture
-                .runAsync(() -> {
-                    try {
-                        saver.readAndSave(packageId, pkgPath);
-                    } catch (Throwable t) {
-                        log.error(ExceptionUtils.getStackTrace(t));
-                        pipeline.close();
-                    }
-                }, subjectSaverExecutor)
-                .thenRun(() -> pipeline.start(executorService));
         } catch (Exception e) {
             semaphore.release();
             throw e;

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

@@ -0,0 +1,10 @@
+package ru.nbch.credit_tracker.service.task;
+
+import java.nio.file.Path;
+import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
+
+public record SetupParam(
+    Long packageId,
+    Path xmlReqPath,
+    PipelineMode mode) {
+}

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

@@ -0,0 +1,96 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
+import ru.nbch.credit_tracker.entities.app.UserPackage;
+import ru.nbch.credit_tracker.enums.PoisonType;
+import ru.nbch.credit_tracker.exceptions.SignalClientException;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
+import ru.nbch.credit_tracker.service.indic.PhoneService2;
+import ru.nbch.credit_tracker.service.task.SetupParam;
+import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
+
+public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    protected final Long packageId;
+    private final Path pkgPath;
+    private final StaxXmlProcessor xmlProcessor;
+    protected final UserPackageMapper userPackageMapper;
+    protected final PhoneService2 phoneService2;
+    private final int BATCH_SIZE;
+    private boolean isInterrupted = false;
+
+
+    public KickPipeLine(StaxXmlProcessor xmlProcessor,
+                        SetupParam setupParam,
+                        UserPackageMapper userPackageMapper,
+                        PhoneService2 phoneService2,
+                        CreditTrackerSettings creditTrackerSettings) {
+        this.packageId = setupParam.packageId();
+        this.pkgPath = setupParam.xmlReqPath();
+        this.xmlProcessor = xmlProcessor;
+        this.userPackageMapper = userPackageMapper;
+        this.phoneService2 = phoneService2;
+        this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize();
+    }
+
+    private static class PipelineStarterException extends SignalClientException {
+        public PipelineStarterException() {
+            super(null, null, null);
+        }
+    }
+
+    @Override
+    public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
+        try {
+            log.info("Start registering subjects. PackageId: {};", packageId);
+            int offset = 0;
+            int limit = BATCH_SIZE;
+            UserPackage userPackage = userPackageMapper.findPackage(packageId);
+            List<PhoneFidMap> page = nextPage(limit, offset);
+            while (!page.isEmpty()) {
+                log.debug("Get page {}", page.size());
+                List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
+                    .mapId(phoneFidMap.getId())
+                    .identifier(phoneFidMap.getSubjectId())
+                    .phone(phoneFidMap.getPhone())
+                    .build()).toList();
+                try {
+                    firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles));
+                } catch (InterruptedException e) {
+                    isInterrupted = true;
+                    log.error("Pipeline starter was interrupted!");
+                    throw new PipelineStarterException();
+                }
+                offset += limit;
+                page = nextPage(limit, offset);
+            }
+        } catch (PipelineStarterException e) {
+            log.info("Pipeline starter shitting down...");
+        } finally {
+            if (isInterrupted) {
+                //noinspection ReturnInsideFinallyBlock
+                return;
+            }
+            log.debug("Send poison pill");
+            try {
+                SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY);
+                firstQueue.put(poison);
+            } catch (InterruptedException e) {
+                log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
+            }
+        }
+    }
+
+    protected abstract PipelineMode mode();
+    protected abstract  List<PhoneFidMap> nextPage(Integer limit, Integer offset);
+}

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

@@ -0,0 +1,31 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.util.List;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.service.indic.PhoneService2;
+import ru.nbch.credit_tracker.service.task.SetupParam;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+
+public class KickPipeLineDefaultImpl extends KickPipeLine {
+
+    public KickPipeLineDefaultImpl(StaxXmlProcessor xmlProcessor,
+                                   SetupParam setupParam,
+                                   UserPackageMapper userPackageMapper,
+                                   PhoneService2 phoneService2,
+                                   CreditTrackerSettings creditTrackerSettings) {
+        super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
+    }
+
+    @Override
+    protected PipelineMode mode() {
+        return PipelineMode.DEFAULT;
+    }
+
+    @Override
+    protected List<PhoneFidMap> nextPage(Integer limit, Integer offset) {
+        return phoneService2.pageByPackageId(packageId, limit, offset);
+    }
+
+}

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

@@ -1,158 +0,0 @@
-package ru.nbch.credit_tracker.service.task.impl;
-
-import java.io.BufferedInputStream;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
-import ru.nbch.credit_tracker.entities.app.UserPackage;
-import ru.nbch.credit_tracker.entities.user_request.monitoring_request.MonitoringRequest;
-import ru.nbch.credit_tracker.exceptions.SignalClientException;
-import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
-import ru.nbch.credit_tracker.enums.PoisonType;
-import ru.nbch.credit_tracker.model.SubjectProfile;
-import ru.nbch.credit_tracker.model.impl.SubjectPayload;
-import ru.nbch.credit_tracker.service.indic.PhoneService2;
-import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter;
-import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
-import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
-
-public class KickPipeLineImpl implements IPipeLineStarter<SubjectPayload> {
-    private final Logger log = LoggerFactory.getLogger(getClass());
-    private final Long packageId;
-    private final Path pkgPath;
-    private final StaxXmlProcessor xmlProcessor;
-    private final UserPackageMapper userPackageMapper;
-    private final PhoneService2 phoneService2;
-    private final int BATCH_SIZE;
-    private boolean isInterrupted = false;
-
-
-    public KickPipeLineImpl(StaxXmlProcessor xmlProcessor,
-                            Long packageId,
-                            Path pkgPath,
-                            UserPackageMapper userPackageMapper,
-                            PhoneService2 phoneService2,
-                            CreditTrackerSettings creditTrackerSettings) {
-        this.packageId = packageId;
-        this.pkgPath = pkgPath;
-        this.xmlProcessor = xmlProcessor;
-        this.userPackageMapper = userPackageMapper;
-        this.phoneService2 = phoneService2;
-        this.BATCH_SIZE = creditTrackerSettings.getPipeline().subjectBatchSize();
-    }
-
-    private static class PipelineStarterException extends SignalClientException {
-        public PipelineStarterException() {
-            super(null, null, null);
-        }
-    }
-
-    @Override
-    public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
-        try {
-            log.info("Start registering subjects. PackageId: {};", packageId);
-            int offset = 0;
-            int limit = BATCH_SIZE;
-            UserPackage userPackage = userPackageMapper.findPackage(packageId);
-            List<PhoneFidMap> page = phoneService2.nextPage(packageId, limit, offset);
-            while (!page.isEmpty()) {
-                log.debug("Get page {}", page.size());
-                List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
-                    .mapId(phoneFidMap.getId())
-                    .identifier(phoneFidMap.getSubjectId())
-                    .phone(phoneFidMap.getPhone())
-                    .build()).toList();
-                try {
-                    firstQueue.put(new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles));
-                } catch (InterruptedException e) {
-                    isInterrupted = true;
-                    log.error("Pipeline starter was interrupted!");
-                    throw new PipelineStarterException();
-                }
-                offset += limit;
-                page = phoneService2.nextPage(packageId, limit, offset);
-            }
-        } catch (PipelineStarterException e) {
-            log.info("Pipeline starter shitting down...");
-        } finally {
-            if (isInterrupted) {
-                //noinspection ReturnInsideFinallyBlock
-                return;
-            }
-            log.debug("Send poison pill");
-            try {
-                SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY);
-                firstQueue.put(poison);
-            } catch (InterruptedException e) {
-                log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
-            }
-        }
-    }
-//    @Override
-    public void startFillingQueueXmnl(BlockingQueue<SubjectPayload> firstQueue) {
-        log.info("Start registering subjects. PackageId: {}; pkg path: {}", packageId, pkgPath);
-        UserPackage userPackage = userPackageMapper.findPackage(packageId);
-        AtomicInteger counter = new AtomicInteger(0);
-        try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(pkgPath), 8 * 1024 * 1024)) {
-            xmlProcessor.read(in, MonitoringRequest.class,
-                request -> {
-                    if (!request.getSubjects().getSubject().isEmpty()) {
-
-                        List<PhoneFidMap> subjectForSave = request.getSubjects().getSubject().stream().map(subject -> PhoneFidMap.builder()
-                            .subjectId(subject.getId())
-                            .packageId(packageId)
-                            .phone(subject.getClientId())
-                            .build()).toList();
-                        List<Long> ids = phoneService2.registerSubjects(subjectForSave);
-                        List<SubjectProfile> subjectProfiles = new ArrayList<>();
-                        for (int i = 0; i < ids.size(); i++) {
-                            PhoneFidMap phoneFidMap = subjectForSave.get(i);
-                            subjectProfiles.add(SubjectProfile.builder()
-                                .identifier(phoneFidMap.getSubjectId())
-                                .phone(phoneFidMap.getPhone())
-                                .mapId(ids.get(i))
-                                .build());
-                        }
-
-                        int count = counter.incrementAndGet();
-                        log.trace("Reading subject's {} batch. Total count: {}. PackageId: {}", count, request.getSubjects().getSubject().size(), packageId);
-                        try {
-                            firstQueue.put(new SubjectPayload(packageId,userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles));
-                        } catch (InterruptedException e) {
-                            isInterrupted = true;
-                            log.error("Pipeline starter was interrupted!");
-                            throw new PipelineStarterException();
-                        }
-                        request.getSubjects().getSubject().clear();
-                    }
-                },
-                request -> request.getSubjects().getSubject().isEmpty(),
-                BATCH_SIZE, "/MonitoringRequest/Subjects/Subject"
-            );
-        } catch (IOException e) {
-            log.error("{}", ExceptionUtils.getStackTrace(e));
-        } catch (PipelineStarterException e) {
-            log.info("Pipeline starter shitting down...");
-        } finally {
-            if (isInterrupted) {
-                //noinspection ReturnInsideFinallyBlock
-                return;
-            }
-            log.debug("Send poison pill");
-            try {
-                SubjectPayload poison = new SubjectPayload(packageId, PoisonType.NORMALLY);
-                firstQueue.put(poison);
-            } catch (InterruptedException e) {
-                log.error("couldn't send poison: {}", ExceptionUtils.getStackTrace(e));
-            }
-        }
-    }
-}

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

@@ -0,0 +1,31 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import java.util.List;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.service.indic.PhoneService2;
+import ru.nbch.credit_tracker.service.task.SetupParam;
+import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
+
+public class KickPipeLineRestoreImpl extends KickPipeLine {
+
+
+    public KickPipeLineRestoreImpl(StaxXmlProcessor xmlProcessor,
+                                   SetupParam setupParam,
+                                   UserPackageMapper userPackageMapper,
+                                   PhoneService2 phoneService2,
+                                   CreditTrackerSettings creditTrackerSettings) {
+        super(xmlProcessor, setupParam, userPackageMapper, phoneService2, creditTrackerSettings);
+    }
+
+    @Override
+    protected PipelineMode mode() {
+        return PipelineMode.RESTORE;
+    }
+
+    @Override
+    protected List<PhoneFidMap> nextPage(Integer limit, Integer offset) {
+        return phoneService2.pageByPackageIdWithNullFids(packageId, limit, offset);
+    }
+}

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

@@ -0,0 +1,6 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+public enum PipelineMode {
+    DEFAULT,
+    RESTORE
+}

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

@@ -92,7 +92,7 @@ public final class XmlMergeCoordinator implements IPipeLineTerminator<PartFilePa
                 w.writeAttribute("processWithErrors", "1");
 
                 w.writeStartElement("Id");
-                w.writeCharacters("%s_%s".formatted(input.getMembercode(), input.getPackageName()));
+                w.writeCharacters("%s_%s".formatted(input.getMembercode().substring(0, 6), input.getPackageName()));
                 w.writeEndElement(); // </Id>
 
                 w.writeStartElement("Auth");

src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java

@@ -10,6 +10,7 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import ru.nbch.credit_tracker.entities.app.UserPackage;
 
 @Slf4j
 public class CTUtil {
@@ -90,4 +91,8 @@ public class CTUtil {
     public static String normalizePhoneNumber(String phoneNumber) {
         return "9" + phoneNumber;
     }
+
+    public static Path workingPathToReqByPackage(String root, UserPackage userPackage) {
+        return Path.of(root, "tmp", userPackage.getUserId().substring(0, 6) + "_" +userPackage.getPackageName() + ".xml");
+    }
 }

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

@@ -29,28 +29,27 @@
         INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at)
         values
         <foreach collection="phoneFidMap" item="phoneFid" separator=",">
-            (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}, #{phoneFid.flagDebt5000}, #{phoneFid.flag90days}, #{phoneFid.calculatedAt})
+            (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}, #{phoneFid.flagDebt5000},
+            #{phoneFid.flag90days}, #{phoneFid.calculatedAt})
         </foreach>
     </insert>
 
-<!--    <insert id="registerJson" parameterType="string">-->
+    <!--    <insert id="registerJson" parameterType="string">-->
-<!--        INSERT INTO signals_phone_fid_map-->
+    <!--        INSERT INTO signals_phone_fid_map-->
-<!--        (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at)-->
+    <!--        (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at)-->
-<!--        SELECT subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at-->
+    <!--        SELECT subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at-->
-<!--        FROM json_to_recordset(#{json}::json) AS t(-->
+    <!--        FROM json_to_recordset(#{json}::json) AS t(-->
-<!--                                                   subject_id   varchar(255),-->
+    <!--                                                   subject_id   varchar(255),-->
-<!--                                                   fid          int,-->
+    <!--                                                   fid          int,-->
-<!--                                                   phone        text,-->
+    <!--                                                   phone        text,-->
-<!--                                                   package_id   bigint,-->
+    <!--                                                   package_id   bigint,-->
-<!--                                                   flag_debt5000 smallint,-->
+    <!--                                                   flag_debt5000 smallint,-->
-<!--                                                   flag_90_12    smallint,-->
+    <!--                                                   flag_90_12    smallint,-->
-<!--                                                   calculated_at timestamp)-->
+    <!--                                                   calculated_at timestamp)-->
-<!--    </insert>-->
+    <!--    </insert>-->
 
     <select id="registerJson" parameterType="string" resultType="long">
-        WITH src AS (
+        WITH src AS (SELECT row_number() OVER () AS rn,
-            SELECT
-                row_number() OVER () AS rn,
                             subject_id,
                             phone,
                             package_id
@@ -58,12 +57,11 @@
                                                                 subject_id varchar(255),
                                                                 phone text,
                                                                 package_id bigint
-                )
+                         ))
-        )
+        INSERT
-        INSERT INTO signals_phone_fid_map
+        INTO signals_phone_fid_map
             (subject_id, phone, package_id)
-        SELECT
+        SELECT subject_id,
-            subject_id,
                phone,
                package_id
         FROM src
@@ -73,8 +71,7 @@
 
     <update id="setFidJson" parameterType="string">
         update signals_phone_fid_map dst
-        set
+        set fid           = src.fid,
-            fid = src.fid,
             calculated_at = src.calculated_at
         FROM json_to_recordset(#{json}::json) AS src(
                                                      id bigint,
@@ -84,6 +81,12 @@
         where dst.id = src.id;
     </update>
 
+    <delete id="deleteByPackageId" parameterType="long">
+        DELETE
+        FROM signals_phone_fid_map
+        WHERE package_id = #{packageId}
+    </delete>
+
     <insert id="registerWithId" parameterType="java.util.List" useGeneratedKeys="true" keyColumn="id" keyProperty="id">
         INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
         values
@@ -111,8 +114,7 @@
 
     <update id="updateOne" parameterType="ru.nbch.credit_tracker.entities.app.PhoneFidMap">
         UPDATE signals_phone_fid_map
-            SET
+        SET package_id    = #{subject.packageId},
-            package_id = #{subject.packageId},
             subject_id    = #{subject.subjectId},
             phone         = #{subject.phone},
             fid           = #{subject.fid2},
@@ -157,7 +159,24 @@
         LIMIT #{limit} OFFSET #{offset}
     </select>
 
-    <select id="findPhoneFidMapById" resultMap="PhoneFidMapResultMap" resultType="ru.nbch.credit_tracker.entities.app.PhoneFidMap">
+    <select id="findNullFidPhoneFidMap" resultMap="PhoneFidMapResultMap" resultType="java.util.List">
+        SELECT id,
+               package_id,
+               subject_id,
+               phone,
+               fid,
+               flag_90_12,
+               flag_debt5000,
+               calculated_at,
+               error_text
+        FROM signals_phone_fid_map
+        WHERE package_id = #{packageId} and fid is null
+        ORDER BY id
+        LIMIT #{limit} OFFSET #{offset}
+    </select>
+
+    <select id="findPhoneFidMapById" resultMap="PhoneFidMapResultMap"
+            resultType="ru.nbch.credit_tracker.entities.app.PhoneFidMap">
         SELECT id,
                package_id,
                subject_id,
@@ -175,7 +194,8 @@
     <select id="hasNonNullFids" resultType="java.lang.Boolean">
         select count(*) > 0
         from signals_phone_fid_map
-        where package_id = #{packageId} AND fid is not null
+        where package_id = #{packageId}
+          AND fid is not null
     </select>