flag calc

pom.xml

@@ -41,6 +41,10 @@
             <artifactId>postgresql</artifactId>
             <scope>runtime</scope>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.ibm.db2</groupId>
             <artifactId>jcc</artifactId>

src/main/java/ru/nbch/credit_tracker/config/task/StageConfig.java

@@ -25,7 +25,7 @@ public class StageConfig {
 
     @Bean("flagQueueFactory")
     public StageQueueFactory<SubjectPayload> flagQueueFactory() {
-        return new StageQueueFactory<>(15);
+        return new StageQueueFactory<>(100);
     }
 
 }

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

@@ -5,9 +5,13 @@ 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.service.indic.FlagsService;
 
 @Mapper
 public interface FlagsMapper {
+    List<FlagsService.RuPastDuarearData> findRuPastDuarearData(@Param("fids") List<Integer> fids);
+
+    List<FlagsService.RuPaymtData> findRuPaymtData(@Param("fids") List<Integer> fids, @Param("reportedDate") LocalDate reportedDate);
 
     List<PaymentsFlag> findFlagsDebt5000(@Param("fids") List<Integer> fids);
 

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

@@ -13,6 +13,7 @@ public interface PhoneFidMapMapper {
 
     void registerSubjects(@Param("subjects") List<Subject> subjects, @Param("packageId") Long packageId);
     void register(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
+    void registerJson(@Param("json") String json);
     void registerWithId(@Param("phoneFidMap") List<PhoneFidMap> phoneFidMap);
 
     List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Long packageId, @Param("offset") int offset, @Param("limit") int limit);

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

@@ -1,7 +1,13 @@
 package ru.nbch.credit_tracker.service.task.impl;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -28,6 +34,10 @@ public class FlagsEnricher implements Runnable {
     private BlockingQueue<SubjectPayload> in;
     private final CompletableFuture<Void> done = new CompletableFuture<>();
     private final ExecutorService ioPool;
+    private final ObjectMapper mapper = new ObjectMapper()
+        .registerModule(new JavaTimeModule())
+        .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
+        .setSerializationInclusion(JsonInclude.Include.NON_NULL);
 
     public FlagsEnricher(FlagsService2 flagsService,
                          PhoneFidMapMapper phoneFidMapMapper) {
@@ -62,8 +72,7 @@ public class FlagsEnricher implements Runnable {
                     done.complete(null);
                     return;
                 }
-                Future<?> future = ioPool.submit(() -> process(subjectPayload));
+                inFlight.add(ioPool.submit(() -> process(subjectPayload)));
-                inFlight.add(future);
             } catch (InterruptedException ie) {
                 Thread.currentThread().interrupt();
             } catch (Exception e) {
@@ -87,14 +96,16 @@ public class FlagsEnricher implements Runnable {
         Map<Integer, Integer> flag90ByRuPaymt = flagsService.findFlag90ByRuPaymt(fids, yearAgo);
 
         List<PhoneFidMap> phoneFidMaps = subjects.stream().map(subject -> {
-            Integer debt500 = null;
+            Integer debt5000 = null;
-            Integer pastDue90 = null;
+            Integer flagDays90 = null;
-            Integer paymt90 = null;
+            LocalDateTime calculatedAt = null;
 
             if (subject.getFid() != null) {
-                debt500 = flagsDebt5000.getOrDefault(subject.getFid(), 0);
+                debt5000 = flagsDebt5000.getOrDefault(subject.getFid(), 0);
-                pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0);
+                int pastDue90 = flag90ByPastDueArrear.getOrDefault(subject.getFid(), 0);
-                paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0);
+                int paymt90 = flag90ByRuPaymt.getOrDefault(subject.getFid(), 0);
+                flagDays90 = Math.max(pastDue90, paymt90);
+                calculatedAt = LocalDateTime.now();
             }
 
             return PhoneFidMap.builder()
@@ -102,11 +113,37 @@ public class FlagsEnricher implements Runnable {
                 .subjectId(subject.getIdentifier())
                 .packageId(subjectPayload.getPackageId())
                 .fid(subject.getFid())
-                .flagDebt5000(debt500)
+                .flagDebt5000(debt5000)
-                .flag90days(paymt90)
+                .flag90days(flagDays90)
+                .calculatedAt(calculatedAt)
                 .build();
         }).toList();
-        phoneFidMapMapper.register(phoneFidMaps);
+
+        List<PhoneFidMap> nullObjs = phoneFidMaps.stream().filter(Objects::isNull).toList();
+        log.debug("null objs: {}", nullObjs.size());
+        if (!nullObjs.isEmpty()) {
+            int i = 0;
+            i++;
+        }
+        String json = null;
+        try {
+            json = mapper.writeValueAsString(
+                phoneFidMaps.stream()
+                    .filter(Objects::nonNull)
+                    .map(p -> new HashMap() {{
+                    put("subject_id", p.getSubjectId());
+                    put("fid", p.getFid());
+                    put("phone", p.getPhone());
+                    put("package_id", p.getPackageId());
+                    put("flag_debt5000", p.getFlagDebt5000());
+                    put("flag_90_12", p.getFlag90days());
+                    put("calculated_at", p.getCalculatedAt());
+                    }}
+                ).toList());
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        phoneFidMapMapper.registerJson(json);
     }
 
     public void setPollingQueue(BlockingQueue<SubjectPayload> queue) {

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

@@ -45,7 +45,7 @@
         with UR
     </select>
 
-    <select id="findFlagsDebt5000" resultType="PaymentsFlagResult">
+    <select id="findFlagsDebt5000" resultMap="PaymentsFlagResult">
         SELECT
         FID as fid,
         (case when (sum(case when AMT_PAST_DUE > 5000 then 1 else 0 end) = 0) then 0 else 1 end) as flag

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

@@ -26,13 +26,27 @@
     </insert>
 
     <insert id="register" parameterType="java.util.List">
-        INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id)
+        INSERT INTO signals_phone_fid_map (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at)
         values
         <foreach collection="phoneFidMap" item="phoneFid" separator=",">
-            (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId})
+            (#{phoneFid.subjectId}, #{phoneFid.fid}, #{phoneFid.phone}, #{phoneFid.packageId}, #{phoneFid.flagDebt5000}, #{phoneFid.flag90days}, #{phoneFid.calculatedAt})
         </foreach>
     </insert>
 
+    <insert id="registerJson" parameterType="string">
+        INSERT INTO signals_phone_fid_map
+        (subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at)
+        SELECT subject_id, fid, phone, package_id, flag_debt5000, flag_90_12, calculated_at
+        FROM json_to_recordset(#{json}::json) AS t(
+                                                   subject_id   varchar(255),
+                                                   fid          int,
+                                                   phone        text,
+                                                   package_id   bigint,
+                                                   flag_debt5000 smallint,
+                                                   flag_90_12    smallint,
+                                                   calculated_at timestamp)
+    </insert>
+
     <insert id="registerWithId" parameterType="java.util.List" useGeneratedKeys="true" keyColumn="id" keyProperty="id">
         INSERT INTO signals_phone_fid_map (subject_id, phone, package_id)
         values