http://jira.mfd.msk:8088/browse/BKI-3241

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();

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);
+        }
+    }
+
+}

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<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory;
     private final Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory;
     private final Function<SetupParam, IPipeLineStarter<SubjectPayload>> pipelineStarterFactory;
@@ -52,6 +54,7 @@ public class PipelinePackageFabric implements InitializingBean {
                                  FidIdEnricher fidIdEnricherHandler,
                                  FidIdUpdater fidIdUpdater,
                                  FidNameUpdater fidNameUpdater,
+                                 FidNameUpdaterMigration fidNameUpdaterMigration,
                                  Supplier<IPipeLineTerminator<PartFilePayload, SignalXmlFilePayload>> pipelineTerminatorFactory,
                                  Supplier<IPipeLineTerminator<VoidPayload, EmptyTerminatePayload>> noOpTerminatorFactory,
                                  Function<SetupParam, IPipeLineStarter<SubjectPayload>> 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());

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<SubjectPayload, SubjectPayload> {
+    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<SubjectProfile> subjects = taskPayload.getSubjects();
+        var ctx = obtainContext(PackageVariable.class, taskPayload);
+        log.debug("NameUpdaterStage handle, size of bucket for save: {}", subjects.size());
+        Map<String, Name> nameByPhone = personService.findNameByNumber(subjects);
+
+        List<SubjectProfile> changingFidPhones = new ArrayList<>();
+        List<Long> notFoundFidPhones = new ArrayList<>();
+        List<Long> 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<Long> 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;
+    }
+}

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),

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
@@ -41,3 +39,5 @@ 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;
+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')