UPDATE - count and delete for date in BatchInfoRepository to take in …

…consideration source system (#1418)

common/persistence/src/main/java/app/coronawarn/server/common/persistence/repository/FederationBatchInfoRepository.java

@@ -39,19 +39,23 @@ void saveDoUpdateStatusOnConflict(
 
   List<FederationBatchInfo> findByStatus(@Param("status") String status);
 
-  @Query("SELECT COUNT(*) FROM federation_batch_info WHERE date<:threshold")
-  int countOlderThan(@Param("threshold") LocalDate date);
+  @Query("SELECT COUNT(*) FROM federation_batch_info WHERE date<:threshold AND source_system=:sourceSystem")
+  int countOlderThan(@Param("threshold") LocalDate date,
+      @Param("sourceSystem") FederationBatchSourceSystem sourceSystem);
 
   @Modifying
-  @Query("DELETE FROM federation_batch_info WHERE date<:threshold")
-  void deleteOlderThan(@Param("threshold") LocalDate date);
+  @Query("DELETE FROM federation_batch_info WHERE date<:threshold AND source_system=:sourceSystem")
+  void deleteOlderThan(@Param("threshold") LocalDate date,
+      @Param("sourceSystem") FederationBatchSourceSystem sourceSystem);
 
-  @Query("SELECT COUNT(*) FROM federation_batch_info WHERE date=:date")
-  int countForDate(@Param("date") LocalDate date);
+  @Query("SELECT COUNT(*) FROM federation_batch_info WHERE date=:date AND source_system=:sourceSystem")
+  int countForDateAndSourceSystem(@Param("date") LocalDate date,
+      @Param("sourceSystem") FederationBatchSourceSystem sourceSystem);
 
   @Modifying
-  @Query("DELETE FROM federation_batch_info WHERE date=:date")
-  void deleteForDate(@Param("date") LocalDate date);
+  @Query("DELETE FROM federation_batch_info WHERE date=:date AND source_system=:sourceSystem")
+  void deleteForDate(@Param("date") LocalDate date,
+      @Param("sourceSystem") FederationBatchSourceSystem sourceSystem);
 
   List<FederationBatchInfo> findByStatusAndSourceSystem(@Param("status") String name,
       @Param("sourceSystem") FederationBatchSourceSystem sourceSystem);

common/persistence/src/main/java/app/coronawarn/server/common/persistence/service/FederationBatchInfoService.java

@@ -74,16 +74,17 @@ public List<FederationBatchInfo> findByStatus(FederationBatchStatus federationBa
    * @throws IllegalArgumentException if {@code daysToRetain} is negative.
    */
   @Transactional
-  public void applyRetentionPolicy(int daysToRetain) {
+  public void applyRetentionPolicy(int daysToRetain,
+      final FederationBatchSourceSystem sourceSystem) {
     if (daysToRetain < 0) {
       throw new IllegalArgumentException("Number of days to retain must be greater or equal to 0.");
     }
 
     LocalDate threshold = LocalDate.now(ZoneOffset.UTC).minus(Period.ofDays(daysToRetain));
-    int numberOfDeletions = federationBatchInfoRepository.countOlderThan(threshold);
+    int numberOfDeletions = federationBatchInfoRepository.countOlderThan(threshold, sourceSystem);
     logger.info("Deleting {} batch info(s) with a date older than {} day(s) ago.",
         numberOfDeletions, daysToRetain);
-    federationBatchInfoRepository.deleteOlderThan(threshold);
+    federationBatchInfoRepository.deleteOlderThan(threshold, sourceSystem);
   }
 
   /**
@@ -92,10 +93,11 @@ public void applyRetentionPolicy(int daysToRetain) {
    * @param date The date for which the batch information entries should be deleted.
    */
   @Transactional
-  public void deleteForDate(LocalDate date) {
-    int numberOfDeletions = federationBatchInfoRepository.countForDate(date);
+  public void deleteForDate(LocalDate date,
+      final FederationBatchSourceSystem sourceSystem) {
+    int numberOfDeletions = federationBatchInfoRepository.countForDateAndSourceSystem(date, sourceSystem);
     logger.info("Deleting {} batch info(s) for date {}.",
         numberOfDeletions, date);
-    federationBatchInfoRepository.deleteForDate(date);
+    federationBatchInfoRepository.deleteForDate(date, sourceSystem);
   }
 }

common/persistence/src/test/java/app/coronawarn/server/common/persistence/repository/FederationBatchInfoRepositoryTest.java

@@ -2,6 +2,7 @@
 
 package app.coronawarn.server.common.persistence.repository;
 
+import static app.coronawarn.server.common.persistence.domain.FederationBatchSourceSystem.EFGS;
 import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -69,18 +70,18 @@ void testCountForDate() {
     federationBatchInfoRepository.saveDoNothingOnConflict(batchTag1, date1, statusUnprocessed, efgsTarget);
     federationBatchInfoRepository.saveDoNothingOnConflict(batchTag2, date2, statusUnprocessed, efgsTarget);
     federationBatchInfoRepository.saveDoNothingOnConflict(batchTag3, date2, statusUnprocessed, efgsTarget);
-    assertThat(federationBatchInfoRepository.countForDate(date1)).isEqualTo(1);
-    assertThat(federationBatchInfoRepository.countForDate(date2)).isEqualTo(2);
+    assertThat(federationBatchInfoRepository.countForDateAndSourceSystem(date1, EFGS)).isEqualTo(1);
+    assertThat(federationBatchInfoRepository.countForDateAndSourceSystem(date2, EFGS)).isEqualTo(2);
   }
 
   @Test
   void testDeleteForDate() {
     federationBatchInfoRepository.saveDoNothingOnConflict(batchTag1, date1, statusUnprocessed, efgsTarget);
     federationBatchInfoRepository.saveDoNothingOnConflict(batchTag2, date2, statusUnprocessed, efgsTarget);
-    assertThat(federationBatchInfoRepository.countForDate(date1)).isEqualTo(1);
-    assertThat(federationBatchInfoRepository.countForDate(date2)).isEqualTo(1);
-    federationBatchInfoRepository.deleteForDate(date1);
-    assertThat(federationBatchInfoRepository.countForDate(date1)).isZero();
-    assertThat(federationBatchInfoRepository.countForDate(date2)).isEqualTo(1);
+    assertThat(federationBatchInfoRepository.countForDateAndSourceSystem(date1, EFGS)).isEqualTo(1);
+    assertThat(federationBatchInfoRepository.countForDateAndSourceSystem(date2, EFGS)).isEqualTo(1);
+    federationBatchInfoRepository.deleteForDate(date1, EFGS);
+    assertThat(federationBatchInfoRepository.countForDateAndSourceSystem(date1, EFGS)).isZero();
+    assertThat(federationBatchInfoRepository.countForDateAndSourceSystem(date2, EFGS)).isEqualTo(1);
   }
 }

common/persistence/src/test/java/app/coronawarn/server/common/persistence/service/FederationBatchInfoServiceTest.java

@@ -107,13 +107,13 @@ void testSaveAndRetrieveOnConflict() {
   @ValueSource(ints = {0, 28})
   @ParameterizedTest
   void testApplyRetentionPolicyForValidNumberOfDays(int daysToRetain) {
-    assertThatCode(() -> federationBatchInfoService.applyRetentionPolicy(daysToRetain))
+    assertThatCode(() -> federationBatchInfoService.applyRetentionPolicy(daysToRetain, EFGS))
         .doesNotThrowAnyException();
   }
 
   @Test
   void testApplyRetentionPolicyForNegativeNumberOfDays() {
-    assertThat(catchThrowable(() -> federationBatchInfoService.applyRetentionPolicy(-1)))
+    assertThat(catchThrowable(() -> federationBatchInfoService.applyRetentionPolicy(-1, EFGS)))
         .isInstanceOf(IllegalArgumentException.class);
   }
 
@@ -123,7 +123,7 @@ void testApplyRetentionPolicyForOneNotApplicableEntry() {
     FederationBatchInfo expectedBatchInfo = new FederationBatchInfo(batchTag, date, EFGS);
 
     federationBatchInfoService.save(expectedBatchInfo);
-    federationBatchInfoService.applyRetentionPolicy(2);
+    federationBatchInfoService.applyRetentionPolicy(2, EFGS);
     List<FederationBatchInfo> actualBatchInfos =
         federationBatchInfoService.findByStatus(FederationBatchStatus.UNPROCESSED,EFGS);
 
@@ -137,7 +137,7 @@ void testApplyRetentionPolicyForOneApplicableEntry() {
     FederationBatchInfo expectedBatchInfo = new FederationBatchInfo(batchTag, date, EFGS);
 
     federationBatchInfoService.save(expectedBatchInfo);
-    federationBatchInfoService.applyRetentionPolicy(1);
+    federationBatchInfoService.applyRetentionPolicy(1, EFGS);
     List<FederationBatchInfo> actualBatchInfos =
         federationBatchInfoService.findByStatus(FederationBatchStatus.UNPROCESSED,EFGS);
 
@@ -151,7 +151,7 @@ void testDeleteForDay() {
     federationBatchInfoService.save(expectedBatchInfo);
 
     assertThat(federationBatchInfoService.findByStatus(FederationBatchStatus.UNPROCESSED,EFGS)).hasSize(1);
-    federationBatchInfoService.deleteForDate(date);
+    federationBatchInfoService.deleteForDate(date, EFGS);
     assertThat(federationBatchInfoService.findByStatus(FederationBatchStatus.UNPROCESSED,EFGS)).isEmpty();
   }
 }

services/download/src/main/java/app/coronawarn/server/services/download/FederationBatchProcessor.java

@@ -92,7 +92,7 @@ public void prepareDownload() throws FatalFederationGatewayException {
     if (config.getEnforceDateBasedDownload()) {
       LocalDate downloadDate = LocalDate.now(ZoneOffset.UTC)
           .minus(Period.ofDays(config.getEnforceDownloadOffsetDays()));
-      batchInfoService.deleteForDate(downloadDate);
+      batchInfoService.deleteForDate(downloadDate, config.getSourceSystem());
       saveFirstBatchInfoForDate(downloadDate);
     }
   }

services/download/src/main/java/app/coronawarn/server/services/download/runner/RetentionPolicy.java

@@ -20,6 +20,7 @@ public class RetentionPolicy implements ApplicationRunner {
   private static final Logger logger = LoggerFactory.getLogger(RetentionPolicy.class);
 
   private final FederationBatchInfoService federationBatchInfoService;
+  private final DownloadServiceConfig downloadServiceConfig;
   private final Integer retentionDays;
 
   /**
@@ -31,13 +32,14 @@ public class RetentionPolicy implements ApplicationRunner {
   public RetentionPolicy(FederationBatchInfoService federationBatchInfoService,
       DownloadServiceConfig downloadServiceConfig) {
     this.federationBatchInfoService = federationBatchInfoService;
+    this.downloadServiceConfig = downloadServiceConfig;
     this.retentionDays = downloadServiceConfig.getRetentionDays();
   }
 
   @Override
   public void run(ApplicationArguments args) {
     try {
-      federationBatchInfoService.applyRetentionPolicy(retentionDays);
+      federationBatchInfoService.applyRetentionPolicy(retentionDays, downloadServiceConfig.getSourceSystem());
       logger.debug("Retention policy applied successfully.");
     } catch (Exception e) {
       logger.error("Application of retention policy failed.", e);

services/download/src/test/java/app/coronawarn/server/services/download/DownloadForChgsIntegrationTest.java

@@ -1,9 +1,11 @@
 package app.coronawarn.server.services.download;
 
+import static app.coronawarn.server.common.persistence.domain.FederationBatchSourceSystem.EFGS;
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import app.coronawarn.server.common.persistence.domain.FederationBatchInfo;
 import app.coronawarn.server.common.persistence.domain.FederationBatchSourceSystem;
@@ -12,6 +14,7 @@
 import app.coronawarn.server.common.protocols.external.exposurenotification.DiagnosisKeyBatch;
 import com.github.tomakehurst.wiremock.http.HttpHeaders;
 import java.time.LocalDate;
+import java.time.ZoneId;
 import java.util.List;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -65,4 +68,11 @@ void downloadShouldRunSuccessfulFor(FederationBatchSourceSystem source) {
     assertThat(errorWontRetryBatches).hasSize(1); // This comes from the "ERROR" batch in the sql-file.
     assertThat(processedBatches).extracting(FederationBatchInfo::getSourceSystem).containsExactly(source);
   }
+
+  @ParameterizedTest
+  @EnumSource(value = FederationBatchSourceSystem.class, names = {"CHGS"})
+  void downloadShouldDelete2Keys(FederationBatchSourceSystem source) {
+    assertEquals(batchInfoRepository.countForDateAndSourceSystem(LocalDate.now(ZoneId.of("UTC")), source), 2);
+    assertEquals(batchInfoRepository.countForDateAndSourceSystem(LocalDate.now(ZoneId.of("UTC")), EFGS), 2);
+  }
 }

services/download/src/test/java/app/coronawarn/server/services/download/FederationBatchProcessorTest.java

@@ -90,7 +90,7 @@ void testWithDateBasedDownload() throws Exception {
 
       LocalDate date = LocalDate.now(ZoneOffset.UTC).minus(Period.ofDays(config.getEnforceDownloadOffsetDays()));
 
-      verify(batchInfoService, times(1)).deleteForDate(date);
+      verify(batchInfoService, times(1)).deleteForDate(date, EFGS);
       verify(batchProcessorSpy, times(1)).saveFirstBatchInfoForDate(date);
     }
 
@@ -100,7 +100,7 @@ void testWithCallbackBasedDownload() throws Exception {
       config.setEnforceDateBasedDownload(false);
       batchProcessorSpy.prepareDownload();
 
-      verify(batchInfoService, never()).deleteForDate(any());
+      verify(batchInfoService, never()).deleteForDate(any(), eq(EFGS));
       verify(batchProcessorSpy, never()).saveFirstBatchInfoForDate(any());
     }
   }
@@ -143,7 +143,7 @@ void testBatchInfoForTodayIsDeleted() throws Exception {
       config.setEnforceDateBasedDownload(true);
       batchProcessor.prepareDownload();
 
-      verify(batchInfoService, times(1)).deleteForDate(date);
+      verify(batchInfoService, times(1)).deleteForDate(date, EFGS);
     }
   }
 

services/download/src/test/java/app/coronawarn/server/services/download/runner/RetentionPolicyTest.java

@@ -1,8 +1,10 @@
 package app.coronawarn.server.services.download.runner;
 
+import static app.coronawarn.server.common.persistence.domain.FederationBatchSourceSystem.EFGS;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import app.coronawarn.server.common.persistence.domain.FederationBatchSourceSystem;
 import app.coronawarn.server.common.persistence.domain.config.TekFieldDerivations;
 import app.coronawarn.server.common.persistence.service.FederationBatchInfoService;
 import app.coronawarn.server.services.download.config.DownloadServiceConfig;
@@ -13,13 +15,15 @@
 import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
 import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
 @EnableConfigurationProperties(value = {DownloadServiceConfig.class, TekFieldDerivations.class})
 @ExtendWith(SpringExtension.class)
 @ContextConfiguration(classes = {RetentionPolicy.class}, initializers = ConfigDataApplicationContextInitializer.class)
 @DirtiesContext
+@ActiveProfiles("connect-efgs")
 class RetentionPolicyTest {
 
   @MockBean
@@ -35,6 +39,7 @@ class RetentionPolicyTest {
   void shouldCallService() {
     retentionPolicy.run(null);
 
-    verify(federationBatchInfoService, times(1)).applyRetentionPolicy(downloadServiceConfig.getRetentionDays());
+    verify(federationBatchInfoService, times(1))
+        .applyRetentionPolicy(downloadServiceConfig.getRetentionDays(), EFGS);
   }
 }

services/download/src/test/resources/db/testdata-error-batch/V99__addErrorBatch.sql

@@ -1,2 +1,7 @@
 INSERT INTO federation_batch_info(batch_tag, date, status,source_system) VALUES ('batchtag', '2021-06-01', 'ERROR','CHGS');
 INSERT INTO federation_batch_info(batch_tag, date, status,source_system) VALUES ('batchtag1', '2021-06-01', 'ERROR','EFGS');
+
+INSERT INTO federation_batch_info(batch_tag, date, status,source_system) VALUES ('shouldBeDeleted1', current_date, 'ERROR','CHGS');
+INSERT INTO federation_batch_info(batch_tag, date, status,source_system) VALUES ('shouldBeDeleted2', current_date, 'ERROR','EFGS');
+INSERT INTO federation_batch_info(batch_tag, date, status,source_system) VALUES ('shouldBeDeleted3', current_date, 'ERROR','CHGS');
+INSERT INTO federation_batch_info(batch_tag, date, status,source_system) VALUES ('shouldBeDeleted4', current_date, 'ERROR','EFGS');