updating flags

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

@@ -1,5 +1,6 @@
 package ru.nbch.credit_tracker.mapper.signal;
 
+import java.util.List;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 import ru.nbch.credit_tracker.entities.app.FlagDetail;
@@ -7,5 +8,7 @@ import ru.nbch.credit_tracker.entities.app.FlagDetail;
 @Mapper
 public interface FlagDetailMapper {
     void registerCalcFlags(@Param("json") String json);
+    void updateCalcFlags(@Param("json") String json);
     FlagDetail getByMapId(@Param("mapId") Long mapId);
+    void deleteFlags(@Param("mapIds") List<Long> mapIds);
 }

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

@@ -29,6 +29,8 @@ import ru.nbch.credit_tracker.entities.external_request.signals.Person;
 import ru.nbch.credit_tracker.entities.external_request.signals.Signals;
 import ru.nbch.credit_tracker.entities.external_response.monitoring_report_error.MonitoringReportErrors;
 import ru.nbch.credit_tracker.enums.Status;
+import ru.nbch.credit_tracker.model.SubjectProfile;
+import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.indic.FlagsService;
 import ru.nbch.credit_tracker.service.indic.PersonService;
 import ru.nbch.credit_tracker.service.indic.PhoneService;
@@ -36,6 +38,8 @@ import ru.nbch.credit_tracker.service.indic.PhoneService2;
 import ru.nbch.credit_tracker.service.signal.HitsService;
 import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
 import ru.nbch.credit_tracker.service.signal.UserPackageDao;
+import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher;
+import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher;
 import ru.nbch.credit_tracker.service.xml.stax.SignalsStaxXmlWriter;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
 import ru.nbch.credit_tracker.utils.CTUtil;
