diff --git a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java index 16a7f19..548e154 100644 --- a/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java +++ b/src/main/java/ru/nbch/credit_tracker/CreditTrackerApplication.java @@ -11,6 +11,11 @@ public class CreditTrackerApplication { public static void main(String[] args) { ConfigurableApplicationContext ctx = SpringApplication.run(CreditTrackerApplication.class, args); +// if (args.length > 0) { +// String arg = args[0]; +// if (arg.equalsIgnoreCase("migration")) { +// } +// } StartupRestoreService updateProcessingPackagesService = ctx.getBean(StartupRestoreService.class); MonitoringService monitoringService = ctx.getBean(MonitoringService.class); updateProcessingPackagesService.run(); diff --git a/src/main/java/ru/nbch/credit_tracker/service/MigrationService.java b/src/main/java/ru/nbch/credit_tracker/service/MigrationService.java new file mode 100644 index 0000000..ebb39e3 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/MigrationService.java @@ -0,0 +1,61 @@ +package ru.nbch.credit_tracker.service; + +import java.util.concurrent.CompletableFuture; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.stereotype.Service; +import ru.nbch.credit_tracker.config.settings.CreditTrackerSettings; +import ru.nbch.credit_tracker.mapper.signal.UserPackageMapper; +import ru.nbch.credit_tracker.service.signal.PhonesMapService; +import ru.nbch.credit_tracker.service.task.Pipeline; +import ru.nbch.credit_tracker.service.task.PipelinePackageFabric; +import ru.nbch.credit_tracker.service.task.PipelinePackageOrchestrator; +import ru.nbch.credit_tracker.service.task.PipelineRecalcPDOrchestrator; +import ru.nbch.credit_tracker.service.task.SetupParam; +import ru.nbch.credit_tracker.service.task.impl.PipelineMode; + +@Slf4j +@Service +@EnableScheduling +public class MigrationService { + private final PipelinePackageFabric pipelinePackageManager; + private final PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator; + private final PipelinePackageOrchestrator pipelinePackageOrchestrator; + private final UserPackageMapper userPackageMapper; + private final PhonesMapService phonesMapService; + + public MigrationService(CreditTrackerSettings config, + PipelinePackageFabric pipelinePackageManager, + PipelineRecalcPDOrchestrator pipelineRecalcPDOrchestrator, + PipelinePackageOrchestrator pipelinePackageOrchestrator, + UserPackageMapper userPackageMapper, + PhonesMapService phonesMapService) { + this.pipelinePackageManager = pipelinePackageManager; + this.pipelineRecalcPDOrchestrator = pipelineRecalcPDOrchestrator; + this.pipelinePackageOrchestrator = pipelinePackageOrchestrator; + this.userPackageMapper = userPackageMapper; + this.phonesMapService = phonesMapService; + } + + public void update() { + log.info("Start updating fids and PD"); + try { + Pipeline pipeline = pipelinePackageManager.buildRecalculationPDMigration(new SetupParam(null, null, PipelineMode.RECALC_PERSONAL_DATA)); + CompletableFuture future = pipelineRecalcPDOrchestrator.startOrReject(pipeline); + future.whenComplete((o, throwable) -> { + try { + if (throwable != null) { + log.error(throwable.getMessage(), throwable); + } + } 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/task/PipelinePackageFabric.java b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java index 9abc35f..ff9c8ab 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java +++ b/src/main/java/ru/nbch/credit_tracker/service/task/PipelinePackageFabric.java @@ -18,6 +18,7 @@ import ru.nbch.credit_tracker.service.task.impl.FidIdEnricher; import ru.nbch.credit_tracker.service.task.impl.FidIdUpdater; import ru.nbch.credit_tracker.service.task.impl.FidNameEnricher; import ru.nbch.credit_tracker.service.task.impl.FidNameUpdater; +import ru.nbch.credit_tracker.service.task.impl.FidNameUpdaterMigration; import ru.nbch.credit_tracker.service.task.impl.FlagsEnricher; import ru.nbch.credit_tracker.service.task.setup.IPipeLineStarter; import ru.nbch.credit_tracker.service.task.setup.IPipeLineTerminator; @@ -37,6 +38,7 @@ public class PipelinePackageFabric implements InitializingBean { private final FidIdUpdater fidIdUpdater; private final FidNameEnricher fidNameEnricher; private final FidNameUpdater fidNameUpdater; + private final FidNameUpdaterMigration fidNameUpdaterMigration; private final Supplier> pipelineTerminatorFactory; private final Supplier> noOpTerminatorFactory; private final Function> pipelineStarterFactory; @@ -52,6 +54,7 @@ public class PipelinePackageFabric implements InitializingBean { FidIdEnricher fidIdEnricherHandler, FidIdUpdater fidIdUpdater, FidNameUpdater fidNameUpdater, + FidNameUpdaterMigration fidNameUpdaterMigration, Supplier> pipelineTerminatorFactory, Supplier> noOpTerminatorFactory, Function> pipelineStarterFactory, @@ -68,6 +71,7 @@ public class PipelinePackageFabric implements InitializingBean { this.fidIdEnricherHandler = fidIdEnricherHandler; this.fidIdUpdater = fidIdUpdater; this.fidNameUpdater = fidNameUpdater; + this.fidNameUpdaterMigration = fidNameUpdaterMigration; this.pipelineTerminatorFactory = pipelineTerminatorFactory; this.noOpTerminatorFactory = noOpTerminatorFactory; this.pipelineStarterFactory = pipelineStarterFactory; @@ -105,6 +109,20 @@ public class PipelinePackageFabric implements InitializingBean { .build(); } + public Pipeline buildRecalculationPDMigration(SetupParam setupParam) throws Exception { + return PipeLineBuilderV2.first(nameUpdaterQueueFactory.getObject(), idUpdaterQueueFactory.getObject()) + .handler(fidNameUpdaterMigration) + .andThen() + .stage(noOpQueueFactory.getObject()) + .handler(fidIdUpdater) + .andThen() + .fillInitialQueue(pipelineStarterFactory.apply(setupParam)) + .pollFinishingQueue(noOpTerminatorFactory.get()) + .backoff(pipelineBackoff) + .context(PackageVariable.class) + .build(); + } + @Override public void afterPropertiesSet() throws Exception { log.info("pipeline settings: {}", creditTrackerSettings.getPipeline()); diff --git a/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java new file mode 100644 index 0000000..4a51923 --- /dev/null +++ b/src/main/java/ru/nbch/credit_tracker/service/task/impl/FidNameUpdaterMigration.java @@ -0,0 +1,87 @@ +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; +import ru.nbch.credit_tracker.entities.indic.Name; +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.service.indic.PersonServiceV2; +import ru.nbch.credit_tracker.service.signal.PhonesMapService; +import ru.nbch.credit_tracker.service.task.Handler; +import ru.nbch.credit_tracker.utils.CTUtil; + +@Service +@Slf4j +public class FidNameUpdaterMigration implements Handler { + private final PersonServiceV2 personService; + private final CreditTrackerSettings creditTrackerSettings; + private final PhonesMapService phonesMapService; + + public FidNameUpdaterMigration(PersonServiceV2 personService, + CreditTrackerSettings creditTrackerSettings, + PhonesMapService phonesMapService) { + this.personService = personService; + this.creditTrackerSettings = creditTrackerSettings; + this.phonesMapService = phonesMapService; + } + + @Override + public SubjectPayload handle(SubjectPayload taskPayload) { + List subjects = taskPayload.getSubjects(); + var ctx = obtainContext(PackageVariable.class, taskPayload); + log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size()); + Map nameByPhone = personService.findNameByNumber(subjects); + + List changingFidPhones = new ArrayList<>(); + List notFoundFidPhones = new ArrayList<>(); + List notChangedPhones = subjects.stream().flatMap(subjectProfile -> { + Name name = nameByPhone.get(CTUtil.normalizePhoneNumber(subjectProfile.getPhone())); + 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()); + return Stream.empty(); + } + return Stream.of(subjectProfile.getPhone()); + }).toList(); + List changingFids = new ArrayList<>(); + if (!notFoundFidPhones.isEmpty()) { + phonesMapService.resetFidAndPDByPhones(notFoundFidPhones); + changingFids.addAll(notFoundFidPhones); + } + + if (!changingFidPhones.isEmpty()) { + phonesMapService.updateNameCachedData(changingFidPhones); + changingFids.addAll(changingFidPhones.stream().map(SubjectProfile::getPhone).toList()); + } + + if (!notChangedPhones.isEmpty()) { + phonesMapService.updateSearchedAt(notChangedPhones); + } + return new SubjectPayload(null, null, null, changingFidPhones); + } + + @Override + public int preferredConcurrency() { + return creditTrackerSettings.getPipeline().nameEnricherStageConcurrency(); + } + + @Override + public SubjectPayload cookPoison() { + return SubjectPayload.POISON; + } +} diff --git a/src/main/resources/sql/credit_tracker_db.ddl b/src/main/resources/sql/credit_tracker_db.ddl index efa00bc..011eccf 100644 --- a/src/main/resources/sql/credit_tracker_db.ddl +++ b/src/main/resources/sql/credit_tracker_db.ddl @@ -9,7 +9,6 @@ create table signals_user_packages deactivated_at timestamp, last_downloaded_report varchar(255), monitoring_at timestamp, - last_searched_at timestamp, signal_types varchar(255), reject_status varchar(255), external_package_name varchar(255), diff --git a/src/main/resources/sql/v2.21.ddl b/src/main/resources/sql/v2.21.ddl index 3b73e0f..332a65c 100644 --- a/src/main/resources/sql/v2.21.ddl +++ b/src/main/resources/sql/v2.21.ddl @@ -1,5 +1,3 @@ -alter table signals_user_packages - add LAST_SEARCHED_AT timestamp; alter table signals_user_packages add SIGNAL_TYPES varchar(255); alter table signals_user_packages @@ -40,4 +38,6 @@ CREATE TABLE signals_phones_map ALTER TABLE signals_phones_map ADD CONSTRAINT signals_phones_map_phone_key UNIQUE (PHONE); -update signals_phone_fid_map set phone_l = phone::BIGINT where phone is not null; \ No newline at end of file +update signals_phone_fid_map set phone_l = phone::BIGINT where phone is not null; +update signals_user_packages set signal_types = '1', external_package_name = concat(substring(membercode, 0, 6), '_', package_name) where signal_types is null; +insert into app_version values ('2.0') \ No newline at end of file