D-Net

package eu.dnetlib.openaire.directindex.api;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import eu.dnetlib.common.rmi.DNetRestDocumentation;

import eu.dnetlib.data.index.CloudIndexClient;

import eu.dnetlib.data.index.CloudIndexClientException;

import eu.dnetlib.data.index.CloudIndexClientFactory;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.openaire.directindex.objects.ResultEntry;

import eu.dnetlib.openaire.directindex.utils.OafToIndexRecordFactory;

import org.apache.commons.io.IOUtils;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.velocity.app.VelocityEngine;

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

import org.springframework.core.io.ClassPathResource;

import org.springframework.http.HttpStatus;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;

import java.io.IOException;

import java.util.List;

import java.util.Map;

import java.util.concurrent.*;

/**

 * Created by michele on 11/11/15.

 */

@Controller

@DNetRestDocumentation

public class OpenaireResultSubmitter {

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

	public class IndexDsInfo {

		private final String indexBaseUrl;

		private final String indexDsId;

		private final String format;

		private final String coll;

		public IndexDsInfo(final String indexBaseUrl, final String indexDsId, final String format, final String coll) {

			this.indexBaseUrl = indexBaseUrl;

			this.indexDsId = indexDsId;

			this.format = format;

			this.coll = coll;

		}

		public String getIndexBaseUrl() {

			return indexBaseUrl;

		}

		public String getIndexDsId() {

			return indexDsId;

		}

		public String getFormat() {

			return format;

		}

		public String getColl() {

			return coll;

		}

	}

	@Value(value = "oaf.schema.location")

	private String oafSchemaLocation;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Resource

	private OafToIndexRecordFactory oafToIndexRecordFactory;

	@Resource

	private RecentResultsQueue recentResultsQueue;

	@Resource(name = "openaireplusApisVelocityEngine")

	private VelocityEngine velocityEngine;

	@Value(value = "${openaire.api.directindex.findSolrIndexUrl.xquery}")

	private ClassPathResource findSolrIndexUrl;

	@Value(value = "${openaire.api.directindex.findIndexDsInfo.xquery}")

	private ClassPathResource findIndexDsInfo;

	/**

	 * Autocommit feature activation flag

	 */

	@Value(value = "${openaire.api.directindex.autocommit.active}")

	private boolean autocommitactive;

	/**

	 * Autocommit frequency (Seconds)

	 */

	@Value(value = "${openaire.api.directindex.autocommit.frequency}")

	private long commitfrquency = 60;

	/**

	 * Minimum autocommit frequency

	 */

	private static long COMMIT_FREQUENCY_MINVALUE = 5;

	/**

	 * Autocommit executor service

	 */

	private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

	/**

	 * Autocommit task handle

	 */

	private ScheduledFuture task = null;

	public OpenaireResultSubmitter() {

		updateSchedule();

	}

	private void updateSchedule() {

		if (task != null) {

			task.cancel(true);

		}

		task = executor.scheduleAtFixedRate(doCommit(), 0, getCommitfrquency(), TimeUnit.SECONDS);

	}

	private Runnable doCommit() {

		return () -> {

			//TODO perform the commit only if there are pending documents, e.g. query /solr/admin/metrics?group=core&prefix=UPDATE.updateHandler to inspect if there are any.

			if (isAutocommitactive()) {

				try {

					for (Map.Entry<IndexDsInfo, CloudIndexClient> e : getClients().entrySet()) {

						final IndexDsInfo i = e.getKey();

						try(final CloudIndexClient client = e.getValue()) {

							log.info("performing commit on " + i.getColl());

							client.commit();

						}

					}

				} catch (DirecIndexApiException | CloudIndexClientException | IOException e) {

					//TODO perhaps we could send an email to notify this failure.

					throw new RuntimeException(e);

				}

			}

		};

	}

	@RequestMapping(value = { "/api/admin/autocommit/active" }, method = RequestMethod.GET)

	public @ResponseBody Boolean getAutocommit() throws DirecIndexApiException {

		return isAutocommitactive();

	}

	@RequestMapping(value = { "/api/admin/autocommit/active" }, method = RequestMethod.POST)

