Setup logic for listener

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

@@ -1,25 +1,37 @@
 package ru.nbch.credit_tracker.service.mode.monitoring;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.SendResult;
+import ru.nbch.credit_tracker.entities.app.PhonesMap;
 import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.model.UserPackagePhoneData;
+import ru.nbch.credit_tracker.service.nbch.SourceService;
+import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
+import ru.nbch.credit_tracker.utils.CTUtil;
 
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
 
 @Slf4j
 //@Service
 public class KafkaMonitoringService {
 
     private final KafkaTemplate<String, String> kafkaTemplate;
+    private final PackageRecordsService packageRecordsService;
+    private final SourceService sourceService;
 
-//    @Value("${ct.tmp.dir}")
+    public KafkaMonitoringService(KafkaTemplate<String, String> kafkaTemplate,
-    private String tmpDir;
+                                  PackageRecordsService packageRecordsService,
-
+                                  SourceService sourceService) {
-    public KafkaMonitoringService(KafkaTemplate<String, String> kafkaTemplate) {
         this.kafkaTemplate = kafkaTemplate;
+        this.packageRecordsService = packageRecordsService;
+        this.sourceService = sourceService;
     }
 
 //    @KafkaListener(
@@ -35,7 +47,70 @@ public class KafkaMonitoringService {
     }
 
     private void process(List<PackageMessage> messages) {
+        List<Integer> fids = messages.stream().map(PackageMessage::getFid).collect(Collectors.toList());
+        List<String> memberCodes = messages.stream()
+                .map(message -> message.getMemberCode().substring(0,6))
+                .toList();
 
+        List<PhonesMap> phonesMaps = packageRecordsService.findCachedByFids(fids);
+
+        // key: fid; value: список флагов просрочек (у одного фида может быть несколько номеров)
+        Map<Integer, List<PhonesMap>> phonesDataMap = phonesMaps.stream()
+                .collect(Collectors.groupingBy(PhonesMap::getFid));
+
+        Set<Long> uniquePhones = phonesMaps.stream().map(PhonesMap::getPhone).collect(Collectors.toSet());
+
+        // прелоад по memberCode категорий
+        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
+
+        // subjectId, packageId, signalTypes, phone только для статуса ON_MONITORING
+        List<UserPackagePhoneData> packagePhoneData = packageRecordsService
+                .selectUserPackagePhoneData(uniquePhones);
+
+        // 1 номер может быть у нескольких пакетов
+        Map<Long, List<UserPackagePhoneData>> userPackagePhoneDataMap = packagePhoneData.stream()
+                .collect(Collectors.groupingBy(UserPackagePhoneData::getPhone));
+
+        for (PackageMessage message : messages) {
+            Integer fid = message.getFid();
+            String memberCode = message.getMemberCode();
+
+            String busCategory = busCategories.get(memberCode.substring(0, 6));
+            if (StringUtils.isBlank(busCategory)) {
+                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", memberCode.substring(0, 6));
+                continue;
+            }
+
+            Integer signalType = CTUtil.checkBusCategory(busCategory);
+
+            List<PhonesMap> phonesMapList = phonesDataMap.get(fid);
+
+            if (phonesMapList == null) {
+                continue;
+            }
+
+            for (PhonesMap phonesMap : phonesMapList) {
+                // флаги просрочек
+                Integer flagDebt5000 = phonesMap.getFlagDebt5000();
+                Integer flag90days = phonesMap.getFlag90days();
+
+                // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
+                List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());
+
+                if (userPackagePhoneDataList == null) {
+                    continue;
+                }
+
+                for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
+                    String subjectId = userPackagePhoneData.getSubjectId();
+                    String signalTypes = userPackagePhoneData.getSignalTypes();
+
+                    if (signalTypes.contains(signalType.toString())) {
+                        // TODO put to queue for process
+                    }
+                }
+            }
+        }
     }
 
     private void send(String id, String data) {

src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java

@@ -1,5 +1,10 @@
 package ru.nbch.credit_tracker.utils;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import ru.nbch.credit_tracker.entities.app.UserPackage;
+import ru.nbch.credit_tracker.entities.indic.Id;
+
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
@@ -8,10 +13,6 @@ import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import ru.nbch.credit_tracker.entities.app.UserPackage;
-import ru.nbch.credit_tracker.entities.indic.Id;
 
 @Slf4j
 public class CTUtil {
@@ -41,9 +42,6 @@ public class CTUtil {
     }
 
     public static Integer checkBusCategory(String busCategory) {
-        if (StringUtils.isBlank(busCategory)) {
-            return -1;
-        }
         switch (busCategory) {
             case "MFO", "MKK", "MFK" -> {
                 return 1;
@@ -52,7 +50,7 @@ public class CTUtil {
                 return 2;
             }
             default -> {
-                return -1;
+                return 3;
             }
         }
     }