ScheduledAccessionArchiver.java

/*
 * Copyright 2026 Global Crop Diversity Trust
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.genesys.server.service.worker;

import java.io.BufferedWriter;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.zip.GZIPOutputStream;

import org.genesys.filerepository.InvalidRepositoryFileDataException;
import org.genesys.filerepository.InvalidRepositoryPathException;
import org.genesys.filerepository.NoSuchRepositoryFileException;
import org.genesys.filerepository.service.RepositoryService;
import org.genesys.server.component.security.AsAdminInvoker;
import org.genesys.server.service.AccessionService;
import org.genesys.server.service.filter.AccessionFilter;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Pageable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.opencsv.CSVWriter;
import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
import org.apache.commons.lang3.time.StopWatch;

@Component
@Slf4j
public class ScheduledAccessionArchiver implements InitializingBean {

	final Map<String, String> fields = new LinkedHashMap<>();

	@Autowired
	private RepositoryService repositoryService;

	@Autowired
	private AccessionService accessionService;

	@Autowired
	private AsAdminInvoker asAdminInvoker;

	@Override
	public void afterPropertiesSet() throws Exception {
		fields.put("instituteCode", "INSTCODE");
		fields.put("accessionNumber", "ACCENUMB");
		fields.put("doi", "DOI");
		fields.put("taxonomy.genus", "GENUS");
		fields.put("taxonomy.species", "SPECIES");
		fields.put("taxonomy.subtaxa", "SUBTAXA");
		fields.put("cropName", "CROPNAME");
		fields.put("acquisitionDate", "ACQDATE");
		fields.put("origCty", "ORIGCTY");
		fields.put("sampStat", "SAMPSTAT");
		fields.put("duplSite", "DUPLSITE");
		fields.put("latitude", "DECLATITUDE");
		fields.put("longitude", "DECLONGITUDE");
		fields.put("coll.collSrc", "COLLSRC");
		fields.put("storage", "STORAGE");
		fields.put("mlsStatus", "MLSSTATUS");

		final Path archivePath = Paths.get("/archive");
		asAdminInvoker.invoke(() -> repositoryService.ensureFolder(archivePath));
	}

	@Scheduled(cron = "0 0 0 1 * *")
	// @Scheduled(initialDelay =  5 * 1000, fixedRate = 30 * 60 * 1000) // Test every 30 min
	@SchedulerLock(name = "org.genesys.server.service.worker.AccessionArchivator", lockAtLeastFor = "PT5M", lockAtMostFor = "PT30M")
	public void archive() throws Exception {

		// File for /archive
		final var date = LocalDate.now(ZoneOffset.UTC);
		final String archiveFileName = String.format("genesys-archive-%04d-%02d-%02d.csv.gz", date.getYear(), date.getMonthValue(), date.getDayOfMonth());

		asAdminInvoker.invoke(() -> {
			try {
				var existingDumpFile = repositoryService.getFile(Path.of("/archive"), archiveFileName);
				if (existingDumpFile != null) {
					repositoryService.removeFile(existingDumpFile);
					log.warn("Existing dump file /archive/{} was removed.", archiveFileName);
				}
			} catch (NoSuchRepositoryFileException e) {
				log.warn("Preparing Genesys CSV dump file /archive/{}...", archiveFileName);
			}
			return true;
		});

		final Path tmpGz;
		try {
			tmpGz = Files.createTempFile("genesys-archive-", ".csv.gz");
		} catch (IOException e) {
			log.error("Cannot create temp gzip file for archiving", e);
			return;
		}

		var stopWatch = StopWatch.createStarted();
		var counter = new AtomicInteger(0);

		// Write TSV into gzip temp file
		try (var fos = new FileOutputStream(tmpGz.toFile());
			 var gout = new GZIPOutputStream(fos);
			 var bw = new BufferedWriter(new OutputStreamWriter(gout, StandardCharsets.UTF_8), 1024 * 1024); // 1M buffer
			 var csvWriter = new CSVWriter(bw, '\t', '"', '\\', "\n")) {

			// Headers
			csvWriter.writeNext(fields.values().toArray(new String[0]), false);

			var selectFields = fields.entrySet().stream().map(entry -> entry.getKey() + " " + entry.getValue()).collect(Collectors.toList());

			var selectedFieldNames = new ArrayList<>(fields.values());
			var columnCount = selectedFieldNames.size();

			asAdminInvoker.invoke(() -> {
				String[] row = new String[columnCount];
				stopWatch.split();;
				log.info("In {} initializing query for active accessions", stopWatch.formatSplitTime());
				accessionService.query(new AccessionFilter(false), selectFields, Pageable.unpaged(), true, (one) -> {
					for (int i = 0; i < columnCount; i++) {
						var val = one.get(selectedFieldNames.get(i));
						row[i] = val == null ? null : Objects.toString(val);
					}
					csvWriter.writeNext(row, false);
					counter.incrementAndGet();
					if (counter.get() % 100000 == 0) {
						stopWatch.split();
						log.warn("In {} processed {} accessions", stopWatch.formatSplitTime(), counter.get());
					}
				});
				return true;
			});
			csvWriter.flush();
		} catch (IOException ioe) {
			log.error("Error writing gzipped TSV archive to temp file {}", tmpGz, ioe);
			try {
				Files.deleteIfExists(tmpGz);
			} catch (IOException ex) {
				log.error("Error deleting temp file {}", tmpGz, ex);
			}
			return;
		} catch (Exception ex) {
			log.error("Error generating gzipped accession archive TSV", ex);
			try {
				Files.deleteIfExists(tmpGz);
			} catch (IOException ex2) {
				log.error("Error deleting temp file {}", tmpGz, ex2);
			}
			return;
		}

		stopWatch.stop();
		log.warn("In {} processed all {} active accession records.", stopWatch.formatTime(), counter.get());
		stopWatch.reset();
		stopWatch.start();

		asAdminInvoker.invoke(() -> {
			try (InputStream uploadIs = Files.newInputStream(tmpGz)) {
				var archiveFile = repositoryService.addFile(Paths.get("/archive"), archiveFileName, "application/gzip", uploadIs, null);
				log.warn("Uploaded monthly accession archive: {} as /archive/{} ({} rows)", archiveFile.getStoragePath(), archiveFileName, counter.get());
			} catch (InvalidRepositoryPathException | InvalidRepositoryFileDataException | IOException e) {
				log.error("Failed to upload archive to repository", e);
			} finally {
				try {
					Files.deleteIfExists(tmpGz);
				} catch (IOException ex) {
					log.error("Error deleting temp file {}", tmpGz, ex);
				}
			}
			return true;
		});

		stopWatch.stop();
		log.warn("Uploaded to storage in {}ms", stopWatch.getTime(TimeUnit.MILLISECONDS));
	}
}