Optimize loading fids and calculating flags

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

@@ -24,7 +24,6 @@ import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.enums.XmlErrorFormat;
 import ru.nbch.credit_tracker.exceptions.SignalClientException;
 import ru.nbch.credit_tracker.service.SignalService;
-import ru.nbch.credit_tracker.service.indic.PhoneService;
 import ru.nbch.credit_tracker.service.scoring.AuthorizationService;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 import ru.nbch.credit_tracker.service.xml.stax.StaxXmlProcessor;
@@ -162,22 +161,20 @@ public class SignalFacade {
         executor.execute(() -> {
             try {
                 signalService.registerSubjects(subjects, packageId);
-                List<PhoneService.FidsPhoneData> phoneFidsData = signalService.registerSubjectsFidAndDebts(packageId);
+                subjects.clear();
-                if (!phoneFidsData.isEmpty()) {
+                signalService.updateSubjects(packageId, null);
-                    Signals signals = signalService.buildSignals(packageId, phoneFidsData);
+                Signals signals = signalService.buildSignals(packageId);
-                    // TODO validate xsd?
+                signals.getConditions().getPersons().getPerson().addAll(signalService.buildPersons(packageId, null));
-                    if (signalRestService.update(signals)) {
+                if (signalRestService.update(signals)) {
-                        log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName);
+                    log.info("Package was successfully registered in signal service. PackageId: {} PackageName: {}", packageId, packageName);
-                        signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
+                    signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
-                    } else {
+                } else {
-                        log.info("Package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName);
+                    log.info("Package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName);
-                        signalService.updatePackageStatus(packageId, Status.ERROR);
+                    signalService.updatePackageStatus(packageId, Status.ERROR);
-                    }
                 }
             } catch (Exception e) {
                 signalService.updatePackageStatus(packageId, Status.ERROR);
                 log.error("Ошибка при поиске фидов или расчете флагов просрочек. Package_id {}", packageId, e);
-                // TODO очистить таблицу signals_phone_fid_map ?
             }
         });
     }

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

@@ -12,10 +12,11 @@ public interface PhoneFidMapMapper {
 
     void registerSubjects(@Param("subjects") List<Subject> subjects, @Param("packageId") Integer packageId);
 
-    List<PhoneFidMap> findPhoneFidMaps(@Param("packageId") Integer packageId);
+    List<PhoneFidMap> findPhoneFidMaps(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit);
 
-    List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Integer packageId);
+    List<PhoneFidMap> findFullPhoneFidMap(@Param("packageId") Integer packageId, @Param("offset") int offset, @Param("limit") int limit);
 
     void updatePhoneFidMaps(@Param("subjects") List<PhoneFidMap> subjects);
 
+    PhoneFidMap findPhoneFidMap(@Param("packageId") Integer packageId, @Param("subjectId") String subjectId);
 }

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

@@ -119,7 +119,6 @@ public class MonitoringService {
         MonitoringSignalResponse response = new MonitoringSignalResponse();
         response.setPackageName(userPackage.getPackageName()); // Имя пакета, переданного при установке
         Subjects subjects = new Subjects();
-        List<PhoneFidMap> phoneFidMapList = signalService.findFullSubjectsData(userPackage.getId());
         for (Person person : persons) {
             Signals signals = new Signals();
             for (Inquiry inquiry : person.getInquiries()) {
@@ -136,7 +135,8 @@ public class MonitoringService {
 
             // Если есть хотя бы один заполненный сигнал, значит заполняем остальные данные
             if (!signals.getSignals().isEmpty()) {
-                phoneFidMapList.stream().filter(p -> p.getSubjectId().equals(person.getUid())).findFirst().ifPresent(subjectData -> {
+                PhoneFidMap subjectData = signalService.findPhoneFidMapBySubjectId(userPackage.getId(), person.getUid());
+                if(subjectData != null) {
                     Subject subject = new Subject();
                     subject.setId(subjectData.getSubjectId()); // Уникальный идентификатор субъекта (равен Id из XML-пакета при установке мониторинга)
                     subject.setClientId(subjectData.getPhone());
@@ -144,7 +144,7 @@ public class MonitoringService {
                     subject.setFlag9012(subjectData.getFlag90days());
                     subject.setSignals(signals);
                     subjects.getSubjects().add(subject);
-                });
+                }
             }
         }
         response.setSubjects(subjects);

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

@@ -21,10 +21,8 @@ import ru.nbch.credit_tracker.service.signal.PhoneFidMapService;
 import ru.nbch.credit_tracker.service.signal.UserPackageService;
 
 import java.time.LocalDateTime;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.List;
+import java.util.function.Predicate;
-import java.util.Map;
-import java.util.Objects;
 
 @Slf4j
 @Service
@@ -80,56 +78,78 @@ public class SignalService {
         log.info("Subjects were registered. PackageId: {}", packageId);
     }
 
-    public List<PhoneService.FidsPhoneData> registerSubjectsFidAndDebts(Integer packageId) {
+    public boolean updateSubjects(Integer packageId, Predicate<PhoneFidMap> filter) {
-        List<PhoneFidMap> subjects = phoneFidMapService.findPhoneFidMaps(packageId);
+        boolean updated = false;
-        return updateFlags(subjects, packageId);
+        int offset = 0;
+        List<PhoneFidMap> batch;
+        do {
+            batch = phoneFidMapService.findPhoneFidMaps(packageId, offset, batchSize);
+
+            List<PhoneFidMap> filteredItems;
+            if (filter != null) {
+                filteredItems = batch.stream().filter(filter).toList();
+                updateFlags(batch.stream().filter(filter).toList(), packageId);
+            } else {
+                filteredItems = batch;
+            }
+
+            if (!updated && !filteredItems.isEmpty()) {
+                updated = true;
+            }
+
+            updateFlags(filteredItems, packageId);
+
+            offset += batchSize;
+        } while (!batch.isEmpty());
+
+        return updated;
     }
 
-    public List<PhoneService.FidsPhoneData> updateFlags(List<PhoneFidMap> subjects, Integer packageId) {
+    public void updateFlags(List<PhoneFidMap> subjects, Integer packageId) {
-        log.info("Start calculating fids and flags for subjects. PackageId: {}", packageId);
+        if (subjects.isEmpty()) {
+            return;
+        }
+        log.info("Start calculating fids and flags for subjects batch. PackageId: {}", packageId);
         List<PhoneService.FidsPhoneData> fidsPhoneDataList = phoneService.findFidsPhoneData(subjects);
         LocalDateTime calculatedAt = LocalDateTime.now();
-        for (int i = 0; i < fidsPhoneDataList.size(); i += batchSize) {
-            int end = Math.min(i + batchSize, fidsPhoneDataList.size());
-            List<PhoneService.FidsPhoneData> currentList = fidsPhoneDataList.subList(i, end);
-            Map<Integer, FlagsService.FlagsData> flagsDataMap = flagsService.findFlags(currentList.stream().map(PhoneService.FidsPhoneData::getFid).filter(Objects::nonNull).toList());
-            List<PhoneFidMap> subjectToUpdate = new ArrayList<>();
 
-            currentList.forEach(data -> {
+        Map<Integer, FlagsService.FlagsData> flagsDataMap = flagsService.findFlags(fidsPhoneDataList.stream().map(PhoneService.FidsPhoneData::getFid).filter(Objects::nonNull).toList());
-                PhoneFidMap subject;
+        List<PhoneFidMap> subjectToUpdate = new ArrayList<>();
-                if (data.getFid() != null) {
-                    FlagsService.FlagsData flagsData = flagsDataMap.get(data.getFid());
-                    subject = PhoneFidMap.builder()
-                            .id(data.getId())
-                            .packageId(packageId)
-                            .subjectId(data.getSubjectId())
-                            .phone(data.getPhone())
-                            .fid(data.getFid())
-                            .flagDebt5000(flagsData.getFlagDebt5000())
-                            .flag90days(flagsData.getFlag90days())
-                            .calculatedAt(calculatedAt)
-                            .build();
-                } else {
-                    subject = PhoneFidMap.builder()
-                            .id(data.getId())
-                            .packageId(packageId)
-                            .subjectId(data.getSubjectId())
-                            .phone(data.getPhone())
-                            .calculatedAt(calculatedAt)
-                            .build();
-                }
-                subjectToUpdate.add(subject);
-            });
 
-            int updateBatchSize = batchSize / 8; // 8 параметров на одну запись для обновления. Максимальное количество параметров ~65000
+        fidsPhoneDataList.forEach(data -> {
-            for (int j = 0; j < subjectToUpdate.size(); j += updateBatchSize) {
+            PhoneFidMap subject;
-                int endIndex = Math.min(j + updateBatchSize, subjectToUpdate.size());
+            if (data.getFid() != null) {
-                phoneFidMapService.updatePhoneFidMaps(subjectToUpdate.subList(j, endIndex));
+                FlagsService.FlagsData flagsData = flagsDataMap.get(data.getFid());
-                log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", i, end, subjectToUpdate.size(), packageId);
+                subject = PhoneFidMap.builder()
+                        .id(data.getId())
+                        .packageId(packageId)
+                        .subjectId(data.getSubjectId())
+                        .phone(data.getPhone())
+                        .fid(data.getFid())
+                        .flagDebt5000(flagsData.getFlagDebt5000())
+                        .flag90days(flagsData.getFlag90days())
+                        .calculatedAt(calculatedAt)
+                        .build();
+            } else {
+                subject = PhoneFidMap.builder()
+                        .id(data.getId())
+                        .packageId(packageId)
+                        .subjectId(data.getSubjectId())
+                        .phone(data.getPhone())
+                        .calculatedAt(calculatedAt)
+                        .build();
             }
+            subjectToUpdate.add(subject);
+        });
+
+        int updateBatchSize = batchSize / 8; // 8 параметров на одну запись для обновления. Максимальное количество параметров ~65000
+        for (int j = 0; j < subjectToUpdate.size(); j += updateBatchSize) {
+            int endIndex = Math.min(j + updateBatchSize, subjectToUpdate.size());
+            phoneFidMapService.updatePhoneFidMaps(subjectToUpdate.subList(j, endIndex));
+            log.trace("Updated fids and flags for subject's batch from: {} to: {}. Total count: {}. PackageId: {}", j, endIndex, subjectToUpdate.size(), packageId);
         }
-        log.info("Finish calculating fids and flags for subjects. PackageId: {}", packageId);
+
-        return fidsPhoneDataList;
+        log.info("Finish calculating fids and flags for subjects. PackageId: {}. Batch size: {}", packageId, subjects.size());
     }
 
     public void rejectPackage(MonitoringRequest request) {
@@ -146,7 +166,7 @@ public class SignalService {
         userPackageService.updatePackageStatus(packageId, status);
     }
 
-    public Signals buildSignals(Integer packageId, List<PhoneService.FidsPhoneData> fids) {
+    public Signals buildSignals(Integer packageId) {
         UserPackage userPackage = userPackageService.findPackage(packageId);
         Signals signals = new Signals();
         signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы)
@@ -154,23 +174,40 @@ public class SignalService {
         Auth auth = new Auth();
         auth.setUser(config.getSignalsuser());
         signals.setAuth(auth);
-        for (int i = 0; i < fids.size(); i += batchSize) {
-            int end = Math.min(i + batchSize, fids.size());
-            List<Person> persons = findPersons(fids.subList(i, end));
-            if (!persons.isEmpty()) {
-                signals.getConditions().getPersons().getPerson().addAll(persons);
-            }
-        }
 
         return signals;
     }
 
-    private List<Person> findPersons(List<PhoneService.FidsPhoneData> fidsPhoneData) {
+    public List<Person> buildPersons(Integer packageId, Predicate<PhoneFidMap> filter) {
+        List<Person> result = new ArrayList<>();
+        int offset = 0;
+        List<PhoneFidMap> batch;
+        do {
+            batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize);
+            List<Person> persons;
+            if (filter != null) {
+                persons = findPersons(batch.stream().filter(filter).toList());
+            } else {
+                persons = findPersons(batch);
+            }
+            if (!persons.isEmpty()) {
+                result.addAll(persons);
+            }
+            offset += batchSize;
+        } while (!batch.isEmpty());
+
+        return result;
+    }
+
+    private List<Person> findPersons(List<PhoneFidMap> subjects) {
+        if (subjects.isEmpty()) {
+            return Collections.emptyList();
+        }
         List<Person> persons = new ArrayList<>();
-        Map<Integer, PersonService.PersonData> personFidMap = personService.findPersonData(fidsPhoneData.stream()
+        Map<Integer, PersonService.PersonData> personFidMap = personService.findPersonData(subjects.stream()
-                .map(PhoneService.FidsPhoneData::getFid)
+                .map(PhoneFidMap::getFid)
                 .filter(Objects::nonNull).toList());
-        fidsPhoneData.forEach(data -> {
+        subjects.forEach(data -> {
             if (data.getFid() != null) {
                 PersonService.PersonData personData = personFidMap.get(data.getFid());
                 Person person = PersonFactoryBuilder.builder()
@@ -201,10 +238,6 @@ public class SignalService {
         return userPackageService.findAllPackagesByStatus(Status.ON_MONITORING);
     }
 
-    public List<PhoneFidMap> findFullSubjectsData(Integer packageId) {
-        return phoneFidMapService.findFullPhoneFidMap(packageId);
-    }
-
     public void insertHits(List<Hit> hits) {
         hitsService.insertHits(hits);
     }
@@ -212,4 +245,8 @@ public class SignalService {
     public void updateLastDownloadedReport(Integer id, String fileName) {
         userPackageService.updatePackageLastDownloadedReport(id, fileName);
     }
+
+    public PhoneFidMap findPhoneFidMapBySubjectId(Integer packageId, String subjectId) {
+        return phoneFidMapService.findPhoneFidMap(packageId, subjectId);
+    }
 }

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

@@ -4,11 +4,9 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.stereotype.Service;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
-import ru.nbch.credit_tracker.entities.dto.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.dto.UserPackage;
 import ru.nbch.credit_tracker.entities.signal_package.Signals;
 import ru.nbch.credit_tracker.enums.Status;
-import ru.nbch.credit_tracker.service.indic.PhoneService;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
 
 import java.time.LocalDateTime;
@@ -46,19 +44,18 @@ public class UpdatingFlagsService {
         List<UserPackage> packages = signalService.findPackagesOnMonitoring();
         for (UserPackage userPackage : packages) {
             log.info("Updating flags for Package {}", userPackage.getPackageName());
-            List<PhoneFidMap> subjects = signalService.findFullSubjectsData(userPackage.getId())
+            updateFlags(userPackage.getId(), userPackage.getPackageName());
-                    .stream().filter(subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt())).toList();
-            updateFlags(userPackage.getId(), subjects, userPackage.getPackageName());
         }
         log.info("Finish updating flags");
     }
 
-    private void updateFlags(Integer packageId, List<PhoneFidMap> subjects, String packageName) {
+    private void updateFlags(Integer packageId, String packageName) {
         try {
             signalService.updatePackageStatus(packageId, Status.PROCESSING);
-            List<PhoneService.FidsPhoneData> phoneFidsData = signalService.updateFlags(subjects, packageId);
+            boolean updated = signalService.updateSubjects(packageId, subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt()));
-            if (!phoneFidsData.isEmpty()) {
+            if (updated) {
-                Signals signals = signalService.buildSignals(packageId, phoneFidsData);
+                Signals signals = signalService.buildSignals(packageId);
+                signals.getConditions().getPersons().getPerson().addAll(signalService.buildPersons(packageId, subject -> subject.getCalculatedAt() != null && hasDaysPassed(subject.getCalculatedAt())));
                 if (signalRestService.update(signals)) {
                     log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName);
                     signalService.updatePackageStatus(packageId, Status.ON_MONITORING);

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

@@ -22,25 +22,21 @@ public class PhoneService {
 
     public List<FidsPhoneData> findFidsPhoneData(List<PhoneFidMap> subjects) {
         List<FidsPhoneData> fidsPhoneDataList = new ArrayList<>();
-        int batchSize = 10000;
-        for (int i = 0; i < subjects.size(); i += batchSize) {
-            int end = Math.min(i + batchSize, subjects.size());
-            List<PhoneFidMap> currentSubjects = subjects.subList(i, end);
-            List<String> phones = currentSubjects.stream().map(subject -> "9" + subject.getPhone()).toList();
-            List<FidsData> fidsData = mapper.findFids(phones);
 
-            Map<String, Integer> fidsMap = convertFidsDataToMap(fidsData);
+        List<String> phones = subjects.stream().map(subject -> "9" + subject.getPhone()).toList();
+        List<FidsData> fidsData = mapper.findFids(phones);
+        Map<String, Integer> fidsMap = convertFidsDataToMap(fidsData);
+
+        subjects.forEach(subject -> {
+            String phone = "9" + subject.getPhone();
+            Integer fid = fidsMap.get(phone);
+            if (fid != null) {
+                fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), fid, subject.getPhone(), subject.getSubjectId()));
+            } else {
+                fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), null, subject.getPhone(), subject.getSubjectId()));
+            }
+        });
 
-            currentSubjects.forEach(subject -> {
-                String phone = "9" + subject.getPhone();
-                Integer fid = fidsMap.get(phone);
-                if (fid != null) {
-                    fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), fid, subject.getPhone(), subject.getSubjectId()));
-                } else {
-                    fidsPhoneDataList.add(new FidsPhoneData(subject.getId(), null, subject.getPhone(), subject.getSubjectId()));
-                }
-            });
-        }
         return fidsPhoneDataList;
     }
 

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

@@ -20,15 +20,19 @@ public class PhoneFidMapService {
         mapper.registerSubjects(subjects, packageId);
     }
 
-    public List<PhoneFidMap> findPhoneFidMaps(Integer packageId) {
+    public List<PhoneFidMap> findPhoneFidMaps(Integer packageId,  int offset,  int limit) {
-        return mapper.findPhoneFidMaps(packageId);
+        return mapper.findPhoneFidMaps(packageId, offset, limit);
     }
 
-    public List<PhoneFidMap> findFullPhoneFidMap(Integer packageId) {
+    public List<PhoneFidMap> findFullPhoneFidMap(Integer packageId,  int offset,  int limit) {
-        return mapper.findFullPhoneFidMap(packageId);
+        return mapper.findFullPhoneFidMap(packageId, offset, limit);
     }
 
     public void updatePhoneFidMaps(List<PhoneFidMap> subjects) {
         mapper.updatePhoneFidMaps(subjects);
     }
+
+    public PhoneFidMap findPhoneFidMap(Integer packageId, String subjectId) {
+        return mapper.findPhoneFidMap(packageId, subjectId);
+    }
 }

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

@@ -44,6 +44,8 @@
         SELECT id, subject_id, phone
         FROM signals_phone_fid_map
         WHERE package_id = #{packageId}
+        ORDER BY id
+        LIMIT #{limit} OFFSET #{offset}
     </select>
 
     <select id="findFullPhoneFidMap" resultMap="PhoneFidMapResultMap" resultType="java.util.List">
@@ -57,6 +59,22 @@
                calculated_at
         FROM signals_phone_fid_map
         WHERE package_id = #{packageId}
+        ORDER BY id
+        LIMIT #{limit} OFFSET #{offset}
+    </select>
+
+    <select id="findPhoneFidMap" resultMap="PhoneFidMapResultMap" resultType="ru.nbch.credit_tracker.entities.dto.PhoneFidMap">
+        SELECT id,
+               package_id,
+               subject_id,
+               phone,
+               fid,
+               flag_90_12,
+               flag_debt5000,
+               calculated_at
+        FROM signals_phone_fid_map
+        WHERE package_id = #{packageId} AND subject_id = #{subjectId}
+        LIMIT 1
     </select>