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

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

@@ -5,7 +5,6 @@ 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.UpdatingFlagsService;
 
 @SpringBootApplication
 public class CreditTrackerApplication {
@@ -14,8 +13,6 @@ public class CreditTrackerApplication {
 		ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args);
 		UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class);
 		MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
-		UpdatingFlagsService updatingFlagsService = ctx.getBean(UpdatingFlagsService.class);
-//		updatingFlagsService.update();
 		updateProcessingPackagesService.run();
 		monitoringService.run();
 	}

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

@@ -18,4 +18,5 @@ public interface PhonesMapMapper {
     void resetFidAndPD(@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

@@ -30,6 +30,7 @@ public interface UserPackageMapper {
     void setFidChanged(@Param("packageIds") List<Long> id);
 
     UserPackage findPackage(@Param("id") Long id);
+    List<UserPackage> findForRepeatedPackages();
 
     List<UserPackage> findUserPackages(@Param("userId") String userId, @Param("statuses") List<String> status);
 

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

@@ -1,10 +1,15 @@
 package ru.nbch.credit_tracker.service;
 
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.UserPackage;
+import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
+import ru.nbch.credit_tracker.service.signal.PhonesMapService;
 import ru.nbch.credit_tracker.service.task.Pipeline;
 import ru.nbch.credit_tracker.service.task.PipelinePackageFabric;
 import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator;
@@ -19,30 +24,49 @@ public class UpdatingFidAndPdService {
     private final PipelinePackageFabric pipelinePackageManager;
     private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator;
     private final PipelinePackageOrchestrator pipelinePackageOrchestrator;
+    private final UserPackageMapper userPackageMapper;
+    private final PhonesMapService phonesMapService;
 
     public UpdatingFidAndPdService(CreditTrackerSettings config,
                                    PipelinePackageFabric pipelinePackageManager,
                                    PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator,
-                                   PipelinePackageOrchestrator pipelinePackageOrchestrator) {
+                                   PipelinePackageOrchestrator pipelinePackageOrchestrator,
+                                   UserPackageMapper userPackageMapper,
+                                   PhonesMapService phonesMapService) {
         this.pipelinePackageManager = pipelinePackageManager;
         this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator;
         this.pipelinePackageOrchestrator = pipelinePackageOrchestrator;
+        this.userPackageMapper = userPackageMapper;
+        this.phonesMapService = phonesMapService;
     }
 
     @Scheduled(cron = "${ct.update-fid-cron-expression}")
     public void update() {
+        log.info("Delete 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));
-            pipelineRecalcPDOrchestrator.startOrReject(pipeline, () -> {
+            CompletableFuture<?> future = pipelineRecalcPDOrchestrator.startOrReject(pipeline);
-                SetupParam setupParam = new SetupParam(null, null, PipelineMode.MAKE_XML_BY_REPEATED);
+            future.whenComplete((o, throwable) -> {
+                if (throwable != null) {
+                    log.error(throwable.getMessage(), throwable);
+                    return;
+                }
+
+                List<UserPackage> packagesRepeated = userPackageMapper.findForRepeatedPackages();
+                packagesRepeated.forEach(userPackage -> {
+                    Long packageId = userPackage.getId();
+                    SetupParam setupParam = new SetupParam(packageId, null, PipelineMode.DEFAULT);
                     try {
                         Pipeline makeXmlByCache = pipelinePackageManager.build(setupParam);
-                    pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, null);
+                        pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId);
                     } catch (Exception e) {
                         log.error("Failed to build pipeline", e);
                     }
                 });
+
+            });
         } catch (Exception e) {
             log.error("Failed to build pipeline", e);
         }

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.indic;
 
+import java.time.LocalDate;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -73,7 +74,7 @@ public class PersonServiceV2 {
             .filter(name ->
                     StringUtils.isNotEmpty(name.getFirst()) &&
                     StringUtils.isNotEmpty(name.getLast()) &&
-                    name.getBirthDate() != null
+                    name.getBirthDate() != null && !LocalDate.of(1, 1, 1).equals(name.getBirthDate())
             )
             .collect(Collectors.toMap(
                 Name::getNumber,

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

@@ -50,4 +50,8 @@ public class PhonesMapService {
             throw new RuntimeException(e);
         }
     }
+
+    public void deleteUnusedPhones() {
+        phonesMapMapper.deleteUnusedPhones();
+    }
 }

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

@@ -80,7 +80,6 @@ public class PipelinePackageFabric implements InitializingBean {
     public Pipeline build(SetupParam setupParam) throws Exception {
         return PipeLineBuilderV2.first(nameEnricherQueueFactory.getObject(), idEnricherQueueFactory.getObject())
                     .handler(fidNameEnricher)
-                    .handlerResult(flagsEnricherV2)
                     .andThen()
                 .stage(sendSignalQueueFactory.getObject())
                     .handler(fidIdEnricherHandler)

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.service.task;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
@@ -24,46 +25,33 @@ public class PipelineRecalcPDOrchestrator {
         this.semaphore = new Semaphore(1);
     }
 
-    public void startOrReject(Pipeline pipeline, Runnable after) {
+    public CompletableFuture<?> startOrReject(Pipeline pipeline) {
         synchronized (lock) {
             if (!semaphore.tryAcquire()) {
                 log.debug("couldn't acquire permit from semaphore. task rejected...");
-                return;
+                return CompletableFuture.failedFuture(new IllegalStateException("couldn't acquire permit from semaphore."));
             }
             log.info("throttler permits after acquire: {}", semaphore.availablePermits());
         }
-        innerStart(pipeline, after);
+        return innerStart(pipeline);
     }
 
-    private void innerStart(Pipeline pipeline, Runnable after) {
+    private CompletableFuture<?> innerStart(Pipeline pipeline) {
         try {
             pipeline.start(pipelineExecutorService);
-            addCallback(pipeline, after);
+            return addCallback(pipeline);
         } catch (Exception e) {
             log.error(ExceptionUtils.getStackTrace(e));
             pipeline.close();
             release();
+            return CompletableFuture.failedFuture(e);
         }
     }
 
-    private void addCallback(Pipeline pipeline, Runnable after) {
+    private CompletableFuture<?> addCallback(Pipeline pipeline) {
-        pipeline.getTerminator()
+        return pipeline.getTerminator()
             .doneFuture()
-            .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS)
+            .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS);
-            .whenComplete((v, e) -> {
-                try {
-                    long finishTime = System.currentTimeMillis();
-                    log.info("pipeline time passed: {} ms", finishTime - pipeline.getStartTime());
-                    after.run();
-                } catch (Exception ex) {
-                    log.error("pipeline finished with error{}", ExceptionUtils.getStackTrace(ex));
-                } finally {
-                    log.info("pipeline finished");
-                    pipeline.close();
-                    release();
-                    log.info("semaphore amount: {}", semaphore.availablePermits());
-                }
-            });
     }
 
     private void release() {

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

@@ -116,7 +116,7 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
             .filter(cr -> cachedRecord.getSerNum() != null && cachedRecord.getIdType() != null)
             .map(cr -> Optional.of(
                 (IdData.builder()
-                    .idNum(cr.getSerNum())
+                    .serNum(cr.getSerNum())
                     .idType(cr.getIdType())
                     .build()
             )))

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

@@ -49,9 +49,9 @@ public class FidNameUpdater implements Handler<SubjectPayload, SubjectPayload> {
         List<Long> notFoundFidPhones = new ArrayList<>();
         subjects.forEach(subjectProfile -> {
             Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone()));
-            if (name == null) {
+            if (subjectProfile.getFid() == null && name == null) {
                 notFoundFidPhones.add(subjectProfile.getPhone());
-            } else if (!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());

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

@@ -43,9 +43,10 @@
                          ))
         INSERT
         INTO signals_packages_records
-            (subject_id, phone, package_id)
+            (subject_id, phone, phone_l, package_id)
         SELECT subject_id,
                phone,
+               phone::BIGINT,
                package_id
         FROM src
         ORDER BY rn

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

@@ -145,7 +145,8 @@
             name_1   = src.name_1,
             first    = src.first,
             middle   = src.middle,
-            birth_dt = src.birth_dt
+            birth_dt = src.birth_dt,
+            searched_at = now()
         FROM json_to_recordset(#{json}::json) AS src(
                                                      fid int,
                                                      phone bigint,
@@ -192,35 +193,26 @@
             ser_num       = null,
             id_type       = null,
             flag_debt5000 = null,
-            flag_90_12    = null
+            flag_90_12    = null,
+            calculated_at = null
         FROM json_to_recordset(#{json}::json) AS src(
                                                      phone bigint)
         where dst.phone = src.phone
     </update>
 
-
+    <delete id="deleteUnusedPhones">
-    <!--    <select id="pageByPackageIdAndOlderThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">-->
+        DELETE
-    <!--        SELECT id,-->
+        FROM signals_phones_map spm
-    <!--               package_id,-->
+        WHERE NOT EXISTS (SELECT 1
-    <!--               subject_id,-->
+                          FROM signals_packages_records spr
-    <!--               phone,-->
+                                   JOIN signals_user_packages sup
-    <!--               error_text-->
+                                        ON sup.id = spr.package_id
-    <!--        FROM signals_phone_fid_map-->
+                          WHERE spr.phone_l = spm.phone
-    <!--        WHERE package_id = #{packageId} and searched_at &lt; now() - make_interval(days => #{daysCount})-->
+                            AND sup.status IN
-    <!--        ORDER BY id-->
+                                (
-    <!--        LIMIT #{limit} OFFSET #{offset}-->
+                                 'ON_MONITORING',
-    <!--    </select>-->
+                                 'PROCESSING',
-
+                                 'PROCESSING_REPEATED'
-    <!--    <select id="pageByPackageIdAndNewerThanN" resultMap="PhoneFidMapResultMap" resultType="java.util.List">-->
+                                    ))
-    <!--        SELECT id,-->
+    </delete>
-    <!--               package_id,-->
-    <!--               subject_id,-->
-    <!--               phone,-->
-    <!--               fid,-->
-    <!--               error_text-->
-    <!--        FROM signals_phone_fid_map-->
-    <!--        WHERE package_id = #{packageId} and searched_at &gt;= now() - make_interval(days => #{daysCount})-->
-    <!--        ORDER BY id-->
-    <!--        LIMIT #{limit} OFFSET #{offset}-->
-    <!--    </select>-->
 </mapper>

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

@@ -76,7 +76,8 @@
 
     <update id="setFidChanged">
         UPDATE signals_user_packages
-        SET fid_changed = 1
+        SET fid_changed = 1,
+            status = 'PROCESSING_REPEATED'
         WHERE id in
         <foreach item="packageId" collection="packageIds" open="(" separator="," close=")">
             #{packageId}
@@ -124,6 +125,22 @@
         LIMIT 1
     </select>
 
+    <select id="findForRepeatedPackages" 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 fid_changed = 1 and status = 'PROCESSING_REPEATED'
+    </select>
+
     <select id="findUserPackages" resultMap="packageDataResultMap">
         SELECT id,
                membercode,