diff --git a/credit-tracker.properties b/credit-tracker.properties index fb3ee0c..4365883 100644 --- a/credit-tracker.properties +++ b/credit-tracker.properties @@ -69,7 +69,7 @@ ct.monitoring-interval=${MONITORING_INTERVAL:15} ct.update-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *} ct.new-flags-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 25 18 * * *} ct.days-to-update-flags=${DAYS_TO_UPDATE_FLAGS:30} -ct.update-fid-cron-expression=${UPDATE_FLAGS_CRON_EXPRESSION:0 10 12 * * *} +ct.update-fid-cron-expression=${UPDATE_FIDS_CRON_EXPRESSION:0 10 12 * * *} ct.days-to-update-fids=${DAYS_TO_UPDATE_FIDS:30} ct.db-version=${DB_VERSION:1234} diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index 97b975d..16a7f19 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -4,14 +4,14 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.ConfigurableApplicationContext; import ru.nbch.credit_tracker.service.MonitoringService; -import ru.nbch.credit_tracker.service.UpdateProcessingPackagesService; +import ru.nbch.credit_tracker.service.StartupRestoreService; @SpringBootApplication public class CreditTrackerApplication { public static void main(String[] args) { ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); - UpdateProcessingPackagesService updateProcessingPackagesService = ctx.getBean(UpdateProcessingPackagesService.class); + StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class); MonitoringService monitoringService = ctx.getBean(MonitoringService.class); updateProcessingPackagesService.run(); monitoringService.run(); diff --git a/src/main/java/ru/nbch/credit_tracker/component/comparator/DeletePackageStatusComparator.java b/src/main/java/ru/nbch/credit_tracker/component/comparator/DeletePackageStatusComparator.java index 8c09ebb..d7bd9a5 100644 --- a/src/main/java/ru/nbch/credit_tracker/component/comparator/DeletePackageStatusComparator.java +++ b/src/main/java/ru/nbch/credit_tracker/component/comparator/DeletePackageStatusComparator.java @@ -15,7 +15,7 @@ public class DeletePackageStatusComparator { try { Status status = Status.valueOf(str); return switch (status) { - case ON_MONITORING, PROCESSING_REPEATED -> 1; + case ON_MONITORING, PROCESSING_REPEATED, NO_FID -> 1; case DELETED -> 0; default -> -1; }; diff --git a/src/main/java/ru/nbch/credit_tracker/enums/Status.java b/src/main/java/ru/nbch/credit_tracker/enums/Status.java index dd35f21..c1368bb 100644 --- a/src/main/java/ru/nbch/credit_tracker/enums/Status.java +++ b/src/main/java/ru/nbch/credit_tracker/enums/Status.java @@ -2,6 +2,7 @@ package ru.nbch.credit_tracker.enums; public enum Status { ON_MONITORING, // пакет стоит на мониторинге в оф. сервисе + NO_FID, REJECTED, // отклонен на стороне НБКИ PROCESSING, // идет поиск телефона и расчет флагов по пакету PROCESSING_REPEATED, diff --git a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java index ffeb32f..b355e68 100644 --- a/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java +++ b/src/main/java/ru/nbch/credit_tracker/facade/SignalFacadeV3.java @@ -6,6 +6,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.StandardCopyOption; import java.time.LocalDateTime; +import java.util.Arrays; import java.util.List; import java.util.concurrent.ExecutorService; import javax.xml.stream.XMLStreamException; @@ -214,10 +215,12 @@ public class SignalFacadeV3 { PackageAll pack = new PackageAll(); pack.setPackageName(data.getPackageName()); pack.setCreatedAt(data.getCreatedAt() != null ? data.getCreatedAt().toString() : null); - String statusForClient = data.getStatus().equals(Status.ACCEPTED.name()) ? Status.PROCESSING.name() : - data.getStatus().equals(Status.PROCESSING_REPEATED.name()) ? Status.ON_MONITORING.name() : data.getStatus(); - pack.setStatus(statusForClient); - if (data.getStatus().equals(Status.DELETED.name())) { + Status packageStatus = EnumUtil.getEnum(Status.class, data.getStatus()); + Status statusForClient = packageStatus == Status.ACCEPTED ? Status.PROCESSING : + Arrays.asList(Status.PROCESSING_REPEATED, Status.NO_FID).contains(packageStatus) ? Status.ON_MONITORING : packageStatus; + assert statusForClient != null; + pack.setStatus(statusForClient.name()); + if (packageStatus == Status.DELETED) { pack.setDeactivatedAt(data.getDeactivatedAt() != null ? data.getDeactivatedAt().toString() : null); } packages.getPackages().add(pack); @@ -234,13 +237,15 @@ public class SignalFacadeV3 { -> clientExcp(SignalError.ERROR_016, XmlErrorFormat.MonitoringPackageDeleteError, packageName)); Status status = EnumUtil.getEnum(Status.class, packageToDelete.getStatus()); - if (!Status.ON_MONITORING.equals(status)) { + if (!Arrays.asList(Status.ON_MONITORING, Status.NO_FID).contains(status)) { throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName); } - boolean successfullyDeleted = signalRestService.deletePackage(username, packageName); - if (!successfullyDeleted) { - throw clientExcp(SignalError.ERROR_017, XmlErrorFormat.MonitoringPackageDeleteError, packageName); + if (status == Status.ON_MONITORING) { + boolean successfullyDeleted = signalRestService.deletePackage(username, packageName); + if (!successfullyDeleted) { + throw clientExcp(SignalError.ERROR_017, XmlErrorFormat.MonitoringPackageDeleteError, packageName); + } } LocalDateTime deactivatedAt = LocalDateTime.now(); @@ -280,7 +285,7 @@ public class SignalFacadeV3 { if (status == null) { throw clientExcp(SignalError.ERROR_017, XmlErrorFormat.MonitoringReportListError); } - if (status.equals(Status.PROCESSING)) { + if (Arrays.asList(Status.PROCESSING, Status.ACCEPTED).contains(status)) { throw clientExcp(SignalError.ERROR_018, XmlErrorFormat.MonitoringReportListError, packageName); } if (status.equals(Status.REJECTED)) { diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java index 6c2a9c3..8bec5bd 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/PhonesMapMapper.java @@ -8,7 +8,7 @@ import ru.nbch.credit_tracker.entities.app.PhonesMap; @Mapper public interface PhonesMapMapper { List getCachedData(@Param("phones") List phones); - List searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("limit") int limit, @Param("offset") int offset); + List searchedAtLessThan(@Param("daysCount") Integer daysCount, @Param("batchSize") int batchSize, @Param("lastId") Long lastId); PhonesMap findByPhone(@Param("phone") Long phone); List findEligibleForFlagsUpdate(@Param("lastId") Long lastId, @Param("batchSize") Integer batchSize); @@ -18,6 +18,7 @@ public interface PhonesMapMapper { void updateNameCachedData(@Param("json") String json); void updateFlagCachedData(@Param("json") String json); void resetFidAndPD(@Param("json") String json); + void updateSearchedAt(@Param("json") String json); void registerJson(@Param("json") String json); void setFidJson(@Param("json") String json); void deleteUnusedPhones(); diff --git a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java index 9ab791e..4323aaa 100644 --- a/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java +++ b/src/main/java/ru/nbch/credit_tracker/mapper/signal/UserPackageMapper.java @@ -11,12 +11,6 @@ public interface UserPackageMapper { boolean isPackageNotUnique(@Param("userId") String userId, @Param("packageName") String packageName); - boolean isPackageExists(@Param("userId") String userId, @Param("packageName") String packageName, @Param("status") String status); - - boolean hasActivePackages(@Param("userId") String userId, @Param("packageName") String packageName, @Param("statuses") List statuses); - - void deletePackage(@Param("userId") String userId, @Param("packageName") String packageName, @Param("status") String newStatus, @Param("previousStatus") String previousStatus, @Param("deactivatedAt") LocalDateTime deactivatedAt); - void deletePackageById(@Param("id") Long id, @Param("status") String newStatus, @Param("deactivatedAt") LocalDateTime deactivatedAt); Long insertPackage(@Param("userPackage") UserPackage userPackage); @@ -26,6 +20,7 @@ public interface UserPackageMapper { void updatePackageLastDownloadedReport(@Param("id") Long id, @Param("fileName") String fileName); void updatePackageOnMonitoringSuccess(@Param("id") Long id, @Param("monitoringAt") LocalDateTime monitoringAt); + void updateNoFidMonitoring(@Param("id") Long id); void setLastSearchedDt(@Param("id") Long id); void setFidChanged(@Param("packageIds") List id); @@ -37,6 +32,7 @@ public interface UserPackageMapper { List findAllPackagesByStatus(@Param("statuses") String... statuses); List findAllPackagesByUserId(@Param("userId") String userId); + List findOnMonitoringPackages(); List findByMemberCodeAndPackageName(@Param("memberCode") String memberCode, @Param("packageName") String packageName); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java b/src/main/java/ru/nbch/credit_tracker/service/StartupRestoreService.java similarity index 66% rename from src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java rename to src/main/java/ru/nbch/credit_tracker/service/StartupRestoreService.java index 96498e4..4de6e2f 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdateProcessingPackagesService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/StartupRestoreService.java @@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.service; import java.nio.file.Path; import java.util.List; -import java.util.concurrent.ExecutorService; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.EnableAspectJAutoProxy; import org.springframework.stereotype.Service; @@ -27,38 +26,43 @@ import ru.nbch.credit_tracker.utils.error.ExceptionUtils; @Slf4j @Service @EnableAspectJAutoProxy(exposeProxy = true) -public class UpdateProcessingPackagesService { +public class StartupRestoreService { private final UserPackageMapper userPackageMapper; private final PackagesRecordsMapper phoneFidMapMapper; private final PipelinePackageFabric pipelinePackageManager; private final PipelinePackageOrchestrator pipelineOrchestrator; private final CreditTrackerSettings creditTrackerSettings; + private final MonitoringErrorProcesingService monitoringErrorProcesingService; private final Saver saver; - private final ExecutorService pipelineExecutorService; - private final ExecutorService saverExecutorService; - public UpdateProcessingPackagesService(UserPackageMapper userPackageMapper, - PackagesRecordsMapper phoneFidMapMapper, - PipelinePackageFabric pipelinePackageManager, PipelinePackageOrchestrator pipelineOrchestrator, - CreditTrackerSettings creditTrackerSettings, - Saver saver, - ExecutorService pipelineExecutorService, - ExecutorService saverExecutorService) { + public StartupRestoreService(UserPackageMapper userPackageMapper, + PackagesRecordsMapper phoneFidMapMapper, + PipelinePackageFabric pipelinePackageManager, + PipelinePackageOrchestrator pipelineOrchestrator, + CreditTrackerSettings creditTrackerSettings, + MonitoringErrorProcesingService monitoringErrorProcesingService, + Saver saver) { this.userPackageMapper = userPackageMapper; this.phoneFidMapMapper = phoneFidMapMapper; this.pipelinePackageManager = pipelinePackageManager; this.pipelineOrchestrator = pipelineOrchestrator; this.creditTrackerSettings = creditTrackerSettings; + this.monitoringErrorProcesingService = monitoringErrorProcesingService; this.saver = saver; - this.pipelineExecutorService = pipelineExecutorService; - this.saverExecutorService = saverExecutorService; } public void run() { + onMonitoringRestore(); acceptedRestore(); processingRestore(); - processingRepeatedRestore(); + } + + private void onMonitoringRestore() { + List packages = userPackageMapper.findOnMonitoringPackages(); + packages.forEach(userPackage -> { + monitoringErrorProcesingService.findErrors(userPackage.getId()); + }); } private void acceptedRestore() { @@ -82,11 +86,11 @@ public class UpdateProcessingPackagesService { } private void processingRestore() { - List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name()); - packages = packages.stream().filter(userPackage -> userPackage.getRejectStatus() == null).toList(); - log.debug("Found {} packages in status PROCESSING and rejected_status = null", packages.size()); + List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name(), Status.PROCESSING_REPEATED.name()); + log.debug("Found {} packages in status PROCESSING", packages.size()); for (UserPackage userPackage : packages) { try { + log.debug("Build pipeline for packageId {}", userPackage.getId()); Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.DEFAULT)); pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); } catch (Exception e) { @@ -95,17 +99,4 @@ public class UpdateProcessingPackagesService { } } - private void processingRepeatedRestore() { - List packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING_REPEATED.name()); - log.debug("Found {} packages in status PROCESSING_REPEATED", packages.size()); - for (UserPackage userPackage : packages) { - try { - Pipeline pipeline = pipelinePackageManager.build(new SetupParam(userPackage.getId(), null, PipelineMode.REPEATED_RESTORE)); - pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId()); - } catch (Exception e) { - log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e)); - } - } - } - } diff --git a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java index 5193514..37b7735 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UpdatingFidAndPdService.java @@ -42,30 +42,39 @@ public class UpdatingFidAndPdService { @Scheduled(cron = "${ct.update-fid-cron-expression}") public void update() { - log.info("Delete delete unused phones"); + if (!pipelineRecalcPDOrchestrator.checkPermits()) { + log.info("The previous task has not yet been completed."); + return; + } + log.info("Delete unused phones"); phonesMapService.deleteUnusedPhones(); log.info("Start updating fids and PD"); try { Pipeline pipeline = pipelinePackageManager.buildRecalculationPD(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); CompletableFuture future = pipelineRecalcPDOrchestrator.startOrReject(pipeline); future.whenComplete((o, throwable) -> { - if (throwable != null) { - log.error(throwable.getMessage(), throwable); - return; - } - - List packagesRepeated = userPackageMapper.findForRepeatedPackages(); - packagesRepeated.forEach(userPackage -> { - Long packageId = userPackage.getId(); - SetupParam setupParam = new SetupParam(packageId, null, PipelineMode.DEFAULT); - try { - Pipeline makeXmlByCache = pipelinePackageManager.build(setupParam); - pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId); - } catch (Exception e) { - log.error("Failed to build pipeline", e); + try { + if (throwable != null) { + log.error(throwable.getMessage(), throwable); + return; } - }); + List packagesRepeated = userPackageMapper.findForRepeatedPackages(); + packagesRepeated.forEach(userPackage -> { + Long packageId = userPackage.getId(); + SetupParam setupParam = new SetupParam(packageId, null, PipelineMode.DEFAULT); + try { + Pipeline makeXmlByCache = pipelinePackageManager.build(setupParam); + pipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId); + } catch (Exception e) { + log.error("Failed to build pipeline", e); + } + }); + } finally { + log.info("pipeline finished"); + pipeline.close(); + pipelineRecalcPDOrchestrator.release(); + } }); } catch (Exception e) { log.error("Failed to build pipeline", e); diff --git a/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java b/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java index a12e6f1..789babc 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/UserPackageService.java @@ -56,7 +56,7 @@ public class UserPackageService { } public List findActivePackages(String userId) { - return userPackageMapper.findUserPackages(userId, Arrays.asList(Status.ON_MONITORING.name())); + return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name(), Status.NO_FID.name())); } public void setFidChanged(List id) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java index 16d3726..c048ffc 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java +++ b/src/main/java/ru/nbch/credit_tracker/service/indic/PersonServiceV2.java @@ -102,6 +102,6 @@ public class PersonServiceV2 { } private boolean checkLength(Id p) { - return ("21".equals(p.getIdType2()) || "21".equals(p.getIdType())) && CTUtil.normalizePassport(p).length() == 10; + return !("21".equals(p.getIdType2()) && "21".equals(p.getIdType())) || CTUtil.normalizePassport(p).length() == 10; } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java index 065784e..8f80fa7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/signal/PhonesMapService.java @@ -33,6 +33,19 @@ public class PhonesMapService { } } + public void updateSearchedAt(List phones) { + try { + String json = entityToJsonMapper.writeValueAsString(phones.stream() + .filter(Objects::nonNull) + .map(phone -> new HashMap() {{ + put("phone", phone); + }}).toList()); + phonesMapMapper.updateSearchedAt(json); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + public void updateNameCachedData(List subjectProfiles) { try { String json = entityToJsonMapper.writeValueAsString(subjectProfiles.stream() diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java index 0908366..40be194 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageOrchestrator.java @@ -118,7 +118,7 @@ public class PipelinePackageOrchestrator { } if (res.signalXmlFilePath() == null) { log.info("No one fid was found by packageId : {}; xml file won't be sent", packageId); - userPackageMapper.updatePackageOnMonitoringSuccess(packageId, LocalDateTime.now()); + userPackageMapper.updateNoFidMonitoring(packageId); return; } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java index 0a92801..a2964ef 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelineRecalcPDOrchestrator.java @@ -36,6 +36,11 @@ public class PipelineRecalcPDOrchestrator { return innerStart(pipeline); } + public boolean checkPermits() { + synchronized (lock) { + return semaphore.availablePermits() > 0; + } + } private CompletableFuture innerStart(Pipeline pipeline) { try { pipeline.start(pipelineExecutorService); @@ -54,10 +59,11 @@ public class PipelineRecalcPDOrchestrator { .orTimeout(creditTrackerSettings.getPipeline().globalTimeout().toMillis(), TimeUnit.MILLISECONDS); } - private void release() { + public void release() { log.debug("checking for delayed pipelines..."); synchronized (lock) { semaphore.release(); + log.info("semaphore amount: {}", semaphore.availablePermits()); } } } diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java index 5123d05..b952151 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdEnricher.java @@ -51,10 +51,12 @@ public class FidIdEnricher implements Handler { log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); List noIdDataFids = subjectsProfile.stream() - .filter(subjectProfile -> subjectProfile.getIdData() == null) + .filter(subjectProfile -> subjectProfile.getIdData() == null && subjectProfile.getFid() != null) .map(SubjectProfile::getFid) .toList(); + long countEmptyFid = subjectsProfile.stream().filter(subjectProfile -> subjectProfile.getFid() == null).count(); + log.info("Count with empty fid: {}", countEmptyFid); Map idByFid; if (!noIdDataFids.isEmpty()) { idByFid = personService.filledPassportsData(noIdDataFids); @@ -64,6 +66,7 @@ public class FidIdEnricher implements Handler { List idCacheOnUpdate = new ArrayList<>(); List personOnProcessing = subjectsProfile.stream() + .filter(subjectProfile -> subjectProfile.getFid() != null) .flatMap(subjectProfile -> { if (subjectProfile.getIdData() == null) { Id id = idByFid.get(subjectProfile.getFid()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java index 896eda6..bce267a 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidIdUpdater.java @@ -1,6 +1,5 @@ package ru.nbch.credit_tracker.service.task.impl; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.stream.Stream; @@ -36,14 +35,13 @@ public class FidIdUpdater implements Handler { @Override public VoidPayload handle(SubjectPayload taskPayload) { List subjectsProfile = taskPayload.getSubjects(); - log.debug("FidIdEnricher handle by packageId: {}; size: {}", taskPayload.getPackageId(), subjectsProfile.size()); + log.debug("FidIdUpdater handle; size: {}", subjectsProfile.size()); List noIdDataFids = subjectsProfile.stream() .map(SubjectProfile::getFid) .toList(); Map idByFid = personService.filledPassportsData(noIdDataFids); - List idCacheOnUpdate = new ArrayList<>(); List personOnProcessing = subjectsProfile.stream() .flatMap(subjectProfile -> { Id id = idByFid.get(subjectProfile.getFid()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java index f778c0c..1e44469 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdater.java @@ -3,6 +3,7 @@ package ru.nbch.credit_tracker.service.task.impl; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.stream.Stream; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; @@ -47,19 +48,24 @@ public class FidNameUpdater implements Handler { List changingFidPhones = new ArrayList<>(); List notFoundFidPhones = new ArrayList<>(); - subjects.forEach(subjectProfile -> { + List notChangedPhones = subjects.stream().flatMap(subjectProfile -> { Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone())); - if (subjectProfile.getFid() == null && name == null) { + if (name != null) { + if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) { + subjectProfile.setFid(name.getFid()); + subjectProfile.getNameData().setFirstName(name.getFirst()); + subjectProfile.getNameData().setLastName(name.getLast()); + subjectProfile.getNameData().setMiddleName(name.getMiddle()); + subjectProfile.getNameData().setBirthDate(name.getBirthDate()); + changingFidPhones.add(subjectProfile); + return Stream.empty(); + } + } else if (subjectProfile.getFid() != null) { notFoundFidPhones.add(subjectProfile.getPhone()); - } else if (subjectProfile.getFid() == null || !subjectProfile.getFid().equals(name.getFid())) { - subjectProfile.setFid(name.getFid()); - subjectProfile.getNameData().setFirstName(name.getFirst()); - subjectProfile.getNameData().setLastName(name.getLast()); - subjectProfile.getNameData().setMiddleName(name.getMiddle()); - subjectProfile.getNameData().setBirthDate(name.getBirthDate()); - changingFidPhones.add(subjectProfile); + return Stream.empty(); } - }); + return Stream.of(subjectProfile.getPhone()); + }).toList(); List changingFids = new ArrayList<>(); if (!notFoundFidPhones.isEmpty()) { phonesMapService.resetFidAndPDByPhones(notFoundFidPhones); @@ -71,6 +77,10 @@ public class FidNameUpdater implements Handler { changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList()); } + if (!notChangedPhones.isEmpty()) { + phonesMapService.updateSearchedAt(notChangedPhones); + } + if (!changingFids.isEmpty()) { List packageIds = packageRecordsService.selectUniquePackageIds(changingFids); if (!packageIds.isEmpty()) { diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java index f066860..e07f862 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/KickRecalcPipeLine.java @@ -64,8 +64,8 @@ public class KickRecalcPipeLine implements IPipeLineStarter { public void startFillingQueue(BlockingQueue firstQueue) { try { log.info("Start kicker: mode:{};", mode()); - int offset = 0; - List page = nextPage(offset); + long lastId = 0; + List page = nextPage(lastId); while (!page.isEmpty()) { log.debug("Get page {}", page.size()); List subjectProfiles = page.stream().map(phoneFidMap -> SubjectProfile.builder() @@ -92,8 +92,8 @@ public class KickRecalcPipeLine implements IPipeLineStarter { log.error("Pipeline starter was interrupted!"); throw new PipelineStarterException(); } - offset += BATCH_SIZE; - page = nextPage(offset); + lastId = page.getLast().getId(); + page = nextPage(lastId); } } catch (PipelineStarterException e) { log.info("Pipeline starter shutting down..."); @@ -115,7 +115,7 @@ public class KickRecalcPipeLine implements IPipeLineStarter { protected PipelineMode mode() { return PipelineMode.REPEATED_RESTORE; } - protected List nextPage(int offset) { - return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, offset); + protected List nextPage(long lastId) { + return phonesMapMapper.searchedAtLessThan(daysToUpdateFids.intValue(), BATCH_SIZE, lastId); } } diff --git a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java index 9c62e99..1664a42 100644 --- a/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java +++ b/src/main/java/ru/nbch/credit_tracker/utils/CTUtil.java @@ -77,7 +77,7 @@ public class CTUtil { if (StringUtils.isBlank(value)) { return null; } - if (idType.equals("21") || idType2.equals("21")) { + if ("21".equals(idType) || "21".equals(idType2)) { return value.replaceAll("[^0-9]", ""); } else { return value.replaceAll("[^А-Яа-яA-Za-z0-9]", ""); diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsJoinMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsJoinMapper.xml index 5efa8b1..fda3809 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsJoinMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsJoinMapper.xml @@ -34,7 +34,7 @@ sup.id_type, sup.flag_debt5000, sup.flag_90_12 - from signals_packages_records spr + from signals_phone_fid_map spr join signals_phones_map sup on cast(spr.phone as bigint) = sup.phone where spr.package_id = #{packageId} and sup.fid is not null diff --git a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml index e113bbd..e9328b0 100644 --- a/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PackageRecordsMapper.xml @@ -19,7 +19,7 @@ subject_id, phone, error_text - FROM signals_packages_records + FROM signals_phone_fid_map WHERE id = #{id} @@ -28,7 +28,10 @@ FROM json_to_recordset(#{json}::json) AS t( phone bigint )) - select distinct (package_id) from signals_packages_records where phone_l in (select * from src) + select distinct (package_id) + FROM signals_phone_fid_map spr + join signals_user_packages sup on sup.id = spr.package_id + where spr.phone_l in (select * from src) and sup.status = 'ON_MONITORING' - update signals_packages_records dst + update signals_phone_fid_map dst set error_text = src.error_text FROM json_to_recordset(#{json}::json) AS src( id bigint, @@ -65,7 +68,7 @@ DELETE - FROM signals_packages_records + FROM signals_phone_fid_map WHERE package_id = #{packageId} @@ -75,7 +78,7 @@ subject_id, phone, error_text - FROM signals_packages_records + FROM signals_phone_fid_map WHERE package_id = #{packageId} ORDER BY id LIMIT #{limit} OFFSET #{offset} diff --git a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml index e880b70..6b245bd 100644 --- a/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml +++ b/src/main/resources/mybatis/mapper/signal/PhonesMapMapper.xml @@ -54,24 +54,24 @@ searched_at, calculated_at FROM signals_phones_map - WHERE searched_at < now() - make_interval(days => #{daysCount}) - ORDER BY id - LIMIT #{limit} OFFSET #{offset} + WHERE searched_at < now() - make_interval(days => #{daysCount}) and id > #{lastId} + order by id + limit #{batchSize} + + +