Refactor Kafka monitoring processing

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

@@ -0,0 +1,94 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.service.signal.HitsService;
+import ru.nbch.credit_tracker.service.signal.SegmentMapService;
+import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+public class KafkaMonitoringBatchProcessor {
+
+    private final SegmentMapService segmentMapService;
+    private final HitsService hitsService;
+    private final KafkaMonitoringMessageFilter messageFilter;
+    private final NaturalPersonKafkaSignalProcessor naturalPersonSignalProcessor;
+    private final LegalPersonKafkaSignalProcessor legalPersonSignalProcessor;
+
+    public KafkaMonitoringBatchProcessor(SegmentMapService segmentMapService,
+                                         HitsService hitsService,
+                                         KafkaMonitoringMessageFilter messageFilter,
+                                         NaturalPersonKafkaSignalProcessor naturalPersonSignalProcessor,
+                                         LegalPersonKafkaSignalProcessor legalPersonSignalProcessor) {
+        this.segmentMapService = segmentMapService;
+        this.hitsService = hitsService;
+        this.messageFilter = messageFilter;
+        this.naturalPersonSignalProcessor = naturalPersonSignalProcessor;
+        this.legalPersonSignalProcessor = legalPersonSignalProcessor;
+    }
+
+    public void process(List<PackageMessage> messages,
+                        ExecutorService processingExecutor,
+                        ExecutorService dbExecutor) {
+        Map<Boolean, List<PackageMessage>> partition = messages.stream()
+                .filter(Objects::nonNull)
+                .filter(messageFilter::isSupported)
+                .collect(Collectors.partitioningBy(messageFilter::isNaturalPerson));
+
+        List<PackageMessage> npMessages = partition.get(true);
+        List<PackageMessage> lpMessages = partition.get(false);
+
+        Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
+
+        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
+            if (npMessages != null && !npMessages.isEmpty()) {
+                return naturalPersonSignalProcessor.process(npMessages, segmentMap);
+            }
+            return Collections.emptyList();
+        }, processingExecutor);
+
+        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
+            if (lpMessages != null && !lpMessages.isEmpty()) {
+                return legalPersonSignalProcessor.process(lpMessages, segmentMap);
+            }
+            return Collections.emptyList();
+        }, processingExecutor);
+
+        CompletableFuture<List<Hit>> combinedFuture = npFuture
+                .thenCombine(lpFuture, (npHits, lpHits) -> {
+                    List<Hit> allHits = new ArrayList<>();
+                    if (npHits != null) allHits.addAll(npHits);
+                    if (lpHits != null) allHits.addAll(lpHits);
+                    return allHits;
+                });
+
+        combinedFuture.thenAcceptAsync(hits -> {
+            if (!hits.isEmpty()) {
+                try {
+                    hitsService.insertHits(hits);
+                } catch (Exception e) {
+                    log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
+                }
+            }
+        }, dbExecutor);
+
+        try {
+            combinedFuture.get(1, TimeUnit.MINUTES);
+        } catch (Exception e) {
+            log.error("Error processing messages", e);
+        }
+    }
+}

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