	public @ResponseBody Boolean setAutocommit(@RequestParam(value = "active", required = true) final Boolean active) throws DirecIndexApiException {

		setAutocommitactive(active);

		log.info(String.format("automatic commit, active '%s', frequency '%s'", isAutocommitactive(), getCommitfrquency()));

		return isAutocommitactive();

	}

	@RequestMapping(value = { "/api/admin/autocommit/frequency" }, method = RequestMethod.GET)

	public @ResponseBody long getAutocommitFrequency() throws DirecIndexApiException {

		return getCommitfrquency();

	}

	@RequestMapping(value = { "/api/admin/autocommit/frequency" }, method = RequestMethod.POST)

	public @ResponseBody long setAutocommitFrequency(@RequestParam(value = "frequency", required = true) final Integer frequency) throws DirecIndexApiException {

		setCommitfrquency(frequency);

		log.info(String.format("automatic commit, active '%s', frequency '%s'", isAutocommitactive(), getCommitfrquency()));

		return getCommitfrquency();

	}

	@Deprecated

	@RequestMapping(value = { "/api/publications/feedJson", "/api/results/feedJson" }, method = RequestMethod.POST)

	public @ResponseBody String feedObjectJson(@RequestParam(value = "json", required = true) final String json,

			@RequestParam(value = "commit", required = false, defaultValue = "true") final boolean commit) throws DirecIndexApiException {

		final ResultEntry pub = new Gson().fromJson(json, ResultEntry.class);

		return feedObject(pub, commit);

	}

	@RequestMapping(value = { "/api/results/feedObject" }, method = RequestMethod.POST)

	public @ResponseBody String feedResult(@RequestBody final ResultEntry pub,

			@RequestParam(value = "commit", required = false, defaultValue = "true") final boolean commit)

			throws DirecIndexApiException {

		return feed(pub, commit);

	}

	@Deprecated

	@RequestMapping(value = { "/api/publications/feedObject" }, method = RequestMethod.POST)

	public @ResponseBody String feedObject(@RequestBody final ResultEntry pub,

			@RequestParam(value = "commit", required = false, defaultValue = "true") final boolean commit)

			throws DirecIndexApiException {

		return feed(pub, commit);

	}

	@RequestMapping(value = "/api/result/{openaireId}", method = RequestMethod.DELETE)

	public @ResponseBody boolean deleteResultWithOpenaireId(

			@PathVariable(value = "openaireId") final String openaireId,

			@RequestParam(value = "commit", required = false, defaultValue = "true") final boolean commit) throws DirecIndexApiException {

		return deleteResult(openaireId, commit);

	}

	@RequestMapping(value = "/api/results", method = RequestMethod.DELETE)

	public @ResponseBody boolean deleteResultWithOriginalId(

			@RequestParam(value = "originalId", required = true) final String originalId,

			@RequestParam(value = "collectedFromId", required = true) final String collectedFromId,

			@RequestParam(value = "commit", required = false, defaultValue = "true") final boolean commit) throws Exception {

		final String openaireId = ResultEntry.calculateOpenaireId(originalId, collectedFromId, serviceLocator.getService(ISLookUpService.class));

		return deleteResult(openaireId, commit);

	}

	@Deprecated

	@RequestMapping(value = { "/api/publications/deleteObject", "/api/results/delete" }, method = RequestMethod.POST)

	public @ResponseBody boolean deleteResultPost(

			@RequestParam(value = "originalId", required = true) final String originalId,

			@RequestParam(value = "collectedFromId", required = true) final String collectedFromId,

			@RequestParam(value = "commit", required = false, defaultValue = "true") final boolean commit) throws Exception {

		final String openaireId = ResultEntry.calculateOpenaireId(originalId, collectedFromId, serviceLocator.getService(ISLookUpService.class));

		return deleteResult(openaireId, commit);

	}

	@Deprecated

	private String feed(final ResultEntry pub, final boolean commit) throws DirecIndexApiException {

		return feed(pub);

	}

