D-Net

package eu.dnetlib.openaire.dsm;

import java.util.*;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;

import java.util.function.Function;

import java.util.stream.Collectors;

import java.util.stream.Stream;

import javax.annotation.PostConstruct;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.common.collect.Queues;

import com.google.common.util.concurrent.*;

import com.google.common.xml.XmlEscapers;

import eu.dnetlib.OpenaireExporterConfig;

import eu.dnetlib.enabling.datasources.common.*;

import eu.dnetlib.openaire.common.ISClient;

import eu.dnetlib.openaire.community.CommunityClient;

import eu.dnetlib.openaire.dsm.dao.*;

import eu.dnetlib.openaire.dsm.dao.utils.DsmMappingUtils;

import eu.dnetlib.openaire.dsm.dao.utils.IndexDsInfo;

import eu.dnetlib.openaire.dsm.dao.utils.IndexRecordsInfo;

import eu.dnetlib.openaire.dsm.domain.*;

import eu.dnetlib.openaire.dsm.domain.db.ApiDbEntry;

import eu.dnetlib.openaire.dsm.domain.db.DatasourceDbEntry;

import eu.dnetlib.openaire.dsm.domain.db.IdentityDbEntry;

import eu.dnetlib.openaire.vocabularies.Country;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.lang3.exception.ExceptionUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

import org.springframework.data.domain.Page;

import org.springframework.stereotype.Component;

import static eu.dnetlib.openaire.common.ExporterConstants.*;

import static eu.dnetlib.openaire.dsm.dao.utils.DsmMappingUtils.*;

@Component

@ConditionalOnProperty(value = "openaire.exporter.enable.dsm", havingValue = "true")

public class DsmCore {

	private static final Log log = LogFactory.getLog(DsmCore.class);

	@Autowired

	private MongoLoggerClient mongoLoggerClient;

	@Autowired

	private ISClient isClient;

	@Autowired

	private ObjectStoreClient objectStoreClient;

	@Autowired

	private DatasourceIndexClient datasourceIndexClient;

	@Autowired

	private VocabularyClient vocabularyClient;

	@Autowired

	private DatasourceDao dsDao;

	@Autowired

	private OpenaireExporterConfig config;

	@Autowired

	private CommunityClient communityClient;

	private ListeningExecutorService executor;

	@PostConstruct

	public void init() {

		executor = MoreExecutors.listeningDecorator(

				new ScheduledThreadPoolExecutor(config.getRequestWorkers(),

						new ThreadFactoryBuilder().setNameFormat("dsm-client-%d").build()));

	}

	public List<Country> listCountries() throws DsmException {

		try {

			return dsDao.listCountries();

		} catch (Throwable e) {

			log.error("error listing countries", e);

			throw e;

		}

	}

	public DatasourceResponse search(RequestSort requestSortBy, RequestSortOrder order, RequestFilter requestFilter, int page, int size)

			throws DsmException {

		try {

			final List<DatasourceInfo> datasourceInfo = Lists.newArrayList();

			final Queue<Throwable> errors = Queues.newLinkedBlockingQueue();

			final CountDownLatch outerLatch = new CountDownLatch(2);

			final Page<DatasourceDbEntry> dsPage = dsDao.search(requestSortBy, order, requestFilter, page, size);

			if (dsPage.getTotalElements() > 0 && dsPage.getNumberOfElements() > 0) {

				dsPage.forEach(d -> datasourceInfo.add(enrichDatasourceInfo(asDetails(d), outerLatch, errors)));

				waitLatch(outerLatch, errors, config.getRequestTimeout());

			}

			if (!errors.isEmpty()) {

				//TODO report on error metrics

				errors.forEach(e -> log.error(e));

			}

			return ResponseUtils.datasourceResponse(datasourceInfo, errors, dsPage.getTotalElements());

		} catch (Throwable e) {

			log.error("error searching datasources", e);

			throw e;

		}

	}

	public DatasourceResponse searchSnippet(RequestSort requestSortBy, RequestSortOrder order, RequestFilter requestFilter, int page, int size)