@@ -0,0 +1,19 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+
+@Component
+public class KafkaMonitoringMessageFilter {
+
+    public boolean isSupported(PackageMessage message) {
+        Integer inquiryType = message.getInquiryType();
+        String billingCode = message.getBillingCode();
+        return inquiryType == 1 || inquiryType == 2 ||
+                (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
+    }
+
+    public boolean isNaturalPerson(PackageMessage message) {
+        return message.getPersons() != null && !message.getPersons().isEmpty();
+    }
+}

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

@@ -2,8 +2,6 @@ package ru.nbch.credit_tracker.service.mode.monitoring;
 
 import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.tuple.Pair;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.kafka.annotation.KafkaListener;
@@ -11,25 +9,12 @@ import org.springframework.kafka.support.Acknowledgment;
 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.app.Hit;
-import ru.nbch.credit_tracker.entities.app.LegalsMap;
-import ru.nbch.credit_tracker.entities.app.PhonesMap;
 import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
 import ru.nbch.credit_tracker.enums.EventType;
 import ru.nbch.credit_tracker.model.SignalKafkaEvent;
-import ru.nbch.credit_tracker.model.UserPackageLegalData;
-import ru.nbch.credit_tracker.model.UserPackagePhoneData;
-import ru.nbch.credit_tracker.service.nbch.SourceService;
-import ru.nbch.credit_tracker.service.signal.HitsService;
-import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
-import ru.nbch.credit_tracker.service.signal.SegmentMapService;
-import ru.nbch.credit_tracker.utils.CTUtil;
-import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
 
-import java.time.LocalDateTime;
-import java.util.*;
 import java.util.concurrent.*;
-import java.util.stream.Collectors;
+import java.util.List;
 
 @Slf4j
 @Service
@@ -45,26 +30,17 @@ public class KafkaMonitoringService {
 
     private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
 
-    private final PackageRecordsService packageRecordsService;
-    private final SourceService sourceService;
-    private final SegmentMapService segmentMapService;
-    private final HitsService hitsService;
+    private final KafkaMonitoringBatchProcessor batchProcessor;
     private final ReportNameRegistry reportNameRegistry;
 
     public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
-                                  PackageRecordsService packageRecordsService,
-                                  SourceService sourceService,
-                                  SegmentMapService segmentMapService,
                                   CreditTrackerSettings settings,
-                                  HitsService hitsService,
+                                  KafkaMonitoringBatchProcessor batchProcessor,
                                   ReportNameRegistry reportNameRegistry) {
         this.signalEventsQueue = signalEventsQueue;
-        this.packageRecordsService = packageRecordsService;
-        this.sourceService = sourceService;
-        this.segmentMapService = segmentMapService;
+        this.batchProcessor = batchProcessor;
         this.reportNameRegistry = reportNameRegistry;
         this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
-        this.hitsService = hitsService;
         this.processingPermits = new Semaphore(permits);
 
         this.processingExecutor = Executors.newFixedThreadPool(
@@ -96,329 +72,13 @@ public class KafkaMonitoringService {
         try {
             acknowledgment.acknowledge();
             long startTime = System.currentTimeMillis();
-            process(records);
+            batchProcessor.process(records, processingExecutor, dbExecutor);
             log.info("Processed {} messages from Kafka. Time {} ms", records.size(), System.currentTimeMillis() - startTime);
         } finally {
             processingPermits.release();
         }
     }
 
-    private void process(List<PackageMessage> messages) {
-        Map<Boolean, List<PackageMessage>> partition = messages.stream()
-                .filter(Objects::nonNull)
-                .filter(this::filterMessages)
-                .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));
-
-        List<PackageMessage> npMessages = partition.get(true);
-        List<PackageMessage> lpMessages = partition.get(false);
-
-        Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
-
-        CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
-            if (npMessages != null && !npMessages.isEmpty()) {
-                return processNpMessages(npMessages, segmentMap);
-            }
-            return Collections.emptyList();
-        }, processingExecutor);
-
-        CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
-            if (lpMessages != null && !lpMessages.isEmpty()) {
-                return processLpMessages(lpMessages, segmentMap);
-            }
-            return Collections.emptyList();
-        }, processingExecutor);
-
-        CompletableFuture<List<Hit>> combinedFuture = npFuture
-                .thenCombine(lpFuture, (npHits, lpHits) -> {
-                    List<Hit> allHits = new ArrayList<>();
-                    if (npHits != null) allHits.addAll(npHits);
-                    if (lpHits != null) allHits.addAll(lpHits);
-                    return allHits;
-                });
-
-        combinedFuture.thenAcceptAsync(hits -> {
-            if (!hits.isEmpty()) {
-                try {
-                    hitsService.insertHits(hits);
-                } catch (Exception e) {
-                    log.error("Failed to save hits. {}", ExceptionUtils.getStackTrace(e));
-                }
-            }
-        }, dbExecutor);
-
-        try {
-            combinedFuture.get(1, TimeUnit.MINUTES);
-        } catch (Exception e) {
-            log.error("Error processing messages", e);
-        }
-    }
-
-    private List<Hit> processNpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
-        List<Hit> hits = new ArrayList<>();
-
-        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;
-            }
-
-            int signalType = CTUtil.checkBusCategory(busCategory);
-
-            List<PhonesMap> phonesMapList = phonesDataMap.get(fid);
-
-            if (phonesMapList == null) {
-                continue;
-            }
-
-            for (PhonesMap phonesMap : phonesMapList) {
-                // в userPackagePhoneDataMap только нужные номера, те что не вошли в userPackagePhoneDataList игнорируются
-                List<UserPackagePhoneData> userPackagePhoneDataList = userPackagePhoneDataMap.get(phonesMap.getPhone());
-
-                if (userPackagePhoneDataList == null) {
-                    continue;
-                }
-
-                for (UserPackagePhoneData userPackagePhoneData : userPackagePhoneDataList) {
-                    String signalTypes = userPackagePhoneData.getSignalTypes();
-
-                    if (signalTypes.contains(Integer.toString(signalType))) {
-                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
-                        SignalKafkaEvent event = buildSignalKafkaEvent(
-                                true, message,
-                                phonesMap, userPackagePhoneData,
-                                null, null, // legal person fields
-                                signalType, busCategory, segment
-                        );
-
-                        String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
-                                + "_" + userPackagePhoneData.getPackageName();
-                        String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
-                        if (reportFileName == null) {
-                            continue; // событие не встало в очередь — hit не пишем
-                        }
-                        Hit hit = buildHit(
-                                true, message,
-                                userPackagePhoneData,
-                                null,
-                                signalType, busCategory,
-                                segment, reportFileName
-                        );
-                        hits.add(hit);
-                    }
-                }
-            }
-        }
-
-        return hits;
-    }
-
-    private List<Hit> processLpMessages(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
-        List<Hit> hits = new ArrayList<>();
-
-        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<LegalsMap> legalsMaps = packageRecordsService.findCachedLegalsMapByFids(fids);
-
-        Map<Integer, List<LegalsMap>> legalsDataMap = legalsMaps.stream()
-                .collect(Collectors.groupingBy(LegalsMap::getFid));
-
-        Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
-                .map(l -> Pair.of(l.getInn(), l.getRegNum()))
-                .collect(Collectors.toSet());
-
-        // прелоад по memberCode категорий
-        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
-
-        // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING
-        List<UserPackageLegalData> packageLegalData = packageRecordsService
-                .selectUserPackageLegalData(uniqueInnRegnumPairs);
-
-        Map<Pair<Long, Long>, List<UserPackageLegalData>> userPackageLegalDataMap = packageLegalData.stream()
-                .collect(Collectors.groupingBy(p -> Pair.of(p.getInn(), p.getRegNum())));
-
-        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;
-            }
-
-            int signalType = CTUtil.checkBusCategory(busCategory);
-
-            List<LegalsMap> legalsMapList = legalsDataMap.get(fid);
-
-            if (legalsMapList == null) {
-                continue;
-            }
-
-            for (LegalsMap legalsMap : legalsMapList) {
-                // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
-                List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()));
-
-                if (userPackageLegalDataList == null) {
-                    continue;
-                }
-
-                for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) {
-                    String signalTypes = userPackageLegalData.getSignalTypes();
-
-                    if (signalTypes.contains(Integer.toString(signalType))) {
-                        Integer segment = segmentMap.get(memberCode.substring(0, 6));
-
-                        SignalKafkaEvent event = buildSignalKafkaEvent(
-                                false, message,
-                                null, null, // np fields
-                                legalsMap,
-                                userPackageLegalData,
-                                signalType, busCategory, segment
-                        );
-
-                        String reportKey = userPackageLegalData.getUserId().substring(0, 6)
-                                + "_" + userPackageLegalData.getPackageName();
-                        String reportFileName = reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
-                        if (reportFileName == null) {
-                            continue; // событие не встало в очередь — hit не пишем
-                        }
-                        Hit hit = buildHit(
-                                false, message,
-                                null,
-                                userPackageLegalData,
-                                signalType, busCategory,
-                                segment, reportFileName
-                        );
-                        hits.add(hit);
-                    }
-                }
-            }
-        }
-
-        return hits;
-    }
-
-    private boolean filterMessages(PackageMessage message) {
-        Integer inquiryType = message.getInquiryType();
-        String billingCode = message.getBillingCode();
-        return inquiryType == 1 || inquiryType == 2 ||
-                (inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
-    }
-
-    private SignalKafkaEvent buildSignalKafkaEvent(boolean isNatural, PackageMessage message,
-                                                   PhonesMap phonesMap, UserPackagePhoneData userPackagePhoneData,
-                                                   LegalsMap legalsMap, UserPackageLegalData userPackageLegalData,
-                                                   int signalType, String busCategory, Integer segment) {
-        if (isNatural) {
-            // флаги просрочек
-            Integer flagDebt5000 = phonesMap.getFlagDebt5000();
-            Integer flag90days = phonesMap.getFlag90days();
-
-            String subjectId = userPackagePhoneData.getSubjectId();
-
-            return SignalKafkaEvent.builder()
-                    .eventType(EventType.WRITE)
-                    .natural(true)
-                    .userPackagePhoneData(userPackagePhoneData)
-                    .packageMessage(message)
-                    .subjectId(subjectId)
-                    .flagDebt5000(flagDebt5000)
-                    .flag90days(flag90days)
-                    .signalType(Integer.toString(signalType))
-                    .busCategory(signalType == 3 ? busCategory : null)
-                    .segment(segment != null ? segment : -1)
-                    .build();
-        } else {
-            // флаги просрочек
-            Integer flag90days = legalsMap.getFlag90days();
-            LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
-
-            String subjectId = userPackageLegalData.getSubjectId();
-            return SignalKafkaEvent.builder()
-                    .eventType(EventType.WRITE)
-                    .natural(false)
-                    .userPackageLegalData(userPackageLegalData)
-                    .packageMessage(message)
-                    .subjectId(subjectId)
-                    .flag90days(flag90days)
-                    .signalType(Integer.toString(signalType))
-                    .busCategory(signalType == 3 ? busCategory : null)
-                    .lastRejectedDate(lastRejectedDate)
-                    .segment(segment != null ? segment : -1)
-                    .build();
-        }
-    }
-
-    private Hit buildHit(boolean isNatural, PackageMessage message,
-                         UserPackagePhoneData userPackagePhoneData,
-                         UserPackageLegalData userPackageLegalData,
-                         int signalType, String busCategory,
-                         Integer segment, String reportFileName) {
-        if (isNatural) {
-            return Hit.builder()
-                    .packageId(userPackagePhoneData.getPackageId())
-                    .subjectId(userPackagePhoneData.getSubjectId())
-                    .signalType(signalType)
-                    .reportFileName(reportFileName.replace(".xml", ".zip"))
-                    .inqT(message.getInquiryType())
-                    .busCategory(busCategory)
-                    .inqDt(message.getInquiryDate())
-                    .inqP(message.getInquiryPurpose())
-                    .inqA(message.getInquiryAmount())
-                    .segment(segment != null ? segment : -1)
-                    .recordedAt(LocalDateTime.now())
-                    .phone(userPackagePhoneData.getPhone())
-                    .build();
-        } else {
-           return Hit.builder()
-                    .packageId(userPackageLegalData.getPackageId())
-                    .subjectId(userPackageLegalData.getSubjectId())
-                    .signalType(signalType)
-                    .reportFileName(reportFileName.replace(".xml", ".zip"))
-                    .inqT(message.getInquiryType())
-                    .busCategory(busCategory)
-                    .inqDt(message.getInquiryDate())
-                    .inqP(message.getInquiryPurpose())
-                    .inqA(message.getInquiryAmount())
-                    .segment(segment != null ? segment : -1)
-                    .recordedAt(LocalDateTime.now())
-                    .inn(userPackageLegalData.getInn())
-                    .regNum(userPackageLegalData.getRegNum())
-                    .build();
-        }
-    }
-
     @PreDestroy
     public void stopKafkaMonitoringService() throws InterruptedException {
         log.info("Stopping KafkaMonitoringService...");

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

@@ -0,0 +1,146 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+import ru.nbch.credit_tracker.entities.app.LegalsMap;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.enums.EventType;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+import ru.nbch.credit_tracker.model.UserPackageLegalData;
+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.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+public class LegalPersonKafkaSignalProcessor {
+
+    private final PackageRecordsService packageRecordsService;
+    private final SourceService sourceService;
+    private final SignalKafkaEventPublisher eventPublisher;
+
+    public LegalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService,
+                                           SourceService sourceService,
+                                           SignalKafkaEventPublisher eventPublisher) {
+        this.packageRecordsService = packageRecordsService;
+        this.sourceService = sourceService;
+        this.eventPublisher = eventPublisher;
+    }
+
+    public List<Hit> process(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
+        List<Hit> hits = new ArrayList<>();
+
+        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<LegalsMap> legalsMaps = packageRecordsService.findCachedLegalsMapByFids(fids);
+
+        Map<Integer, List<LegalsMap>> legalsDataMap = legalsMaps.stream()
+                .collect(Collectors.groupingBy(LegalsMap::getFid));
+
+        Set<Pair<Long, Long>> uniqueInnRegnumPairs = legalsMaps.stream()
+                .map(l -> Pair.of(l.getInn(), l.getRegNum()))
+                .collect(Collectors.toSet());
+
+        // прелоад по memberCode категорий
+        Map<String, String> busCategories = sourceService.findBusCategories(memberCodes);
+
+        // subjectId, packageId, signalTypes, inn, regNum только для статуса ON_MONITORING
+        List<UserPackageLegalData> packageLegalData = packageRecordsService
+                .selectUserPackageLegalData(uniqueInnRegnumPairs);
+
+        Map<Pair<Long, Long>, List<UserPackageLegalData>> userPackageLegalDataMap = packageLegalData.stream()
+                .collect(Collectors.groupingBy(p -> Pair.of(p.getInn(), p.getRegNum())));
+
+        for (PackageMessage message : messages) {
+            Integer fid = message.getFid();
+            String memberCode = message.getMemberCode();
+            String commonMemberCode = memberCode.substring(0, 6);
+
+            String busCategory = busCategories.get(commonMemberCode);
+            if (StringUtils.isBlank(busCategory)) {
+                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode);
+                continue;
+            }
+
+            int signalType = CTUtil.checkBusCategory(busCategory);
+
+            List<LegalsMap> legalsMapList = legalsDataMap.get(fid);
+
+            if (legalsMapList == null) {
+                continue;
+            }
+
+            for (LegalsMap legalsMap : legalsMapList) {
+                // флаги просрочек
+                Integer flag90days = legalsMap.getFlag90days();
+                LocalDateTime lastRejectedDate = legalsMap.getLastRejectedDate();
+
+                // в userPackageLegalDataMap только нужные inn, regnum, те что не вошли в userPackageLegalDataList игнорируются
+                List<UserPackageLegalData> userPackageLegalDataList = userPackageLegalDataMap.get(Pair.of(legalsMap.getInn(), legalsMap.getRegNum()));
+
+                if (userPackageLegalDataList == null) {
+                    continue;
+                }
+
+                for (UserPackageLegalData userPackageLegalData : userPackageLegalDataList) {
+                    String subjectId = userPackageLegalData.getSubjectId();
+                    String signalTypes = userPackageLegalData.getSignalTypes();
+
+                    if (signalTypes.contains(Integer.toString(signalType))) {
+                        SignalKafkaEvent event = new SignalKafkaEvent();
+                        event.setEventType(EventType.WRITE);
+                        event.setNatural(false);
+                        event.setUserPackageLegalData(userPackageLegalData);
+                        event.setPackageMessage(message);
+                        event.setSubjectId(subjectId);
+                        event.setFlag90days(flag90days);
+                        event.setSignalType(Integer.toString(signalType));
+                        event.setBusCategory(signalType == 3 ? busCategory : null);
+                        event.setLastRejectedDate(lastRejectedDate);
+
+                        Integer segment = segmentMap.get(commonMemberCode);
+                        event.setSegment(segment != null ? segment : -1);
+
+                        String reportKey = userPackageLegalData.getUserId().substring(0, 6)
+                                + "_" + userPackageLegalData.getPackageName();
+                        String reportFileName = eventPublisher.resolveNameAndEnqueue(reportKey, event);
+                        if (reportFileName == null) {
+                            continue; // событие не встало в очередь — hit не пишем
+                        }
+                        Hit hit = Hit.builder()
+                                .packageId(userPackageLegalData.getPackageId())
+                                .subjectId(subjectId)
+                                .signalType(signalType)
+                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                .inqT(message.getInquiryType())
+                                .busCategory(busCategory)
+                                .inqDt(message.getInquiryDate())
+                                .inqP(message.getInquiryPurpose())
+                                .inqA(message.getInquiryAmount())
+                                .segment(segment != null ? segment : -1)
+                                .recordedAt(LocalDateTime.now())
+                                .inn(userPackageLegalData.getInn())
+                                .regNum(userPackageLegalData.getRegNum())
+                                .build();
+                        hits.add(hit);
+                    }
+                }
+            }
+        }
+
+        return hits;
+    }
+}

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

