[UPBE-600] added data factory integration into generate report api

GiovanaSolorzano · GiovanaSolorzano · commit 6ccff2f5cbfc · 2026-02-13T16:01:40.000+01:00
diff --git a/src/main/java/it/gov/pagopa/idpay/transactions/data/factory/DataFactoryService.java b/src/main/java/it/gov/pagopa/idpay/transactions/data/factory/DataFactoryService.java
@@ -4,5 +4,5 @@
 import reactor.core.publisher.Mono;
 
 public interface DataFactoryService {
-    Mono<String> generateReport(Report report);
+    Mono<String> triggerTransactionReportPipeline(Report report);
 }
diff --git a/src/main/java/it/gov/pagopa/idpay/transactions/data/factory/DataFactoryServiceImpl.java b/src/main/java/it/gov/pagopa/idpay/transactions/data/factory/DataFactoryServiceImpl.java
@@ -39,9 +39,9 @@ public DataFactoryServiceImpl(DataFactoryManager dataFactoryManager,
     }
 
     @Override
-    public Mono<String> generateReport(Report report) {
+    public Mono<String> triggerTransactionReportPipeline(Report report) {
         Mono<String> callMono = Mono.fromCallable(() -> {
-                    log.info("[CALLING_DATA_FACTORY] Running pipeline for report {}", report.getId());
+                    log.info("[CALLING_DATA_FACTORY] Starting pipeline execution for Report {}", report.getId());
                     Response<CreateRunResponse> resp = dataFactoryManager.pipelines().createRunWithResponse(
                             resourceGroup,
                             factoryName,
@@ -62,7 +62,7 @@ public Mono<String> generateReport(Report report) {
                     if (body == null) {
                         throw new IllegalStateException("ADF createRun returned empty body");
                     }
-                    log.info("[CALLING_DATA_FACTORY] Identification run pipeline {} for report {}", body.runId(), report.getId());
+                    log.info("[CALLING_DATA_FACTORY] Report {} generation request sent successfully. Run ID: {}", report.getId(), body.runId());
                     return body.runId();
                 })
                 .subscribeOn(Schedulers.boundedElastic());
diff --git a/src/main/java/it/gov/pagopa/idpay/transactions/service/ReportServiceImpl.java b/src/main/java/it/gov/pagopa/idpay/transactions/service/ReportServiceImpl.java
@@ -11,6 +11,7 @@
 import it.gov.pagopa.idpay.transactions.enums.ReportStatus;
 import it.gov.pagopa.idpay.transactions.enums.ReportType;
 import it.gov.pagopa.idpay.transactions.enums.RewardBatchAssignee;
+import it.gov.pagopa.idpay.transactions.exception.AzureConnectingErrorException;
 import it.gov.pagopa.idpay.transactions.model.Report;
 import it.gov.pagopa.idpay.transactions.repository.ReportRepository;
 import it.gov.pagopa.idpay.transactions.utils.Utilities;
@@ -26,7 +27,6 @@
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
-import java.util.Locale;
 
 import static it.gov.pagopa.idpay.transactions.utils.ExceptionConstants.ExceptionCode.*;
 import static it.gov.pagopa.idpay.transactions.utils.ExceptionConstants.ExceptionMessage.*;
@@ -163,6 +163,14 @@ public Mono<ReportDTO> generateMerchantTransactionsReport(String merchantId,
 
                     return reportRepository.save(reportEntity);
                 })
+                .flatMap(report ->
+                        triggerTransactionReportPipeline(report)
+                                .thenReturn(report)
+                                .onErrorResume(AzureConnectingErrorException.class, ex -> {
+                                    report.setReportStatus(ReportStatus.FAILED);
+                                    return reportRepository.save(report);
+                                })
+                )
                 .map(reportMapper::toDTO)
                 .doOnSuccess(saved -> log.info("[GENERATE_REPORT] Saved report {} for merchant {}",
                         saved.getFileName(), Utilities.sanitizeString(merchantId)));
@@ -203,13 +211,15 @@ public Mono<ReportDTO> patchReport(String initiativeId,
     public Mono<List<Report2RunDto>> forceGenerateReports(ReportGenerateForce reportGenerateForce) {
         log.info("[RUN_GENERATE_REPORT] Request generate report {}", reportGenerateForce.getReportsId());
         return reportRepository.findAllById(reportGenerateForce.getReportsId())
-                .flatMap(report ->
-                        dataFactoryService.generateReport(report)
-                                .map(runId ->
-                                        Report2RunDto.builder()
-                                                .reportId(report.getId())
-                                                .runId(runId).build()))
+                .flatMap(this::triggerTransactionReportPipeline)
                 .collectList();
     }
 
+    private Mono<Report2RunDto> triggerTransactionReportPipeline(Report report) {
+        return dataFactoryService.triggerTransactionReportPipeline(report)
+                .map(runId ->
+                        Report2RunDto.builder()
+                                .reportId(report.getId())
+                                .runId(runId).build());
+    }
 }
diff --git a/src/test/java/it/gov/pagopa/idpay/transactions/data/factory/DataFactoryServiceImplTest.java b/src/test/java/it/gov/pagopa/idpay/transactions/data/factory/DataFactoryServiceImplTest.java
@@ -48,7 +48,7 @@ void setUp() {
     }
 
     @Test
-    void generateReport_success() {
+    void triggerTransactionReport_Pipeline_success() {
 
         LocalDateTime now = LocalDateTime.now();
         Report report = Report.builder()
@@ -59,7 +59,6 @@ void generateReport_success() {
                 .endPeriod(now)
                 .fileName("file.csv").build();
 
-        // mock Azure chain
         when(dataFactoryManager.pipelines()).thenReturn(pipelines);
         when(pipelines.createRunWithResponse(
                 anyString(),
@@ -77,7 +76,7 @@ void generateReport_success() {
         when(response.getValue()).thenReturn(createRunResponse);
         when(createRunResponse.runId()).thenReturn("RUN_ID");
 
-        StepVerifier.create(service.generateReport(report))
+        StepVerifier.create(service.triggerTransactionReportPipeline(report))
                 .expectNext("RUN_ID")
                 .verifyComplete();
 
@@ -95,7 +94,7 @@ void generateReport_success() {
     }
 
     @Test
-    void generateReport_error_shouldThrowAzureConnectingError() {
+    void triggerTransactionReport_Pipeline_error_shouldThrowAzureConnectingError() {
 
         Report report = mock(Report.class);
         when(report.getId()).thenReturn("REPORT_ID");
@@ -114,7 +113,7 @@ void generateReport_error_shouldThrowAzureConnectingError() {
                 any()
         )).thenThrow(new RuntimeException("Connection error"));
 
-        StepVerifier.create(service.generateReport(report))
+        StepVerifier.create(service.triggerTransactionReportPipeline(report))
                 .expectError(AzureConnectingErrorException.class)
                 .verify();
     }
diff --git a/src/test/java/it/gov/pagopa/idpay/transactions/service/ReportServiceImplTest.java b/src/test/java/it/gov/pagopa/idpay/transactions/service/ReportServiceImplTest.java
@@ -13,6 +13,7 @@
 import it.gov.pagopa.idpay.transactions.enums.ReportStatus;
 import it.gov.pagopa.idpay.transactions.enums.ReportType;
 import it.gov.pagopa.idpay.transactions.enums.RewardBatchAssignee;
+import it.gov.pagopa.idpay.transactions.exception.AzureConnectingErrorException;
 import it.gov.pagopa.idpay.transactions.model.Report;
 import it.gov.pagopa.idpay.transactions.repository.ReportRepository;
 import org.junit.jupiter.api.BeforeEach;
@@ -285,6 +286,8 @@ void generateMerchantTransactionsReport_success() {
         when(reportRepository.save(any()))
                 .thenReturn(Mono.just(savedReport));
 
+        when(dataFactoryServiceMock.triggerTransactionReportPipeline(savedReport)).thenReturn(Mono.just("RUN_ID"));
+
         when(reportMapper.toDTO(savedReport))
                 .thenReturn(mappedDto);
 
@@ -371,6 +374,7 @@ void generateMerchantTransactionsReport_fileNameGeneratedCorrectly() {
                     .build();
 
             when(reportRepository.save(any())).thenReturn(Mono.just(saved));
+            when(dataFactoryServiceMock.triggerTransactionReportPipeline(saved)).thenReturn(Mono.just("RUN_ID"));
             when(reportMapper.toDTO(saved)).thenReturn(ReportDTO.builder().id("R200").fileName(saved.getFileName()).build());
 
             StepVerifier.create(service.generateMerchantTransactionsReport(MERCHANT_ID, ORGANIZATION_ROLE, INITIATIVE_ID, request))
@@ -381,6 +385,42 @@ void generateMerchantTransactionsReport_fileNameGeneratedCorrectly() {
             assertEquals("Report_01022026123045", captor.getValue().getFileName());
         }
     }
+    @Test
+    void generateMerchantTransactionsReport_TriggerPipelineError() {
+        ReportRequest request = new ReportRequest();
+        request.setStartPeriod(LocalDateTime.of(2026, 1, 1, 0, 0));
+        request.setEndPeriod(LocalDateTime.of(2026, 1, 31, 23, 59));
+        request.setReportType(ReportType.MERCHANT_TRANSACTIONS);
+
+        MerchantDetailDTO merchant = new MerchantDetailDTO();
+        merchant.setBusinessName("Business");
+
+        LocalDateTime fixedNow = LocalDateTime.of(2026, 2, 1, 12, 30, 45);
+
+        try (MockedStatic<LocalDateTime> mocked = mockStatic(LocalDateTime.class, CALLS_REAL_METHODS)) {
+            mocked.when(LocalDateTime::now).thenReturn(fixedNow);
+
+            when(merchantRestClient.getMerchantDetail(MERCHANT_ID, INITIATIVE_ID))
+                    .thenReturn(Mono.just(merchant));
+
+            Report saved = Report.builder()
+                    .id("R200")
+                    .fileName("Report_01022026123045")
+                    .build();
+
+            Report saved2 = Report.builder()
+                    .id("R200_2")
+                    .build();
+
+            when(reportRepository.save(any())).thenReturn(Mono.just(saved)).thenReturn(Mono.just(saved2));
+            when(dataFactoryServiceMock.triggerTransactionReportPipeline(saved)).thenReturn(Mono.error(new AzureConnectingErrorException("DUMMY_ERROR", new RuntimeException())));
+            when(reportMapper.toDTO(saved2)).thenReturn(ReportDTO.builder().id("R200_2").build());
+
+            StepVerifier.create(service.generateMerchantTransactionsReport(MERCHANT_ID, ORGANIZATION_ROLE, INITIATIVE_ID, request))
+                    .assertNext(dto -> assertEquals("R200_2", dto.getId()))
+                    .verifyComplete();
+        }
+    }
 
     @Test
     void patchReport_success_updatesStatus() {
@@ -508,8 +548,8 @@ void forceGenerateReports() {
         when(reportRepository.findAllById(anyIterable()))
                 .thenReturn(Flux.just(report1, report2));
 
-        when(dataFactoryServiceMock.generateReport(report1)).thenReturn(Mono.just("RUN1"));
-        when(dataFactoryServiceMock.generateReport(report2)).thenReturn(Mono.just("RUN2"));
+        when(dataFactoryServiceMock.triggerTransactionReportPipeline(report1)).thenReturn(Mono.just("RUN1"));
+        when(dataFactoryServiceMock.triggerTransactionReportPipeline(report2)).thenReturn(Mono.just("RUN2"));
 
         Mono<List<Report2RunDto>> resultMono = service.forceGenerateReports(request);
 
@@ -524,6 +564,6 @@ void forceGenerateReports() {
                 .verifyComplete();
 
         verify(reportRepository, times(1)).findAllById(anyList());
-        verify(dataFactoryServiceMock, times(2)).generateReport(any());
+        verify(dataFactoryServiceMock, times(2)).triggerTransactionReportPipeline(any());
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -4,5 +4,5 @@`
`4`	`4`	`import reactor.core.publisher.Mono;`
`5`	`5`
`6`	`6`	`public interface DataFactoryService {`
`7`		`- Mono<String> generateReport(Report report);`
	`7`	`+ Mono<String> triggerTransactionReportPipeline(Report report);`
`8`	`8`	`}`