			throws DsmException {

		try {

			final Page<DatasourceDbEntry> dsPage = dsDao.search(requestSortBy, order, requestFilter, page, size);

			return ResponseUtils.datasourceResponse(dsPage.map(d -> asSnippet(d)).getContent(), Queues.newLinkedBlockingQueue(), dsPage.getTotalElements());

		} catch (Throwable e) {

			log.error("error searching datasources", e);

			throw e;

		}

	}

	public List<String> findBaseURLs(final RequestFilter requestFilter, final int page, final int size) throws DsmException {

		try {

			return dsDao.findApiBaseURLs(requestFilter, page, size);

		} catch (Throwable e) {

			log.error("error searching datasource base urls", e);

			throw e;

		}

	}

	public ApiDetailsResponse getApis(final String dsId) throws DsmException {

		try {

			final List<ApiDbEntry> apis = dsDao.getApis(dsId);

			final List<ApiDetails> api = apis.stream()

					.map(DsmMappingUtils::asDetails)

					.collect(Collectors.toList());

			return ResponseUtils.apiResponse(api, api.size());

		} catch (Throwable e) {

			log.error(String.format("error searching datasource api %s", dsId), e);

			throw e;

		}

	}

	public void setManaged(final String dsId, final boolean managed) throws DsmException {

		log.info(String.format("updated api '%s' managed with '%s'", dsId, managed));

		dsDao.setManaged(dsId, managed);

	}

	public boolean isManaged(final String dsId) throws DsmException {

		return dsDao.isManaged(dsId);

	}

	public boolean exist(final DatasourceDetails d) throws DsmException {

		return dsDao.existDs(d.getId());

	}

	public void save(final DatasourceDetails d) throws DsmException {

		try {

			dsDao.saveDs(asDbEntry(d));

			isClient.registerDS(d);

		} catch(Throwable e) {

			log.error(ExceptionUtils.getStackTrace(e));

			throw e;

		}

	}

	public void updateDatasource(final DatasourceDetailsUpdate d) throws DsmException, DsmNotFoundException {

		try {

			//initialize with current values from DB

			final Datasource ds = dsDao.getDs(d.getId());

			final DatasourceDbEntry dbEntry = (DatasourceDbEntry) ds;

			if (dbEntry == null) {

				throw new DsmNotFoundException(String.format("ds '%s' does not exist", d.getId()));

			}

			final DatasourceDbEntry update = asDbEntry(d);

			if (d.getIdentities() != null) {

				final Set<IdentityDbEntry> identities = (

						new HashSet<>(

								Stream.of(

										update.getIdentities(),

										dbEntry.getIdentities())

										.flatMap(Collection::stream)

										.collect(

												Collectors.toMap(

														i -> i.getIssuertype() + i.getPid(),

														Function.identity(),

														(i1, i2) -> i1))

										.values()));

				copyNonNullProperties(update, dbEntry);

				dbEntry.setIdentities(identities);

			} else {

				copyNonNullProperties(update, dbEntry);

			}

			dsDao.saveDs(dbEntry);

			isClient.updateDatasourceFields(d.getId(), asMapOfChanges(d));

		} catch (Throwable e) {

			log.error(ExceptionUtils.getStackTrace(e));

			throw e;

		}

	}

	@Deprecated

	public void updateDatasourcename(final String dsId, final String officialname, final String englishname) throws DsmException {

		log.info(String.format("updated datasource '%s' with officialname '%s' and englishname '%s'", dsId, officialname, englishname));

		dsDao.updateName(dsId, officialname, englishname);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(OFFICIAL_NAME, XmlEscapers.xmlContentEscaper().escape(officialname));

		changes.put(ENGLISH_NAME, XmlEscapers.xmlContentEscaper().escape(englishname));

		isClient.updateDatasourceFields(dsId, changes);

	}

	@Deprecated

	public void updateDatasourceLogoUrl(final String dsId, final String logourl) throws DsmException {

		log.info(String.format("updated datasource '%s' with logo URL '%s'", dsId, logourl));

		dsDao.updateLogoUrl(dsId, logourl);

	}

	@Deprecated

