Merge remote-tracking branch 'origin/dev' into dev

credit-tracker.properties

@@ -69,7 +69,7 @@ ct.monitoring-interval=${MONITORING_INTERVAL:15}
 ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *}
 ct.new-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *}
 ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30}
-ct.update-fid-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 10 12 * * *}
+ct.update-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *}
 ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30}
 
 ct.db-version=${DB_VERSION:1234}

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

@@ -4,14 +4,14 @@ 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;
+import ru.nbch.credit_tracker.service.StartupRestoreService;
 
 @SpringBootApplication
 public class CreditTrackerApplication {
 
 	public static void main(String[] args) {
 		ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args);
-		UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
+		StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class);
 		MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
 		updateProcessingPackagesService.run();
 		monitoringService.run();

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

@@ -15,7 +15,7 @@ public class DeletePackageStatusComparator {
         try {
             Status status = Status.valueOf(str);
             return switch (status) {
-                case ON_MONITORING, PROCESSING_REPEATED -> 1;
+                case ON_MONITORING, PROCESSING_REPEATED, NO_FID -> 1;
                 case DELETED -> 0;
                 default -> -1;
             };

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

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

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

@@ -6,6 +6,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.time.LocalDateTime;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import javax.xml.stream.XMLStreamException;
@@ -214,10 +215,12 @@ public class SignalFacadeV3 {
             PackageAll pack = new PackageAll();
             pack.setPackageName(data.getPackageName());
             pack.setCreatedAt(data.getCreatedAt() != null ? data.getCreatedAt().toString() : null);
-            String statusForClient = data.getStatus().equals(Status.ACCEPTED.name()) ? Status.PROCESSING.name() :
+            Status packageStatus = EnumUtil.getEnum(Status.class, data.getStatus());
-                data.getStatus().equals(Status.PROCESSING_REPEATED.name()) ? Status.ON_MONITORING.name() : data.getStatus();
+            Status statusForClient = packageStatus == Status.ACCEPTED ? Status.PROCESSING :
-            pack.setStatus(statusForClient);
+                Arrays.asList(Status.PROCESSING_REPEATED, Status.NO_FID).contains(packageStatus) ? Status.ON_MONITORING : packageStatus;
-            if (data.getStatus().equals(Status.DELETED.name())) {
+            assert statusForClient != null;
+            pack.setStatus(statusForClient.name());
+            if (packageStatus == Status.DELETED) {
                 pack.setDeactivatedAt(data.getDeactivatedAt() != null ? data.getDeactivatedAt().toString() : null);
             }
             packages.getPackages().add(pack);
@@ -234,14 +237,16 @@ public class SignalFacadeV3 {
                 -> clientExcp(SignalError.ERROR_016, XmlErrorFormat.MonitoringPackageDeleteError, packageName));
 
         Status status = EnumUtil.getEnum(Status.class, packageToDelete.getStatus());
-        if (!Status.ON_MONITORING.equals(status)) {
+        if (!Arrays.asList(Status.ON_MONITORING, Status.NO_FID).contains(status)) {
             throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName);
         }
 
+        if (status == Status.ON_MONITORING) {
             boolean successfullyDeleted = signalRestService.deletePackage(username, packageName);
             if (!successfullyDeleted) {
                 throw clientExcp(SignalError.ERROR_017, XmlErrorFormat.MonitoringPackageDeleteError, packageName);
             }
+        }
 
         LocalDateTime deactivatedAt = LocalDateTime.now();
         userPackageService.deletePackageById(packageToDelete.getId(), deactivatedAt);
@@ -280,7 +285,7 @@ public class SignalFacadeV3 {
             if (status == null) {
                 throw clientExcp(SignalError.ERROR_017, XmlErrorFormat.MonitoringReportListError);
             }
-            if (status.equals(Status.PROCESSING)) {
+            if (Arrays.asList(Status.PROCESSING, Status.ACCEPTED).contains(status)) {
                 throw clientExcp(SignalError.ERROR_018, XmlErrorFormat.MonitoringReportListError, packageName);
             }
             if (status.equals(Status.REJECTED)) {

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

@@ -8,7 +8,7 @@ import ru.nbch.credit_tracker.entities.app.PhonesMap;
 @Mapper
 public interface PhonesMapMapper {
     List<PhonesMap> getCachedData(@Param("phones") List<Long> phones);
-    List<PhonesMap> searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("limit") int limit, @Param("offset") int offset);
+    List<PhonesMap> searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("batchSize") int batchSize, @Param("lastId") Long lastId);
     PhonesMap findByPhone(@Param("phone") Long phone);
     List<PhonesMap> findEligibleForFlagsUpdate(@Param("lastId") Long lastId,
                                                @Param("batchSize") Integer batchSize);
@@ -18,6 +18,7 @@ public interface PhonesMapMapper {
     void updateNameCachedData(@Param("json") String json);
     void updateFlagCachedData(@Param("json") String json);
     void resetFidAndPD(@Param("json") String json);
+    void updateSearchedAt(@Param("json") String json);
     void registerJson(@Param("json") String json);
     void setFidJson(@Param("json") String json);
     void deleteUnusedPhones();

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

@@ -11,12 +11,6 @@ public interface UserPackageMapper {
 
     boolean isPackageNotUnique(@Param("userId") String userId, @Param("packageName") String packageName);
 
-    boolean isPackageExists(@Param("userId") String userId, @Param("packageName") String packageName, @Param("status") String status);
-
-    boolean hasActivePackages(@Param("userId") String userId, @Param("packageName")  String packageName, @Param("statuses") List<String> statuses);
-
-    void deletePackage(@Param("userId") String userId, @Param("packageName") String packageName, @Param("status") String newStatus, @Param("previousStatus") String previousStatus, @Param("deactivatedAt") LocalDateTime deactivatedAt);
-
     void deletePackageById(@Param("id") Long id, @Param("status") String newStatus, @Param("deactivatedAt") LocalDateTime deactivatedAt);
 
     Long insertPackage(@Param("userPackage") UserPackage userPackage);
@@ -26,6 +20,7 @@ public interface UserPackageMapper {
     void updatePackageLastDownloadedReport(@Param("id") Long id, @Param("fileName") String fileName);
 
     void updatePackageOnMonitoringSuccess(@Param("id") Long id, @Param("monitoringAt") LocalDateTime monitoringAt);
+    void updateNoFidMonitoring(@Param("id") Long id);
     void setLastSearchedDt(@Param("id") Long id);
     void setFidChanged(@Param("packageIds") List<Long> id);
 
@@ -37,6 +32,7 @@ public interface UserPackageMapper {
     List<UserPackage> findAllPackagesByStatus(@Param("statuses") String... statuses);
 
     List<UserPackage> findAllPackagesByUserId(@Param("userId") String userId);
+    List<UserPackage> findOnMonitoringPackages();
     List<UserPackage> findByMemberCodeAndPackageName(@Param("memberCode") String memberCode,
                                                      @Param("packageName") String packageName);
 

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

@@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.service;
 
 import java.nio.file.Path;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.stereotype.Service;
@@ -27,38 +26,43 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 @Slf4j
 @Service
 @EnableAspectJAutoProxy(exposeProxy = true)
-public class UpdateProcessingPackagesService {
+public class StartupRestoreService {
 
     private final UserPackageMapper userPackageMapper;
     private final PackagesRecordsMapper phoneFidMapMapper;
     private final PipelinePackageFabric pipelinePackageManager;
     private final PipelinePackageOrchestrator pipelineOrchestrator;
     private final CreditTrackerSettings creditTrackerSettings;
+    private final MonitoringErrorProcesingService monitoringErrorProcesingService;
     private final Saver saver;
-    private final ExecutorService pipelineExecutorService;
-    private final ExecutorService saverExecutorService;
 
-    public UpdateProcessingPackagesService(UserPackageMapper userPackageMapper,
+    public StartupRestoreService(UserPackageMapper userPackageMapper,
                                  PackagesRecordsMapper phoneFidMapMapper,
-                                           PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator,
+                                 PipelinePackageFabric pipelinePackageManager,
+                                 PipelinePackageOrchestrator pipelineOrchestrator,
                                  CreditTrackerSettings creditTrackerSettings,
-                                           Saver saver,
+                                 MonitoringErrorProcesingService monitoringErrorProcesingService,
-                                           ExecutorService pipelineExecutorService,
+                                 Saver saver) {
-                                           ExecutorService saverExecutorService) {
         this.userPackageMapper = userPackageMapper;
         this.phoneFidMapMapper = phoneFidMapMapper;
         this.pipelinePackageManager = pipelinePackageManager;
         this.pipelineOrchestrator = pipelineOrchestrator;
         this.creditTrackerSettings = creditTrackerSettings;
+        this.monitoringErrorProcesingService = monitoringErrorProcesingService;
         this.saver = saver;
-        this.pipelineExecutorService = pipelineExecutorService;
-        this.saverExecutorService = saverExecutorService;
     }
 
     public void run() {
+        onMonitoringRestore();
         acceptedRestore();
         processingRestore();
-        processingRepeatedRestore();
+    }
+
+    private void onMonitoringRestore() {
+        List<UserPackage> packages = userPackageMapper.findOnMonitoringPackages();
+        packages.forEach(userPackage -> {
+            monitoringErrorProcesingService.findErrors(userPackage.getId());
+        });
     }
 
     private void acceptedRestore() {
@@ -82,11 +86,11 @@ public class UpdateProcessingPackagesService {
     }
 
     private void processingRestore() {
-        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name());
+        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name(), Status.PROCESSING_REPEATED.name());
-        packages = packages.stream().filter(userPackage -> userPackage.getRejectStatus() == null).toList();
+        log.debug("Found {} packages in status PROCESSING", packages.size());
-        log.debug("Found {} packages in status PROCESSING and rejected_status = null", packages.size());
         for (UserPackage userPackage : packages) {
             try {
+                log.debug("Build pipeline for packageId {}", userPackage.getId());
                 Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT));
                 pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
             } catch (Exception e) {
@@ -95,17 +99,4 @@ public class UpdateProcessingPackagesService {
         }
     }
 
-    private void processingRepeatedRestore() {
-        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING_REPEATED.name());
-        log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size());
-        for (UserPackage userPackage : packages) {
-            try {
-                Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE));
-                pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
-            } catch (Exception e) {
-                log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
-            }
-        }
-    }
-
 }

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

@@ -42,13 +42,18 @@ public class UpdatingFidAndPdService {
 
     @Scheduled(cron = "${ct.update-fid-cron-expression}")
     public void update() {
-        log.info("Delete delete unused phones");
+        if (!pipelineRecalcPDOrchestrator.checkPermits()) {
+            log.info("The previous task has not yet been completed.");
+            return;
+        }
+        log.info("Delete unused phones");
         phonesMapService.deleteUnusedPhones();
         log.info("Start updating fids and PD");
         try {
             Pipeline pipeline = pipelinePackageManager.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA));
             CompletableFuture<?> future = pipelineRecalcPDOrchestrator.startOrReject(pipeline);
             future.whenComplete((o, throwable) -> {
+                try {
                     if (throwable != null) {
                         log.error(throwable.getMessage(), throwable);
                         return;
@@ -65,7 +70,11 @@ public class UpdatingFidAndPdService {
                             log.error("Failed to build pipeline", e);
                         }
                     });
-
+                } finally {
+                    log.info("pipeline finished");
+                    pipeline.close();
+                    pipelineRecalcPDOrchestrator.release();
+                }
             });
         } catch (Exception e) {
             log.error("Failed to build pipeline", e);

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

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

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

@@ -102,6 +102,6 @@ public class PersonServiceV2 {
     }
 
     private boolean checkLength(Id p) {
-        return ("21".equals(p.getIdType2()) || "21".equals(p.getIdType())) && CTUtil.normalizePassport(p).length() == 10;
+        return !("21".equals(p.getIdType2()) && "21".equals(p.getIdType())) || CTUtil.normalizePassport(p).length() == 10;
     }
 }

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

@@ -33,6 +33,19 @@ public class PhonesMapService {
         }
     }
 
+    public void updateSearchedAt(List<Long> phones) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(phones.stream()
+                .filter(Objects::nonNull)
+                .map(phone -> new HashMap<String, Object>() {{
+                    put("phone", phone);
+                }}).toList());
+            phonesMapMapper.updateSearchedAt(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     public void updateNameCachedData(List<SubjectProfile> subjectProfiles) {
         try {
             String json = entityToJsonMapper.writeValueAsString(subjectProfiles.stream()

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

@@ -118,7 +118,7 @@ public class PipelinePackageOrchestrator {
                     }
                     if (res.signalXmlFilePath() == null) {
                         log.info("No one fid was found by packageId : {}; xml file won't be sent", packageId);
-                        userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
+                        userPackageMapper.updateNoFidMonitoring(packageId);
                         return;
                     }
 

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

@@ -36,6 +36,11 @@ public class PipelineRecalcPDOrchestrator {
         return innerStart(pipeline);
     }
 
+    public boolean checkPermits() {
+        synchronized (lock) {
+            return semaphore.availablePermits() > 0;
+        }
+    }
     private CompletableFuture<?> innerStart(Pipeline pipeline) {
         try {
             pipeline.start(pipelineExecutorService);
@@ -54,10 +59,11 @@ public class PipelineRecalcPDOrchestrator {
             .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS);
     }
 
-    private void release() {
+    public void release() {
         log.debug("checking for delayed pipelines...");
         synchronized (lock) {
             semaphore.release();
+            log.info("semaphore amount: {}", semaphore.availablePermits());
         }
     }
 }

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

@@ -51,10 +51,12 @@ public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
         log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
 
         List<Integer> noIdDataFids = subjectsProfile.stream()
-            .filter(subjectProfile -> subjectProfile.getIdData() == null)
+            .filter(subjectProfile -> subjectProfile.getIdData() == null && subjectProfile.getFid() != null)
             .map(SubjectProfile::getFid)
             .toList();
 
+        long countEmptyFid = subjectsProfile.stream().filter(subjectProfile -> subjectProfile.getFid() == null).count();
+        log.info("Count with empty fid: {}", countEmptyFid);
         Map<Integer, Id> idByFid;
         if (!noIdDataFids.isEmpty()) {
             idByFid = personService.filledPassportsData(noIdDataFids);
@@ -64,6 +66,7 @@ public class FidIdEnricher implements Handler<SubjectPayload, PartFilePayload> {
 
         List<IdData> idCacheOnUpdate = new ArrayList<>();
         List<Person> personOnProcessing = subjectsProfile.stream()
+            .filter(subjectProfile -> subjectProfile.getFid() != null)
             .flatMap(subjectProfile -> {
                 if (subjectProfile.getIdData() == null) {
                     Id id = idByFid.get(subjectProfile.getFid());

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

@@ -1,6 +1,5 @@
 package ru.nbch.credit_tracker.service.task.impl;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Stream;
@@ -36,14 +35,13 @@ public class FidIdUpdater implements Handler<SubjectPayload, VoidPayload> {
     @Override
     public VoidPayload handle(SubjectPayload taskPayload) {
         List<SubjectProfile> subjectsProfile = taskPayload.getSubjects();
-        log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size());
+        log.debug("FidIdUpdater handle; size: {}", subjectsProfile.size());
 
         List<Integer> noIdDataFids = subjectsProfile.stream()
             .map(SubjectProfile::getFid)
             .toList();
         Map<Integer, Id> idByFid = personService.filledPassportsData(noIdDataFids);
 
-        List<IdData> idCacheOnUpdate = new ArrayList<>();
         List<IdData> personOnProcessing = subjectsProfile.stream()
             .flatMap(subjectProfile -> {
                 Id id = idByFid.get(subjectProfile.getFid());

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

@@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.service.task.impl;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
@@ -47,19 +48,24 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
 
         List<SubjectProfile> changingFidPhones = new ArrayList<>();
         List<Long> notFoundFidPhones = new ArrayList<>();
-        subjects.forEach(subjectProfile -> {
+        List<Long> notChangedPhones = subjects.stream().flatMap(subjectProfile -> {
             Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone()));
-            if (subjectProfile.getFid() == null && name == null) {
+            if (name != null) {
-                notFoundFidPhones.add(subjectProfile.getPhone());
+                if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) {
-            } else if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) {
                     subjectProfile.setFid(name.getFid());
                     subjectProfile.getNameData().setFirstName(name.getFirst());
                     subjectProfile.getNameData().setLastName(name.getLast());
                     subjectProfile.getNameData().setMiddleName(name.getMiddle());
                     subjectProfile.getNameData().setBirthDate(name.getBirthDate());
                     changingFidPhones.add(subjectProfile);
+                    return Stream.empty();
                 }
-        });
+            } else if (subjectProfile.getFid() != null) {
+                notFoundFidPhones.add(subjectProfile.getPhone());
+                return Stream.empty();
+            }
+            return Stream.of(subjectProfile.getPhone());
+        }).toList();
         List<Long> changingFids = new ArrayList<>();
         if (!notFoundFidPhones.isEmpty()) {
             phonesMapService.resetFidAndPDByPhones(notFoundFidPhones);
@@ -71,6 +77,10 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
             changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList());
         }
 
+        if (!notChangedPhones.isEmpty()) {
+            phonesMapService.updateSearchedAt(notChangedPhones);
+        }
+
         if (!changingFids.isEmpty()) {
             List<Long> packageIds = packageRecordsService.selectUniquePackageIds(changingFids);
             if (!packageIds.isEmpty()) {

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

@@ -64,8 +64,8 @@ public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
     public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
         try {
             log.info("Start kicker: mode:{};", mode());
-            int offset = 0;
+            long lastId = 0;
-            List<PhonesMap> page = nextPage(offset);
+            List<PhonesMap> page = nextPage(lastId);
             while (!page.isEmpty()) {
                 log.debug("Get page {}", page.size());
                 List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
@@ -92,8 +92,8 @@ public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
                     log.error("Pipeline starter was interrupted!");
                     throw new PipelineStarterException();
                 }
-                offset += BATCH_SIZE;
+                lastId = page.getLast().getId();
-                page = nextPage(offset);
+                page = nextPage(lastId);
             }
         } catch (PipelineStarterException e) {
             log.info("Pipeline starter shutting down...");
@@ -115,7 +115,7 @@ public class KickRecalcPipeLine implements IPipeLineStarter<SubjectPayload> {
     protected PipelineMode mode() {
         return PipelineMode.REPEATED_RESTORE;
     }
-    protected List<PhonesMap> nextPage(int offset) {
+    protected List<PhonesMap> nextPage(long lastId) {
-        return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, offset);
+        return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId);
     }
 }

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

@@ -77,7 +77,7 @@ public class CTUtil {
         if (StringUtils.isBlank(value)) {
             return null;
         }
-        if (idType.equals("21") || idType2.equals("21")) {
+        if ("21".equals(idType) || "21".equals(idType2)) {
             return value.replaceAll("[^0-9]", "");
         } else {
             return value.replaceAll("[^А-Яа-яA-Za-z0-9]", "");

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

@@ -34,7 +34,7 @@
                sup.id_type,
                sup.flag_debt5000,
                sup.flag_90_12
-        from signals_packages_records spr
+        from signals_phone_fid_map spr
                  join signals_phones_map sup on cast(spr.phone as bigint) = sup.phone
         where spr.package_id = #{packageId}
           and sup.fid is not null

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

@@ -19,7 +19,7 @@
                subject_id,
                phone,
                error_text
-        FROM signals_packages_records
+        FROM signals_phone_fid_map
         WHERE id = #{id}
     </select>
 
@@ -28,7 +28,10 @@
                      FROM json_to_recordset(#{json}::json) AS t(
                                                                 phone bigint
                          ))
-        select distinct (package_id) from signals_packages_records where phone_l in (select * from src)
+        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="registerJson" parameterType="string" resultType="long">
@@ -42,7 +45,7 @@
                                                                 package_id bigint
                          ))
         INSERT
-        INTO signals_packages_records
+        INTO signals_phone_fid_map
             (subject_id, phone, phone_l, package_id)
         SELECT subject_id,
                phone,
@@ -54,7 +57,7 @@
     </select>
 
     <update id="setErrorJson" parameterType="string">
-        update signals_packages_records dst
+        update signals_phone_fid_map dst
         set error_text = src.error_text
         FROM json_to_recordset(#{json}::json) AS src(
                                                      id bigint,
@@ -65,7 +68,7 @@
 
     <delete id="deleteByPackageId" parameterType="long">
         DELETE
-        FROM signals_packages_records
+        FROM signals_phone_fid_map
         WHERE package_id = #{packageId}
     </delete>
 
@@ -75,7 +78,7 @@
                subject_id,
                phone,
                error_text
-        FROM signals_packages_records
+        FROM signals_phone_fid_map
         WHERE package_id = #{packageId}
         ORDER BY id
         LIMIT #{limit} OFFSET #{offset}

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

@@ -54,9 +54,9 @@
                searched_at,
                calculated_at
         FROM signals_phones_map
-        WHERE searched_at < now() - make_interval(days => #{daysCount})
+        WHERE searched_at < now() - make_interval(days => #{daysCount})  and id > #{lastId}
-        ORDER BY id
+        order by id
-        LIMIT #{limit} OFFSET #{offset}
+        limit #{batchSize}
     </select>
 
     <select id="getCachedData" resultMap="PhoneMap" resultType="java.util.List">
@@ -204,7 +204,16 @@
             id_type       = null,
             flag_debt5000 = null,
             flag_90_12    = null,
-            calculated_at = null
+            calculated_at = null,
+            searched_at = now()
+        FROM json_to_recordset(#{json}::json) AS src(
+                                                     phone bigint)
+        where dst.phone = src.phone
+    </update>
+
+    <update id="updateSearchedAt" parameterType="string">
+        update signals_phones_map dst
+        set searched_at = now()
         FROM json_to_recordset(#{json}::json) AS src(
                                                      phone bigint)
         where dst.phone = src.phone
@@ -214,7 +223,7 @@
         DELETE
         FROM signals_phones_map spm
         WHERE NOT EXISTS (SELECT 1
-                          FROM signals_packages_records spr
+                          FROM signals_phone_fid_map spr
                                    JOIN signals_user_packages sup
                                         ON sup.id = spr.package_id
                           WHERE spr.phone_l = spm.phone

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

@@ -92,7 +92,13 @@
 
     <update id="updatePackageOnMonitoringSuccess">
         UPDATE signals_user_packages
-        SET monitoring_at = #{monitoringAt}, status = 'ON_MONITORING'
+        SET monitoring_at = #{monitoringAt}, 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>
 
@@ -141,6 +147,23 @@
         WHERE fid_changed = 1 and status = 'PROCESSING_REPEATED'
     </select>
 
+    <select id="findOnMonitoringPackages" resultMap="packageDataResultMap">
+        SELECT id,
+               membercode,
+               package_name,
+               created_at,
+               status,
+               deactivated_at,
+               last_downloaded_report,
+               monitoring_at,
+               last_searched_at,
+               reject_status,
+               signal_types
+        FROM signals_user_packages
+        WHERE status = 'ON_MONITORING' and reject_status is null
+    </select>
+
+
     <select id="findUserPackages" resultMap="packageDataResultMap">
         SELECT id,
                membercode,

src/main/resources/sql/credit_tracker_db.ddl

@@ -1,43 +1,94 @@
--- Create signals_user_packages
+create table signals_user_packages
--- Хранит основную информацию о каждом пакете мониторинга, установленном клиентом
+(
-CREATE TABLE signals_user_packages (
+    id                     bigserial
-                                       id SERIAL PRIMARY KEY, -- внутренний идентификатор в таблице
+        primary key,
-                                       membercode VARCHAR(255), -- логин клиента
+    membercode             varchar(255),
-                                       package_name VARCHAR(255), -- имя пакета
+    package_name           varchar(255),
-                                       created_at TIMESTAMP, -- дата и время первичной регистрации пакета
+    created_at             timestamp,
-                                       status VARCHAR(255), -- статус пакета: ON_MONITORING, REJECTED, PROCESSING, DELETED, ERROR
+    status                 varchar(255),
-                                       deactivated_at TIMESTAMP, -- дата и время удаления пакета
+    deactivated_at         timestamp,
-                                       last_downloaded_report VARCHAR(255) -- имя последнего успешно загруженного отчёта из оф. сервиса по пакету
+    last_downloaded_report varchar(255),
+    monitoring_at          timestamp,
+    last_searched_at       timestamp,
+    signal_types           varchar(255),
+    reject_status          varchar(255),
+    external_package_name  varchar(255),
+    fid_changed            smallint
 );
 
--- Create signals_phone_fid_map
+alter table signals_user_packages
-CREATE TABLE signals_phone_fid_map (
+    owner to credit_tracker_user;
-                                       id SERIAL PRIMARY KEY, -- внутренний идентификатор в таблице
+
-                                       package_id  INTEGER NOT NULL, -- идентификатор пакета
+create table signals_hits
-                                       subject_id VARCHAR(255), -- идентификатор субъекта из XML
+(
-                                       phone VARCHAR(255), -- номер телефона
+    id               bigserial
-                                       fid VARCHAR(255), -- найденный FID или NULL
+        primary key,
-                                       flag_debt5000 SMALLINT, -- 1, если есть активная просрочка > 5000 руб., иначе 0
+    package_id       bigint not null
-                                       flag_90_12 SMALLINT, -- 1, если по субъекту когда-либо была просрочка 90+ дней за последний год, иначе – 0
+        constraint fk_signals_user_packages
-                                       calculated_at TIMESTAMP, -- дата последнего расчёта флагов
+            references signals_user_packages,
-                                       error_text TEXT,
+    subject_id       varchar(255),
-                                       CONSTRAINT fk_signals_user_packages
+    signal_type      integer,
-                                           FOREIGN KEY (package_id)
+    response_sent_at timestamp,
-                                               REFERENCES signals_user_packages(id)
+    report_file_name varchar(255),
+    phone            bigint
 );
 
--- Create signals_hits
+alter table signals_hits
-CREATE TABLE signals_hits (
+    owner to credit_tracker_user;
-                              id SERIAL PRIMARY KEY, -- внутренний идентификатор в таблице
+
-                              package_id  INTEGER NOT NULL, -- идентификатор пакета
+create index idx_signals_hits_response_sent_at
-                              subject_id VARCHAR(255), -- идентификатор субъекта, по которому сработал сигнал
+    on signals_hits (response_sent_at);
-                              signal_type INTEGER, -- тип сигнала
+
-                              response_sent_at TIMESTAMP, -- дата и время формирования отчета с данным сигналом
+create table signals_phone_fid_map
-                              report_file_name VARCHAR(255), -- имя XML-файла отчёта, в который вошел данный сигнал
+(
-                              CONSTRAINT fk_signals_user_packages
+    id            bigserial
-                                  FOREIGN KEY (package_id)
+        primary key,
-                                      REFERENCES signals_user_packages(id)
+    package_id    bigint not null
+        constraint fk_signals_user_packages
+            references signals_user_packages,
+    subject_id    varchar(255),
+    phone         varchar(255),
+    fid           varchar(255),
+    flag_debt5000 smallint,
+    flag_90_12    smallint,
+    calculated_at timestamp,
+    error_text    text,
+    phone_l       bigint
 );
 
--- Create index on response_sent_at
+alter table signals_phone_fid_map
-CREATE INDEX idx_signals_hits_response_sent_at ON signals_hits(response_sent_at);
+    owner to credit_tracker_user;
+
+create index idx_signals_phone_fid_map_calculated_at
+    on signals_phone_fid_map (calculated_at);
+
+create table app_version
+(
+    db_version varchar(12)
+);
+
+alter table app_version
+    owner to credit_tracker_user;
+
+create table signals_phones_map
+(
+    id            bigserial
+        primary key,
+    fid           integer,
+    phone         bigint
+        unique,
+    name_1        varchar(255),
+    first         varchar(255),
+    middle        varchar(255),
+    birth_dt      timestamp,
+    ser_num       varchar(255),
+    id_type       varchar(255),
+    flag_debt5000 smallint,
+    flag_90_12    smallint,
+    searched_at   timestamp,
+    calculated_at timestamp
+);
+
+alter table signals_phones_map
+    owner to credit_tracker_user;
+

src/main/resources/sql/v2.21.ddl

@@ -1,8 +1,3 @@
-CREATE TABLE app_version
-(
-    DB_VERSION varchar(12)
-);
-
 alter table signals_user_packages
     add LAST_SEARCHED_AT timestamp;
 alter table signals_user_packages
@@ -13,13 +8,17 @@ alter table signals_user_packages
     add EXTERNAL_PACKAGE_NAME varchar(255);
 alter table signals_user_packages
     add FID_CHANGED smallint;
-alter table signals_packages_records
-    add phone_l bigint;
-
 
 alter table signals_hits
     add PHONE bigint;
 
+alter table signals_phone_fid_map
+    add phone_l bigint;
+
+CREATE TABLE app_version
+(
+    DB_VERSION varchar(12)
+);
 
 CREATE TABLE signals_phones_map
 (
@@ -41,13 +40,4 @@ CREATE TABLE signals_phones_map
 ALTER TABLE signals_phones_map
     ADD CONSTRAINT signals_phones_map_phone_key UNIQUE (PHONE);
 
-
+update signals_phone_fid_map set phone_l = phone::BIGINT where phone is not null;
-
-alter table signals_phone_fid_map rename to signals_packages_records;
-alter table signals_packages_records drop column fid;
-alter table signals_packages_records drop column flag_debt5000;
-alter table signals_packages_records drop column flag_90_12;
-alter table signals_packages_records drop column calculated_at;
-alter table signals_packages_records drop column rec_uuid;
-alter table signals_packages_records drop column searched_at;
--- todo alter indexes and other constrains