Compare commits
No commits in common. "59b074b2daa1a3fa80498534b706017d15ce98db" and "2c0525d28d387a67a20a6f9dd11efe68715d982f" have entirely different histories.
59b074b2da
...
2c0525d28d
37 changed files with 345 additions and 367 deletions
|
|
@ -22,9 +22,9 @@ ct.indic-data-source-settings.leak-detection-threshold=${LEAK_DETECTION_THRESHOL
|
|||
|
||||
# signal
|
||||
ct.signal-data-source-settings.driver=org.postgresql.Driver
|
||||
ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.200.200.133:5432/credit_tracker?currentSchema=test_3}
|
||||
ct.signal-data-source-settings.url=${DB_URL_SIGNAL:jdbc:postgresql://10.230.227.125:5432/credit_tracker?currentSchema=test}
|
||||
ct.signal-data-source-settings.username=${USERNAME_SIGNAL:credit_tracker_user}
|
||||
ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:-}
|
||||
ct.signal-data-source-settings.password=${PASSWORD_SIGNAL:password}
|
||||
ct.signal-data-source-settings.maximum-pool-size=${MAXIMUM_POOL_SIZE_SIGNAL:40}
|
||||
ct.signal-data-source-settings.minimum-idle=${MINIMUM_IDLE_SIGNAL:10}
|
||||
ct.signal-data-source-settings.connection-timeout=${CONNECTION_TIMEOUT_SIGNAL:10000}
|
||||
|
|
|
|||
|
|
@ -1,10 +1,9 @@
|
|||
package ru.nbch.credit_tracker.component.comparator;
|
||||
|
||||
import ru.nbch.credit_tracker.entities.app.UserPackage;
|
||||
import ru.nbch.credit_tracker.enums.Status;
|
||||
|
||||
import java.util.Comparator;
|
||||
import java.util.function.ToIntFunction;
|
||||
import ru.nbch.credit_tracker.entities.app.UserPackage;
|
||||
import ru.nbch.credit_tracker.enums.Status;
|
||||
|
||||
public class DeletePackageStatusComparator {
|
||||
|
||||
|
|
@ -16,7 +15,7 @@ public class DeletePackageStatusComparator {
|
|||
try {
|
||||
Status status = Status.valueOf(str);
|
||||
return switch (status) {
|
||||
case ON_MONITORING -> 1;
|
||||
case ON_MONITORING, PROCESSING_REPEATED, NO_FID -> 1;
|
||||
case DELETED -> 0;
|
||||
default -> -1;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,16 +1,15 @@
|
|||
package ru.nbch.credit_tracker.component.comparator;
|
||||
|
||||
import ru.nbch.credit_tracker.entities.app.UserPackage;
|
||||
import ru.nbch.credit_tracker.enums.Status;
|
||||
|
||||
import java.util.Comparator;
|
||||
import java.util.EnumSet;
|
||||
import java.util.function.ToIntFunction;
|
||||
import ru.nbch.credit_tracker.entities.app.UserPackage;
|
||||
import ru.nbch.credit_tracker.enums.Status;
|
||||
|
||||
public class ReportListPackageStatusComparator {
|
||||
|
||||
public static final EnumSet<Status> reportedStatuses = EnumSet.of(
|
||||
Status.ON_MONITORING
|
||||
Status.ON_MONITORING, Status.PROCESSING_REPEATED
|
||||
);
|
||||
|
||||
public static Comparator<UserPackage> INSTANCE = Comparator
|
||||
|
|
@ -25,7 +24,7 @@ public class ReportListPackageStatusComparator {
|
|||
try {
|
||||
Status status = Status.valueOf(str);
|
||||
return switch (status) {
|
||||
case ON_MONITORING -> 1;
|
||||
case ON_MONITORING, PROCESSING_REPEATED -> 1;
|
||||
default -> -1;
|
||||
};
|
||||
} catch (Throwable e) {
|
||||
|
|
|
|||
|
|
@ -1,13 +1,14 @@
|
|||
package ru.nbch.credit_tracker.config.kafka;
|
||||
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.kafka.annotation.EnableKafka;
|
||||
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
|
||||
import org.springframework.kafka.core.ConsumerFactory;
|
||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
|
||||
import org.springframework.kafka.core.*;
|
||||
import org.springframework.kafka.listener.ContainerProperties;
|
||||
import org.springframework.kafka.support.serializer.JsonDeserializer;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
|
|
@ -26,6 +27,7 @@ public class KafkaConfig {
|
|||
|
||||
@Bean("signalKafkaEventQueue")
|
||||
public BlockingQueue<SignalKafkaEvent> signalKafkaEventQueue(CreditTrackerSettings settings) {
|
||||
// TODO add capacity
|
||||
return new LinkedBlockingQueue<>();
|
||||
}
|
||||
|
||||
|
|
@ -64,4 +66,19 @@ public class KafkaConfig {
|
|||
);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ProducerFactory<String, String> producerFactory() {
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
|
||||
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
|
||||
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
||||
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
||||
return new DefaultKafkaProducerFactory<>(props);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
|
||||
return new KafkaTemplate<>(producerFactory);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,40 +10,20 @@ import java.util.concurrent.Executors;
|
|||
@Configuration
|
||||
public class ExecutorsConfig {
|
||||
|
||||
@Bean(name = "npPipelineExecutorService",
|
||||
destroyMethod = "shutdown")
|
||||
public ExecutorService npPipelineExecutorService() {
|
||||
@Bean(destroyMethod = "shutdown")
|
||||
public ExecutorService pipelineExecutorService() {
|
||||
return Executors.newFixedThreadPool(15, r -> {
|
||||
Thread t = new Thread(r);
|
||||
t.setName("np-pipe-starter-%d".formatted(t.threadId()));
|
||||
t.setName("pipe-starter-%d".formatted(t.threadId()));
|
||||
return t;
|
||||
});
|
||||
}
|
||||
|
||||
@Bean(name = "lpPipelineExecutorService",
|
||||
destroyMethod = "shutdown")
|
||||
public ExecutorService lpPipelineExecutorService() {
|
||||
return Executors.newFixedThreadPool(15, r -> {
|
||||
Thread t = new Thread(r);
|
||||
t.setName("lp-pipe-starter-%d".formatted(t.threadId()));
|
||||
return t;
|
||||
});
|
||||
}
|
||||
|
||||
@Bean("npSaverExecutorService")
|
||||
public ExecutorService npSaverExecutorService(CreditTrackerSettings creditTrackerSettings) {
|
||||
@Bean
|
||||
public ExecutorService saverExecutorService(CreditTrackerSettings creditTrackerSettings) {
|
||||
return Executors.newFixedThreadPool(creditTrackerSettings.getNpPipeline().parallelRequests(), r -> {
|
||||
Thread t = new Thread(r);
|
||||
t.setName("np-subject-saver-%d".formatted(t.threadId()));
|
||||
return t;
|
||||
});
|
||||
}
|
||||
|
||||
@Bean("lpSaverExecutorService")
|
||||
public ExecutorService lpSaverExecutorService(CreditTrackerSettings creditTrackerSettings) {
|
||||
return Executors.newFixedThreadPool(creditTrackerSettings.getNpPipeline().parallelRequests(), r -> {
|
||||
Thread t = new Thread(r);
|
||||
t.setName("lp-subject-saver-%d".formatted(t.threadId()));
|
||||
t.setName("subject-saver-%d".formatted(t.threadId()));
|
||||
return t;
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,8 +2,10 @@ package ru.nbch.credit_tracker.enums;
|
|||
|
||||
public enum Status {
|
||||
ON_MONITORING, // пакет стоит на мониторинге в оф. сервисе
|
||||
NO_FID,
|
||||
REJECTED, // отклонен на стороне НБКИ
|
||||
PROCESSING, // идет поиск телефона и расчет флагов по пакету
|
||||
PROCESSING_REPEATED,
|
||||
ACCEPTED,
|
||||
DELETED, // пакет удален
|
||||
ERROR; // произошла ошибка при постановке пакета на мониторинг в оф. сервис
|
||||
|
|
|
|||
|
|
@ -223,7 +223,8 @@ public class SignalFacade {
|
|||
pack.setPackageName(data.getPackageName());
|
||||
pack.setCreatedAt(data.getCreatedAt() != null ? data.getCreatedAt().toString() : null);
|
||||
Status packageStatus = EnumUtil.getEnum(Status.class, data.getStatus());
|
||||
Status statusForClient = packageStatus == Status.ACCEPTED ? Status.PROCESSING : packageStatus;
|
||||
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) {
|
||||
|
|
@ -243,7 +244,7 @@ public class SignalFacade {
|
|||
-> clientExcp(SignalError.ERROR_016, XmlErrorFormat.MonitoringPackageDeleteError, packageName));
|
||||
|
||||
Status status = EnumUtil.getEnum(Status.class, packageToDelete.getStatus());
|
||||
if (Status.ON_MONITORING != status) {
|
||||
if (!Arrays.asList(Status.ON_MONITORING, Status.NO_FID).contains(status)) {
|
||||
throw clientExcp(deleteStatusError(status), XmlErrorFormat.MonitoringPackageDeleteError, packageName);
|
||||
}
|
||||
|
||||
|
|
@ -265,6 +266,7 @@ public class SignalFacade {
|
|||
case PROCESSING, ACCEPTED -> SignalError.ERROR_023;
|
||||
case REJECTED, ERROR -> SignalError.ERROR_024;
|
||||
case DELETED -> SignalError.ERROR_025;
|
||||
case PROCESSING_REPEATED -> SignalError.ERROR_026;
|
||||
default -> throw new IllegalStateException("Unexpected value: " + status);
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,6 +14,4 @@ public interface LegalsFidMapMapper {
|
|||
List<SignalLegalFidMap> findFullLegalFidMap(@Param("packageId") Long packageId,
|
||||
@Param("limit") int limit,
|
||||
@Param("offset") int offset);
|
||||
|
||||
Long deleteByPackageId(@Param("packageId") Long id);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ public interface LegalsMapMapper {
|
|||
@Param("batchSize") int batchSize,
|
||||
@Param("lastId") Long lastId);
|
||||
|
||||
List<LegalsMap> findCachedByInnAndRegNum(@Param("json") String json);
|
||||
List<LegalsMap> findCachedByInnAndRegNUm(@Param("json") String json);
|
||||
|
||||
void updateSearchedAt(@Param("json") String json);
|
||||
|
||||
|
|
|
|||
|
|
@ -17,6 +17,10 @@ public interface PackagesRecordsMapper {
|
|||
@Param("limit") int limit,
|
||||
@Param("offset") int offset);
|
||||
|
||||
List<Long> uniquePackageIdByPhones(@Param("json") String json);
|
||||
|
||||
List<Long> uniquePackageIdByInnAndRegNum(@Param("json") String json);
|
||||
|
||||
List<UserPackagePhoneData> selectUserPackagePhoneData(@Param("json") String json);
|
||||
|
||||
List<UserPackageLegalData> selectUserPackageLegalData(@Param("json") String json);
|
||||
|
|
|
|||
|
|
@ -25,8 +25,14 @@ public interface UserPackageMapper {
|
|||
void updatePackageOnMonitoringSuccess(@Param("id") Long id,
|
||||
@Param("monitoringAt") LocalDateTime monitoringAt);
|
||||
|
||||
void updateNoFidMonitoring(@Param("id") Long id);
|
||||
|
||||
void setFidChanged(@Param("packageIds") List<Long> id);
|
||||
|
||||
UserPackage findPackage(@Param("id") Long id);
|
||||
|
||||
List<UserPackage> findForRepeatedPackages();
|
||||
|
||||
List<UserPackage> findUserPackages(@Param("userId") String userId,
|
||||
@Param("statuses") List<String> status);
|
||||
|
||||
|
|
@ -41,5 +47,7 @@ public interface UserPackageMapper {
|
|||
List<UserPackage> findByMemberCodeAndPackageName(@Param("memberCode") String memberCode,
|
||||
@Param("packageName") String packageName);
|
||||
|
||||
void setRejectedStatus(@Param("id") Long id, @Param("status") String status);
|
||||
|
||||
void updatePackageLastDownloadedReport(@Param("json") String json);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ import ru.nbch.credit_tracker.entities.external_request.kafka.PackageMessage;
|
|||
import ru.nbch.credit_tracker.enums.EventType;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Map;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
|
|
@ -14,7 +13,6 @@ import java.util.Map;
|
|||
@AllArgsConstructor
|
||||
public class SignalKafkaEvent {
|
||||
private EventType eventType;
|
||||
private Map<String, String> reportNames;
|
||||
|
||||
private UserPackagePhoneData userPackagePhoneData;
|
||||
private UserPackageLegalData userPackageLegalData;
|
||||
|
|
|
|||
|
|
@ -17,8 +17,6 @@ public class SubjectPayload implements StagePayload {
|
|||
public static final SubjectPayload POISON = new SubjectPayload(null, PoisonType.NORMALLY);
|
||||
private final PoisonType poison;
|
||||
|
||||
private boolean hasFids = false;
|
||||
|
||||
public SubjectPayload(Long packageId, PoisonType poison) {
|
||||
this.memberCode = null;
|
||||
this.packageName = null;
|
||||
|
|
@ -61,14 +59,6 @@ public class SubjectPayload implements StagePayload {
|
|||
return packageName;
|
||||
}
|
||||
|
||||
public boolean isHasFids() {
|
||||
return hasFids;
|
||||
}
|
||||
|
||||
public void setHasFids(boolean hasFids) {
|
||||
this.hasFids = hasFids;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isPoison() {
|
||||
return poison == PoisonType.NORMALLY;
|
||||
|
|
|
|||
|
|
@ -59,13 +59,17 @@ public class UserPackageService {
|
|||
}
|
||||
|
||||
public List<UserPackage> findActivePackages(String userId) {
|
||||
return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name()));
|
||||
return userPackageMapper.findUserPackages(userId.substring(0, 6), Arrays.asList(Status.ON_MONITORING.name(), Status.NO_FID.name()));
|
||||
}
|
||||
|
||||
public UserPackage findByPackageName(String packageName) {
|
||||
return userPackageMapper.findByPackageName(packageName);
|
||||
}
|
||||
|
||||
public void setFidChanged(List<Long> id) {
|
||||
userPackageMapper.setFidChanged(id);
|
||||
}
|
||||
|
||||
public List<UserPackage> findAllPackages(String userId) {
|
||||
return userPackageMapper.findAllPackagesByUserId(userId.substring(0, 6));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,10 +6,11 @@ import org.springframework.stereotype.Service;
|
|||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.entities.app.UserPackage;
|
||||
import ru.nbch.credit_tracker.enums.Status;
|
||||
import ru.nbch.credit_tracker.mapper.signal.LegalsFidMapMapper;
|
||||
import ru.nbch.credit_tracker.mapper.signal.PackagesRecordsMapper;
|
||||
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
|
||||
import ru.nbch.credit_tracker.service.task.*;
|
||||
import ru.nbch.credit_tracker.service.task.NpPipelinePackageFabric;
|
||||
import ru.nbch.credit_tracker.service.task.NpPipelinePackageOrchestrator;
|
||||
import ru.nbch.credit_tracker.service.task.SetupParam;
|
||||
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
|
||||
import ru.nbch.credit_tracker.service.task.impl.Saver;
|
||||
import ru.nbch.credit_tracker.utils.CTUtil;
|
||||
|
|
@ -30,30 +31,21 @@ public class StartupRestoreService {
|
|||
|
||||
private final UserPackageMapper userPackageMapper;
|
||||
private final PackagesRecordsMapper phoneFidMapMapper;
|
||||
private final LegalsFidMapMapper legalsFidMapMapper;
|
||||
private final NpPipelinePackageFabric npPipelinePackageManager;
|
||||
private final NpPipelinePackageOrchestrator npPipelineOrchestrator;
|
||||
private final LpPipelinePackageFabric lpPipelinePackageManager;
|
||||
private final LpPipelinePackageOrchestrator lpPipelineOrchestrator;
|
||||
private final NpPipelinePackageOrchestrator pipelineOrchestrator;
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
private final Saver saver;
|
||||
|
||||
public StartupRestoreService(UserPackageMapper userPackageMapper,
|
||||
PackagesRecordsMapper phoneFidMapMapper,
|
||||
LegalsFidMapMapper legalsFidMapMapper,
|
||||
NpPipelinePackageFabric npPipelinePackageManager,
|
||||
NpPipelinePackageOrchestrator npPipelineOrchestrator,
|
||||
LpPipelinePackageFabric lpPipelinePackageManager,
|
||||
LpPipelinePackageOrchestrator lpPipelineOrchestrator,
|
||||
NpPipelinePackageOrchestrator pipelineOrchestrator,
|
||||
CreditTrackerSettings creditTrackerSettings,
|
||||
Saver saver) {
|
||||
this.userPackageMapper = userPackageMapper;
|
||||
this.phoneFidMapMapper = phoneFidMapMapper;
|
||||
this.legalsFidMapMapper = legalsFidMapMapper;
|
||||
this.npPipelinePackageManager = npPipelinePackageManager;
|
||||
this.npPipelineOrchestrator = npPipelineOrchestrator;
|
||||
this.lpPipelinePackageManager = lpPipelinePackageManager;
|
||||
this.lpPipelineOrchestrator = lpPipelineOrchestrator;
|
||||
this.pipelineOrchestrator = pipelineOrchestrator;
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
this.saver = saver;
|
||||
}
|
||||
|
|
@ -61,64 +53,39 @@ public class StartupRestoreService {
|
|||
public void run() {
|
||||
acceptedRestore();
|
||||
processingRestore();
|
||||
// TODO LP
|
||||
}
|
||||
|
||||
private void acceptedRestore() {
|
||||
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.ACCEPTED.name());
|
||||
log.debug("Found {} packages in status ACCEPTED", packages.size());
|
||||
for (UserPackage userPackage : packages) {
|
||||
if (userPackage.getSubjectType() == 1) {
|
||||
try {
|
||||
Long count = phoneFidMapMapper.deleteByPackageId(userPackage.getId());
|
||||
log.trace("Deleted {} np rows", count);
|
||||
log.trace("Deleted {} rows", count);
|
||||
Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage);
|
||||
log.debug("Try to read and build pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
|
||||
Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT));
|
||||
npPipelineOrchestrator.startOrDelay(
|
||||
pipelineOrchestrator.startOrDelay(
|
||||
() -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
|
||||
pipeline, userPackage.getId()
|
||||
);
|
||||
} catch (Exception e) {
|
||||
log.error("Error build default np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
Long count = legalsFidMapMapper.deleteByPackageId(userPackage.getId());
|
||||
log.trace("Deleted {} lp rows", count);
|
||||
Path xmlReqPath = CTUtil.workingPathToReqByPackage(creditTrackerSettings.getLoadDir(), userPackage);
|
||||
log.debug("Try to read and build lp pipeline for packageId {}; req xml path: {}", userPackage.getId(), xmlReqPath);
|
||||
Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));
|
||||
lpPipelineOrchestrator.startOrDelay(
|
||||
() -> saver.readAndSaveNpPackage(userPackage.getId(), xmlReqPath),
|
||||
pipeline, userPackage.getId()
|
||||
);
|
||||
} catch (Exception e) {
|
||||
log.error("Error build default lp pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
log.error("Error build default pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void processingRestore() {
|
||||
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name());
|
||||
List<UserPackage> packages = userPackageMapper.findAllPackagesByStatus(Status.PROCESSING.name(), Status.PROCESSING_REPEATED.name());
|
||||
log.debug("Found {} packages in status PROCESSING", packages.size());
|
||||
for (UserPackage userPackage : packages) {
|
||||
if (userPackage.getSubjectType() == 1) {
|
||||
try {
|
||||
log.debug("Build np pipeline for packageId {}", userPackage.getId());
|
||||
log.debug("Build pipeline for packageId {}", userPackage.getId());
|
||||
Pipeline pipeline = npPipelinePackageManager.build(new SetupParam(userPackage.getId(), 1, PipelineMode.DEFAULT));
|
||||
npPipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
|
||||
pipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
|
||||
} catch (Exception e) {
|
||||
log.error("Error build restore np pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
log.debug("Build lp pipeline for packageId {}", userPackage.getId());
|
||||
Pipeline pipeline = lpPipelinePackageManager.build(new SetupParam(userPackage.getId(), 2, PipelineMode.DEFAULT));
|
||||
lpPipelineOrchestrator.startOrDelay(pipeline, userPackage.getId());
|
||||
} catch (Exception e) {
|
||||
log.error("Error build restore lp pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
log.error("Error build restore pipeline for packageId {}: {}", userPackage.getId(), ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,12 +5,15 @@ import org.springframework.scheduling.annotation.EnableScheduling;
|
|||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Service;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.entities.app.UserPackage;
|
||||
import ru.nbch.credit_tracker.mapper.signal.LegalsMapMapper;
|
||||
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
|
||||
import ru.nbch.credit_tracker.service.signal.PhonesMapService;
|
||||
import ru.nbch.credit_tracker.service.task.*;
|
||||
import ru.nbch.credit_tracker.service.task.impl.PipelineMode;
|
||||
import ru.nbch.pipeline.engine.core.Pipeline;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
@Slf4j
|
||||
|
|
@ -21,6 +24,9 @@ public class UpdatingFidService {
|
|||
private final LpPipelinePackageFabric lpPipelinePackageFabric;
|
||||
private final NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator;
|
||||
private final LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator;
|
||||
private final NpPipelinePackageOrchestrator npPipelinePackageOrchestrator;
|
||||
private final LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator;
|
||||
private final UserPackageMapper userPackageMapper;
|
||||
private final PhonesMapService phonesMapService;
|
||||
private final LegalsMapMapper legalsMapMapper;
|
||||
|
||||
|
|
@ -29,12 +35,18 @@ public class UpdatingFidService {
|
|||
LpPipelinePackageFabric lpPipelinePackageFabric,
|
||||
NpPipelineRecalcOrchestrator npPipelineRecalcOrchestrator,
|
||||
LpPipelineRecalcOrchestrator lpPipelineRecalcOrchestrator,
|
||||
NpPipelinePackageOrchestrator npPipelinePackageOrchestrator,
|
||||
LpPipelinePackageOrchestrator lpPipelinePackageOrchestrator,
|
||||
UserPackageMapper userPackageMapper,
|
||||
PhonesMapService phonesMapService,
|
||||
LegalsMapMapper legalsMapMapper) {
|
||||
this.npPipelinePackageFabric = npPipelinePackageFabric;
|
||||
this.lpPipelinePackageFabric = lpPipelinePackageFabric;
|
||||
this.npPipelineRecalcOrchestrator = npPipelineRecalcOrchestrator;
|
||||
this.lpPipelineRecalcOrchestrator = lpPipelineRecalcOrchestrator;
|
||||
this.npPipelinePackageOrchestrator = npPipelinePackageOrchestrator;
|
||||
this.lpPipelinePackageOrchestrator = lpPipelinePackageOrchestrator;
|
||||
this.userPackageMapper = userPackageMapper;
|
||||
this.phonesMapService = phonesMapService;
|
||||
this.legalsMapMapper = legalsMapMapper;
|
||||
}
|
||||
|
|
@ -47,7 +59,7 @@ public class UpdatingFidService {
|
|||
}
|
||||
log.info("Delete unused phones");
|
||||
phonesMapService.deleteUnusedPhones();
|
||||
log.info("Start updating fids");
|
||||
log.info("Start updating fids and PD");
|
||||
try {
|
||||
Pipeline pipeline = npPipelinePackageFabric.buildRecalculationPD(new SetupParam(null, 1, PipelineMode.RECALC_FID));
|
||||
CompletableFuture<?> future = npPipelineRecalcOrchestrator.startOrReject(pipeline);
|
||||
|
|
@ -58,7 +70,17 @@ public class UpdatingFidService {
|
|||
return;
|
||||
}
|
||||
|
||||
log.info("Updating NP fids successfully completed.");
|
||||
List<UserPackage> packagesRepeated = userPackageMapper.findForRepeatedPackages();
|
||||
packagesRepeated.forEach(userPackage -> {
|
||||
Long packageId = userPackage.getId();
|
||||
SetupParam setupParam = new SetupParam(packageId, 1, PipelineMode.RESEND);
|
||||
try {
|
||||
Pipeline makeXmlByCache = npPipelinePackageFabric.build(setupParam);
|
||||
npPipelinePackageOrchestrator.startOrDelay(makeXmlByCache, packageId);
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to build pipeline", e);
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
log.info("pipeline finished");
|
||||
pipeline.close();
|
||||
|
|
@ -76,7 +98,7 @@ public class UpdatingFidService {
|
|||
log.info("The previous task has not yet been completed.");
|
||||
return;
|
||||
}
|
||||
log.info("Delete unused lp maps");
|
||||
log.info("Delete unused phones");
|
||||
legalsMapMapper.deleteUnusedMaps();
|
||||
log.info("Start updating fids");
|
||||
try {
|
||||
|
|
@ -89,7 +111,17 @@ public class UpdatingFidService {
|
|||
return;
|
||||
}
|
||||
|
||||
log.info("Updating LP fids successfully completed.");
|
||||
List<UserPackage> packagesRepeated = userPackageMapper.findForRepeatedPackages();
|
||||
packagesRepeated.forEach(userPackage -> {
|
||||
Long packageId = userPackage.getId();
|
||||
SetupParam setupParam = new SetupParam(packageId, 2, PipelineMode.RESEND);
|
||||
try {
|
||||
Pipeline setupPipeline = lpPipelinePackageFabric.build(setupParam);
|
||||
lpPipelinePackageOrchestrator.startOrDelay(setupPipeline, packageId);
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to build pipeline", e);
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
log.info("pipeline finished");
|
||||
pipeline.close();
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
|
|||
import org.springframework.context.annotation.DependsOn;
|
||||
import org.springframework.kafka.annotation.KafkaListener;
|
||||
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;
|
||||
|
|
@ -24,21 +23,20 @@ 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.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@Slf4j
|
||||
@Service
|
||||
@DependsOn(value = "signalKafkaEventWriterService")
|
||||
public class KafkaMonitoringService {
|
||||
|
||||
private final ExecutorService processingExecutor;
|
||||
private final ExecutorService dbExecutor;
|
||||
|
||||
private volatile boolean isShuttingdDown = false;
|
||||
private final int permits;
|
||||
private final Semaphore processingPermits;
|
||||
|
|
@ -49,32 +47,20 @@ public class KafkaMonitoringService {
|
|||
private final SourceService sourceService;
|
||||
private final SegmentMapService segmentMapService;
|
||||
private final HitsService hitsService;
|
||||
private final ReportNameRegistry reportNameRegistry;
|
||||
|
||||
public KafkaMonitoringService(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
|
||||
PackageRecordsService packageRecordsService,
|
||||
SourceService sourceService,
|
||||
SegmentMapService segmentMapService,
|
||||
CreditTrackerSettings settings,
|
||||
HitsService hitsService,
|
||||
ReportNameRegistry reportNameRegistry) {
|
||||
HitsService hitsService) {
|
||||
this.signalEventsQueue = signalEventsQueue;
|
||||
this.packageRecordsService = packageRecordsService;
|
||||
this.sourceService = sourceService;
|
||||
this.segmentMapService = segmentMapService;
|
||||
this.reportNameRegistry = reportNameRegistry;
|
||||
this.permits = Integer.parseInt(settings.getKafkaConsumerSettings().concurency()); // количество равно concurency листенера
|
||||
this.hitsService = hitsService;
|
||||
this.processingPermits = new Semaphore(permits);
|
||||
|
||||
this.processingExecutor = Executors.newFixedThreadPool(
|
||||
permits * 2,
|
||||
new CustomizableThreadFactory("kafka-processor-")
|
||||
);
|
||||
this.dbExecutor = Executors.newFixedThreadPool(
|
||||
permits,
|
||||
new CustomizableThreadFactory("kafka-db-writer-")
|
||||
);
|
||||
}
|
||||
|
||||
@KafkaListener(
|
||||
|
|
@ -104,6 +90,8 @@ public class KafkaMonitoringService {
|
|||
}
|
||||
|
||||
private void process(List<PackageMessage> messages) {
|
||||
List<Hit> hits = new ArrayList<>();
|
||||
|
||||
Map<Boolean, List<PackageMessage>> partition = messages.stream()
|
||||
.filter(this::filterMessages)
|
||||
.collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));
|
||||
|
|
@ -113,43 +101,17 @@ public class KafkaMonitoringService {
|
|||
|
||||
Map<String, Integer> segmentMap = segmentMapService.loadSegmentMap();
|
||||
|
||||
CompletableFuture<List<Hit>> npFuture = CompletableFuture.supplyAsync(() -> {
|
||||
if (npMessages != null && !npMessages.isEmpty()) {
|
||||
return processNpMessages(npMessages, segmentMap);
|
||||
List<Hit> npHits = processNpMessages(npMessages, segmentMap);
|
||||
hits.addAll(npHits);
|
||||
}
|
||||
return Collections.emptyList();
|
||||
}, processingExecutor);
|
||||
|
||||
CompletableFuture<List<Hit>> lpFuture = CompletableFuture.supplyAsync(() -> {
|
||||
if (lpMessages != null && !lpMessages.isEmpty()) {
|
||||
return processLpMessages(lpMessages, segmentMap);
|
||||
List<Hit> lpHits = processLpMessages(lpMessages, segmentMap);
|
||||
hits.addAll(lpHits);
|
||||
}
|
||||
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) {
|
||||
|
|
@ -228,17 +190,12 @@ public class KafkaMonitoringService {
|
|||
Integer segment = segmentMap.get(memberCode.substring(0, 6));
|
||||
event.setSegment(segment != null ? segment : -1);
|
||||
|
||||
String reportKey = userPackagePhoneData.getUserId().substring(0, 6)
|
||||
+ "_" + userPackagePhoneData.getPackageName();
|
||||
String reportFileName = resolveNameAndEnqueue(reportKey, event);
|
||||
if (reportFileName == null) {
|
||||
continue; // событие не встало в очередь — hit не пишем
|
||||
}
|
||||
putToQueue(event);
|
||||
Hit hit = Hit.builder()
|
||||
.packageId(userPackagePhoneData.getPackageId())
|
||||
.subjectId(subjectId)
|
||||
.signalType(signalType)
|
||||
.reportFileName(reportFileName.replace(".xml", ".zip"))
|
||||
.reportFileName("") // TODO
|
||||
.inqT(message.getInquiryType())
|
||||
.busCategory(busCategory)
|
||||
.inqDt(message.getInquiryDate())
|
||||
|
|
@ -333,17 +290,12 @@ public class KafkaMonitoringService {
|
|||
Integer segment = segmentMap.get(memberCode.substring(0, 6));
|
||||
event.setSegment(segment != null ? segment : -1);
|
||||
|
||||
String reportKey = userPackageLegalData.getUserId().substring(0, 6)
|
||||
+ "_" + userPackageLegalData.getPackageName();
|
||||
String reportFileName = resolveNameAndEnqueue(reportKey, event);
|
||||
if (reportFileName == null) {
|
||||
continue; // событие не встало в очередь — hit не пишем
|
||||
}
|
||||
putToQueue(event);
|
||||
Hit hit = Hit.builder()
|
||||
.packageId(userPackageLegalData.getPackageId())
|
||||
.subjectId(subjectId)
|
||||
.signalType(signalType)
|
||||
.reportFileName(reportFileName.replace(".xml", ".zip"))
|
||||
.reportFileName("") // TODO
|
||||
.inqT(message.getInquiryType())
|
||||
.busCategory(busCategory)
|
||||
.inqDt(message.getInquiryDate())
|
||||
|
|
@ -370,18 +322,12 @@ public class KafkaMonitoringService {
|
|||
(inquiryType == 3 && billingCode.equalsIgnoreCase("CHSP"));
|
||||
}
|
||||
|
||||
/**
|
||||
* Резолвит имя результирующего файла и ставит событие в очередь.
|
||||
*
|
||||
* @return имя xml-файла или null, если поток был прерван и событие не встало в очередь
|
||||
*/
|
||||
private String resolveNameAndEnqueue(String reportKey, SignalKafkaEvent event) {
|
||||
private void putToQueue(SignalKafkaEvent event) {
|
||||
try {
|
||||
return reportNameRegistry.resolveNameAndEnqueue(reportKey, event);
|
||||
signalEventsQueue.put(event);
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Error while putting signal event", e);
|
||||
Thread.currentThread().interrupt();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -393,7 +339,9 @@ public class KafkaMonitoringService {
|
|||
processingPermits.acquire(permits);
|
||||
|
||||
try {
|
||||
reportNameRegistry.enqueueRotate();
|
||||
SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
|
||||
rotateEvent.setEventType(EventType.ROTATE);
|
||||
signalEventsQueue.put(rotateEvent);
|
||||
log.info("ROTATE event sent to queue");
|
||||
|
||||
SignalKafkaEvent stopEvent = new SignalKafkaEvent();
|
||||
|
|
|
|||
|
|
@ -1,72 +0,0 @@
|
|||
package ru.nbch.credit_tracker.service.mode.monitoring;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.nbch.credit_tracker.enums.EventType;
|
||||
import ru.nbch.credit_tracker.model.SignalKafkaEvent;
|
||||
import ru.nbch.credit_tracker.utils.CTUtil;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.HashMap;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.locks.ReadWriteLock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
|
||||
/**
|
||||
* Реестр имён результирующих файлов текущего цикла ротации.
|
||||
* Имя фиксируется при первом сигнале цикла для пакета (timestamp — момент первого сигнала).
|
||||
* <p>
|
||||
* событие, получившее имя эпохи N, встаёт в очередь раньше, чем ROTATE,
|
||||
* закрывающий эпоху N. Поэтому WRITE-события кладутся в очередь только через
|
||||
* {@link #resolveNameAndEnqueue}, а ROTATE — только через {@link #enqueueRotate}.
|
||||
*/
|
||||
@Component
|
||||
public class ReportNameRegistry {
|
||||
|
||||
// ключ: membercode_packageName; значение: финальное имя xml-файла
|
||||
private final ConcurrentHashMap<String, String> names = new ConcurrentHashMap<>();
|
||||
private final ReadWriteLock lock = new ReentrantReadWriteLock();
|
||||
|
||||
private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
|
||||
|
||||
public ReportNameRegistry(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue) {
|
||||
this.signalEventsQueue = signalEventsQueue;
|
||||
}
|
||||
|
||||
/**
|
||||
* Резолвит имя файла для пакета и атомарно с этим ставит WRITE-событие в очередь.
|
||||
*
|
||||
* @param key membercode_packageName
|
||||
* @return имя xml-файла, в который попадёт сигнал
|
||||
*/
|
||||
public String resolveNameAndEnqueue(String key, SignalKafkaEvent event) throws InterruptedException {
|
||||
lock.readLock().lock();
|
||||
try {
|
||||
String name = names.computeIfAbsent(key, k ->
|
||||
String.format("%s_%s.xml", k,
|
||||
LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter)));
|
||||
signalEventsQueue.put(event);
|
||||
return name;
|
||||
} finally {
|
||||
lock.readLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Закрывает текущую эпоху: снапшот имён уезжает внутри ROTATE-события,
|
||||
* реестр очищается атомарно с постановкой события в очередь.
|
||||
*/
|
||||
public void enqueueRotate() throws InterruptedException {
|
||||
lock.writeLock().lock();
|
||||
try {
|
||||
SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
|
||||
rotateEvent.setEventType(EventType.ROTATE);
|
||||
rotateEvent.setReportNames(new HashMap<>(names));
|
||||
names.clear();
|
||||
signalEventsQueue.put(rotateEvent);
|
||||
} finally {
|
||||
lock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -3,10 +3,14 @@ package ru.nbch.credit_tracker.service.mode.monitoring;
|
|||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.enums.EventType;
|
||||
import ru.nbch.credit_tracker.model.SignalKafkaEvent;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
|
@ -19,12 +23,12 @@ public class RotateSheduler {
|
|||
);
|
||||
private volatile boolean isShuttingdDown = false;
|
||||
|
||||
private final ReportNameRegistry reportNameRegistry;
|
||||
private final BlockingQueue<SignalKafkaEvent> signalEventsQueue;
|
||||
private final CreditTrackerSettings config;
|
||||
|
||||
public RotateSheduler(ReportNameRegistry reportNameRegistry,
|
||||
public RotateSheduler(@Qualifier("signalKafkaEventQueue") BlockingQueue<SignalKafkaEvent> signalEventsQueue,
|
||||
CreditTrackerSettings config) {
|
||||
this.reportNameRegistry = reportNameRegistry;
|
||||
this.signalEventsQueue = signalEventsQueue;
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
|
|
@ -40,7 +44,9 @@ public class RotateSheduler {
|
|||
return;
|
||||
}
|
||||
try {
|
||||
reportNameRegistry.enqueueRotate();
|
||||
SignalKafkaEvent rotateEvent = new SignalKafkaEvent();
|
||||
rotateEvent.setEventType(EventType.ROTATE);
|
||||
signalEventsQueue.put(rotateEvent);
|
||||
} catch (InterruptedException e) {
|
||||
log.error("Error while putting signal event", e);
|
||||
Thread.currentThread().interrupt();
|
||||
|
|
|
|||
|
|
@ -34,7 +34,6 @@ import java.time.format.DateTimeFormatter;
|
|||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
|
@ -86,7 +85,7 @@ public class SignalKafkaEventWriterService {
|
|||
}
|
||||
|
||||
if (event.getEventType() == EventType.ROTATE) {
|
||||
rotate(event.getReportNames());
|
||||
rotate();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -135,16 +134,14 @@ public class SignalKafkaEventWriterService {
|
|||
}
|
||||
|
||||
/**
|
||||
* Упаковывает все результирующие xml в архив.
|
||||
*
|
||||
* @param reportNames снапшот имён файлов закрываемой эпохи (ключ: membercode_packageName)
|
||||
* Упаковывает все результирующие xml в архив
|
||||
*/
|
||||
private void rotate(Map<String, String> reportNames) {
|
||||
private void rotate() {
|
||||
List<UserPackage> packagesToUpdate = new ArrayList<>();
|
||||
List<UserPackage> userPackages = userPackageMapper.findAllPackagesByStatus(Status.ON_MONITORING.name());
|
||||
for (UserPackage userPackage : userPackages) {
|
||||
try {
|
||||
String reportName = movedAndZip(userPackage, reportNames);
|
||||
String reportName = movedAndZip(userPackage);
|
||||
userPackage.setLastDownloadedReport(reportName);
|
||||
packagesToUpdate.add(userPackage);
|
||||
} catch (IOException e) {
|
||||
|
|
@ -170,7 +167,7 @@ public class SignalKafkaEventWriterService {
|
|||
}
|
||||
}
|
||||
|
||||
private String movedAndZip(UserPackage userPackage, Map<String, String> reportNames) throws IOException {
|
||||
private String movedAndZip(UserPackage userPackage) throws IOException {
|
||||
String finalName = "";
|
||||
String commonMembercode = userPackage.getUserId().substring(0, 6);
|
||||
Path tmpFilePath = Path.of(config.getLoadDir(),
|
||||
|
|
@ -187,17 +184,11 @@ public class SignalKafkaEventWriterService {
|
|||
if (!Files.exists(rootReportPath)) {
|
||||
Files.createDirectories(rootReportPath);
|
||||
}
|
||||
|
||||
String reportKey = commonMembercode + "_" + userPackage.getPackageName();
|
||||
finalName = reportNames != null ? reportNames.get(reportKey) : null;
|
||||
if (finalName == null) {
|
||||
// tmp-файл есть, а имени в реестре нет —
|
||||
// генерируем имя по времени ротации
|
||||
LocalDateTime newReportDate = LocalDateTime.now();
|
||||
finalName = String.format("%s_%s_%s.xml",
|
||||
commonMembercode,
|
||||
userPackage.getPackageName(),
|
||||
LocalDateTime.now().format(CTUtil.responseReportDateTimeFormatter));
|
||||
}
|
||||
newReportDate.format(CTUtil.responseReportDateTimeFormatter));
|
||||
|
||||
Path responsePath = rootReportPath.resolve(finalName);
|
||||
|
||||
|
|
|
|||
|
|
@ -56,13 +56,12 @@ public class PackageRecordsService {
|
|||
|
||||
public Map<Pair<Long, Long>, LegalsMap> findCachedLegalsByInnAndRegNums(List<SubjectProfile> subjects) {
|
||||
try {
|
||||
String json = entityToJsonMapper.writeValueAsString(subjects.stream()
|
||||
.filter(Objects::nonNull)
|
||||
.map(s -> new HashMap<String, Object>() {{
|
||||
put("inn", s.getInn());
|
||||
put("reg_num", s.getRegNum());
|
||||
}}).toList());
|
||||
return legalsMapMapper.findCachedByInnAndRegNum(json)
|
||||
List<InnRegnumPair> innAndRegNums = subjects.stream()
|
||||
.map(s -> new InnRegnumPair(s.getInn(), s.getRegNum()))
|
||||
.toList();
|
||||
String json = entityToJsonMapper.writeValueAsString(innAndRegNums);
|
||||
|
||||
return legalsMapMapper.findCachedByInnAndRegNUm(json)
|
||||
.stream()
|
||||
.collect(Collectors.toMap(
|
||||
lm -> Pair.of(lm.getInn(), lm.getRegNum()),
|
||||
|
|
@ -110,6 +109,33 @@ public class PackageRecordsService {
|
|||
}
|
||||
}
|
||||
|
||||
public List<Long> selectNpUniquePackageIds(List<Long> phones) {
|
||||
try {
|
||||
String json = entityToJsonMapper.writeValueAsString(phones.stream()
|
||||
.filter(Objects::nonNull)
|
||||
.map(phone -> new HashMap<String, Object>() {{
|
||||
put("phone", phone);
|
||||
}}).toList());
|
||||
return packagesRecordsMapper.uniquePackageIdByPhones(json);
|
||||
} catch (JsonProcessingException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
public List<Long> selectLpUniquePackageIds(List<LegalsMap> legalsMaps) {
|
||||
try {
|
||||
List<InnRegnumPair> pairs = legalsMaps.stream()
|
||||
.filter(Objects::nonNull)
|
||||
.map(legalMap -> new InnRegnumPair(legalMap.getInn(), legalMap.getRegNum()))
|
||||
.toList();
|
||||
|
||||
String json = entityToJsonMapper.writeValueAsString(pairs);
|
||||
return packagesRecordsMapper.uniquePackageIdByInnAndRegNum(json);
|
||||
} catch (JsonProcessingException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
public List<UserPackagePhoneData> selectUserPackagePhoneData(Set<Long> phones) {
|
||||
try {
|
||||
String json = entityToJsonMapper.writeValueAsString(phones.stream()
|
||||
|
|
@ -198,13 +224,11 @@ public class PackageRecordsService {
|
|||
|
||||
public List<UserPackageLegalData> selectUserPackageLegalData(Set<Pair<Long, Long>> uniqueInnRegnumPairs) {
|
||||
try {
|
||||
String json = entityToJsonMapper.writeValueAsString(uniqueInnRegnumPairs.stream()
|
||||
.filter(Objects::nonNull)
|
||||
.map(s -> new HashMap<String, Object>() {{
|
||||
put("inn", s.getKey());
|
||||
put("reg_num", s.getValue());
|
||||
}}).toList());
|
||||
List<InnRegnumPair> pairs = uniqueInnRegnumPairs.stream()
|
||||
.map(pair -> new InnRegnumPair(pair.getKey(), pair.getValue()))
|
||||
.toList();
|
||||
|
||||
String json = entityToJsonMapper.writeValueAsString(pairs);
|
||||
return packagesRecordsMapper.selectUserPackageLegalData(json);
|
||||
} catch (JsonProcessingException e) {
|
||||
throw new RuntimeException(e);
|
||||
|
|
@ -225,4 +249,6 @@ public class PackageRecordsService {
|
|||
}
|
||||
}
|
||||
|
||||
public record InnRegnumPair(Long inn, Long regNum) {
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -47,12 +47,11 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
|
|||
log.debug("LpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
|
||||
//найдем inn + regNum в табличном кэше
|
||||
Map<Pair<Long, Long>, LegalsMap> cachedRecords = packageRecordsService.findCachedLegalsByInnAndRegNums(subjects);
|
||||
boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
|
||||
taskPayload.setHasFids(hasFids);
|
||||
processByNumber(subjects, cachedRecords, ctx);
|
||||
return taskPayload;
|
||||
}
|
||||
|
||||
private boolean processByNumber(List<SubjectProfile> subjects,
|
||||
private void processByNumber(List<SubjectProfile> subjects,
|
||||
Map<Pair<Long, Long>, LegalsMap> cachedRecords,
|
||||
PipelineContext<PackageVariable> context) {
|
||||
List<SubjectProfile> noFidsData = subjects.stream()
|
||||
|
|
@ -126,10 +125,7 @@ public class LpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
|
|||
.filter(legalsMap -> seen.add(Pair.of(legalsMap.getInn(), legalsMap.getRegNum())))
|
||||
.toList();
|
||||
|
||||
|
||||
packageRecordsService.saveLegalsToCache(subjectForSave);
|
||||
|
||||
return subjectForSave.stream().anyMatch(s -> s.getFid() != null) || (noFidsData.isEmpty() && !cachedRecords.isEmpty());
|
||||
}
|
||||
|
||||
private void fillByCache(SubjectProfile subjectProfile, LegalsMap cachedRecord) {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
package ru.nbch.credit_tracker.service.task;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Service;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.config.settings.PipelineSettings;
|
||||
|
|
@ -10,12 +9,14 @@ import java.util.concurrent.ExecutorService;
|
|||
|
||||
@Service
|
||||
public class LpPipelinePackageOrchestrator extends PackageOrchestrator {
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
|
||||
public LpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
|
||||
UserPackageMapper userPackageMapper,
|
||||
@Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService,
|
||||
@Qualifier("lpSaverExecutorService") ExecutorService lpSaverExecutorService) {
|
||||
super(creditTrackerSettings, userPackageMapper, lpPipelineExecutorService, lpSaverExecutorService);
|
||||
UserPackageMapper userPackageMapper, // TODO
|
||||
ExecutorService pipelineExecutorService, // TODO
|
||||
ExecutorService saverExecutorService) { // TODO
|
||||
super(userPackageMapper, pipelineExecutorService, saverExecutorService);
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
package ru.nbch.credit_tracker.service.task;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Service;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.config.settings.PipelineSettings;
|
||||
|
|
@ -10,9 +9,12 @@ import java.util.concurrent.ExecutorService;
|
|||
@Service
|
||||
public class LpPipelineRecalcOrchestrator extends RecalcOrchestrator{
|
||||
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
|
||||
public LpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
|
||||
@Qualifier("lpPipelineExecutorService") ExecutorService lpPipelineExecutorService) {
|
||||
super(creditTrackerSettings, lpPipelineExecutorService);
|
||||
ExecutorService pipelineExecutorService) { // TODO another executor
|
||||
super(pipelineExecutorService);
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
package ru.nbch.credit_tracker.service.task;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Service;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.config.settings.PipelineSettings;
|
||||
|
|
@ -10,12 +9,14 @@ import java.util.concurrent.ExecutorService;
|
|||
|
||||
@Service
|
||||
public class NpPipelinePackageOrchestrator extends PackageOrchestrator {
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
|
||||
public NpPipelinePackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
|
||||
UserPackageMapper userPackageMapper,
|
||||
@Qualifier("npPipelineExecutorService") ExecutorService npPipelineExecutorService,
|
||||
@Qualifier("npSaverExecutorService") ExecutorService npSaverExecutorService) {
|
||||
super(creditTrackerSettings, userPackageMapper, npPipelineExecutorService, npSaverExecutorService);
|
||||
ExecutorService pipelineExecutorService,
|
||||
ExecutorService saverExecutorService) {
|
||||
super(userPackageMapper, pipelineExecutorService, saverExecutorService);
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
package ru.nbch.credit_tracker.service.task;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Service;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.config.settings.PipelineSettings;
|
||||
|
|
@ -10,9 +9,11 @@ import java.util.concurrent.ExecutorService;
|
|||
@Service
|
||||
public class NpPipelineRecalcOrchestrator extends RecalcOrchestrator {
|
||||
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
public NpPipelineRecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
|
||||
@Qualifier("npPipelineExecutorService")ExecutorService npPipelineExecutorService) {
|
||||
super(creditTrackerSettings, npPipelineExecutorService);
|
||||
ExecutorService pipelineExecutorService) {
|
||||
super(pipelineExecutorService);
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.service.task;
|
|||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.config.settings.PipelineSettings;
|
||||
import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper;
|
||||
import ru.nbch.credit_tracker.model.impl.PackageVariable;
|
||||
|
|
@ -21,7 +20,6 @@ import java.util.concurrent.atomic.AtomicInteger;
|
|||
|
||||
public abstract class PackageOrchestrator {
|
||||
private final Logger log = LoggerFactory.getLogger(getClass());
|
||||
protected final CreditTrackerSettings creditTrackerSettings;
|
||||
private final UserPackageMapper userPackageMapper;
|
||||
private final ExecutorService pipelineExecutorService;
|
||||
private final ExecutorService saverExecutorService;
|
||||
|
|
@ -29,11 +27,9 @@ public abstract class PackageOrchestrator {
|
|||
private final java.util.Deque<PackageOrchestrator.Launch> pending = new java.util.ArrayDeque<>();
|
||||
private final Semaphore semaphore;
|
||||
|
||||
public PackageOrchestrator(CreditTrackerSettings creditTrackerSettings,
|
||||
UserPackageMapper userPackageMapper,
|
||||
public PackageOrchestrator(UserPackageMapper userPackageMapper,
|
||||
ExecutorService pipelineExecutorService,
|
||||
ExecutorService saverExecutorService) {
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
this.userPackageMapper = userPackageMapper;
|
||||
this.pipelineExecutorService = pipelineExecutorService;
|
||||
this.saverExecutorService = saverExecutorService;
|
||||
|
|
@ -112,6 +108,8 @@ public abstract class PackageOrchestrator {
|
|||
}
|
||||
if (!res.hasFids()) {
|
||||
log.info("No one fid was found by packageId : {}.", packageId);
|
||||
userPackageMapper.updateNoFidMonitoring(packageId);
|
||||
return;
|
||||
}
|
||||
|
||||
PipelineMode mode = ctx.get(PackageVariable.mode);
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ package ru.nbch.credit_tracker.service.task;
|
|||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings;
|
||||
import ru.nbch.credit_tracker.config.settings.PipelineSettings;
|
||||
import ru.nbch.credit_tracker.utils.error.ExceptionUtils;
|
||||
import ru.nbch.pipeline.engine.core.Pipeline;
|
||||
|
|
@ -14,14 +13,11 @@ import java.util.concurrent.TimeUnit;
|
|||
|
||||
public abstract class RecalcOrchestrator {
|
||||
private final Logger log = LoggerFactory.getLogger(getClass());
|
||||
protected final CreditTrackerSettings creditTrackerSettings;
|
||||
private final ExecutorService pipelineExecutorService;
|
||||
private final Object lock = new Object();
|
||||
private final Semaphore semaphore;
|
||||
|
||||
public RecalcOrchestrator(CreditTrackerSettings creditTrackerSettings,
|
||||
ExecutorService pipelineExecutorService) {
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
public RecalcOrchestrator(ExecutorService pipelineExecutorService) {
|
||||
this.pipelineExecutorService = pipelineExecutorService;
|
||||
this.semaphore = new Semaphore(1);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -107,6 +107,12 @@ public class LpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
|
|||
legalsMapService.updateSearchedAt(notChangingFidLegalMaps);
|
||||
}
|
||||
|
||||
if (!changingFids.isEmpty()) {
|
||||
List<Long> packageIds = packageRecordsService.selectLpUniquePackageIds(changingFids);
|
||||
if (!packageIds.isEmpty()) {
|
||||
userPackageService.setFidChanged(packageIds);
|
||||
}
|
||||
}
|
||||
return new VoidPayload();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -46,12 +46,11 @@ public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
|
|||
log.debug("NpFidEnricherStage handle by packageId: {}, size of bucket for save: {}", packageId, subjects.size());
|
||||
//найдем телефоны в табличном кэше
|
||||
Map<Long, PhonesMap> cachedRecords = packageRecordsService.findCachedPhones(subjects);
|
||||
boolean hasFids = processByNumber(subjects, cachedRecords, ctx);
|
||||
taskPayload.setHasFids(hasFids);
|
||||
processByNumber(subjects, cachedRecords, ctx);
|
||||
return taskPayload;
|
||||
}
|
||||
|
||||
private boolean processByNumber(List<SubjectProfile> subjects,
|
||||
private void processByNumber(List<SubjectProfile> subjects,
|
||||
Map<Long, PhonesMap> cachedRecords,
|
||||
PipelineContext<PackageVariable> context) {
|
||||
List<SubjectProfile> noFidsData = subjects.stream()
|
||||
|
|
@ -98,8 +97,6 @@ public class NpFidEnricher implements Handler<SubjectPayload, SubjectPayload> {
|
|||
.toList();
|
||||
|
||||
packageRecordsService.saveToCache(subjectForSave);
|
||||
|
||||
return subjectForSave.stream().anyMatch(s -> s.getFid() != null) || (noFidsData.isEmpty() && !cachedRecords.isEmpty());
|
||||
}
|
||||
|
||||
private void fillByCache(SubjectProfile subjectProfile, PhonesMap cachedRecord) {
|
||||
|
|
|
|||
|
|
@ -8,7 +8,9 @@ import ru.nbch.credit_tracker.model.SubjectProfile;
|
|||
import ru.nbch.credit_tracker.model.impl.PackageVariable;
|
||||
import ru.nbch.credit_tracker.model.impl.SubjectPayload;
|
||||
import ru.nbch.credit_tracker.model.impl.VoidPayload;
|
||||
import ru.nbch.credit_tracker.service.UserPackageService;
|
||||
import ru.nbch.credit_tracker.service.indic.PersonService;
|
||||
import ru.nbch.credit_tracker.service.signal.PackageRecordsService;
|
||||
import ru.nbch.credit_tracker.service.signal.PhonesMapService;
|
||||
import ru.nbch.credit_tracker.utils.CTUtil;
|
||||
import ru.nbch.pipeline.engine.core.Handler;
|
||||
|
|
@ -23,14 +25,20 @@ import java.util.stream.Stream;
|
|||
public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
|
||||
private final PersonService personService;
|
||||
private final CreditTrackerSettings creditTrackerSettings;
|
||||
private final PackageRecordsService packageRecordsService;
|
||||
private final PhonesMapService phonesMapService;
|
||||
private final UserPackageService userPackageService;
|
||||
|
||||
public NpFidUpdater(PersonService personService,
|
||||
CreditTrackerSettings creditTrackerSettings,
|
||||
PhonesMapService phonesMapService) {
|
||||
PackageRecordsService packageRecordsService,
|
||||
PhonesMapService phonesMapService,
|
||||
UserPackageService userPackageService) {
|
||||
this.personService = personService;
|
||||
this.creditTrackerSettings = creditTrackerSettings;
|
||||
this.packageRecordsService = packageRecordsService;
|
||||
this.phonesMapService = phonesMapService;
|
||||
this.userPackageService = userPackageService;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -71,6 +79,12 @@ public class NpFidUpdater implements Handler<SubjectPayload, VoidPayload> {
|
|||
phonesMapService.updateSearchedAt(notChangedPhones);
|
||||
}
|
||||
|
||||
if (!changingFids.isEmpty()) {
|
||||
List<Long> packageIds = packageRecordsService.selectNpUniquePackageIds(changingFids);
|
||||
if (!packageIds.isEmpty()) {
|
||||
userPackageService.setFidChanged(packageIds);
|
||||
}
|
||||
}
|
||||
return new VoidPayload();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ public final class SetupMonitoringPipelineTerminator implements IPipeLineTermina
|
|||
continue;
|
||||
}
|
||||
if (!part.isPoison() && !hasFids.get()) {
|
||||
hasFids.set(part.isHasFids());
|
||||
hasFids.set(part.getSubjects().stream().anyMatch(s -> s.getFid() != null));
|
||||
}
|
||||
if (part.isPoison()) {
|
||||
log.debug("Got poison");
|
||||
|
|
|
|||
|
|
@ -49,10 +49,4 @@
|
|||
LIMIT #{limit} OFFSET #{offset}
|
||||
</select>
|
||||
|
||||
<delete id="deleteByPackageId" parameterType="long">
|
||||
DELETE
|
||||
FROM signals_legal_fid_map
|
||||
WHERE package_id = #{packageId}
|
||||
</delete>
|
||||
|
||||
</mapper>
|
||||
|
|
@ -8,7 +8,7 @@
|
|||
<resultMap id="LegalsMap" type="ru.nbch.credit_tracker.entities.app.LegalsMap">
|
||||
<result property="id" column="id"/>
|
||||
<result property="inn" column="inn"/>
|
||||
<result property="regNum" column="reg_num"/>
|
||||
<result property="regNUm" column="reg_num"/>
|
||||
<result property="fid" column="fid"/>
|
||||
<result property="searchedAt" column="searched_at"/>
|
||||
<result property="flag90days" column="flag_90_12"/>
|
||||
|
|
@ -58,11 +58,11 @@
|
|||
</foreach>
|
||||
</select>
|
||||
|
||||
<select id="findCachedByInnAndRegNum" resultMap="LegalsMap">
|
||||
WITH src AS (SELECT inn, reg_num
|
||||
<select id="findCachedByInnAndRegNUm" resultMap="LegalsMap">
|
||||
WITH src AS (SELECT inn, regNum
|
||||
FROM json_to_recordset(#{json}::json) AS t(
|
||||
inn bigint,
|
||||
reg_num bigint
|
||||
regNum bigint
|
||||
))
|
||||
SELECT slm.id AS id,
|
||||
slm.fid as fid,
|
||||
|
|
@ -73,7 +73,7 @@
|
|||
slm.calculated_at AS calculated_at,
|
||||
slm.last_rejected_date AS last_rejected_date
|
||||
FROM signals_legals_map slm
|
||||
WHERE (slm.inn, slm.reg_num) IN (SELECT inn, reg_num FROM src)
|
||||
WHERE (slm.inn, slm.reg_num) IN (SELECT inn, regNum FROM src)
|
||||
</select>
|
||||
|
||||
<select id="findEligibleForFlagsUpdate" resultMap="LegalsMap">
|
||||
|
|
@ -231,7 +231,8 @@
|
|||
AND sup.status IN
|
||||
(
|
||||
'ON_MONITORING',
|
||||
'PROCESSING'
|
||||
'PROCESSING',
|
||||
'PROCESSING_REPEATED'
|
||||
))
|
||||
</delete>
|
||||
|
||||
|
|
|
|||
|
|
@ -42,6 +42,30 @@
|
|||
</foreach>
|
||||
</select>
|
||||
|
||||
<select id="uniquePackageIdByPhones" resultType="java.lang.Long">
|
||||
WITH src AS (SELECT phone
|
||||
FROM json_to_recordset(#{json}::json) AS t(
|
||||
phone bigint
|
||||
))
|
||||
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'
|
||||
</select>
|
||||
|
||||
<select id="uniquePackageIdByInnAndRegNum" resultType="java.lang.Long">
|
||||
WITH src AS (SELECT inn, regNum
|
||||
FROM json_to_recordset(#{json}::json) AS t(
|
||||
inn bigint,
|
||||
regNum bigint
|
||||
))
|
||||
select distinct (package_id)
|
||||
FROM signals_legal_fid_map slfm
|
||||
JOIN signals_user_packages sup ON sup.id = slfm.package_id
|
||||
WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, regNum FROM src)
|
||||
AND sup.status = 'ON_MONITORING'
|
||||
</select>
|
||||
|
||||
<select id="selectUserPackagePhoneData" resultType="ru.nbch.credit_tracker.model.UserPackagePhoneData">
|
||||
WITH src AS (SELECT phone
|
||||
FROM json_to_recordset(#{json}::json) AS t(
|
||||
|
|
@ -54,17 +78,17 @@
|
|||
</select>
|
||||
|
||||
<select id="selectUserPackageLegalData" resultType="ru.nbch.credit_tracker.model.UserPackageLegalData">
|
||||
WITH src AS (SELECT inn, reg_num
|
||||
WITH src AS (SELECT inn, regNum
|
||||
FROM json_to_recordset(#{json}::json) AS t(
|
||||
inn bigint,
|
||||
reg_num bigint
|
||||
regNum bigint
|
||||
))
|
||||
SELECT sup.id AS packageId, sup.signal_types AS signalTypes, sup.membercode AS userId,
|
||||
sup.package_name AS packageName, slfm.inn AS inn, slfm.reg_num AS regNum,
|
||||
slfm.subject_id AS subjectId
|
||||
FROM signals_legal_fid_map slfm
|
||||
JOIN signals_user_packages sup ON sup.id = slfm.package_id
|
||||
WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, reg_num FROM src)
|
||||
WHERE (slfm.inn, slfm.reg_num) IN (SELECT inn, regNum FROM src)
|
||||
AND sup.status = 'ON_MONITORING'
|
||||
</select>
|
||||
|
||||
|
|
|
|||
|
|
@ -210,7 +210,8 @@
|
|||
AND sup.status IN
|
||||
(
|
||||
'ON_MONITORING',
|
||||
'PROCESSING'
|
||||
'PROCESSING',
|
||||
'PROCESSING_REPEATED'
|
||||
))
|
||||
</delete>
|
||||
</mapper>
|
||||
|
|
@ -14,7 +14,9 @@
|
|||
<result property="deactivatedAt" column="deactivated_at"/>
|
||||
<result property="lastDownloadedReport" column="last_downloaded_report"/>
|
||||
<result property="monitoringAt" column="monitoring_at"/>
|
||||
<result property="rejectStatus" column="reject_status"/>
|
||||
<result property="signalTypes" column="signal_types"/>
|
||||
<result property="externalPackageName" column="external_package_name"/>
|
||||
</resultMap>
|
||||
|
||||
<select id="isPackageNotUnique" resultType="boolean">
|
||||
|
|
@ -61,8 +63,8 @@
|
|||
|
||||
<select id="insertPackage" resultType="java.lang.Long"
|
||||
parameterType="ru.nbch.credit_tracker.entities.app.UserPackage">
|
||||
INSERT INTO signals_user_packages (membercode, package_name, subject_type, signal_types, created_at, status)
|
||||
values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.subjectType}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}) RETURNING id
|
||||
INSERT INTO signals_user_packages (membercode, package_name, subject_type, signal_types, created_at, status, external_package_name)
|
||||
values (#{userPackage.userId}, #{userPackage.packageName}, #{userPackage.subjectType}, #{userPackage.signalTypes}, #{userPackage.createdAt}, #{userPackage.status}, concat(substring(#{userPackage.userId}, 1, 6), '_', #{userPackage.packageName})) RETURNING id
|
||||
</select>
|
||||
|
||||
<update id="updatePackageStatus">
|
||||
|
|
@ -71,6 +73,16 @@
|
|||
WHERE id = #{packageId}
|
||||
</update>
|
||||
|
||||
<update id="setFidChanged">
|
||||
UPDATE signals_user_packages
|
||||
SET fid_changed = 1,
|
||||
status = 'PROCESSING_REPEATED'
|
||||
WHERE id in
|
||||
<foreach item="packageId" collection="packageIds" open="(" separator="," close=")">
|
||||
#{packageId}
|
||||
</foreach>
|
||||
</update>
|
||||
|
||||
<update id="updatePackageLastDownloadedReport" parameterType="string">
|
||||
WITH src AS (
|
||||
SELECT
|
||||
|
|
@ -93,7 +105,20 @@
|
|||
<if test="monitoringAt != null">
|
||||
monitoring_at = #{monitoringAt},
|
||||
</if>
|
||||
status = 'ON_MONITORING'
|
||||
status = 'ON_MONITORING',
|
||||
fid_changed = null
|
||||
WHERE id = #{id}
|
||||
</update>
|
||||
|
||||
<update id="updateNoFidMonitoring">
|
||||
UPDATE signals_user_packages
|
||||
SET monitoring_at = created_at, status = 'NO_FID', fid_changed = null
|
||||
WHERE id = #{id}
|
||||
</update>
|
||||
|
||||
<update id="setRejectedStatus">
|
||||
UPDATE signals_user_packages
|
||||
SET reject_status = #{status}
|
||||
WHERE id = #{id}
|
||||
</update>
|
||||
|
||||
|
|
@ -106,6 +131,7 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE id = #{id}
|
||||
|
|
@ -121,12 +147,28 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE package_name = #{packageName}
|
||||
LIMIT 1
|
||||
</select>
|
||||
|
||||
<select id="findForRepeatedPackages" resultMap="packageDataResultMap">
|
||||
SELECT id,
|
||||
membercode,
|
||||
package_name,
|
||||
created_at,
|
||||
status,
|
||||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE fid_changed = 1 and status = 'PROCESSING_REPEATED'
|
||||
</select>
|
||||
|
||||
<select id="findOnMonitoringPackages" resultMap="packageDataResultMap">
|
||||
SELECT id,
|
||||
membercode,
|
||||
|
|
@ -136,9 +178,10 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE status = 'ON_MONITORING'
|
||||
WHERE status = 'ON_MONITORING' and reject_status is null
|
||||
</select>
|
||||
|
||||
|
||||
|
|
@ -151,6 +194,7 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE membercode IS NOT NULL
|
||||
|
|
@ -170,6 +214,7 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE membercode IS NOT NULL
|
||||
|
|
@ -185,6 +230,7 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE status in
|
||||
|
|
@ -202,6 +248,7 @@
|
|||
deactivated_at,
|
||||
last_downloaded_report,
|
||||
monitoring_at,
|
||||
reject_status,
|
||||
signal_types
|
||||
FROM signals_user_packages
|
||||
WHERE LEFT(membercode, 6) = #{memberCode} AND package_name = #{packageName}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue