DMP-4312: Dets Clean up automated Task

src/integrationTest/java/uk/gov/hmcts/darts/testutils/stubs/DartsDatabaseStub.java

@@ -336,6 +336,7 @@ public void clearDatabaseInThisOrder() {
             transcriptionRepository.deleteAllInBatch();
             transcriptionWorkflowRepository.deleteAllInBatch();
             retentionConfidenceCategoryMapperRepository.deleteAllInBatch();
+            objectStateRecordRepository.deleteAllInBatch();
         });
     }
 

src/integrationTest/java/uk/gov/hmcts/darts/testutils/stubs/DartsPersistence.java

@@ -79,6 +79,7 @@
 import uk.gov.hmcts.darts.common.repository.ObjectAdminActionRepository;
 import uk.gov.hmcts.darts.common.repository.ObjectHiddenReasonRepository;
 import uk.gov.hmcts.darts.common.repository.ObjectRecordStatusRepository;
+import uk.gov.hmcts.darts.common.repository.ObjectStateRecordRepository;
 import uk.gov.hmcts.darts.common.repository.ProsecutorRepository;
 import uk.gov.hmcts.darts.common.repository.RegionRepository;
 import uk.gov.hmcts.darts.common.repository.RetentionConfidenceCategoryMapperRepository;
