argos/dmp-backend/core/src/main/java/eu/eudat/service/storage/StorageFileCleanupTask.java

package eu.eudat.service.storage;

import eu.eudat.commons.fake.FakeRequestScope;
import eu.eudat.commons.scope.tenant.TenantScope;
import eu.eudat.data.StorageFileEntity;
import eu.eudat.data.TenantEntity;
import eu.eudat.data.TenantEntityManager;
import eu.eudat.model.StorageFile;
import eu.eudat.query.StorageFileQuery;
import eu.eudat.query.TenantQuery;
import gr.cite.tools.data.query.Ordering;
import gr.cite.tools.data.query.QueryFactory;
import gr.cite.tools.logging.LoggerService;
import jakarta.persistence.EntityManager;
import jakarta.persistence.EntityManagerFactory;
import jakarta.persistence.EntityTransaction;
import jakarta.persistence.OptimisticLockException;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.Closeable;
import java.io.IOException;
import java.time.Instant;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Service
@Scope(value = ConfigurableBeanFactory.SCOPE_SINGLETON)
public class StorageFileCleanupTask  implements Closeable, ApplicationListener<ApplicationReadyEvent> {
	private class CandidateInfo
	{
		private UUID id;
		private Instant createdAt;

		public UUID getId() {
			return id;
		}

		public void setId(UUID id) {
			this.id = id;
		}

		public Instant getCreatedAt() {
			return createdAt;
		}

		public void setCreatedAt(Instant createdAt) {
			this.createdAt = createdAt;
		}
	}

	private static final LoggerService logger = new LoggerService(LoggerFactory.getLogger(StorageFileCleanupTask.class));
	private final StorageFileCleanupProperties _config;
	private final ApplicationContext applicationContext;
	private ScheduledExecutorService scheduler;

	public StorageFileCleanupTask(
			StorageFileCleanupProperties config, 
			ApplicationContext applicationContext) {
		this._config = config;
		this.applicationContext = applicationContext;

		
	}

	@Override
	public void onApplicationEvent(ApplicationReadyEvent event) {
		long intervalSeconds = this._config .getIntervalSeconds();
		if (this._config .getEnable() && intervalSeconds > 0) {
			logger.info("File clean up run in {} seconds", intervalSeconds);

			scheduler = Executors.newScheduledThreadPool(1);
			scheduler.scheduleAtFixedRate(this::process, 10, intervalSeconds, TimeUnit.SECONDS);
		} else {
			scheduler = null;
		}
	}
	
	@Override
	public void close() throws IOException {
		if (scheduler != null) this.scheduler.close();
	}

	protected void process() {
		if (!this._config.getEnable()) return;
		try {
			Instant lastCandidateCreationTimestamp = null;
			while (true) {

				CandidateInfo candidate = this.candidate(lastCandidateCreationTimestamp);
				if (candidate == null) break;
				lastCandidateCreationTimestamp = candidate.getCreatedAt();

				logger.debug("Clean up file: {}", candidate.getId());
				
				boolean successfullyProcessed = this.processStorageFile(candidate.getId());
				if (!successfullyProcessed) {
					logger.error("Problem processing file cleanups. {}", candidate.getId());
				}
			}
		} catch (Exception ex) {
			logger.error("Problem processing file cleanups. Breaking for next interval", ex);
		}
	}

	private boolean processStorageFile(UUID fileId) {
		EntityTransaction transaction = null;
		EntityManager entityManager = null;
		boolean success = false;
		try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
			try {
				QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
				EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
				
				entityManager = entityManagerFactory.createEntityManager();
				
				transaction = entityManager.getTransaction();
				transaction.begin();

				StorageFileEntity item = queryFactory.query(StorageFileQuery.class).ids(fileId).isPurged(false).first();
				success = true;
				
				if (item != null) {
					TenantScope tenantScope = this.applicationContext.getBean(TenantScope.class);
					TenantEntityManager tenantEntityManager = this.applicationContext.getBean(TenantEntityManager.class);
					tenantEntityManager.setEntityManager(entityManager);
					
					StorageFileService storageFileService = this.applicationContext.getBean(StorageFileService.class);
					try {
						if (item.getTenantId() != null) {
							TenantEntity tenant = queryFactory.query(TenantQuery.class).ids(item.getTenantId()).first();
							tenantScope.setTempTenant(entityManager, tenant.getId(), tenant.getCode());
						} else {
							tenantScope.setTempTenant(entityManager, null, tenantScope.getDefaultTenantCode());
						}
						storageFileService.purgeSafe(fileId);
					} finally {
						tenantScope.removeTempTenant(entityManager);
					}
				}

				transaction.commit();
			} catch (OptimisticLockException ex) {
				// we get this if/when someone else already modified the notifications. We want to essentially ignore this, and keep working
				logger.debug("Concurrency exception getting file. Skipping: {} ", ex.getMessage());
				if (transaction != null) transaction.rollback();
				success = false;
			} catch (Exception ex) {
				logger.error("Problem getting list of file. Skipping: {}", ex.getMessage(), ex);
				if (transaction != null) transaction.rollback();
				success = false;
			} finally {
				if (entityManager != null) entityManager.close();
			}
		} catch (Exception ex) {
			logger.error("Problem getting list of file. Skipping: {}", ex.getMessage(), ex);
		}

		return success;
	}

	private CandidateInfo candidate(Instant lastCandidateCreationTimestamp) {
		EntityTransaction transaction = null;
		EntityManager entityManager = null;
		CandidateInfo candidate = null;
		try (FakeRequestScope fakeRequestScope = new FakeRequestScope()) {
			try {
				QueryFactory queryFactory = this.applicationContext.getBean(QueryFactory.class);
				EntityManagerFactory entityManagerFactory = this.applicationContext.getBean(EntityManagerFactory.class);
				entityManager = entityManagerFactory.createEntityManager();

				transaction = entityManager.getTransaction();
				transaction.begin();

				StorageFileQuery query = queryFactory.query(StorageFileQuery.class)
						.canPurge(true)
						.isPurged(false)
						.createdAfter(lastCandidateCreationTimestamp);
				query.setOrder(new Ordering().addAscending(StorageFile._createdAt));
				StorageFileEntity item = query.first();

				if (item != null) {
					entityManager.flush();

					candidate = new CandidateInfo();
					candidate.setId(item.getId());
					candidate.setCreatedAt(item.getCreatedAt());
				}

				transaction.commit();
			} catch (OptimisticLockException ex) {
				// we get this if/when someone else already modified the notifications. We want to essentially ignore this, and keep working
				logger.debug("Concurrency exception getting file. Skipping: {} ", ex.getMessage());
				if (transaction != null) transaction.rollback();
				candidate = null;
			} catch (Exception ex) {
				logger.error("Problem getting list of file. Skipping: {}", ex.getMessage(), ex);
				if (transaction != null) transaction.rollback();
				candidate = null;
			} finally {
				if (entityManager != null) entityManager.close();
			}
		} catch (Exception ex) {
			logger.error("Problem getting list of file. Skipping: {}", ex.getMessage(), ex);
		}

		return candidate;
	}
	
}