@@ -55,8 +59,10 @@ public class SignalService {
     private final FlagsService flagsService;
     private final PersonService personService;
     private final StaxXmlProcessor staxXmlProcessor;
+    private final FidNameEnricher fidNameEnricher;
+    private final FlagsEnricher flagsEnricher;
 
-    private final int batchSize = 10000;
+    private final int batchSize;
 
     public SignalService(CreditTrackerSettings config,
                          HitsService hitsService,
@@ -66,7 +72,7 @@ public class SignalService {
                          PhoneService phoneService,
                          FlagsService flagsService,
                          PersonService personService,
-                         StaxXmlProcessor staxXmlProcessor) {
+                         StaxXmlProcessor staxXmlProcessor, FidNameEnricher fidNameEnricher, FlagsEnricher flagsEnricher) {
         this.config = config;
         this.hitsService = hitsService;
         this.userPackageService = userPackageService;
@@ -76,6 +82,36 @@ public class SignalService {
         this.flagsService = flagsService;
         this.personService = personService;
         this.staxXmlProcessor = staxXmlProcessor;
+        this.batchSize = config.getPipeline().subjectBatchSize();
+        this.fidNameEnricher = fidNameEnricher;
+        this.flagsEnricher = flagsEnricher;
+    }
+
+    public UpdateResult updateFlags2(UserPackage userPackage) {
+        log.debug("updating flags for package [{} {}]", userPackage.getPackageName(), userPackage.getId());
+        boolean updated = false;
+        boolean noFidsFound = true;
+        Long packageId = userPackage.getId();
+        int offset = 0;
+        List<PhoneFidMap> page = phoneService2.pageByPackageId(packageId, batchSize, offset);
+        int counter = 1;
+        while (!page.isEmpty()) {
+            log.debug("got page number {}, size {}", counter++, page.size());
+            List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
+                .mapId(phoneFidMap.getId())
+                .identifier(phoneFidMap.getSubjectId())
+                .phone(phoneFidMap.getPhone())
+                .build()).toList();
+            SubjectPayload payload = new SubjectPayload(packageId, userPackage.getUserId(), userPackage.getPackageName(), subjectProfiles);
+            log.debug("page number {}: updating fids...", counter);
+            fidNameEnricher.handle(payload);
+            log.debug("page number {}: updating flags...", counter);
+            flagsEnricher.process(payload, true);
+            offset += batchSize;
+            log.debug("loading next page...");
+            page = phoneService2.pageByPackageId(packageId, batchSize, offset);
+        }
+        return new UpdateResult(updated, noFidsFound);
     }
 
     /**

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

@@ -1,12 +1,12 @@
 package ru.nbch.credit_tracker.service.indic;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import java.time.LocalDate;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.entities.app.FlagDetail;
@@ -44,6 +44,28 @@ public class FlagsService2 {
             .collect(Collectors.toMap(PaymentsFlag::getFid, PaymentsFlag::getFlag));
     }
 
+    public void updateCalculationFlags(List<FlagDetail> flagDetails) {
+        try {
+            //fixme, временное решение
+            String jsonFlagDetail = entityToJsonMapper.writeValueAsString(flagDetails.stream()
+                .map(flagDetail -> new HashMap<String, Object>() {{
+                    put("fid", flagDetail.getFid());
+                    put("map_id", flagDetail.getMapId());
+                    put("flag_debt5000", flagDetail.getFlagDebt5000());
+                    put("flag_90_12", flagDetail.getFlag90days());
+                    put("calculated_at", flagDetail.getCalculatedAt());
+                }}).toList());
+
+            flagDetailMapper.updateCalcFlags(jsonFlagDetail);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void deleteFlags(List<Long> mapIds) {
+        flagDetailMapper.deleteFlags(mapIds);
+    }
+
     public void registerCalculationFlags(List<FlagDetail> flagDetails) {
         try {
             //fixme, временное решение
@@ -62,7 +84,6 @@ public class FlagsService2 {
         } catch (JsonProcessingException e) {
             throw new RuntimeException(e);
         }
-
     }
 
 }

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

@@ -51,6 +51,8 @@ public class FidNameEnricher implements Handler<SubjectPayload, SubjectPayload>
 
                 subject.setFid(name.getFid());
                 subject.setNameData(nameData);
+            } else {
+                subject.setFid(null);
             }
 
             return PhoneFidMap.builder()

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

@@ -14,7 +14,6 @@ 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.FlagDetail;
-import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.model.SubjectProfile;
 import ru.nbch.credit_tracker.model.impl.SubjectPayload;
 import ru.nbch.credit_tracker.service.indic.FlagsService2;
@@ -43,6 +42,10 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
     }
 
     public void process(SubjectPayload subjectPayload) {
+        process(subjectPayload, false);
+    }
+
+    public void process(SubjectPayload subjectPayload, boolean update) {
         List<SubjectProfile> subjects = subjectPayload.getSubjects();
         log.debug("FlagsEnricher handle by packageId: {}, size: {}", subjectPayload.getPackageId(), subjects.size());
 
@@ -84,7 +87,22 @@ public class FlagsEnricher implements Consumer<SubjectPayload> {
                 .calculatedAt(finalCalculatedAt)
                 .build();
         }).toList();
-        flagsService.registerCalculationFlags(flagDetails);
+        if (update) {
+            log.debug("updating calculated flags size {}", flagDetails.size());
+            flagsService.updateCalculationFlags(flagDetails);
+            //если signals_phone_fid_map#fid == null, удалить signals_flags с таким map_id
+            List<Long> phoneFidMapIdsWithNullFid = subjects.stream()
+                .filter(subject -> subject.getFid() == null)
+                .map(SubjectProfile::getMapId)
+                .filter(Objects::nonNull)
+                .toList();
+            if (!phoneFidMapIdsWithNullFid.isEmpty()) {
+                log.debug("deleting flags for signals_phone_fid_map#fid == null; amount: {}", phoneFidMapIdsWithNullFid.size());
+                flagsService.deleteFlags(phoneFidMapIdsWithNullFid);
+            }
+        } else {
+            flagsService.registerCalculationFlags(flagDetails);
+        }
     }
 
     @Override

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

@@ -28,6 +28,32 @@
                                                    calculated_at timestamp)
     </insert>
 
+    <insert id="updateCalcFlags" parameterType="string">
+        INSERT INTO signals_flags
+        (map_id, fid, flag_debt5000, flag_90_12, calculated_at)
+        SELECT map_id, fid, flag_debt5000, flag_90_12, calculated_at
+        FROM json_to_recordset(#{json}::json) AS t(
+                                                   map_id bigint,
+                                                   fid int,
+                                                   flag_debt5000 smallint,
+                                                   flag_90_12 smallint,
+                                                   calculated_at timestamp)
+        ON CONFLICT (map_id) DO UPDATE
+            SET
+                fid           = EXCLUDED.fid,
+                flag_debt5000 = EXCLUDED.flag_debt5000,
+                flag_90_12    = EXCLUDED.flag_90_12,
+                calculated_at = EXCLUDED.calculated_at;
+    </insert>
+
+    <delete id="deleteFlags" parameterType="java.util.List">
+        DELETE FROM signals_flags
+        WHERE map_id IN
+        <foreach collection="mapIds" item="mapId" open="(" separator="," close=")">
+            #{mapId}
+        </foreach>
+    </delete>
+
     <select id="getByMapId" resultMap="FlagDetail">
         SELECT id,
                map_id,

src/main/resources/sql/v2.21.ddl

@@ -21,4 +21,7 @@ CREATE TABLE signals_flags (
     FLAG_DEBT5000 smallint,
     FLAG_90_12 smallint,
     CALCULATED_AT timestamp
-);
+);
+
+ALTER TABLE signals_flags
+ADD CONSTRAINT signals_flags_map_id_key UNIQUE(map_id);