	private String feed(final ResultEntry pub) throws DirecIndexApiException {

		try {

			final String oafRecord = pub.asOafRecord(velocityEngine, serviceLocator.getService(ISLookUpService.class), oafSchemaLocation);

			for(Map.Entry<IndexDsInfo, CloudIndexClient> e : getClients().entrySet()) {

				final IndexDsInfo idx = e.getKey();

				try(final CloudIndexClient client = e.getValue()) {

					client.feed(oafRecord, idx.getIndexDsId(), oafToIndexRecordFactory.newTransformer(idx.getFormat()), false);

				}

			}

			recentResultsQueue.add(oafRecord);

			return pub.getOpenaireId();

		} catch (final Throwable e) {

			log.error("Error saving record", e);

			log.debug(pub.toString());

			throw new DirecIndexApiException("Error adding publication: " + e.getMessage(), e);

		}

	}

	private boolean deleteResult(final String openaireId, final boolean commit) throws DirecIndexApiException {

		try {

			for(Map.Entry<IndexDsInfo, CloudIndexClient> e : getClients().entrySet()) {

				final IndexDsInfo idx = e.getKey();

				try(final CloudIndexClient client = e.getValue()) {

					client.remove(openaireId, false);

					log.info("Deleted result with id: " + openaireId + " from: " + idx.getIndexBaseUrl());

				}

			}

			recentResultsQueue.remove(openaireId);

			return true;

		} catch (Throwable e) {

			throw new DirecIndexApiException("Error deleting publication: " + e.getMessage(), e);

		}

	}

	private Map<IndexDsInfo, CloudIndexClient> getClients() throws DirecIndexApiException {

		try {

			final List<IndexDsInfo> idxList = calculateCurrentIndexDsInfo();

			if (idxList == null || idxList.isEmpty()) {

				throw new DirecIndexApiException("Cannot commit index: no public Search Service found");

			}

			if (idxList.size() > 1) {

				log.warn("Found more than 1 public search service");

			}

			final Map<IndexDsInfo, CloudIndexClient> clients = Maps.newHashMap();

			for(IndexDsInfo i : idxList) {

				clients.put(i, CloudIndexClientFactory.newIndexClient(i.getIndexBaseUrl(), i.getColl(), false));

			}

			return clients;

		} catch (IOException | ISLookUpException | CloudIndexClientException e) {

			throw new DirecIndexApiException(e);

		}

	}

	private List<IndexDsInfo> calculateCurrentIndexDsInfo() throws IOException, ISLookUpException {

		final List<IndexDsInfo> list = Lists.newArrayList();

		final String queryUrl = IOUtils.toString(findSolrIndexUrl.getInputStream());

		final String queryDs = IOUtils.toString(findIndexDsInfo.getInputStream());

		final ISLookUpService lu = serviceLocator.getService(ISLookUpService.class);

		final String indexBaseUrl = lu.getResourceProfileByQuery(queryUrl);

		final List<String> idxDs = lu.quickSearchProfile(queryDs);

		for (final String idx : idxDs) {

			final String[] arr = idx.split("@@@");

			list.add(new IndexDsInfo(indexBaseUrl, arr[0].trim(), arr[1].trim(), arr[2].trim()));

		}

		return list;

	}

	@ExceptionHandler(Exception.class)

	@ResponseStatus(value = HttpStatus.INTERNAL_SERVER_ERROR)

	public @ResponseBody ErrorMessage handleException(final Exception e) {

		log.error("Error in direct index API", e);

		return new ErrorMessage(e);

	}

	public class ErrorMessage {

		private final String message;

		private final String stacktrace;

		public ErrorMessage(final Exception e) {

			this(e.getMessage(), ExceptionUtils.getStackTrace(e));

		}

		public ErrorMessage(final String message, final String stacktrace) {

			this.message = message;

			this.stacktrace = stacktrace;

		}

		public String getMessage() {

			return this.message;

		}

		public String getStacktrace() {

			return this.stacktrace;

		}

	}

	public boolean isAutocommitactive() {

		return autocommitactive;

	}

	public synchronized void setAutocommitactive(boolean autocommitactive) {

		this.autocommitactive = autocommitactive;

	}

	public long getCommitfrquency() {

		return commitfrquency;

	}

	public synchronized void setCommitfrquency(long commitfrquency) {

		if (commitfrquency < COMMIT_FREQUENCY_MINVALUE) {

			throw new RuntimeException("cannot set autocommit frequency: minimum accepted value (inclusive) is " + COMMIT_FREQUENCY_MINVALUE);

		}

		this.commitfrquency = commitfrquency;

		updateSchedule();

	}

}