fix recalc flags

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

@@ -1,48 +1,25 @@
 package ru.nbch.credit_tracker.service;
 
 import java.io.BufferedInputStream;
-import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.UUID;
-import java.util.function.Consumer;
-import java.util.function.Predicate;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
-import ru.nbch.credit_tracker.builder.PersonFactoryBuilder;
 import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
 import ru.nbch.credit_tracker.entities.app.Hit;
 import ru.nbch.credit_tracker.entities.app.PhoneFidMap;
 import ru.nbch.credit_tracker.entities.app.UserPackage;
-import ru.nbch.credit_tracker.entities.external_request.signals.Auth;
-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;
 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;
 
 @Slf4j
 @Service
@@ -55,13 +32,7 @@ public class SignalService {
     private final UserPackageDao userPackageService;
     private final PhoneFidMapService phoneFidMapService;
     private final PhoneService2 phoneService2;
-    private final PhoneService phoneService;
-    private final FlagsService flagsService;
-    private final PersonService personService;
     private final StaxXmlProcessor staxXmlProcessor;
-    private final FidNameEnricher fidNameEnricher;
-    private final FlagsEnricher flagsEnricher;
-
     private final int batchSize;
 
     public SignalService(CreditTrackerSettings config,
@@ -69,137 +40,14 @@ public class SignalService {
                          UserPackageDao userPackageService,
                          PhoneFidMapService phoneFidMapService,
                          PhoneService2 phoneService2,
-                         PhoneService phoneService,
+                         StaxXmlProcessor staxXmlProcessor) {
-                         FlagsService flagsService,
-                         PersonService personService,
-                         StaxXmlProcessor staxXmlProcessor, FidNameEnricher fidNameEnricher, FlagsEnricher flagsEnricher) {
         this.config = config;
         this.hitsService = hitsService;
         this.userPackageService = userPackageService;
         this.phoneFidMapService = phoneFidMapService;
         this.phoneService2 = phoneService2;
-        this.phoneService = phoneService;
-        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());
-        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(true, false);
-    }
-
-    /**
-     * Обрабатывает все PhoneFidMap по packageId, выгружая их пачками и фильтруя по переданному фильтру
-     */
-    public UpdateResult updateSubjects(Long packageId, Predicate<PhoneFidMap> filter) {
-        boolean updated = false;
-        boolean noFidsFound = true;
-        int offset = 0;
-        List<PhoneFidMap> batch;
-        Map<Integer, PhoneFidMap> phonesByFidToProcessing = new HashMap<>();
-        do {
-            batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize);
-
-            List<PhoneFidMap> filteredItems;
-            if (filter != null) {
-                filteredItems = batch.stream().filter(filter).toList();
-            } else {
-                filteredItems = batch;
-            }
-
-            if (!updated && !filteredItems.isEmpty()) {
-                updated = true;
-            }
-
-            if (!filteredItems.isEmpty()) {
-                Map<Integer, PhoneFidMap> fidsMapByBatch = phoneService.defineFidByPhone(filteredItems);
-                phoneService.defineFidByPhone(filteredItems);
-                log.debug("Actual fids size by phone's batch: {}", fidsMapByBatch.size());
-
-                if (fidsMapByBatch.size() + phonesByFidToProcessing.size() <= batchSize) {
-                    phonesByFidToProcessing.putAll(fidsMapByBatch);
-                } else {
-                    //todo execute find name and id -> forming xml
-                    boolean fidsFound = updateFlags(phonesByFidToProcessing, packageId);
-                    if (noFidsFound) {
-                        noFidsFound = !fidsFound;
-                    }
-                    phonesByFidToProcessing = new HashMap<>(fidsMapByBatch);
-                }
-            }
-
-            offset += batchSize;
-        } while (!batch.isEmpty());
-
-        if (!phonesByFidToProcessing.isEmpty()) {
-            boolean fidsFound = updateFlags(phonesByFidToProcessing, packageId);
-            if (noFidsFound) {
-                noFidsFound = !fidsFound;
-            }
-        }
-
-        return new UpdateResult(updated, noFidsFound);
-    }
-
-    /**
-     * Обрабатывает пачку PhoneFidMap: обновляет фиды и флаги просрочек
-     * @return fidsFound true - если хотя бы по одному номеру был найден фид, false - в противном случае
-     */
-    // TODO Добавить отдельный метод. Если не нашелся фид, надо обновить в базе fid = null и флаги = null calculated_at=now()
-    public boolean updateFlags(Map<Integer, PhoneFidMap> fidsMap, Long packageId) {
-        log.info("Start calculating flags for subjects batch. PackageId: {}", packageId);
-        Map<Integer, FlagsService.FlagsData> flagsDataMap = flagsService.findFlags(fidsMap.keySet().stream().toList());
-        List<PhoneFidMap> subjectToUpdate = new ArrayList<>();
-        for (Map.Entry<Integer, FlagsService.FlagsData> entry : flagsDataMap.entrySet()) {
-            Integer fid = entry.getKey();
-            FlagsService.FlagsData flagsData = entry.getValue();
-            PhoneFidMap phoneFidMap = fidsMap.get(fid);
-
-            phoneFidMap.setFid(fid);
-            phoneFidMap.setFlagDebt5000(flagsData.getFlagDebt5000());
-            phoneFidMap.setFlag90days(flagsData.getFlag90days());
-            phoneFidMap.setCalculatedAt(LocalDateTime.now());
-
-            subjectToUpdate.add(phoneFidMap);
-        }
-
-        log.debug("Start updating subject's batch");
-        if (!subjectToUpdate.isEmpty()) {
-            int updateBatchSize = 7000; // 9 параметров на одну запись для обновления. Максимальное количество параметров ~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.debug("Complete updating subject's batch");
-
-        log.info("Finish calculating flags for subjects. PackageId: {}. Batch size: {}", packageId, fidsMap.size());
-        return !fidsMap.isEmpty();
     }
 
 
@@ -207,87 +55,6 @@ public class SignalService {
         userPackageService.updatePackageStatus(packageId, status);
     }
 
-    public void updatePackageOnMonitoringSuccess(Long packageId, LocalDateTime monitoringAt) {
-        userPackageService.updatePackageOnMonitoringSuccess(packageId, monitoringAt);
-    }
-
-    /**
-     * Формирует файл запроса в официальный сервис сигналов, выгружая паспортные данные пачками и сразу записывая их в файл
-     *
-     * @return Путь к файлу запроса к официальному сервису сигналов на постановку на мониторинг
-     */
-    public String writeSignals(Long packageId) {
-        String folderPath = config.getLoadDir() + File.separator + "tmp_folder" + File.separator;
-        String filePath = folderPath + UUID.randomUUID() + "_output.xml";
-        CTUtil.createFolder(folderPath);
-
-        Signals signals = buildSignals(packageId);
-
-        SignalsStaxXmlWriter signalsStaxXmlWriter = new SignalsStaxXmlWriter(config.getEncoding());
-
-        signalsStaxXmlWriter.writeWithStax(filePath,
-                signals,
-                writer -> buildPersons(packageId, writer));
-
-        return filePath;
-    }
-
-    public Signals buildSignals(Long packageId) {
-        UserPackage userPackage = userPackageService.findPackage(packageId);
-        Signals signals = new Signals();
-        signals.setOnline("2"); // Всегда=2 (только онлайн-сигналы)
-        signals.setProcessWithErrors("1"); // Всегда=1 (обработать пакет полностью, по обнаруженным ошибкам (при наличии) сформировать файл отказа, записи без ошибок установить на мониторинг)
-        signals.setId(userPackage.getUserId().substring(0, 6) + "_" + userPackage.getPackageName()); // <MEMBERCODE>_<PackageName>
-        Auth auth = new Auth();
-        auth.setUser(config.getSignalsuser());
-        signals.setAuth(auth);
-        return signals;
-    }
-
-    /**
-     * Обрабатывает все PhoneFidMap по packageId, выгружая их пачками и фильтруя по переданному фильтру.
-     * writer записывает пачку сразу в файл
-     */
-    private void buildPersons(Long packageId, Consumer<List<Person>> writer) {
-        int offset = 0;
-        List<PhoneFidMap> batch;
-        do {
-            batch = phoneFidMapService.findFullPhoneFidMap(packageId, offset, batchSize);
-            List<Person> persons = findPersons(batch);
-            if (!persons.isEmpty()) {
-                writer.accept(persons); // записывает пачку Persons в файл
-            }
-            offset += batchSize;
-        } while (!batch.isEmpty());
-
-    }
-
-    /**
-     * Формирует список Person для постановки на мониторинг в официальный сервис сигналов на основе списка PhoneFidMap
-     */
-    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(subjects.stream()
-                .map(PhoneFidMap::getFid)
-                .filter(Objects::nonNull).toList());
-        subjects.forEach(data -> {
-            if (data.getFid() != null) {
-                PersonService.PersonData personData = personFidMap.get(data.getFid());
-                Person person = PersonFactoryBuilder.builder()
-                        .personData(personData)
-                        .uid(data.getId())
-                        .build();
-                if (person != null) {
-                    persons.add(person);
-                }
-            }
-        });
-        return persons;
-    }
-
     public void updateSubjectsErrors(Path filePath, Long packageId) {
         log.info("Start updating subjects errors. PackageId: {}", packageId);
         try (BufferedInputStream in = new BufferedInputStream(Files.newInputStream(filePath), 8 * 1024 * 1024)) {
@@ -314,14 +81,6 @@ public class SignalService {
         log.info("Subject's errors were updated. PackageId: {}", packageId);
     }
 
-    public List<UserPackage> findPackagesOnMonitoring() {
-        return userPackageService.findAllPackagesByStatus(Status.ON_MONITORING);
-    }
-
-    public List<UserPackage> findPackagesOnProcessing() {
-        return userPackageService.findAllPackagesByStatus(Status.PROCESSING);
-    }
-
     public void insertHits(List<Hit> hits) {
         hitsService.insertHits(hits);
     }

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

@@ -1,8 +1,7 @@
 package ru.nbch.credit_tracker.service;
 
-import java.time.LocalDateTime;
-import java.time.temporal.ChronoUnit;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.MDC;
 import org.springframework.scheduling.annotation.EnableScheduling;
@@ -14,6 +13,10 @@ import ru.nbch.credit_tracker.enums.Status;
 import ru.nbch.credit_tracker.mapper.signal.PhoneFidMapMapper;
 import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
 import ru.nbch.credit_tracker.service.signal.SignalRestService;
+import ru.nbch.credit_tracker.service.task.Pipeline;
+import ru.nbch.credit_tracker.service.task.PipelinePackageManager;
+import ru.nbch.credit_tracker.service.task.SetupParam;
+import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
 
 /**
  * Сервис периодического обновления фидов и флагов просрочек по пакетам находящимся в статусе ON_MONITORING
@@ -30,66 +33,46 @@ public class UpdatingFlagsService {
     private final Long daysToUpdateFlags;
     private final PhoneFidMapMapper phoneFidMapMapper;
     private final UserPackageMapper userPackageMapper;
+    private final PipelinePackageManager pipelinePackageManager;
+    private final ExecutorService pipelineExecutorService;
 
     public UpdatingFlagsService(SignalService signalService,
                                 SignalRestService signalRestService,
                                 MonitoringErrorProcesingService monitoringErrorProcesingService,
                                 CreditTrackerSettings config,
-                                PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper) {
+                                PhoneFidMapMapper phoneFidMapMapper, UserPackageMapper userPackageMapper,
+                                PipelinePackageManager pipelinePackageManager,
+                                ExecutorService pipelineExecutorService) {
         this.signalService = signalService;
         this.signalRestService = signalRestService;
         this.monitoringErrorProcesingService = monitoringErrorProcesingService;
         this.daysToUpdateFlags = config.getDaysToUpdateFlags() != null ? config.getDaysToUpdateFlags() : 1;
         this.phoneFidMapMapper = phoneFidMapMapper;
         this.userPackageMapper = userPackageMapper;
+        this.pipelinePackageManager = pipelinePackageManager;
+        this.pipelineExecutorService = pipelineExecutorService;
     }
 
     @Scheduled(cron = "${ct.update-flags-cron-expression}")
     private void update() {
         log.info("Start updating flags");
-        List<UserPackage> packages = signalService.findPackagesOnMonitoring();
+        List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
         for (UserPackage userPackage : packages) {
             MDC.put("ru.nbch.credit_tracker.log.mdc.key", "[" + userPackage.getPackageName() + "]");
             log.info("Check on updating flags for Package {}", userPackage.getPackageName());
             boolean hasOldRow = phoneFidMapMapper.hasOldRow(userPackage.getId(), daysToUpdateFlags.intValue());
             log.info("There are {} old rows", hasOldRow ? "exist" : "doesn't exist");
             if (hasOldRow) {
-                updateFlags(userPackage);
+                try {
+                    userPackageMapper.updatePackageStatus(userPackage.getId(), Status.PROCESSING_REPEATED.name());
+                    Pipeline pipeline = pipelinePackageManager.buildAndRun(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT));
+                    pipeline.start(pipelineExecutorService);
+                } catch (Exception e) {
+                    log.error("Failed to build pipeline", e);
+                }
             }
             MDC.remove("ru.nbch.credit_tracker.log.mdc.key");
         }
         log.info("Finish updating flags");
     }
-
-    private void updateFlags(UserPackage userPackage) {
-        Long packageId = userPackage.getId();
-        String packageName = userPackage.getPackageName();
-        try {
-            //todo set reject_status = null
-            userPackageMapper.updatePackageStatus(packageId, Status.PROCESSING_REPEATED.name());
-            SignalService.UpdateResult updateResult = signalService.updateFlags2(userPackage);
-            if (updateResult.isUpdated()) {
-                if (signalRestService.setupMonitoring(packageId)) {
-//                    LocalDateTime monitoringAt = userPackage.getMonitoringAt() != null ? userPackage.getMonitoringAt() : LocalDateTime.now();
-                    log.info("Flags were updated successfully. PackageId: {} PackageName: {}. Package status changed to ON_MONITORING", packageId, packageName);
-                    signalService.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now());
-                    monitoringErrorProcesingService.findErrors(packageId);
-                } else {
-                    log.info("Updated package registration was rejected by signal service. PackageId: {} PackageName: {}", packageId, packageName);
-                    signalService.updatePackageStatus(packageId, Status.ERROR);
-                }
-            } else {
-                log.info("No flags were updated for package id {}. Package status changed to ON_MONITORING", packageId);
-                signalService.updatePackageStatus(packageId, Status.ON_MONITORING);
-            }
-        } catch (Exception e) {
-            log.error("Ошибка при поиске фидов или перерасчете флагов просрочек. Package_id {}", packageId, e);
-        }
-    }
-
-    private boolean hasDaysPassed(LocalDateTime pastDateTime) {
-        LocalDateTime now = LocalDateTime.now();
-        long daysPassed = ChronoUnit.DAYS.between(pastDateTime, now);
-        return daysPassed >= daysToUpdateFlags;
-    }
 }

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

@@ -31,6 +31,8 @@ public class PersonService {
             return new HashMap<>();
         }
         Map<Integer, PersonData> personDataMap = new HashMap<>();
+
+        //делаем тоже самое что сделали signalService.updateFlags2(userPackage);
         Map<Integer, NameData> nameDataMap = findNameData(fids);
         Map<Integer, PassportData> passortDataMap = findPassportData(fids);
 

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

@@ -66,15 +66,6 @@ public class SignalRestService {
         this.authToken = buildToken(config.getSignalsToken(), config.getSignalsuser());
     }
 
-    /**
-     * Ставит пакет на мониторинг в официальный сервис сигналов
-     * @return true - если пакет успешно поставился на мониторинг. false - в противном случае
-     */
-    public boolean setupMonitoring(Long packageId) throws Exception {
-        String path = signalService.writeSignals(packageId);
-        return setupMonitoring(Path.of(path));
-    }
-
     public boolean setupMonitoring(Path signalFilePath) throws Exception {
         Path gzip = ZipUtils.packGzip(signalFilePath);
 

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

@@ -55,7 +55,7 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
     @Override
     public void startFillingQueue(BlockingQueue<SubjectPayload> firstQueue) {
         try {
-            log.info("Start registering subjects. PackageId: {};", packageId);
+            log.info("Start kicker: mode:{}, packageId: {};", mode(), packageId);
             int limit = BATCH_SIZE;
             UserPackage userPackage = userPackageMapper.findPackage(packageId);
             List<PhoneFidMap> page = nextPage(limit);
@@ -64,6 +64,8 @@ public abstract class KickPipeLine implements IPipeLineStarter<SubjectPayload> {
                 List<SubjectProfile> subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder()
                     .mapId(phoneFidMap.getId())
                     .identifier(phoneFidMap.getSubjectId())
+                    //наличие этого поля зависит от выборки, в определенных select намеренно опущено это поле
+                    .fid(phoneFidMap.getFid())
                     .phone(phoneFidMap.getPhone())
                     .build()).toList();
                 try {

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

@@ -22,7 +22,7 @@ public class KickPipeLineRestoreImpl extends KickPipeLine {
 
     @Override
     protected PipelineMode mode() {
-        return PipelineMode.RESTORE;
+        return PipelineMode.REPEATED_RESTORE;
     }
 
     @Override

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

@@ -158,7 +158,6 @@
                package_id,
                subject_id,
                phone,
-               fid,
                error_text
         FROM signals_phone_fid_map
         WHERE package_id = #{packageId}