	public void updateCoordinates(final String dsId, final Double latitude, final Double longitude) throws DsmException {

		log.info(String.format("updated datasource '%s' with coordinates Lat:'%s', Lon:'%s'", dsId, latitude, longitude));

		dsDao.updateCoordinates(dsId, latitude, longitude);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(LATITUDE, XmlEscapers.xmlContentEscaper().escape(String.valueOf(latitude)));

		changes.put(LONGITUDE, XmlEscapers.xmlContentEscaper().escape(String.valueOf(longitude)));

		isClient.updateDatasourceFields(dsId, changes);

	}

	@Deprecated

	public void updateTimezone(final String dsId, final String timezone) throws DsmException {

		log.info(String.format("updated datasource '%s' timezone with '%s'", dsId, timezone));

		dsDao.updateTimezone(dsId, timezone);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(TIMEZONE, XmlEscapers.xmlContentEscaper().escape(timezone));

		isClient.updateDatasourceFields(dsId, changes);

	}

	@Deprecated

	public void updateDsTypology(final String dsId, final String typology) throws DsmException {

		log.info(String.format("updated datasource '%s' typology with '%s'", dsId, typology));

		dsDao.updateTypology(dsId, typology);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(TYPOLOGY, XmlEscapers.xmlContentEscaper().escape(typology));

		isClient.updateDatasourceFields(dsId, changes);

	}

	@Deprecated

	public void updateDsRegisteringUser(final String dsId, final String registeredBy) throws DsmException {

		log.info(String.format("setting datasource '%s' registering user with '%s'", dsId, registeredBy));

		dsDao.updateRegisteringUser(dsId, registeredBy);

	}

	@Deprecated

	public void updateDsPlatform(final String dsId, final String platform) throws DsmException {

		log.info(String.format("updated datasource '%s' platform with '%s'", dsId, platform));

		dsDao.updatePlatform(dsId, platform);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(PLATFORM, XmlEscapers.xmlContentEscaper().escape(platform)); // this is not a typo, Repository profiles map the platform in the DATASOURCE_TYPE field.

		isClient.updateDatasourceFields(dsId, changes);

	}

	//TODO remove if unused

	public void deleteDs(final String dsId) throws DsmException {

		log.info(String.format("deleted datasource '%s'", dsId));

		dsDao.deleteDs(dsId);

	}

	// API

	public void updateApiOaiSet(final String dsId, final String apiId, final String oaiSet) throws DsmException {

		boolean insert = dsDao.upsertApiOaiSet(apiId, oaiSet);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(OAI_SET, XmlEscapers.xmlContentEscaper().escape(oaiSet));

		if (!insert) {

			isClient.updateAPIField(dsId, apiId, changes);

		} else {

			isClient.addAPIAttribute(dsId, apiId, changes);

		}

	}

	public void updateApiBaseurl(final String dsId, final String apiId, final String baseUrl) throws DsmException {

		log.info(String.format("updated api '%s' baseurl with '%s'", apiId, baseUrl));

		dsDao.updateApiBaseUrl(apiId, baseUrl);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(BASE_URL, XmlEscapers.xmlContentEscaper().escape(baseUrl));

		isClient.updateAPIField(dsId, apiId, changes);

	}

	public void updateApiCompatibility(final String dsId, final String apiId, final String compliance, final boolean override) throws DsmException {

		log.info(String.format("updated api '%s' compliance with '%s'", apiId, compliance));

		dsDao.updateCompliance(null, apiId, compliance, override);

		final Map<String, String> changes = Maps.newHashMap();

		changes.put(COMPLIANCE, XmlEscapers.xmlAttributeEscaper().escape(compliance));

		isClient.updateAPIField(dsId, apiId, changes);

	}

	public void addApi(final ApiDetails api) throws DsmException {

		if (StringUtils.isBlank(api.getId())) {

			api.setId(createId(api));

			log.info(String.format("missing api id, created '%s'", api.getId()));

		}

		dsDao.addApi(asDbEntry(api));

		isClient.registerAPI(api);

	}