@@ -170,6 +171,7 @@ public class DartsPersistence {
     private final ObjectAdminActionRepository objectAdminActionRepository;
     private final EventLinkedCaseRepository eventLinkedCaseRepository;
     private final RetentionConfidenceCategoryMapperRepository retentionConfidenceCategoryMapperRepository;
+    private final ObjectStateRecordRepository objectStateRecordRepository;
 
     private final EntityManager entityManager;
     private final CurrentTimeHelper currentTimeHelper;
@@ -911,6 +913,14 @@ public void overrideLastModifiedBy(MediaRequestEntity mediaRequestEntity, Offset
             .executeUpdate();
     }
 
+    @Transactional
+    public void overrideLastModifiedBy(ExternalObjectDirectoryEntity externalObjectDirectoryEntity, OffsetDateTime lastModifiedDate) {
+        entityManager.createNativeQuery("UPDATE external_object_directory SET last_modified_ts = :lastModifiedDate WHERE eod_id = :id")
+            .setParameter("lastModifiedDate", lastModifiedDate)
+            .setParameter("id", externalObjectDirectoryEntity.getId())
+            .executeUpdate();
+    }
+
     @Transactional
     public <T extends HasId<?> & CreatedBy> void updateCreatedBy(T object, OffsetDateTime createdDateTime) {
         Table table = object.getClass().getAnnotation(Table.class);

src/main/java/uk/gov/hmcts/darts/arm/service/CleanUpDetsDataProcessor.java

@@ -0,0 +1,8 @@
+package uk.gov.hmcts.darts.arm.service;
+
+import uk.gov.hmcts.darts.task.config.CleanUpDetsDataAutomatedTaskConfig;
+
+@FunctionalInterface
+public interface CleanUpDetsDataProcessor {
+    void processCleanUpDetsData(int batchSize, CleanUpDetsDataAutomatedTaskConfig minimumStoredAge);
+}

src/main/java/uk/gov/hmcts/darts/arm/service/impl/CleanUpDetsDataProcessorImpl.java

@@ -0,0 +1,143 @@
+package uk.gov.hmcts.darts.arm.service.impl;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.collections4.ListUtils;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import uk.gov.hmcts.darts.arm.service.CleanUpDetsDataProcessor;
+import uk.gov.hmcts.darts.common.repository.ExternalObjectDirectoryRepository;
+import uk.gov.hmcts.darts.common.repository.ObjectStateRecordRepository;
+import uk.gov.hmcts.darts.dets.service.DetsApiService;
+import uk.gov.hmcts.darts.task.config.CleanUpDetsDataAutomatedTaskConfig;
+import uk.gov.hmcts.darts.util.AsyncUtil;
+
+import java.time.Clock;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class CleanUpDetsDataProcessorImpl implements CleanUpDetsDataProcessor {
+
+    private final CleanUpDetsDataBatchProcessor cleanUpDetsDataBatchProcessor;
+    private final Clock clock;
+
+    @Override
+    //Required for async batch processing
+    @SuppressWarnings({"PMD.DoNotUseThreads"})
+    public void processCleanUpDetsData(int batchSize, CleanUpDetsDataAutomatedTaskConfig config) {
+        log.info("Processing clean up of DETS data with batch size: {}", batchSize);
+
+        OffsetDateTime minimumStoredAge = OffsetDateTime.now(clock).minus(config.getMinimumStoredAge());
+        int chunkSize = config.getChunkSize();
+
+        int totalProcessed = 0;
+
+        while (totalProcessed < batchSize && chunkSize > 0) {
+            log.info("Processing clean up of DETS data with chunk size: {}", chunkSize);
+
+            List<CleanUpDetsDataProcessorImpl.CleanUpDetsProcedureResponse> eodIdsToCleanUp =
+                cleanUpDetsDataBatchProcessor.callDetsCleanUpStoredProcedure(chunkSize, minimumStoredAge);
+
+            if (eodIdsToCleanUp.isEmpty()) {
+                log.info("No more DETS data to clean up. Ending process.");
+                break;
+            }
+
+            List<List<CleanUpDetsDataProcessorImpl.CleanUpDetsProcedureResponse>> batchesToDeleteBlobStoreRecordFor =
+                ListUtils.partition(eodIdsToCleanUp, config.getChunkSize() / config.getThreads());
+
+            List<Callable<Void>> tasks = batchesToDeleteBlobStoreRecordFor
+                .stream()
+                .map(eodsForBatch -> (Callable<Void>) () -> {
+                    cleanUpDetsDataBatchProcessor.process(eodsForBatch);
+                    return null;
+                })
+                .toList();
+
+            try {
+                AsyncUtil.invokeAllAwaitTermination(tasks, config);
+            } catch (InterruptedException e) {
+                log.error("Clean up dets data batch processing interrupted", e);
+                Thread.currentThread().interrupt();
+            } catch (Exception e) {
+                log.error("Clean up dets data unexpected exception", e);
+            }
+            //Update total processed count and adjust chunk size for next iteration if needed
+            totalProcessed += eodIdsToCleanUp.size();
+            //Ensure we do not exceed the batch size in the next iteration
+            //Takes into account the possibility that the procedure may return more records than requested
+            if (totalProcessed + chunkSize > batchSize) {
+                chunkSize = batchSize - totalProcessed;
+            }
+            log.info("Processed batch of DETS data clean up. Total processed so far: {}. Batch size: {}", totalProcessed, eodIdsToCleanUp.size());
+        }
+        log.info("Completed processing clean up of DETS data. Total processed: {}", totalProcessed);
+    }
+
+    @Component
+    @RequiredArgsConstructor
+    public static class CleanUpDetsDataBatchProcessor {
+
+        private final ExternalObjectDirectoryRepository externalObjectDirectoryRepository;
+        private final ObjectStateRecordRepository objectStateRecordRepository;
+        private final DetsApiService detsApiService;
+
+        @Transactional(propagation = Propagation.REQUIRES_NEW)
+        public void process(List<CleanUpDetsDataProcessorImpl.CleanUpDetsProcedureResponse> eodsCleanedUp) {
+            List<Long> objectStateRecordsForDetsRecordsCleanedUpSuccessfully = new ArrayList<>();
+            List<Long> objectStateRecordsForDetsRecordsFailedToCleanUp = new ArrayList<>();
+
+            if (eodsCleanedUp.isEmpty()) {
+                return;
+            }
+
+            for (CleanUpDetsProcedureResponse response : eodsCleanedUp) {
+                try {
+                    log.debug("Processing clean up response for EOD ID: {}, Location: {}", response.getOsrUuid(), response.getDetsLocation());
+                    if (detsApiService.deleteBlobDataFromContainer(response.getDetsLocation())) {
+                        log.debug("Successfully deleted DETS blob for EOD ID: {}, Location: {}", response.getOsrUuid(), response.getDetsLocation());
+                        objectStateRecordsForDetsRecordsCleanedUpSuccessfully.add(response.getOsrUuid());
+                        continue;
+                    } else {
+                        log.error("Failed to delete DETS blob for EOD ID: {}, Location: {}. Blob may not exist or deletion failed.",
+                                  response.getOsrUuid(), response.getDetsLocation());
+                    }
+                } catch (Exception exception) {
+                    log.error("Failed to delete DETS blob for EOD Location: {}, object state record id: {}.",
+                              response.getDetsLocation(), response.getOsrUuid(), exception);
+                }
+                objectStateRecordsForDetsRecordsFailedToCleanUp.add(response.getOsrUuid());
+            }
+            objectStateRecordRepository.markDetsCleanupStatusAsComplete(objectStateRecordsForDetsRecordsCleanedUpSuccessfully);
+            log.info("Marked object state records as clean up complete for EOD IDs: {}", objectStateRecordsForDetsRecordsCleanedUpSuccessfully);
+
+            if (CollectionUtils.isNotEmpty(objectStateRecordsForDetsRecordsFailedToCleanUp)) {
+                log.error("Dets clean up failed for Object state record Ids: {}", objectStateRecordsForDetsRecordsFailedToCleanUp);
+            }
+        }
+
+        @Transactional(propagation = Propagation.REQUIRES_NEW)
+        public List<CleanUpDetsDataProcessorImpl.CleanUpDetsProcedureResponse> callDetsCleanUpStoredProcedure(int chunkSize, OffsetDateTime minimumStoredAge) {
+            return objectStateRecordRepository.cleanUpDetsDataProcedure(chunkSize, minimumStoredAge);
+        }
+    }
+
+    @AllArgsConstructor
+    @Getter
+    @Setter
+    public static class CleanUpDetsProcedureResponse {
+        private Long osrUuid;
+        private String detsLocation;
+    }
+}

src/main/java/uk/gov/hmcts/darts/common/repository/ExternalObjectDirectoryRepository.java

@@ -278,7 +278,6 @@ List<Long> findIdsForAudioToBeDeletedFromUnstructured(ObjectRecordStatusEntity s
                                                           OffsetDateTime unstructuredLastModifiedBefore,
                                                           Limit limit);
 
-
     @Query(
         """                       
             SELECT eod FROM ExternalObjectDirectoryEntity eod

src/main/java/uk/gov/hmcts/darts/common/repository/ObjectStateRecordRepository.java

@@ -1,10 +1,27 @@
 package uk.gov.hmcts.darts.common.repository;
 
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import uk.gov.hmcts.darts.arm.service.impl.CleanUpDetsDataProcessorImpl;
 import uk.gov.hmcts.darts.common.entity.ObjectStateRecordEntity;
 
+import java.time.OffsetDateTime;
+import java.util.List;
 import java.util.Optional;
 
 public interface ObjectStateRecordRepository extends JpaRepository<ObjectStateRecordEntity, Long> {
     Optional<ObjectStateRecordEntity> findByArmEodId(long armEodId);
+
+    @Modifying
+    @Query("UPDATE ObjectStateRecordEntity o SET o.flagFileDetsCleanupStatus = true where o.uuid in :uuids")
+    void markDetsCleanupStatusAsComplete(List<Long> uuids);
+
+
+    @Query(value = "SELECT osr_uuid AS osrUuid, dets_location AS detsLocation  " +
+        "FROM dets_cleanup_eod_osr_rows(:limit, :last_modified_before_ts)", nativeQuery = true)
+    List<CleanUpDetsDataProcessorImpl.CleanUpDetsProcedureResponse> cleanUpDetsDataProcedure(
+        @Param("limit") Integer limit, @Param("last_modified_before_ts") OffsetDateTime lastModifiedBefore);
+
 }

src/main/java/uk/gov/hmcts/darts/dets/service/impl/DetsApiServiceImpl.java

@@ -86,6 +86,7 @@ public String saveBlobData(BinaryData binaryData, String fileName) {
         return client.getBlobName();
     }
 
+
     @Override
     public boolean deleteBlobDataFromContainer(String blobId) throws AzureDeleteBlobException {
         try {
@@ -107,7 +108,6 @@ public boolean deleteBlobDataFromContainer(String blobId) throws AzureDeleteBlob
                                                configuration.getContainerName(), blobId, httpStatus);
                 throw new AzureDeleteBlobException(message);
             }
-
         } catch (RuntimeException e) {
             throw new AzureDeleteBlobException(
                 "Could not delete from storage container=" + configuration.getContainerName() + ", blobId=" + blobId, e

src/main/java/uk/gov/hmcts/darts/task/api/AutomatedTaskName.java

@@ -43,7 +43,8 @@ public enum AutomatedTaskName {
     ARM_MISSING_RESPONSE_REPLY_TASK_NAME("ArmMissingResponseReplay"),
     DETS_CLEANUP_ARM_RESPONSE_FILES("DETSCleanupArmResponseFiles"),
     MEDIA_REQUEST_CLEANUP("MediaRequestCleanUp"),
-    ARM_RPO_BACKLOG_CATCHUP("ArmRpoBacklogCatchup", Constants.AUTOMATED_TASK_PROCESS_E2E_ARM_RPO_PENDING_PROCESS_E2E_ARM_RPO_FALSE);
+    ARM_RPO_BACKLOG_CATCHUP("ArmRpoBacklogCatchup", Constants.AUTOMATED_TASK_PROCESS_E2E_ARM_RPO_PENDING_PROCESS_E2E_ARM_RPO_FALSE),
+    CLEAN_UP_DETS_DATA("CleanUpDetsData");
 
     private final String taskName;
     private final String conditionalOnSpEL;

src/main/java/uk/gov/hmcts/darts/task/config/CleanUpDetsDataAutomatedTaskConfig.java

@@ -0,0 +1,18 @@
+package uk.gov.hmcts.darts.task.config;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+import java.time.Duration;
+
+@ConfigurationProperties("darts.automated.task.clean-up-dets-data")
+@Getter
+@Setter
+@Configuration
+public class CleanUpDetsDataAutomatedTaskConfig extends AbstractAsyncAutomatedTaskConfig {
+
+    private Duration minimumStoredAge;
+    private int chunkSize;
+}

src/main/java/uk/gov/hmcts/darts/task/runner/impl/CleanUpDetsDataAutomatedTask.java

@@ -0,0 +1,42 @@
+package uk.gov.hmcts.darts.task.runner.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import uk.gov.hmcts.darts.arm.service.CleanUpDetsDataProcessor;
+import uk.gov.hmcts.darts.common.repository.AutomatedTaskRepository;
+import uk.gov.hmcts.darts.log.api.LogApi;
+import uk.gov.hmcts.darts.task.api.AutomatedTaskName;
+import uk.gov.hmcts.darts.task.config.CleanUpDetsDataAutomatedTaskConfig;
+import uk.gov.hmcts.darts.task.runner.AutoloadingManualTask;
+import uk.gov.hmcts.darts.task.service.LockService;
+
+import static uk.gov.hmcts.darts.task.api.AutomatedTaskName.CLEAN_UP_DETS_DATA;
+
+@Slf4j
+@Component
+public class CleanUpDetsDataAutomatedTask
+    extends AbstractLockableAutomatedTask<CleanUpDetsDataAutomatedTaskConfig>
+    implements AutoloadingManualTask {
+
+    private final CleanUpDetsDataProcessor cleanUpDetsDataProcessor;
+
+    @Autowired
+    public CleanUpDetsDataAutomatedTask(AutomatedTaskRepository automatedTaskRepository,
+                                        CleanUpDetsDataAutomatedTaskConfig automatedTaskConfigurationProperties,
+                                        CleanUpDetsDataProcessor processor,
+                                        LogApi logApi, LockService lockService) {
+        super(automatedTaskRepository, automatedTaskConfigurationProperties, logApi, lockService);
+        this.cleanUpDetsDataProcessor = processor;
+    }
+
+    @Override
+    public AutomatedTaskName getAutomatedTaskName() {
+        return CLEAN_UP_DETS_DATA;
+    }
+
+    @Override
+    protected void runTask() {
+        cleanUpDetsDataProcessor.processCleanUpDetsData(getAutomatedTaskBatchSize(), getConfig());
+    }
+}

src/main/resources/application.yaml

@@ -440,6 +440,15 @@ darts:
         lock:
           at-least-for: PT1M
           at-most-for: PT90M
+      clean-up-dets-data:
+        system-user-email: systemCleanUpDetsData@hmcts.net
+        lock:
+          at-least-for: PT1M
+          at-most-for: PT90M
+        minimum-stored-age: 7D
+        chunk-size: 10000
+        async-timeout: 20M
+        threads: 20
       cleanup-arm-response-files:
         system-user-email: system_CleanupArmResponseFiles@hmcts.net
         lock:

src/main/resources/db/migration/common/V1_505__CleanUpDetsData_automated_task.sql

@@ -0,0 +1,10 @@
+INSERT INTO darts.automated_task (aut_id, task_name, task_description, cron_expression, cron_editable, batch_size,
+                                  created_ts, created_by, last_modified_ts, last_modified_by, task_enabled)
+VALUES (36, 'CleanUpDetsData', 'Cleans up Dets files that have successfully been stored in ARM', '0 24 0-6,19-23 ? * *', true, 100_000,
+        current_timestamp, 0, current_timestamp, 0, false);
+
+INSERT INTO user_account (usr_id, user_name, user_email_address, description, created_ts, last_modified_ts, last_modified_by, created_by, is_system_user,
+                          is_active, user_full_name)
+
+VALUES (-51, 'systemCleanUpDetsDataAutomatedTask', 'systemCleanUpDetsDataAutomatedTask@hmcts.net', 'systemCleanUpDetsDataAutomatedTask',
+        '2024-01-01 00:00:00+00', '2024-01-01 00:00:00+00', 0, 0, true, true, 'systemCleanUpDetsDataAutomatedTask');