@@ -0,0 +1,144 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.entities.app.Hit;
+import ru.nbch.credit_tracker.entities.app.PhonesMap;
+import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
+import ru.nbch.credit_tracker.enums.EventType;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+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.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+public class NaturalPersonKafkaSignalProcessor {
+
+    private final PackageRecordsService packageRecordsService;
+    private final SourceService sourceService;
+    private final SignalKafkaEventPublisher eventPublisher;
+
+    public NaturalPersonKafkaSignalProcessor(PackageRecordsService packageRecordsService,
+                                             SourceService sourceService,
+                                             SignalKafkaEventPublisher eventPublisher) {
+        this.packageRecordsService = packageRecordsService;
+        this.sourceService = sourceService;
+        this.eventPublisher = eventPublisher;
+    }
+
+    public List<Hit> process(List<PackageMessage> messages, Map<String, Integer> segmentMap) {
+        List<Hit> hits = new ArrayList<>();
+
+        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 commonMemberCode = memberCode.substring(0, 6);
+
+            String busCategory = busCategories.get(commonMemberCode);
+            if (StringUtils.isBlank(busCategory)) {
+                log.error("Для memberсode = {} отсутствует bus_category или не найден memberсode", commonMemberCode);
+                continue;
+            }
+
+            int 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(Integer.toString(signalType))) {
+                        SignalKafkaEvent event = new SignalKafkaEvent();
+                        event.setEventType(EventType.WRITE);
+                        event.setNatural(true);
+                        event.setUserPackagePhoneData(userPackagePhoneData);
+                        event.setPackageMessage(message);
+                        event.setSubjectId(subjectId);
+                        event.setFlagDebt5000(flagDebt5000);
+                        event.setFlag90days(flag90days);
+                        event.setSignalType(Integer.toString(signalType));
+                        event.setBusCategory(signalType == 3 ? busCategory : null);
+
+                        Integer segment = segmentMap.get(commonMemberCode);
+                        event.setSegment(segment != null ? segment : -1);
+
+                        String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
+                                + "_" + userPackagePhoneData.getPackageName();
+                        String reportFileName = eventPublisher.resolveNameAndEnqueue(reportKey, event);
+                        if (reportFileName == null) {
+                            continue; // событие не встало в очередь — hit не пишем
+                        }
+                        Hit hit = Hit.builder()
+                                .packageId(userPackagePhoneData.getPackageId())
+                                .subjectId(subjectId)
+                                .signalType(signalType)
+                                .reportFileName(reportFileName.replace(".xml", ".zip"))
+                                .inqT(message.getInquiryType())
+                                .busCategory(busCategory)
+                                .inqDt(message.getInquiryDate())
+                                .inqP(message.getInquiryPurpose())
+                                .inqA(message.getInquiryAmount())
+                                .segment(segment != null ? segment : -1)
+                                .recordedAt(LocalDateTime.now())
+                                .phone(userPackagePhoneData.getPhone())
+                                .build();
+                        hits.add(hit);
+                    }
+                }
+            }
+        }
+
+        return hits;
+    }
+}

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

@@ -0,0 +1,23 @@
+package ru.nbch.credit_tracker.service.mode.monitoring;
+
+import org.springframework.stereotype.Component;
+import ru.nbch.credit_tracker.model.SignalKafkaEvent;
+
+@Component
+public class SignalKafkaEventPublisher {
+
+    private final ReportNameRegistry reportNameRegistry;
+
+    public SignalKafkaEventPublisher(ReportNameRegistry reportNameRegistry) {
+        this.reportNameRegistry = reportNameRegistry;
+    }
+
+    /**
+     * Резолвит имя результирующего файла и ставит событие в очередь.
+     *
+     * @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
+     */
+    public String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
+        return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
+    }
+}