http://jira.mfd.msk:8088/browse/BKI-3241

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);
@@ -16,6 +16,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/service/UpdateProcessingPackagesService.java

@@ -94,17 +94,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,7 +42,11 @@ 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 {

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/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);

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,7 +48,7 @@ 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 (name != null) {
                 if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) {
@@ -57,11 +58,14 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
                     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);
@@ -73,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/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">
@@ -201,6 +201,14 @@
         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
+    </update>
+
     <delete id="deleteUnusedPhones">
         DELETE
         FROM signals_phones_map spm

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

@@ -92,7 +92,7 @@
 
     <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>