LP: added last rejecting date calculating

credit-tracker.properties

@@ -122,6 +122,11 @@ ct.pipeline.name-updater-queue-capacity=${PIPELINE_NAME_UPDATER_QUEUE_CAPACITY:2
 ct.flag-updater.concurrency=${FLAG_UPDATER_CONCURRENCY:10}
 ct.flag-updater.queue-capacity=${FLAG_UPDATER_QUEUE_CAPACITY:100}
 
+#select and calc last rejected date
+ct.update-rejected-date-cron-expression=${UPDATE_REJECTED_DATE_CRON_EXPRESSION:0 25 18 * * *}
+ct.rejected-date-updater.concurrency=${REJECTED_DATE_UPDATER_CONCURRENCY:10}
+ct.rejected-date-updater.queue-capacity=${REJECTED_DATE_UPDATER_QUEUE_CAPACITY:100}
+
 #Backoff settings
 ct.pipeline.error-attempt-count=${PIPELINE_ERROR_ATTEMPT_COUNT:5}
 ct.pipeline.error-attempt-delay=${PIPELINE_ERROR_ATTEMPT_DELAY:2S}

src/main/java/ru/nbch/credit_tracker/config/settings/CreditTrackerSettings.java

@@ -30,6 +30,8 @@ public class CreditTrackerSettings {
     @NestedConfigurationProperty
     private FlagUpdaterSettings flagUpdater;
     @NestedConfigurationProperty
+    private RejectedDateUpdaterSettings rejectedDateUpdater;
+    @NestedConfigurationProperty
     private KafkaConsumerSettings kafkaConsumerSettings;
 
     private Long maxXmlFileSize;

src/main/java/ru/nbch/credit_tracker/config/settings/RejectedDateUpdaterSettings.java

@@ -0,0 +1,9 @@
+package ru.nbch.credit_tracker.config.settings;
+
+import org.springframework.boot.context.properties.bind.DefaultValue;
+
+public record RejectedDateUpdaterSettings(
+    @DefaultValue("10") Integer concurrency,
+    @DefaultValue("100") Integer queueCapacity
+) {
+}

src/main/java/ru/nbch/credit_tracker/entities/indic/RuApplication.java

@@ -0,0 +1,13 @@
+package ru.nbch.credit_tracker.entities.indic;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDate;
+
+@Getter
+@Setter
+public class RuApplication {
+    private Integer fid;
+    private LocalDate rejectedDate;
+}

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

@@ -153,13 +153,13 @@ public class SignalFacade {
             log.info("Package registered. PackageId {}. Status: {}", packageId, Status.ACCEPTED.name());
 
             if (subjectType == 1) {
-                Pipeline pipeline = npPipelinePackageFabric.build(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT));
+                Pipeline pipeline = npPipelinePackageFabric.build(new SetupParam(packageId, 1, PipelineMode.DEFAULT));
                 pipelineOrchestrator.startOrDelay(() -> {
                     saver.readAndSaveNpPackage(packageId, parsedFilePath);
                     CtFileUtils.deleteQuietly(parsedFilePath);
                 }, pipeline, packageId);
             } else {
-                Pipeline pipeline = lpPipelinePackageFabric.build(new SetupParam(packageId, parsedFilePath, PipelineMode.DEFAULT));
+                Pipeline pipeline = lpPipelinePackageFabric.build(new SetupParam(packageId, 2, PipelineMode.DEFAULT));
                 pipelineOrchestrator.startOrDelay(() -> {
                     saver.readAndSaveLpPackage(packageId, parsedFilePath);
                     CtFileUtils.deleteQuietly(parsedFilePath);

src/main/java/ru/nbch/credit_tracker/mapper/indic/IndicFlagsMapper.java

@@ -1,10 +1,12 @@
 package ru.nbch.credit_tracker.mapper.indic;
 
-import java.time.LocalDate;
-import java.util.List;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 import ru.nbch.credit_tracker.entities.indic.PaymentsFlag;
+import ru.nbch.credit_tracker.entities.indic.RuApplication;
+
+import java.time.LocalDate;
+import java.util.List;
 
 @Mapper
 public interface IndicFlagsMapper {
@@ -16,4 +18,5 @@ public interface IndicFlagsMapper {
 
     List<PaymentsFlag> findFlag90ByRuAllArrears(@Param("fids") List<Integer> fids, @Param("reportedDate") LocalDate reportedDate);
 
+    List<RuApplication> findLastRejecteDates(@Param("fids") List<Integer> fids);
 }

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

@@ -15,10 +15,15 @@ public interface LegalsMapMapper {
 
     void updateFlagCachedData(@Param("json") String json);
 
+    void updateLastRejectedDateCache(@Param("json")  String json);
+
     List<LegalsMap> findEligibleForFlagsUpdate(@Param("lastId") Long lastId,
                                                @Param("interval") Integer interval,
                                                @Param("batchSize") Integer batchSize);
 
+    List<LegalsMap> findEligibleForRejectedDatesToUpdate(@Param("lastId") Long lastId,
+                                                         @Param("batchSize") Integer batchSize);
+
     List<LegalsMap> findEligibleForFlagsInsert(@Param("lastId") Long lastId,
                                                @Param("batchSize") Integer batchSize);
 

src/main/java/ru/nbch/credit_tracker/model/FlagData.java

@@ -5,6 +5,8 @@ import lombok.Builder;
 import lombok.Getter;
 import lombok.Setter;
 
+import java.time.LocalDate;
+
 @Getter
 @Setter
 @Builder
@@ -14,4 +16,5 @@ public class FlagData {
     private Long regNum;
     private Integer flag_debt5000;
     private Integer flag_90_12;
+    private LocalDate lastRejectedDate;
 }

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

@@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.service.indic;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.indic.PaymentsFlag;
+import ru.nbch.credit_tracker.entities.indic.RuApplication;
 import ru.nbch.credit_tracker.mapper.indic.IndicFlagsMapper;
 
 import java.time.LocalDate;
@@ -38,4 +39,9 @@ public class FlagsService {
         return flagsMapper.findFlag90ByRuAllArrears(fids, date).stream()
             .collect(Collectors.toMap(PaymentsFlag::getFid, PaymentsFlag::getFlag));
     }
+
+    public Map<Integer, LocalDate> findLastRejecteDates(List<Integer> fids) {
+        return flagsMapper.findLastRejecteDates(fids).stream()
+                .collect(Collectors.toMap(RuApplication::getFid, RuApplication::getRejectedDate));
+    }
 }

src/main/java/ru/nbch/credit_tracker/service/mode/common/UpdatingLastRejectDateService.java

@@ -0,0 +1,54 @@
+package ru.nbch.credit_tracker.service.mode.common;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.entities.app.LegalsMap;
+import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper;
+import ru.nbch.credit_tracker.service.task.impl.LastRejectedDateEnricher;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Slf4j
+@Service
+public class UpdatingLastRejectDateService {
+    private final static int BATCH_SIZE = 1000;
+    private final LegalsMapMapper legalsMapMapper;
+    private final LastRejectedDateEnricher lastRejectedDateEnricher;
+    private final AtomicBoolean updateIsRunning = new AtomicBoolean(false);
+
+    public UpdatingLastRejectDateService(LegalsMapMapper legalsMapMapper,
+                                         LastRejectedDateEnricher lastRejectedDateEnricher) {
+        this.legalsMapMapper = legalsMapMapper;
+        this.lastRejectedDateEnricher = lastRejectedDateEnricher;
+    }
+
+    @Scheduled(cron = "${ct.update-rejected-date-cron-expression}")
+    private void updateFlags() {
+        if (!updateIsRunning.compareAndSet(false, true)) {
+            log.error("update flags processor already running... skipping scheduled update");
+            return;
+        }
+        try {
+            log.info("start update flags processor");
+            long lastLegalId = 0L;
+            int batchNumber = 1;
+            List<LegalsMap> legalsMaps;
+            while (true) {
+                legalsMaps = legalsMapMapper.findEligibleForRejectedDatesToUpdate(lastLegalId, BATCH_SIZE);
+                log.trace("update batch {}, signals_legals_map records loaded: {}.", batchNumber, legalsMaps.size());
+                if (legalsMaps.isEmpty()) {
+                    break;
+                }
+
+                lastRejectedDateEnricher.accept(legalsMaps);
+                lastLegalId = legalsMaps.getLast().getId();
+                log.trace("update batch {}, lastLegalId {}", batchNumber++, lastLegalId);
+            }
+            log.info("end update flags processor");
+        } finally {
+            updateIsRunning.set(false);
+        }
+    }
+}

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

@@ -158,10 +158,10 @@ public class PackageRecordsService {
         try {
             String json = entityToJsonMapper.writeValueAsString(flagData.stream()
                     .filter(Objects::nonNull)
-                    .map(phoneMap -> new HashMap<String, Object>() {{
+                    .map(flag -> new HashMap<String, Object>() {{
-                        put("inn", phoneMap.getInn());
+                        put("inn", flag.getInn());
-                        put("reg_num", phoneMap.getRegNum());
+                        put("reg_num", flag.getRegNum());
-                        put("flag_90_12", phoneMap.getFlag_90_12());
+                        put("flag_90_12", flag.getFlag_90_12());
                     }}).toList());
             legalsMapMapper.updateFlagCachedData(json);
         } catch (JsonProcessingException e) {
@@ -169,6 +169,21 @@ public class PackageRecordsService {
         }
     }
 
+    public void updateLpLastRejectedDateCache(List<FlagData> flagData) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(flagData.stream()
+                    .filter(Objects::nonNull)
+                    .map(flag -> new HashMap<String, Object>() {{
+                        put("inn", flag.getInn());
+                        put("reg_num", flag.getRegNum());
+                        put("last_rejected_date", flag.getLastRejectedDate());
+                    }}).toList());
+            legalsMapMapper.updateLastRejectedDateCache(json);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     public PhonesMap findCachedByPhone(Long phone) {
         return phonesMapMapper.findByPhone(phone);
     }

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

@@ -46,7 +46,8 @@ public class FlagsEnricher {
     }
 
     public void process(List<PhonesMap> phonesMaps, List<LegalsMap> legalsMaps) {
-        log.debug("FlagsEnricher handle: size: {}", phonesMaps.size());
+        log.debug("FlagsEnricher handle: np size: {}", phonesMaps.size());
+        log.debug("FlagsEnricher handle: lp size: {}", legalsMaps.size());
 
         List<Integer> noFlagDataFids = phonesMaps.stream()
             .map(PhonesMap::getFid)

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

@@ -0,0 +1,74 @@
+package ru.nbch.credit_tracker.service.task.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
+import ru.nbch.credit_tracker.entities.app.LegalsMap;
+import ru.nbch.credit_tracker.model.FlagData;
+import ru.nbch.credit_tracker.service.indic.FlagsService;
+import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
+
+import java.time.LocalDate;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+@Slf4j
+@Service
+public class LastRejectedDateEnricher {
+
+    private final FlagsService flagsService;
+    private final PackageRecordsService packageRecordsService;
+    private final ExecutorService ioPool;
+
+    public LastRejectedDateEnricher(FlagsService flagsService,
+                                    PackageRecordsService packageRecordsService,
+                                    CreditTrackerSettings creditTrackerSettings) {
+        this.flagsService = flagsService;
+        this.packageRecordsService = packageRecordsService;
+        this.ioPool = new ThreadPoolExecutor(creditTrackerSettings.getRejectedDateUpdater().concurrency(),
+                creditTrackerSettings.getRejectedDateUpdater().concurrency(),
+                0L, TimeUnit.MINUTES,
+                new LinkedBlockingQueue<>(creditTrackerSettings.getRejectedDateUpdater().queueCapacity()),
+                r -> {
+                    Thread t = new Thread(r);
+                    t.setName("last-rejected-fate-enricher-%d".formatted(t.threadId()));
+                    return t;
+                },
+                new ThreadPoolExecutor.CallerRunsPolicy()
+        );
+    }
+
+    private void process(List<LegalsMap> legalsMaps) {
+        log.debug("FlagsEnricher handle: lp size: {}", legalsMaps.size());
+
+        List<Integer> ligalsNoRejecedDateFids = legalsMaps.stream()
+                .map(LegalsMap::getFid)
+                .toList();
+
+        Map<Integer, LocalDate> fidDatesMap = flagsService.findLastRejecteDates(ligalsNoRejecedDateFids);
+        List<FlagData> lpFlagCacheOnUpdate = legalsMaps.stream()
+                .flatMap(legalMap -> {
+                    Integer fid = legalMap.getFid();
+                    LocalDate lastRejectedDate = fidDatesMap.get(fid);
+                    FlagData flagData = FlagData.builder()
+                            .lastRejectedDate(lastRejectedDate)
+                            .inn(legalMap.getInn())
+                            .regNum(legalMap.getRegNum())
+                            .build();
+                    return Stream.of(flagData);
+                }).toList();
+
+        packageRecordsService.updateLpLastRejectedDateCache(lpFlagCacheOnUpdate);
+    }
+
+    public void accept(List<LegalsMap> legalsMaps) {
+        ThreadPoolExecutor casted = (ThreadPoolExecutor) ioPool;
+        log.info("submitting payload. active tasks number: {}; queue size: {}", casted.getActiveCount(), casted.getQueue().size());
+        this.ioPool.submit(() -> process(legalsMaps));
+    }
+}

src/main/resources/mybatis/mapper/indic/FlagsMapper.xml

@@ -10,6 +10,11 @@
         <result property="flag" column="flag" />
     </resultMap>
 
+    <resultMap id="RuApplicationResult" type="ru.nbch.credit_tracker.entities.indic.RuApplication">
+        <result property="fid" column="fid" />
+        <result property="rejectedDate" column="REJECTED_DATE" />
+    </resultMap>
+
     <select id="findFlagsDebt5000" resultMap="PaymentsFlagResult">
         SELECT FID as fid,
         (case
@@ -87,4 +92,16 @@
         GROUP BY FID
         WITH UR
     </select>
+
+    <select id="findLastRejecteDates" resultMap="RuApplicationResult">
+        SELECT
+        FID as fid,
+        REJECTED_DATE as rejectedDate
+        FROM RU_APPLICATION
+        WHERE FID IN
+        <foreach item="fid" collection="fids" open="(" separator="," close=")">
+            #{fid}
+        </foreach>
+        WITH UR
+    </select>
 </mapper>

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

@@ -97,8 +97,18 @@
         limit #{batchSize}
     </select>
 
+    <select id="findEligibleForRejectedDatesToUpdate" resultMap="LegalsMap">
+        select id, fid, inn, reg_num
+        from signals_legals_map
+        where fid is not null
+          and id > #{lastId}
+        order by id
+        limit #{batchSize}
+    </select>
+
     <update id="updateFlagCachedData" parameterType="string">
-        WITH src AS (SELECT DISTINCT ON (inn, reg_num) inn,
+        WITH src AS (SELECT DISTINCT ON (inn, reg_num)
+                                     inn,
                                      reg_num,
                                      flag_90_12
                      FROM json_to_recordset(#{json}::json) AS src(
@@ -126,6 +136,35 @@
           AND dst.id IN (SELECT id FROM to_lock);
     </update>
 
+    <update id="updateLastRejectedDateCache" parameterType="string">
+        WITH src AS (SELECT DISTINCT ON (inn, reg_num)
+                                     inn,
+                                     reg_num,
+                                     last_rejected_date
+                     FROM json_to_recordset(#{json}::json) AS src(
+                                                                  inn bigint,
+                                                                  reg_num text,
+                                                                  last_rejected_date timestamp
+                         )
+                     ORDER BY inn, reg_num),
+             to_lock AS (SELECT dst.id
+                         FROM signals_legals_map dst
+                                  JOIN src ON src.inn = dst.inn AND src.reg_num = dst.reg_num
+                         ORDER BY dst.inn, dst.reg_num
+                             FOR UPDATE)
+
+        UPDATE signals_legals_map dst
+        SET last_rejected_date    = src.last_rejected_date
+        FROM json_to_recordset(#{json}::json) AS src(
+            inn bigint,
+            reg_num text,
+            last_rejected_date timestamp
+            )
+        WHERE dst.inn = src.inn
+          AND dst.reg_num = src.reg_num
+          AND dst.id IN (SELECT id FROM to_lock);
+    </update>
+
 
     <select id="searchedAtLessThan" resultMap="LegalsMap">
         select id,