job cleaner cron added

src/main/java/ru/nbch/indicsearchservice/config/IndicSearchConfig.java

@@ -0,0 +1,26 @@
+package ru.nbch.indicsearchservice.config;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.nbch.indicsearchservice.model.SearchJob;
+
+@Configuration
+public class IndicSearchConfig {
+    private final ExecutorService searchJobExecutor = Executors.newFixedThreadPool(5);
+    private final Map<UUID, SearchJob> searchJobs = new ConcurrentHashMap<>();
+
+    @Bean("searchJobExecutor")
+    public ExecutorService searchJobExecutor() {
+        return searchJobExecutor;
+    }
+
+    @Bean("searchJobsMap")
+    public Map<UUID, SearchJob> searchJobsMap() {
+        return searchJobs;
+    }
+}

src/main/java/ru/nbch/indicsearchservice/config/settings/ApplicationSettings.java

@@ -5,6 +5,7 @@ import java.nio.file.Path;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 import org.springframework.context.annotation.Configuration;
+import ru.nbch.indicsearchservice.config.settings.cron.JobCleanerSettings;
 
 @Configuration
 @ConfigurationProperties(prefix = "app")
@@ -13,9 +14,11 @@ public class ApplicationSettings {
     private DataSourceSettings dataSource;
 
     private DataSourceSettings indicDataSource = new DataSourceSettings();
-
     private File storageDirectory = Path.of("/opt/app/indic-search-service-storage").toAbsolutePath().toFile();
 
+    @NestedConfigurationProperty
+    private JobCleanerSettings jobCleaner = new JobCleanerSettings();
+
     public DataSourceSettings getDataSource() {
         return dataSource;
     }
@@ -39,4 +42,12 @@ public class ApplicationSettings {
     public void setStorageDirectory(File storageDirectory) {
         this.storageDirectory = storageDirectory;
     }
+
+    public JobCleanerSettings getJobCleaner() {
+        return jobCleaner;
+    }
+
+    public void setJobCleaner(JobCleanerSettings jobCleaner) {
+        this.jobCleaner = jobCleaner;
+    }
 }

src/main/java/ru/nbch/indicsearchservice/config/settings/cron/JobCleanerSettings.java

@@ -0,0 +1,22 @@
+package ru.nbch.indicsearchservice.config.settings.cron;
+
+public class JobCleanerSettings {
+    private boolean enabled = true;
+    private String cron;
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
+    public void setEnabled(boolean enabled) {
+        this.enabled = enabled;
+    }
+
+    public String getCron() {
+        return cron;
+    }
+
+    public void setCron(String cron) {
+        this.cron = cron;
+    }
+}

src/main/java/ru/nbch/indicsearchservice/cron/CronChangeListener.java

@@ -0,0 +1,19 @@
+package ru.nbch.indicsearchservice.cron;
+
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+import ru.nbch.indicsearchservice.event.CronChangedEvent;
+
+@Component
+public class CronChangeListener {
+    private final CronJobManager cronJobManager;
+
+    public CronChangeListener(CronJobManager cronJobManager) {
+        this.cronJobManager = cronJobManager;
+    }
+
+    @EventListener
+    public void onCronChanged(CronChangedEvent event) {
+        cronJobManager.rescheduleNow(event.jobName(), event.newCron());
+    }
+}

src/main/java/ru/nbch/indicsearchservice/cron/CronJob.java

@@ -0,0 +1,5 @@
+package ru.nbch.indicsearchservice.cron;
+
+public interface CronJob {
+    CronJobName getCronJobName();
+}

src/main/java/ru/nbch/indicsearchservice/cron/CronJobManager.java

@@ -0,0 +1,78 @@
+package ru.nbch.indicsearchservice.cron;
+
+import jakarta.annotation.PostConstruct;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
+import java.util.function.Supplier;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.stereotype.Component;
+import ru.nbch.indicsearchservice.config.settings.ApplicationSettings;
+import ru.nbch.indicsearchservice.service.SettingsService;
+
+@Component
+@EnableScheduling
+public class CronJobManager {
+    private final TaskScheduler taskScheduler;
+    private final ApplicationSettings settings;
+    private final List<CronJob> cronJobs;
+    private final SettingsService settingsService;
+
+    private final Map<CronJobName, Runnable> tasks = new ConcurrentHashMap<>();
+    private final Map<CronJobName, ScheduledFuture<?>> futures = new ConcurrentHashMap<>();
+
+    public CronJobManager(TaskScheduler taskScheduler,
+                          ApplicationSettings settings,
+                          List<CronJob> cronJobs,
+                          SettingsService settingsService) {
+        this.taskScheduler = taskScheduler;
+        this.settings = settings;
+        this.cronJobs = cronJobs;
+        this.settingsService = settingsService;
+
+        for (CronJob cronJob : cronJobs) {
+            if (cronJob instanceof Runnable runnableCronJob) {
+                tasks.put(cronJob.getCronJobName(), runnableCronJob);
+            }
+        }
+    }
+
+    @PostConstruct
+    public void scheduleAll() {
+        for (CronJob cronJob : cronJobs) {
+            if (cronJob instanceof Runnable) {
+                if (cronJob instanceof JobCleaner) {
+                    Optional<String> cron = settingsService.getOptionalSettingsValueByName("jobCleaner.cron");
+                    if (cron.isPresent()) {
+                        settings.getJobCleaner().setCron(cron.get());
+                        schedule(cronJob.getCronJobName(), cron::get);
+                    }
+                } else {
+                    return;
+                }
+            }
+        }
+    }
+
+    private void schedule(CronJobName jobName, Supplier<String> cronSupplier) {
+        Runnable task = tasks.get(jobName);
+        CronTrigger trigger = new CronTrigger(cronSupplier.get());
+        ScheduledFuture<?> f = taskScheduler.schedule(task, trigger);
+        futures.put(jobName, f);
+    }
+
+    public void rescheduleNow(CronJobName jobName, String newCron) {
+        ScheduledFuture<?> f = futures.get(jobName);
+        if (f != null) {
+            f.cancel(false);
+        }
+        CronTrigger trigger = CronTrigger.resumeLenientExecution(newCron, Instant.now());
+        ScheduledFuture<?> nf = taskScheduler.schedule(tasks.get(jobName), trigger);
+        futures.put(jobName, nf);
+    }
+}

src/main/java/ru/nbch/indicsearchservice/cron/CronJobName.java

@@ -0,0 +1,5 @@
+package ru.nbch.indicsearchservice.cron;
+
+public enum CronJobName {
+    JOB_CLEANER
+}

src/main/java/ru/nbch/indicsearchservice/cron/JobCleaner.java

@@ -0,0 +1,97 @@
+package ru.nbch.indicsearchservice.cron;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.time.LocalDateTime;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.commons.io.FilenameUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+import ru.nbch.indicsearchservice.config.settings.ApplicationSettings;
+import ru.nbch.indicsearchservice.exception.InternalException;
+import ru.nbch.indicsearchservice.model.SearchJob;
+import ru.nbch.indicsearchservice.service.SettingsService;
+
+@Component
+public class JobCleaner implements Runnable, CronJob {
+    private static final Logger log = LoggerFactory.getLogger(JobCleaner.class);
+
+    private final ApplicationSettings settings;
+    private final Map<UUID, SearchJob> searchJobs;
+
+    public JobCleaner(ApplicationSettings settings,
+                      SettingsService settingsService,
+                      @Qualifier("searchJobsMap") Map<UUID, SearchJob> searchJobs) {
+        this.settings = settings;
+        this.searchJobs = searchJobs;
+
+        Optional<String> enabled = settingsService.getOptionalSettingsValueByName("jobCleaner.enabled");
+        if (enabled.isPresent()) {
+            settings.getJobCleaner().setEnabled(Boolean.parseBoolean(enabled.get()));
+        } else {
+            log.warn("There is no jobCleaner.enabled in SETTINGS table!");
+        }
+    }
+
+    @Override
+    public void run() {
+        LocalDateTime hourBefore = LocalDateTime.now().minusMinutes(1);
+
+        // Очищаем Map
+        Iterator<Map.Entry<UUID, SearchJob>> iterator = searchJobs.entrySet().iterator();
+        while (iterator.hasNext()) {
+            Map.Entry<UUID, SearchJob> entry = iterator.next();
+            SearchJob searchJob = entry.getValue();
+
+            if (searchJob.getFinishedAt().isBefore(hourBefore)) {
+                iterator.remove();
+            }
+        }
+
+        // Очищаем папку
+        for (File file : Optional.ofNullable(settings.getStorageDirectory().listFiles()).orElse(new File[0])) {
+            String fileNameWithoutExtension = FilenameUtils.getBaseName(file.getName());
+            String fileExtension = FilenameUtils.getExtension(file.getName());
+
+            UUID parsedUuid = null;
+            if (fileNameWithoutExtension.startsWith("AGGREGATED-")) {
+                try {
+                    parsedUuid = UUID.fromString(fileNameWithoutExtension.replace("AGGREGATED-", ""));
+                } catch (Exception e) {
+                    log.warn("Could not parse UUID from file name: {}, removing", fileNameWithoutExtension, e);
+                }
+            } else if (fileExtension.startsWith("json") || fileExtension.startsWith("xlsx")) {
+                try {
+                    parsedUuid = UUID.fromString(fileNameWithoutExtension);
+                } catch (Exception e) {
+                    log.warn("Could not parse UUID from file name: {}, removing", fileNameWithoutExtension, e);
+                }
+            } else {
+                try {
+                    parsedUuid = UUID.fromString(fileNameWithoutExtension.substring(fileNameWithoutExtension.lastIndexOf('_') + 1));
+                } catch (Exception e) {
+                    log.warn("Could not parse UUID from file name: {}, removing", fileNameWithoutExtension, e);
+                }
+            }
+
+            if (parsedUuid == null || !searchJobs.containsKey(parsedUuid)) {
+                try {
+                    Files.deleteIfExists(file.toPath());
+                } catch (IOException e) {
+                    throw new InternalException("Failed to delete file " + file + " " + e);
+                }
+            }
+        }
+    }
+
+    @Override
+    public CronJobName getCronJobName() {
+        return CronJobName.JOB_CLEANER;
+    }
+}

src/main/java/ru/nbch/indicsearchservice/event/CronChangedEvent.java

@@ -0,0 +1,9 @@
+package ru.nbch.indicsearchservice.event;
+
+import ru.nbch.indicsearchservice.cron.CronJobName;
+
+public record CronChangedEvent(
+    CronJobName jobName,
+    String newCron
+) {
+}

src/main/java/ru/nbch/indicsearchservice/manager/IndicSearchJobManager.java

@@ -2,9 +2,8 @@ package ru.nbch.indicsearchservice.manager;
 
 import java.util.Map;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
 import ru.nbch.indicsearchservice.dto.request.AddSearchJobRequestDto;
@@ -15,11 +14,15 @@ import ru.nbch.indicsearchservice.processor.IndicSearchProcessor;
 
 @Component
 public class IndicSearchJobManager {
-    private final ExecutorService executor = Executors.newFixedThreadPool(5);
+    private final ExecutorService searchJobExecutor;
-    private final Map<UUID, SearchJob> searchJobs = new ConcurrentHashMap<>();
+    private final Map<UUID, SearchJob> searchJobs;
     private final IndicSearchProcessor indicSearchProcessor;
 
-    public IndicSearchJobManager(IndicSearchProcessor indicSearchProcessor) {
+    public IndicSearchJobManager(@Qualifier("searchJobExecutor") ExecutorService searchJobExecutor,
+                                 @Qualifier("searchJobsMap") Map<UUID, SearchJob> searchJobs,
+                                 IndicSearchProcessor indicSearchProcessor) {
+        this.searchJobExecutor = searchJobExecutor;
+        this.searchJobs = searchJobs;
         this.indicSearchProcessor = indicSearchProcessor;
     }
 
@@ -34,7 +37,7 @@ public class IndicSearchJobManager {
             request.getMemberCode()
         );
         searchJobs.put(uuid, newSearchJob);
-        executor.submit(() -> indicSearchProcessor.aggregateTables(newSearchJob));
+        searchJobExecutor.submit(() -> indicSearchProcessor.aggregateTables(newSearchJob));
 
         return uuid;
     }

src/main/java/ru/nbch/indicsearchservice/processor/IndicSearchProcessor.java

@@ -170,9 +170,6 @@ public class IndicSearchProcessor {
         } catch (IOException e) {
             log.error("Failed to get temp excel file size!", e);
             try {
-                if (aggregatedFile != null) {
-                    Files.deleteIfExists(aggregatedFile);
-                }
                 Files.deleteIfExists(exportedFile);
             } catch (IOException ignored) {
             }
@@ -185,18 +182,12 @@ public class IndicSearchProcessor {
                 @Override
                 public void close() throws IOException {
                     super.close();
-                    if (aggregatedFile != null) {
-                        Files.deleteIfExists(aggregatedFile);
-                    }
                     Files.deleteIfExists(exportedFile);
                 }
             });
         } catch (IOException e) {
             log.error("Failed to create exported file resource!", e);
             try {
-                if (aggregatedFile != null) {
-                    Files.deleteIfExists(aggregatedFile);
-                }
                 Files.deleteIfExists(exportedFile);
             } catch (IOException ignored) {
             }

src/main/java/ru/nbch/indicsearchservice/service/SettingsService.java

@@ -11,6 +11,8 @@ import java.util.Optional;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import ru.nbch.indicsearchservice.config.settings.ApplicationSettings;
+import ru.nbch.indicsearchservice.cron.CronJobName;
+import ru.nbch.indicsearchservice.event.CronChangedEvent;
 import ru.nbch.indicsearchservice.event.DataSourceChangedEvent;
 import ru.nbch.indicsearchservice.exception.InternalException;
 import ru.nbch.indicsearchservice.model.Settings;
@@ -19,6 +21,9 @@ import ru.nbch.indicsearchservice.util.DataSourceUpdater;
 
 @Service
 public class SettingsService {
+    private static final String STORAGE_SETTINGS = "indic-search-service-storage"; // todo: change key
+    private static final String JOB_CLEANER_SETTINGS = "jobCleaner";
+
     private final SettingsMapper settingsMapper;
     private final ApplicationSettings localSettings;
     private final ApplicationEventPublisher publisher;
@@ -31,12 +36,6 @@ public class SettingsService {
         this.publisher = publisher;
     }
 
-    public String getSettingsValueByName(String name) {
-        return settingsMapper.findByName(name)
-            .orElseThrow(() -> new InternalException("Setting with key: " + name + " was not found in database!"))
-            .getValue();
-    }
-
     public Map<String, Optional<String>> getSettingsValuesByNames(List<String> names) {
         if (names == null || names.isEmpty()) {
             return Map.of();
@@ -68,17 +67,29 @@ public class SettingsService {
     }
 
     public void updateLocalSettings() {
-        Optional<String> newStorageDirectory = getOptionalSettingsValueByName("indic-search-service-storage"); // todo: change key
+        Map<String, Optional<String>> newSettings = getSettingsValuesByNames(List.of(
+            STORAGE_SETTINGS,
+            JOB_CLEANER_SETTINGS + ".enabled",
+            JOB_CLEANER_SETTINGS + ".cron"
+        ));
+
+        if (newSettings.get(STORAGE_SETTINGS).isPresent()) {
+            String newStorageDirectory = newSettings.get(STORAGE_SETTINGS).get();
+            try {
+                Files.createDirectories(Paths.get(newStorageDirectory));
+            } catch (IOException e) {
+                throw new InternalException("Failed to create directory " + newStorageDirectory + " " + e);
+            }
+            localSettings.setStorageDirectory(new File(newStorageDirectory));
+        }
+
+        if (newSettings.get(JOB_CLEANER_SETTINGS + ".enabled").isPresent()) {
+            localSettings.getJobCleaner().setEnabled(Boolean.parseBoolean(newSettings.get(JOB_CLEANER_SETTINGS + ".enabled").get()));
+        }
+        if (newSettings.get(JOB_CLEANER_SETTINGS + ".cron").isPresent()) {
+            publisher.publishEvent(new CronChangedEvent(CronJobName.JOB_CLEANER, newSettings.get(JOB_CLEANER_SETTINGS + ".cron").get()));
+        }
 
         publisher.publishEvent(new DataSourceChangedEvent(DataSourceUpdater.DataSourceType.INDIC));
-
-        if (newStorageDirectory.isPresent()) {
-            try {
-                Files.createDirectories(Paths.get(newStorageDirectory.get()));
-            } catch (IOException e) {
-                throw new InternalException("Failed to create directory " + newStorageDirectory.get() + " " + e);
-            }
-            localSettings.setStorageDirectory(new File(newStorageDirectory.get()));
-        }
     }
 }