View Javadoc
1   /*
2    * Copyright (C) 2026 B3Partners B.V.
3    *
4    * SPDX-License-Identifier: MIT
5    */
6   package org.tailormap.api.service;
7   
8   import ch.rasc.sse.eventbus.SseEvent;
9   import ch.rasc.sse.eventbus.SseEventBus;
10  import jakarta.annotation.PostConstruct;
11  import java.io.File;
12  import java.io.IOException;
13  import java.io.UncheckedIOException;
14  import java.lang.invoke.MethodHandles;
15  import java.nio.charset.StandardCharsets;
16  import java.nio.file.Files;
17  import java.nio.file.Path;
18  import java.time.Instant;
19  import java.util.ArrayList;
20  import java.util.Comparator;
21  import java.util.List;
22  import java.util.Map;
23  import java.util.Objects;
24  import java.util.Set;
25  import java.util.concurrent.TimeUnit;
26  import java.util.concurrent.atomic.AtomicInteger;
27  import java.util.stream.Stream;
28  import org.apache.commons.lang3.StringUtils;
29  import org.geotools.api.data.FeatureEvent;
30  import org.geotools.api.data.FileDataStore;
31  import org.geotools.api.data.Query;
32  import org.geotools.api.data.SimpleFeatureSource;
33  import org.geotools.api.data.SimpleFeatureStore;
34  import org.geotools.api.data.Transaction;
35  import org.geotools.api.feature.simple.SimpleFeatureType;
36  import org.geotools.api.filter.Filter;
37  import org.geotools.api.filter.FilterFactory;
38  import org.geotools.api.filter.sort.SortOrder;
39  import org.geotools.data.DataUtilities;
40  import org.geotools.data.DefaultTransaction;
41  import org.geotools.data.csv.CSVDataStoreFactory;
42  import org.geotools.data.geojson.store.GeoJSONDataStoreFactory;
43  import org.geotools.data.shapefile.ShapefileDumper;
44  import org.geotools.factory.CommonFactoryFinder;
45  import org.geotools.feature.SchemaException;
46  import org.geotools.geopkg.FeatureEntry;
47  import org.geotools.geopkg.GeoPackage;
48  import org.geotools.util.factory.GeoTools;
49  import org.jspecify.annotations.NonNull;
50  import org.jspecify.annotations.Nullable;
51  import org.slf4j.Logger;
52  import org.slf4j.LoggerFactory;
53  import org.springframework.beans.factory.annotation.Qualifier;
54  import org.springframework.beans.factory.annotation.Value;
55  import org.springframework.scheduling.annotation.Async;
56  import org.springframework.scheduling.annotation.Scheduled;
57  import org.springframework.stereotype.Service;
58  import org.springframework.transaction.annotation.Transactional;
59  import org.tailormap.api.controller.LayerExtractController;
60  import org.tailormap.api.geotools.collection.ProgressReportingFeatureCollection;
61  import org.tailormap.api.geotools.data.excel.ExcelDataStore;
62  import org.tailormap.api.geotools.data.excel.ExcelDataStoreFactory;
63  import org.tailormap.api.geotools.featuresources.FeatureSourceFactoryHelper;
64  import org.tailormap.api.persistence.TMFeatureType;
65  import org.tailormap.api.util.UUIDv7;
66  import org.tailormap.api.viewer.model.ServerSentEventResponse;
67  import tools.jackson.databind.SerializationFeature;
68  import tools.jackson.databind.json.JsonMapper;
69  
70  @Service
71  public class CreateLayerExtractService {
72    private static final Logger logger =
73        LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
74    private final SseEventBus eventBus;
75    private final JsonMapper jsonMapper;
76    private final FeatureSourceFactoryHelper featureSourceFactoryHelper;
77    private final ZipService zipService;
78    private final FilterFactory ff = CommonFactoryFinder.getFilterFactory(GeoTools.getDefaultHints());
79  
80    private static final String EXTRACT_SUBDIRECTORY = "tm-extracts";
81    // we can safely use the tmp dir as a default here because we are running in a docker container without a shell so
82    // access is limited
83    // Base directory from config; actual export dir is <base>/tm-extracts
84    @Value("${tailormap-api.extract.location:#{systemProperties['java.io.tmpdir']}}")
85    private String exportFilesBaseLocation;
86  
87    private String exportFilesLocation;
88  
89    @Value("${tailormap-api.extract.cleanup-minutes:120}")
90    private int cleanupIntervalMinutes;
91  
92    @Value("#{T(java.lang.Math).max(1, ${tailormap-api.extract.progress-report-interval:100})}")
93    private int progressReportInterval;
94  
95    @Value("${tailormap-api.features.wfs_count_exact:false}")
96    private boolean exactWfsCounts;
97  
98    @PostConstruct
99    void initializeExtractDirectory() {
100     try {
101       Path exportRoot = Path.of(exportFilesBaseLocation, EXTRACT_SUBDIRECTORY);
102       Files.createDirectories(exportRoot);
103       this.exportFilesLocation = exportRoot.toRealPath().toString();
104       logger.info("Using extract output directory: {}", this.exportFilesLocation);
105     } catch (IOException e) {
106       throw new UncheckedIOException(
107           "Failed to initialize extract directory under base path: " + exportFilesBaseLocation, e);
108     }
109   }
110 
111   public CreateLayerExtractService(
112       @Qualifier("viewerSseEventBus") SseEventBus eventBus,
113       JsonMapper jsonMapper,
114       FeatureSourceFactoryHelper featureSourceFactoryHelper,
115       ZipService zipService) {
116     this.eventBus = eventBus;
117     this.featureSourceFactoryHelper = featureSourceFactoryHelper;
118     this.zipService = zipService;
119     // force unindented/single line output for SSE messages, because we may have set
120     // spring.jackson.serialization.indent_output=true for debugging/development/test
121     if (jsonMapper.isEnabled(SerializationFeature.INDENT_OUTPUT)) {
122       this.jsonMapper = jsonMapper
123           .rebuild()
124           .configure(SerializationFeature.INDENT_OUTPUT, false)
125           .build();
126     } else {
127       this.jsonMapper = jsonMapper;
128     }
129   }
130 
131   public String getExportFilesLocation() {
132     return exportFilesLocation;
133   }
134 
135   private void emitError(@NonNull String clientId, String details) {
136     eventBus.handleEvent(SseEvent.builder()
137         .addClientId(clientId)
138         .data(jsonMapper.writeValueAsString(new ServerSentEventResponse()
139             .eventType(ServerSentEventResponse.EventTypeEnum.EXTRACT_FAILED)
140             .id(UUIDv7.randomV7())
141             .details(Map.of(
142                 "message", "An error occurred during extract creation", "explanation", details))))
143         .build());
144   }
145 
146   public void emitProgress(
147       @NonNull String clientId,
148       @Nullable String fileId,
149       int progress,
150       boolean completed,
151       @Nullable String message) {
152     message = StringUtils.isBlank(message) ? "Extract task started" : message;
153     fileId = StringUtils.isBlank(fileId) ? "" : fileId;
154     logger.debug("Emitting progress {}% for client [{}], message: '{}'", progress, clientId, message);
155 
156     eventBus.handleEvent(SseEvent.builder()
157         .addClientId(clientId)
158         .data(jsonMapper.writeValueAsString(new ServerSentEventResponse()
159             .eventType(
160                 completed
161                     ? ServerSentEventResponse.EventTypeEnum.EXTRACT_COMPLETED
162                     : ServerSentEventResponse.EventTypeEnum.EXTRACT_PROGRESS)
163             .id(UUIDv7.randomV7())
164             .details(Map.of(
165                 "progress",
166                 progress,
167                 "message",
168                 completed ? "Extract task completed" : message,
169                 "downloadId",
170                 fileId))))
171         .build());
172   }
173 
174   /**
175    * Check that the sse client id is valid and exists.
176    *
177    * @param clientId the SSE client id
178    * @throws IllegalArgumentException when the SSE client id is invalid or not found on the event bus
179    */
180   public void validateClientId(@NonNull String clientId) throws IllegalArgumentException {
181     if (!clientId.matches("[A-Za-z0-9_-]+")) {
182       logger.warn("Invalid clientId for SSE connection: {}", clientId);
183       throw new IllegalArgumentException("Invalid clientId");
184     }
185 
186     // validate the given clientId is known on the event bus
187     this.eventBus.getAllClientIds().stream()
188         .filter(id -> Objects.equals(id, clientId))
189         .findFirst()
190         .ifPresentOrElse(id -> logger.debug("Validated clientId {}", id), () -> {
191           throw new IllegalArgumentException("No active subscription found for clientId " + clientId);
192         });
193   }
194 
195   /**
196    * Create a validated filename for an extract. The naming follows the pattern
197    * {@code "%s_%s_%s%s".formatted(sourceFT.getName(), clientId, UUIDv7.randomV7(), outputFormat.getExtension()) }
198    * where the first part is the source feature type name (this is cleaned from some characters), the second part is
199    * the SSE client id, the third part is a random UUIDv7 and the fourth part is the file extension based on the
200    * requested output format.
201    *
202    * @param clientId the SSE client id
203    * @param sourceFT the source featuretype for the extract
204    * @param outputFormat the required format of the extract
205    * @return the filename used to create an extract
206    * @throws IllegalArgumentException when the SSE clientId is invalid or not found on the event bus
207    */
208   public String createExtractFilename(
209       @NonNull String clientId,
210       @NonNull TMFeatureType sourceFT,
211       LayerExtractController.@NonNull ExtractOutputFormat outputFormat)
212       throws IllegalArgumentException {
213 
214     this.validateClientId(clientId);
215 
216     String cleanFTName = sourceFT.getName();
217     if (cleanFTName.contains(":")) {
218       // clip off the WFS namespace part
219       cleanFTName = cleanFTName.substring(cleanFTName.lastIndexOf(":") + 1);
220       // remove: '.' and '_' which are used as separators in the filename and could cause issues when parsing the
221       // filename later on
222       cleanFTName = cleanFTName.replaceAll("[._]", "");
223     }
224     return "%s_%s_%s%s".formatted(cleanFTName, clientId, UUIDv7.randomV7(), outputFormat.getExtension());
225   }
226 
227   @Async("extractTaskExecutor")
228   @Transactional
229   public void createLayerExtract(
230       @NonNull String clientId,
231       @NonNull TMFeatureType inputTmFeatureType,
232       @NonNull Set<String> attributes,
233       @Nullable Filter filter,
234       String sortBy,
235       SortOrder sortOrder,
236       LayerExtractController.@NonNull ExtractOutputFormat extractOutputFormat,
237       @NonNull String outputFileName) {
238 
239     this.emitProgress(clientId, outputFileName, 0, false, "Starting extract");
240 
241     switch (extractOutputFormat) {
242       case GEOPACKAGE ->
243         this.handleGeoPackage(
244             clientId, inputTmFeatureType, attributes, filter, sortBy, sortOrder, outputFileName);
245       case SHAPE ->
246         this.handleWithShapeDumper(
247             clientId, inputTmFeatureType, attributes, filter, sortBy, sortOrder, outputFileName);
248       case CSV, GEOJSON, XLSX ->
249         this.handleSingleFileFormats(
250             clientId,
251             inputTmFeatureType,
252             attributes,
253             filter,
254             sortBy,
255             sortOrder,
256             extractOutputFormat,
257             outputFileName);
258     }
259   }
260 
261   private void handleGeoPackage(
262       @NonNull String clientId,
263       @NonNull TMFeatureType inputTmFeatureType,
264       @NonNull Set<String> attributes,
265       Filter filter,
266       String sortBy,
267       SortOrder sortOrder,
268       @NonNull String outputFileName) {
269 
270     SimpleFeatureSource inputFeatureSource = null;
271     File outputFile;
272     try {
273       outputFile = getValidatedOutputFile(outputFileName);
274       if (!logger.isDebugEnabled()) {
275         // delete in production after JVM exit because the event bus will be reset when the JVM exits, and then
276         // we are unlikely to have a reference to the file anymore.
277         // In debug/development mode we want to keep the file for inspection.
278         outputFile.deleteOnExit();
279       }
280     } catch (IOException e) {
281       emitError(clientId, e.getMessage());
282       logger.error("Creating extract failed", e);
283       return;
284     }
285 
286     try (GeoPackage geopkg = new GeoPackage(outputFile)) {
287       geopkg.init();
288 
289       inputFeatureSource = featureSourceFactoryHelper.openGeoToolsFeatureSource(inputTmFeatureType);
290 
291       Query q = createQuery(inputFeatureSource, attributes, filter, sortBy, sortOrder);
292 
293       int featCount = getFeatureCount(inputFeatureSource, q);
294       if (featCount < 0) {
295         logger.warn("Could not determine feature count for extract, progress reporting will be inaccurate");
296       }
297       final boolean hasKnownFeatureCount = featCount > 0;
298 
299       SimpleFeatureType fType =
300           DataUtilities.createSubType(inputFeatureSource.getSchema(), attributes.toArray(new String[0]));
301 
302       FeatureEntry entry = new FeatureEntry();
303       entry.setTableName(fType.getTypeName());
304       entry.setDescription(fType.getTypeName());
305 
306       AtomicInteger lastProgress = new AtomicInteger(0);
307       geopkg.add(
308           entry,
309           new ProgressReportingFeatureCollection(
310               inputFeatureSource.getFeatures(q), progressReportInterval, processed -> {
311                 int progress = hasKnownFeatureCount ? (int) ((processed / (double) featCount) * 99) : 0;
312                 lastProgress.set(progress);
313                 String progressMessage = hasKnownFeatureCount
314                     ? "Extracting geopackage: %d/%d features processed"
315                         .formatted(processed, featCount)
316                     : "Extracting geopackage: %d features processed".formatted(processed);
317                 this.emitProgress(clientId, outputFileName, progress, false, progressMessage);
318               }));
319 
320       this.emitProgress(
321           clientId,
322           outputFileName,
323           Math.max(99, lastProgress.get()),
324           false,
325           "Extract geopackage created successfully");
326       geopkg.createSpatialIndex(entry);
327       geopkg.close();
328       this.emitProgress(clientId, outputFileName, 100, true, "Extract completed successfully");
329     } catch (SchemaException | IOException | IllegalArgumentException e) {
330       emitError(clientId, e.getMessage());
331       logger.error("Creating extract failed", e);
332     } finally {
333       if (inputFeatureSource != null) {
334         try {
335           inputFeatureSource.getDataStore().dispose();
336         } catch (Exception e) {
337           logger.warn("Error disposing datastore for feature source {}", inputFeatureSource.getName(), e);
338         }
339       }
340     }
341   }
342 
343   private void handleSingleFileFormats(
344       @NonNull String clientId,
345       @NonNull TMFeatureType inputTmFeatureType,
346       @NonNull Set<String> attributes,
347       Filter filter,
348       String sortBy,
349       SortOrder sortOrder,
350       LayerExtractController.@NonNull ExtractOutputFormat extractOutputFormat,
351       @NonNull String outputFileName) {
352 
353     SimpleFeatureSource inputFeatureSource = null;
354     FileDataStore outputDataStore = null;
355     try (Transaction outputTransaction = new DefaultTransaction("tailormap-extract-output")) {
356       inputFeatureSource = featureSourceFactoryHelper.openGeoToolsFeatureSource(inputTmFeatureType);
357 
358       Query q = createQuery(inputFeatureSource, attributes, filter, sortBy, sortOrder);
359 
360       int featCount = getFeatureCount(inputFeatureSource, q);
361 
362       if (extractOutputFormat == LayerExtractController.ExtractOutputFormat.XLSX
363           && featCount >= ExcelDataStore.getMaxRows()) {
364         this.emitError(
365             clientId,
366             "Extract result contains %d features, which exceeds the maximum of %d for Excel output format. Please refine your filter or choose a different output format."
367                 .formatted(featCount, ExcelDataStore.getMaxRows()));
368         logger.error(
369             "Extract result contains {} features, which exceeds the maximum of {} for Excel output format. Please refine your filter or choose a different output format.",
370             featCount,
371             ExcelDataStore.getMaxRows());
372         // nothing we can do now as we are in a background/async process, so we just return without creating an
373         // extract file.
374         // The client will receive no extract completed event, and we have already emitted an error message with
375         // details.
376         return;
377       }
378 
379       outputDataStore = this.getExtractDataStore(
380           extractOutputFormat, outputFileName, clientId, inputTmFeatureType.getName());
381       SimpleFeatureType fType =
382           DataUtilities.createSubType(inputFeatureSource.getSchema(), attributes.toArray(new String[0]));
383       outputDataStore.createSchema(fType);
384 
385       final AtomicInteger featsAdded = new AtomicInteger();
386       if (outputDataStore.getFeatureSource() instanceof SimpleFeatureStore featureStore) {
387         featureStore.setTransaction(outputTransaction);
388         featureStore.addFeatureListener(event -> {
389           if (event.getType().equals(FeatureEvent.Type.ADDED)) {
390             featsAdded.getAndIncrement();
391           }
392           if (featCount > 0) {
393             if (featsAdded.get() % progressReportInterval == 0) {
394               this.emitProgress(
395                   clientId,
396                   outputFileName,
397                   (int) ((featsAdded.doubleValue() / featCount) * 100),
398                   false,
399                   null);
400             }
401           }
402         });
403         featureStore.addFeatures(inputFeatureSource.getFeatures(q));
404         outputTransaction.commit();
405         outputDataStore.dispose();
406         this.emitProgress(clientId, outputFileName, 100, true, "Extract completed successfully");
407       } else {
408         outputDataStore.dispose();
409         this.emitError(clientId, "Output datastore is not a SimpleFeatureStore, cannot write features");
410         logger.error("Output datastore is not a SimpleFeatureStore, cannot write features");
411       }
412     } catch (IOException | SchemaException | IllegalArgumentException | NullPointerException e) {
413       emitError(clientId, e.getMessage());
414       logger.error("Creating extract failed", e);
415     } finally {
416       if (outputDataStore != null) {
417         outputDataStore.dispose();
418       }
419       if (inputFeatureSource != null) {
420         try {
421           inputFeatureSource.getDataStore().dispose();
422         } catch (Exception e) {
423           logger.warn("Error disposing datastore for feature source {}", inputFeatureSource.getName(), e);
424         }
425       }
426     }
427   }
428 
429   private File getValidatedOutputFile(String outputFileName) throws IOException {
430     Path exportRoot = Path.of(exportFilesLocation).toRealPath();
431     Path outputPath = exportRoot.resolve(outputFileName).normalize();
432     if (!outputPath.startsWith(exportRoot)) {
433       throw new IOException("Invalid file path");
434     }
435     Path createdFilePath = Files.createFile(outputPath).toRealPath();
436     if (!createdFilePath.startsWith(exportRoot)) {
437       throw new IOException("Invalid file path");
438     }
439     return createdFilePath.toFile();
440   }
441 
442   /**
443    * Create a writable GeoTools {@link FileDataStore} for the requested extract format. The format must be must be
444    * supported by a {@link FileDataStore} implementation, for example CSV, Excel or GeoJSON. For unsupported formats
445    * (for example Shapefile) a custom handling is used in the calling method.
446    *
447    * <p>The output file is validated to ensure it is created under the configured extract location.
448    *
449    * @param extractOutputFormat the requested extract output format
450    * @param outputFileName the target output filename
451    * @param clientId the SSE client id, used for error reporting
452    * @param typeName the source feature type name, used to derive format-specific metadata (for example Excel sheet
453    *     name)
454    * @return a newly created {@link FileDataStore} configured for the requested format
455    * @throws IOException when the output file path is invalid or the datastore cannot be created
456    */
457   private FileDataStore getExtractDataStore(
458       LayerExtractController.ExtractOutputFormat extractOutputFormat,
459       String outputFileName,
460       String clientId,
461       String typeName)
462       throws IOException {
463 
464     final File outputFile = getValidatedOutputFile(outputFileName);
465     if (!logger.isDebugEnabled()) {
466       outputFile.deleteOnExit();
467     }
468 
469     return switch (extractOutputFormat) {
470       case CSV ->
471         (FileDataStore) new CSVDataStoreFactory()
472             .createNewDataStore(Map.of(
473                 CSVDataStoreFactory.FILE_PARAM.key,
474                 outputFile,
475                 CSVDataStoreFactory.STRATEGYP.key,
476                 CSVDataStoreFactory.WKT_STRATEGY,
477                 CSVDataStoreFactory.WKTP.key,
478                 "the_geom_wkt",
479                 CSVDataStoreFactory.WRITEPRJ.key,
480                 false,
481                 CSVDataStoreFactory.QUOTEALL.key,
482                 true));
483       case XLSX -> {
484         String processedTypeName = typeName.contains(":")
485             ? typeName.substring(typeName.lastIndexOf(":") + 1).replaceAll("[\\\\/?*\\[\\]:]", "_")
486             : typeName.replaceAll("[\\\\/?*\\[\\]:]", "_");
487         processedTypeName = processedTypeName.substring(0, Math.min(processedTypeName.length(), 31));
488         yield (FileDataStore) new ExcelDataStoreFactory()
489             .createNewDataStore(Map.of(
490                 ExcelDataStoreFactory.FILE_PARAM.key,
491                 outputFile,
492                 ExcelDataStoreFactory.SHEET_PARAM.key,
493                 processedTypeName));
494       }
495       case GEOJSON ->
496         (FileDataStore) new GeoJSONDataStoreFactory()
497             .createNewDataStore(Map.of(GeoJSONDataStoreFactory.FILE_PARAM.key, outputFile));
498       default -> {
499         emitError(clientId, "Unknown output format: " + extractOutputFormat);
500         logger.error("Unknown output format: {}", extractOutputFormat);
501         throw new IllegalArgumentException("Unknown output format: " + extractOutputFormat);
502       }
503     };
504   }
505 
506   private int getFeatureCount(SimpleFeatureSource source, Query query) throws IOException {
507     int count = source.getCount(query);
508     logger.debug("Filtered source counts {} features", count);
509     if (count < 0 && exactWfsCounts) {
510       count = source.getFeatures(query).size();
511     }
512     return count;
513   }
514 
515   private void handleWithShapeDumper(
516       @NonNull String clientId,
517       @NonNull TMFeatureType inputTmFeatureType,
518       @NonNull Set<String> attributes,
519       Filter filter,
520       String sortBy,
521       SortOrder sortOrder,
522       @NonNull String outputFileName) {
523     SimpleFeatureSource inputFeatureSource = null;
524     File outputDirectory = null;
525     try {
526       File outputFile = getValidatedOutputFile(outputFileName);
527       String baseName = outputFile
528           .getName()
529           .substring(
530               0,
531               outputFile
532                   .getName()
533                   .lastIndexOf(LayerExtractController.ExtractOutputFormat.SHAPE.getExtension()));
534       outputDirectory = outputFile
535           .getParentFile()
536           .toPath()
537           .resolve(baseName)
538           .toFile()
539           .getCanonicalFile();
540       if (!logger.isDebugEnabled()) {
541         // delete in production after JVM exit because the event bus will be reset when the JVM exits, and then
542         // we are unlikely to have a reference to the file anymore.
543         // In debug/development mode we want to keep the directory for inspection.
544         outputDirectory.deleteOnExit();
545       }
546       Files.createDirectories(outputDirectory.toPath());
547 
548       ShapefileDumper dumper = new ShapefileDumper(outputDirectory);
549       dumper.setCharset(StandardCharsets.UTF_8);
550       dumper.setEmptyShapefileAllowed(false);
551 
552       inputFeatureSource = featureSourceFactoryHelper.openGeoToolsFeatureSource(inputTmFeatureType);
553 
554       Query q = createQuery(inputFeatureSource, attributes, filter, sortBy, sortOrder);
555 
556       final int featCount = getFeatureCount(inputFeatureSource, q);
557       final boolean hasKnownFeatureCount = featCount > 0;
558 
559       AtomicInteger lastProgress = new AtomicInteger(0);
560 
561       dumper.dump(new ProgressReportingFeatureCollection(
562           inputFeatureSource.getFeatures(q), progressReportInterval, processed -> {
563             int progress = hasKnownFeatureCount ? (int) ((processed / (double) featCount) * 99) : 0;
564             lastProgress.set(progress);
565             String progressMessage = hasKnownFeatureCount
566                 ? "Extracting shapes: %d/%d features processed".formatted(processed, featCount)
567                 : "Extracting shapes: %d features processed".formatted(processed);
568             this.emitProgress(clientId, outputFileName, progress, false, progressMessage);
569           }));
570       this.emitProgress(
571           clientId,
572           outputFileName,
573           Math.max(99, lastProgress.get()),
574           false,
575           "Extract shapes dumped successfully");
576 
577       zipService.zipDirectory(outputDirectory.toPath(), outputFile.toPath());
578       this.emitProgress(clientId, outputFileName, 100, true, "Extract completed successfully");
579     } catch (IOException | IllegalArgumentException e) {
580       emitError(clientId, e.getMessage());
581       logger.error("Creating extract failed", e);
582     } finally {
583       if (outputDirectory != null) {
584         try {
585           deleteDirectoryRecursively(outputDirectory.toPath());
586         } catch (IOException e) {
587           logger.error("Failed to delete output directory {}", outputDirectory, e);
588         }
589       }
590       if (inputFeatureSource != null) {
591         try {
592           inputFeatureSource.getDataStore().dispose();
593         } catch (Exception e) {
594           logger.warn("Error disposing datastore for feature source {}", inputFeatureSource.getName(), e);
595         }
596       }
597     }
598   }
599 
600   private Query createQuery(
601       SimpleFeatureSource inputFeatureSource,
602       Set<String> attributes,
603       Filter filter,
604       String sortBy,
605       SortOrder sortOrder) {
606     Query q = new Query(inputFeatureSource.getName().toString());
607     if (!attributes.isEmpty()) {
608       q.setPropertyNames(attributes.toArray(new String[0]));
609     }
610 
611     if (filter != null) {
612       q.setFilter(filter);
613     }
614     if (!StringUtils.isBlank(sortBy)) {
615       q.setSortBy(ff.sort(sortBy, Objects.requireNonNullElse(sortOrder, SortOrder.ASCENDING)));
616     }
617     return q;
618   }
619 
620   /**
621    * Cleanup expired extract files. Filenames are created in {@link CreateLayerExtractService#createExtractFilename }
622    * and follow the pattern {@code "%s_%s_%s.%s".formatted(sourceFT.getName(), clientId, UUIDv7.randomV7(),
623    * outputFormat.getExtension()) }
624    */
625   @Scheduled(fixedDelay = 5, timeUnit = TimeUnit.MINUTES, initialDelay = 15)
626   public void cleanupExpiredExtracts() {
627     logger.debug("Running expired extracts cleanup in {}", exportFilesLocation);
628     List<FileWithAttributes> oldDownloadFilesOnDisk = new ArrayList<>();
629     Set<String> validClientIds = eventBus.getAllClientIds();
630 
631     // list download files in export location and delete those that are not bound to an active sse stream client
632     try (Stream<Path> stream = Files.walk(Path.of(exportFilesLocation))) {
633       stream.filter(Files::isRegularFile).forEach(path -> {
634         File file = path.toFile();
635         String filename = file.getName();
636         String[] parts = filename.split("[_.]", -1);
637         if (parts.length < 4) {
638           logger.warn("Unexpected file in extract location: {}", filename);
639           return;
640         }
641         String clientId = parts[1];
642         if (!validClientIds.contains(clientId)) {
643           if (!file.delete()) {
644             logger.error("Failed to delete unattached extract file {}", filename);
645           }
646         } else {
647           try {
648             Instant timestampPart = UUIDv7.timestampAsInstant(UUIDv7.fromString(parts[2]));
649             oldDownloadFilesOnDisk.add(new FileWithAttributes(file, timestampPart, clientId));
650           } catch (IllegalArgumentException ignored) {
651             // not a valid v7 uuid
652           }
653         }
654       });
655 
656       try (Stream<Path> paths = Files.walk(Path.of(exportFilesLocation))) {
657         paths.filter(Files::isDirectory)
658             .filter(path -> !path.equals(Path.of(exportFilesLocation)))
659             .forEach(path -> {
660               logger.debug("Checking directory {} for expired extracts", path);
661               File file = path.toFile();
662               String filename = file.getName();
663               String[] parts = filename.split("[_]", -1);
664               if (parts.length < 3) {
665                 logger.warn("Unexpected directory in extract location: {}", filename);
666                 return;
667               }
668               String clientId = parts[1];
669               if (!validClientIds.contains(clientId)) {
670                 try {
671                   deleteDirectoryRecursively(file.toPath());
672                 } catch (IOException e) {
673                   logger.error("Failed to delete unattached extract directory {}", filename);
674                 }
675               } else {
676                 try {
677                   Instant timestampPart = UUIDv7.timestampAsInstant(UUIDv7.fromString(parts[2]));
678                   oldDownloadFilesOnDisk.add(new FileWithAttributes(file, timestampPart, clientId));
679                 } catch (IllegalArgumentException ignored) {
680                   // not a valid v7 uuid
681                 }
682               }
683             });
684       }
685 
686       // delete any files/directories are older than the cutoff
687       oldDownloadFilesOnDisk.stream()
688           .filter(f -> f.timestamp()
689               .isBefore(Instant.now().minusSeconds(TimeUnit.MINUTES.toSeconds(cleanupIntervalMinutes))))
690           .forEach(f -> {
691             if (f.file.isDirectory()) {
692               try {
693                 deleteDirectoryRecursively(f.file().toPath());
694               } catch (IOException ignored) {
695                 logger.warn("Failed to delete directory {}", f.file());
696               }
697             } else {
698               if (!f.file().delete()) {
699                 logger.error(
700                     "Failed to delete expired extract file {}",
701                     f.file().getName());
702               }
703             }
704           });
705     } catch (IOException e) {
706       logger.error("Error while cleaning up expired extracts", e);
707     }
708   }
709 
710   private void deleteDirectoryRecursively(Path directory) throws IOException {
711     try (Stream<Path> paths = Files.walk(directory)) {
712       paths.sorted(Comparator.reverseOrder()).forEach(path -> {
713         try {
714           logger.debug("Deleting path {}", path);
715           Files.deleteIfExists(path);
716         } catch (IOException e) {
717           throw new RuntimeException("Failed to delete path: " + path, e);
718         }
719       });
720     } catch (RuntimeException e) {
721       if (e.getCause() instanceof IOException ioException) {
722         throw ioException;
723       }
724       throw e;
725     }
726   }
727 
728   private record FileWithAttributes(File file, Instant timestamp, String clientId) {}
729 }