	public void deleteApi(final String apiId) throws DsmForbiddenException, DsmException {

		//TODO handle the api removal in case of associated workflows.

		isClient.removeAPI(apiId);

		dsDao.deleteApi(null, apiId);

	}

	public void dropCaches() {

		mongoLoggerClient.dropCache();

		isClient.dropCache();

		vocabularyClient.dropCache();

		communityClient.dropCache();

	}

	// HELPERS //////////////

	private DatasourceInfo enrichDatasourceInfo(final DatasourceDetails d, final CountDownLatch outerLatch, final Queue<Throwable> errors) {

		final DatasourceInfo dsInfo = new DatasourceInfo().setDatasource(d);

		getAggregationHistory(d.getId(), outerLatch, errors, dsInfo);

		getIndexDsInfo(d.getId(), outerLatch, errors, dsInfo);

		return dsInfo;

	}

	private void getAggregationHistory(final String dsId,

			final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final DatasourceInfo datasourceInfo) {

		Futures.addCallback(

				executor.submit(() -> mongoLoggerClient.getAggregationHistory(dsId)),

				new FutureCallback<List<AggregationInfo>>() {

					public void onSuccess(List<AggregationInfo> info) {

						setAggregationHistory(datasourceInfo, info);

						outerLatch.countDown();

					}

					public void onFailure(Throwable e) {

						log.error(ExceptionUtils.getStackTrace(e));

						errors.offer(e);

						outerLatch.countDown();

					}

				}, executor);

	}

	private void setAggregationHistory(final DatasourceInfo datasourceInfo, final List<AggregationInfo> info) {

		datasourceInfo.setAggregationHistory(info);

		if (!info.isEmpty()) {

			datasourceInfo

					.setLastCollection(info.stream().filter(a -> AggregationStage.COLLECT.equals(a.getAggregationStage())).findFirst().get())

					.setLastTransformation(info.stream().filter(a -> AggregationStage.TRANSFORM.equals(a.getAggregationStage())).findFirst().get());

		}

	}

	private void getIndexDsInfo(final String dsId,

			final CountDownLatch outerLatch,

			final Queue<Throwable> errors,

			final DatasourceInfo datasourceInfo) {

		Futures.addCallback(

				executor.submit(() -> isClient.calculateCurrentIndexDsInfo()),

				new FutureCallback<IndexDsInfo>() {

					public void onSuccess(final IndexDsInfo info) {

						final CountDownLatch innerLatch = new CountDownLatch(2);

						Futures.addCallback(

								executor.submit(() -> datasourceIndexClient.getIndexInfo(dsId, info, errors)),

								new FutureCallback<IndexRecordsInfo>() {

									public void onSuccess(IndexRecordsInfo info) {

										datasourceInfo

												.setIndexRecords(info.getTotal())

												.setFundedContent(info.getFunded())

												.setLastIndexingDate(info.getDate());

										innerLatch.countDown();

									}

									public void onFailure(Throwable e) {

										errors.offer(e);

										innerLatch.countDown();

									}

								}, executor);

						Futures.addCallback(

								executor.submit(() ->

										objectStoreClient.getObjectStoreSize(isClient.getObjectStoreId(dsId))),

								new FutureCallback<Long>() {

									@Override

									public void onSuccess(final Long objectStoreSize) {

										datasourceInfo.setFulltexts(objectStoreSize);

										innerLatch.countDown();

									}

									@Override

									public void onFailure(final Throwable e) {

										errors.offer(e);

										innerLatch.countDown();

									}

								}, executor);

						waitLatch(innerLatch, errors, config.getRequestTimeout());

						outerLatch.countDown();

					}

					public void onFailure(final Throwable e) {

						//log.error(ExceptionUtils.getStackTrace(e));

						errors.offer(e);

						outerLatch.countDown();

					}

				}, executor);

	}

	private void waitLatch(final CountDownLatch latch, final Queue<Throwable> errors, final int waitSeconds) {

		try {

			if (!latch.await(waitSeconds, TimeUnit.SECONDS)) {

				errors.offer(new TimeoutException("Waiting for requests to complete has timed out."));

			}

		} catch (final InterruptedException e) {

			errors.offer(e);

		}

	}

}