LP: added flags update by cron

src/main/java/ru/nbch/credit_tracker/entities/app/LegalsMap.java

@@ -12,7 +12,7 @@ import java.time.LocalDateTime;
 public class LegalsMap {
     private Long id;
     private Long inn;
-    private Long regNUm;
+    private Long regNum;
     private Integer fid;
     private LocalDateTime searchedAt;
     private Integer flag90days;

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

@@ -11,4 +11,12 @@ public interface LegalsMapMapper {
 
     List<LegalsMap> findByFids(@Param("fids") List<Integer> fids);
 
+    void updateFlagCachedData(@Param("json") String json);
+
+    List<LegalsMap> findEligibleForFlagsUpdate(@Param("lastId") Long lastId,
+                                               @Param("interval") Integer interval,
+                                               @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

@@ -10,6 +10,8 @@ import lombok.Setter;
 @Builder
 public class FlagData {
     private Long phone;
+    private Long inn;
+    private Long regNum;
     private Integer flag_debt5000;
     private Integer flag_90_12;
 }

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

@@ -5,7 +5,9 @@ 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.LegalsMap;
 import ru.nbch.credit_tracker.entities.app.PhonesMap;
+import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper;
 import ru.nbch.credit_tracker.mapper.signal.PhonesMapMapper;
 import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
 
@@ -22,16 +24,18 @@ import java.util.concurrent.atomic.AtomicBoolean;
 public class UpdatingFlagsService {
     private final static int BATCH_SIZE = 1000;
     private final PhonesMapMapper phonesMapMapper;
-    private final FlagsEnricher flagsEnricherV2;
+    private final LegalsMapMapper legalsMapMapper;
+    private final FlagsEnricher flagsEnricher;
     private final Integer updateDaysInterval;
     private final AtomicBoolean insertIsRunning = new AtomicBoolean(false);
     private final AtomicBoolean updateIsRunning = new AtomicBoolean(false);
 
     public UpdatingFlagsService(CreditTrackerSettings settings,
-                                PhonesMapMapper phonesMapMapper,
+                                PhonesMapMapper phonesMapMapper, LegalsMapMapper legalsMapMapper,
-                                FlagsEnricher flagsEnricherV2) {
+                                FlagsEnricher flagsEnricher) {
         this.phonesMapMapper = phonesMapMapper;
-        this.flagsEnricherV2 = flagsEnricherV2;
+        this.legalsMapMapper = legalsMapMapper;
+        this.flagsEnricher = flagsEnricher;
         this.updateDaysInterval = settings.getDaysToUpdateFlags();
     }
 
@@ -43,19 +47,23 @@ public class UpdatingFlagsService {
         }
         try {
             log.info("start update flags processor");
-            long lastId = 0L;
+            long lastPhoneId = 0L;
+            long lastLegalId = 0L;
             int batchNumber = 1;
             List<PhonesMap> phonesMaps;
+            List<LegalsMap> legalsMaps;
             while (true) {
-                phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastId, updateDaysInterval, BATCH_SIZE);
+                phonesMaps = phonesMapMapper.findEligibleForFlagsUpdate(lastPhoneId, updateDaysInterval, BATCH_SIZE);
-                log.trace("update batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size());
+                legalsMaps = legalsMapMapper.findEligibleForFlagsUpdate(lastLegalId, updateDaysInterval, BATCH_SIZE);
-                if (phonesMaps.isEmpty()) {
+                log.trace("update batch {}, signals_phones_map records loaded: {}. signals_legals_map records loaded: {}.", batchNumber, phonesMaps.size(), legalsMaps.size());
+                if (phonesMaps.isEmpty() && legalsMaps.isEmpty()) {
                     break;
                 }
 
-                flagsEnricherV2.accept(phonesMaps);
+                flagsEnricher.accept(phonesMaps, legalsMaps);
-                lastId = phonesMaps.getLast().getId();
+                lastPhoneId = phonesMaps.getLast().getId();
-                log.trace("update batch {}, lastId {}", batchNumber++, lastId);
+                lastLegalId = legalsMaps.getLast().getId();
+                log.trace("update batch {}, lastPhoneId {}, lastLegalId {}", batchNumber++, lastPhoneId, lastLegalId);
             }
             log.info("end update flags processor");
         } finally {
@@ -71,19 +79,23 @@ public class UpdatingFlagsService {
         }
         try {
             log.info("start processing new flags");
-            long lastId = 0L;
+            long lastPhoneId = 0L;
+            long lastLegalId = 0L;
             int batchNumber = 1;
             List<PhonesMap> phonesMaps;
+            List<LegalsMap> legalsMaps;
             while (true) {
-                phonesMaps = phonesMapMapper.findEligibleForFlagsInsert(lastId, BATCH_SIZE);
+                phonesMaps = phonesMapMapper.findEligibleForFlagsInsert(lastPhoneId, BATCH_SIZE);
-                log.trace("new batch {}, signals_phones_map records loaded: {}", batchNumber, phonesMaps.size());
+                legalsMaps = legalsMapMapper.findEligibleForFlagsInsert(lastLegalId, BATCH_SIZE);
+                log.trace("new batch {}, signals_phones_map records loaded: {}. signals_legals_map records loaded: {}.", batchNumber, phonesMaps.size(), legalsMaps.size());
                 if (phonesMaps.isEmpty()) {
                     break;
                 }
 
-                flagsEnricherV2.accept(phonesMaps);
+                flagsEnricher.accept(phonesMaps, legalsMaps);
-                lastId = phonesMaps.getLast().getId();
+                lastPhoneId = phonesMaps.getLast().getId();
-                log.trace("new batch {}, lastId {}", batchNumber++, lastId);
+                lastLegalId = legalsMaps.getLast().getId();
+                log.trace("new batch {}, lastPhoneId {}, lastLegalId {}", batchNumber++, lastPhoneId, lastLegalId);
             }
             log.info("end new flags processor");
         } finally {

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java

@@ -206,7 +206,7 @@ public class KafkaMonitoringService {
                 .collect(Collectors.groupingBy(LegalsMap::getFid));
 
         Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
-                .map(l -> Pair.of(l.getInn(), l.getRegNUm()))
+                .map(l -> Pair.of(l.getInn(), l.getRegNum()))
                 .collect(Collectors.toSet());
 
         // прелоад по memberCode категорий
@@ -243,7 +243,7 @@ public class KafkaMonitoringService {
                 LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
 
                 // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
-                List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNUm()));
+                List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()));
 
                 if (userPackageLegalDataList == null) {
                     continue;

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

@@ -113,7 +113,7 @@ public class PackageRecordsService {
         }
     }
 
-    public void updateFlagCache(List<FlagData> flagData) {
+    public void updateNpFlagCache(List<FlagData> flagData) {
         try {
             String json = entityToJsonMapper.writeValueAsString(flagData.stream()
                     .filter(Objects::nonNull)
@@ -128,6 +128,21 @@ public class PackageRecordsService {
         }
     }
 
+    public void updateLpFlagCache(List<FlagData> flagData) {
+        try {
+            String json = entityToJsonMapper.writeValueAsString(flagData.stream()
+                    .filter(Objects::nonNull)
+                    .map(phoneMap -> new HashMap<String, Object>() {{
+                        put("inn", phoneMap.getInn());
+                        put("reg_num", phoneMap.getRegNum());
+                        put("flag_90_12", phoneMap.getFlag_90_12());
+                    }}).toList());
+            legalsMapMapper.updateFlagCachedData(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

@@ -3,6 +3,7 @@ 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.entities.app.PhonesMap;
 import ru.nbch.credit_tracker.model.FlagData;
 import ru.nbch.credit_tracker.service.indic.FlagsService;
@@ -15,6 +16,7 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 @Slf4j
@@ -43,12 +45,18 @@ public class FlagsEnricher {
         );
     }
 
-    public void process(List<PhonesMap> phonesMaps) {
+    public void process(List<PhonesMap> phonesMaps, List<LegalsMap> legalsMaps) {
         log.debug("FlagsEnricher handle: size: {}", phonesMaps.size());
 
         List<Integer> noFlagDataFids = phonesMaps.stream()
             .map(PhonesMap::getFid)
-            .toList();
+            .collect(Collectors.toList());
+
+        List<Integer> ligalsNoFlagDataFids = legalsMaps.stream()
+                .map(LegalsMap::getFid)
+                .toList();
+
+        noFlagDataFids.addAll(ligalsNoFlagDataFids);
 
         LocalDate yearAgo = LocalDate.now().minusYears(1);
 
@@ -72,13 +80,30 @@ public class FlagsEnricher {
                     .build();
                 return Stream.of(flagData);
             }).toList();
-        packageRecordsService.updateFlagCache(flagCacheOnUpdate);
+
+
+        List<FlagData> lpFlagCacheOnUpdate = legalsMaps.stream()
+                .flatMap(legalMap -> {
+                    Integer fid = legalMap.getFid();
+                    int pastDue90 = flag90ByPastDueArrear.getOrDefault(fid, 0);
+                    int paymt90 = flag90ByRuPaymt.getOrDefault(fid, 0);
+                    int allArrears90 = flag90ByRuAllArrear.getOrDefault(fid, 0);
+                    Integer flagDays90 = Math.max(allArrears90, Math.max(pastDue90, paymt90));
+                    FlagData flagData = FlagData.builder()
+                            .flag_90_12(flagDays90)
+                            .inn(legalMap.getInn())
+                            .regNum(legalMap.getRegNum())
+                            .build();
+                    return Stream.of(flagData);
+                }).toList();
+
+        packageRecordsService.updateNpFlagCache(flagCacheOnUpdate);
     }
 
-    public void accept(List<PhonesMap> phonesMaps) {
+    public void accept(List<PhonesMap> phonesMaps, 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(phonesMaps));
+        this.ioPool.submit(() -> process(phonesMaps, legalsMaps));
     }
 
 }

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

@@ -32,5 +32,54 @@
         </foreach>
     </select>
 
+    <select id="findEligibleForFlagsUpdate" resultMap="LegalsMap">
+        select id, fid, inn, reg_num
+        from signals_legals_map
+        where fid is not null
+          and calculated_at is not null
+          and calculated_at &lt; (current_date - make_interval(days => #{interval}))
+          and id > #{lastId}
+        order by id
+        limit #{batchSize}
+    </select>
+
+    <select id="findEligibleForFlagsInsert" resultMap="LegalsMap">
+        select id, fid, inn, reg_num
+        from signals_legals_map
+        where fid is not null
+          and calculated_at is 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,
+                                     reg_num,
+                                     flag_90_12
+                     FROM json_to_recordset(#{json}::json) AS src(
+                                                                  inn bigint,
+                                                                  reg_num text,
+                                                                  flag_90_12 smallint
+                         )
+                     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 flag_90_12    = src.flag_90_12,
+            calculated_at = now()
+        FROM json_to_recordset(#{json}::json) AS src(
+            inn bigint,
+            reg_num text,
+            flag_90_12 smallint
+            )
+        WHERE dst.inn = src.inn
+          AND dst.reg_num = src.reg_num
+          AND dst.id IN (SELECT id FROM to_lock);
+    </update>
 
 </mapper>

src/test/java/ru/nbch/credit_tracker/service/task/impl/FlagsEnricherTest.java

@@ -9,6 +9,7 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.config.settings.FlagUpdaterSettings;
+import ru.nbch.credit_tracker.entities.app.LegalsMap;
 import ru.nbch.credit_tracker.entities.app.PhonesMap;
 import ru.nbch.credit_tracker.model.FlagData;
 import ru.nbch.credit_tracker.service.indic.FlagsService;
@@ -125,9 +126,9 @@ public class FlagsEnricherTest {
         when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class)))
                 .thenReturn(flag90ByRuAllArrears);
 
-        flagsEnricher.process(phonesMaps);
+        flagsEnricher.process(phonesMaps, List.of());
 
-        verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture());
+        verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture());
         List<FlagData> actualFlagCache = flagCacheCaptor.getValue();
 
         assertNotNull(actualFlagCache);
@@ -148,15 +149,16 @@ public class FlagsEnricherTest {
     @Test
     void process_withEmptyList_shouldCallUpdateFlagCacheWithEmptyList() {
         List<PhonesMap> emptyList = List.of();
+        List<LegalsMap> legalsEmptyList = List.of();
 
         when(flagsService.findFlagsDebt5000(anyList())).thenReturn(Map.of());
         when(flagsService.findFlag90ByPastDueArrear(anyList(), any(LocalDate.class))).thenReturn(Map.of());
         when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(Map.of());
         when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(Map.of());
 
-        flagsEnricher.process(emptyList);
+        flagsEnricher.process(emptyList, legalsEmptyList);
 
-        verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture());
+        verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture());
         List<FlagData> actualFlagCache = flagCacheCaptor.getValue();
         assertTrue(actualFlagCache.isEmpty());
     }
@@ -178,9 +180,9 @@ public class FlagsEnricherTest {
         when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(zeroMap);
         when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(zeroMap);
 
-        flagsEnricher.process(phonesMaps);
+        flagsEnricher.process(phonesMaps, List.of());
 
-        verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture());
+        verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture());
         List<FlagData> actualFlagCache = flagCacheCaptor.getValue();
 
         assertEquals(2, actualFlagCache.size());
@@ -210,9 +212,9 @@ public class FlagsEnricherTest {
         when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(flag90ByRuPaymt);
         when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(flag90ByRuAllArrears);
 
-        flagsEnricher.process(phonesMaps);
+        flagsEnricher.process(phonesMaps,List.of());
 
-        verify(packageRecordsService).updateFlagCache(flagCacheCaptor.capture());
+        verify(packageRecordsService).updateNpFlagCache(flagCacheCaptor.capture());
         List<FlagData> actualFlagCache = flagCacheCaptor.getValue();
 
         assertFlagData(actualFlagCache, 79123456781L, 1, 1);
@@ -228,7 +230,7 @@ public class FlagsEnricherTest {
         when(flagsService.findFlag90ByRuPaymt(anyList(), any(LocalDate.class))).thenReturn(Map.of());
         when(flagsService.findFlag90ByRuAllArrears(anyList(), any(LocalDate.class))).thenReturn(Map.of());
 
-        flagsEnricher.process(phonesMaps);
+        flagsEnricher.process(phonesMaps, List.of());
 
         verify(flagsService).findFlagsDebt5000(expectedFids);
         verify(flagsService).findFlag90ByPastDueArrear(eq(expectedFids), any(LocalDate.class));