package eu.dnetlib.ariadneplus.workflows.nodes;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import eu.dnetlib.data.collector.ThreadSafeIterator;
import eu.dnetlib.rmi.data.CollectorServiceRuntimeException;
import org.apache.commons.lang3.StringEscapeUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.client.utils.URIBuilder;
import org.springframework.http.*;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

/**
 * Created by Alessia Bardi on 31/01/2018.
 *
 * @author Alessia Bardi
 */
public class VirtuosoAriadnePlusIterator extends ThreadSafeIterator {

	private static final Log log = LogFactory.getLog(VirtuosoAriadnePlusIterator.class);
	protected static final String ANY_TIME_QUERY_MS = "1800000"; //1800000 == 3 mins
	protected static final int QUEUE_TIMEOUT_SECONDS = 600;
	public final static String TERMINATOR = "ARNOLD";
	public final static String ERROR_TERMINATOR = "SCHWARZ";
	protected final static int SLEEP_MS = 5000;
	protected final static int MAX_RETRIES = 3;
	protected final static int LIMIT = 100;

	private String datasourceName;
	private String datasourceInterface;
	private String virtuosoReaderAPIUrl;
	private boolean started = false;
	private Map<String, Integer> errors = Maps.newHashMap();
	private List<String> listForClass = Lists.newArrayList();
	private BlockingQueue<String> elements = Queues.newArrayBlockingQueue(10);

	private String currentElement = null;
	private ExecutorService executor = Executors.newSingleThreadExecutor();

	private RestTemplate restTemplate;


	private synchronized void verifyStarted(){
		if (!this.started) {
			this.started = true;
			fillQueue();
			getNextElement(MAX_RETRIES);
		}
	}

	protected void fillQueue(){
		log.info("Virtuoso reader at : " + getVirtuosoReaderAPIUrl());
		getExecutor().submit(() -> {
			try {
				int offset = 0;
				boolean again;
				do {
					List<String> subjectList = getSubjectList(offset);
					for (String subject : subjectList) {
						String xmlFile = tryGetRDF(subject, MAX_RETRIES);
						if (StringUtils.isBlank(xmlFile)) {
							log.warn("Skipping blank RDF for " + subject);
						} else {
							getElements().offer(xmlFile, QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
						}
					}
					again = subjectList.size() == LIMIT;
					offset += LIMIT;
				} while(again);
				log.debug("End of subject list, adding terminator to the queue");
				getElements().offer(TERMINATOR, QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
			} catch (Exception e) {
				log.error(e.getMessage());
				try {
					getElements().offer(ERROR_TERMINATOR, QUEUE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
				} catch (InterruptedException e1) {
					log.error(e1.getMessage());
				}
			}

		});
		getExecutor().shutdown();
	}


	protected String tryGetRDF(final String subjectURL, int attempt) throws URISyntaxException, InterruptedException, VirtuosoAriadnePlusException {
		//this is something George said: if it has no http URI, then it is not to be considered relevant by itself
		if(!subjectURL.startsWith("http")){
			log.debug("Skipping as non-http url: "+subjectURL);
			errors.merge("Non-http URLs", 1, Integer::sum);
			return null;
		}
		log.debug("Querying Api, remaining attempts: "+attempt);
		if (attempt <= 0) {
			errors.merge("Failed tryGetRDF", 1, Integer::sum);
			return null;
		}
		ResponseEntity<String> response = null;
		try {
			response = getRDF(subjectURL);
		}catch(ResourceAccessException e){
			//request timed out --> sleep and try again
			log.warn("Request timeout for "+subjectURL+": I'll sleep and then try again");
			Thread.sleep(SLEEP_MS);
			return tryGetRDF(subjectURL, --attempt);
		}
		HttpStatus responseStatus = response.getStatusCode();
		if (responseStatus.is2xxSuccessful()) {
			String rdfFile = response.getBody();
			if(StringUtils.isBlank(rdfFile)){
				log.warn("Got blank RDF for "+subjectURL+" , let's try again...");
				Thread.sleep(SLEEP_MS);
				return tryGetRDF(subjectURL, --attempt);
			}
			else {
				final String xmlFile = completeXML(rdfFile, subjectURL);
				return xmlFile;
			}
		} else {
			if (responseStatus.is5xxServerError()) {
				//sleep for a while and re-try
				log.warn("HTTP ERROR: "+responseStatus.value() + ": " + responseStatus.getReasonPhrase()+": I'll sleep and then try again");
				Thread.sleep(SLEEP_MS);
				return tryGetRDF(subjectURL, --attempt);
			} else {
				log.error("ERROR: Can't get the RDF for " + subjectURL + " " + responseStatus.value() + ": " + responseStatus.getReasonPhrase());
				errors.merge(responseStatus.value() + ": " + responseStatus.getReasonPhrase(), 1, Integer::sum);
			}
		}
		return null;
	}

	protected URI getURIForSubjectList(final int offset) throws URISyntaxException {
		URIBuilder builder = new URIBuilder(getVirtuosoReaderAPIUrl() + "/apiSubjects");
		builder.addParameter("api", getDatasourceInterface());
		builder.addParameter("limit", Integer.toString(LIMIT));
		builder.addParameter("offset", Integer.toString(offset));
		return builder.build();
	}

	protected List<String> getSubjectList(final int offset) throws URISyntaxException, VirtuosoAriadnePlusException {
		URI uri = getURIForSubjectList(offset);
		log.debug("fillQueue -- Calling for subject list: " + uri.toString());
		List<String> subjectList;
		try {
			subjectList = getRestTemplate().getForObject(uri, getListForClass().getClass());
		}catch(RestClientException rce){
			throw new VirtuosoAriadnePlusException(rce);
		}
		return subjectList;
	}

	protected URI getURIForRDFRequest(final String subjectURL) throws URISyntaxException {
		URIBuilder builder = new URIBuilder(getVirtuosoReaderAPIUrl() + "/subject").addParameter("subjectURL", subjectURL).addParameter("timeout", ANY_TIME_QUERY_MS);
		return builder.build();
	}

	protected ResponseEntity<String> getRDF(final String subjectURL) throws URISyntaxException {
		HttpHeaders headers = new HttpHeaders();
		headers.setAccept(Lists.newArrayList(MediaType.APPLICATION_XML));
		URI uri = getURIForRDFRequest(subjectURL);
		log.debug("fillQueue -- Calling for subject RDF: " + uri.toString());
		HttpEntity<String> entity = new HttpEntity<>("parameters", headers);
		return restTemplate.exchange(uri, HttpMethod.GET, entity, String.class);
	}


	public String completeXML(final String rdfFile, final String url) {
		String xmlEscapedURL = StringEscapeUtils.escapeXml11(url);
		String rdfFileNoXmlDecl = rdfFile.replaceAll("\\<\\?xml(.+?)\\?\\>", "").trim();
		return "<?xml version=\"1.0\" encoding=\"UTF-8\"?><record xmlns=\"http://www.openarchives.org/OAI/2.0/\"><header xmlns:dri=\"http://www.driver-repository.eu/namespace/dri\"><dri:objIdentifier>"
				+ xmlEscapedURL + "</dri:objIdentifier><dri:datasourceapi>" + datasourceInterface + "</dri:datasourceapi><dri:datasourcename>" + datasourceName
				+ "</dri:datasourcename></header><metadata>" + rdfFileNoXmlDecl + "</metadata></record>";
	}

	@Override
	public boolean doHasNext() {
		try {
			verifyStarted();
		} catch (Exception e) {
			getExecutor().shutdownNow();
			throw new CollectorServiceRuntimeException(e);
		}
		switch(currentElement){
		case TERMINATOR:
			 if(!executor.isTerminated()) executor.shutdownNow();
			return false;
		case ERROR_TERMINATOR:
			 executor.shutdownNow();
			throw new CollectorServiceRuntimeException("Error getting elements from virtuoso");
		default:
			return true;
		}
	}

	@Override
	public String doNext() {
		if(!hasNext()) {
			log.error("Next called but hasNext is false", new NoSuchElementException());
			throw new NoSuchElementException();
		}
		switch(currentElement){
		case TERMINATOR:
		case ERROR_TERMINATOR:
			executor.shutdownNow();
			throw new NoSuchElementException();
		default:
			String res = currentElement;
			getNextElement(MAX_RETRIES);
			return res;
		}
	}

	private void getNextElement(int attempt){
		log.debug("polling from queue, remaining attempts: "+attempt);
		if(attempt <= 0) currentElement = ERROR_TERMINATOR;
		else{
			try {
				currentElement = elements.take();
			} catch (InterruptedException e) {
				//current thread interrupted. Let's end.
				currentElement = ERROR_TERMINATOR;
				executor.shutdownNow();
			}
		}
	}

	@Override
	public void remove() {
		throw new UnsupportedOperationException();
	}

	public VirtuosoAriadnePlusIterator datasourceInterface(final String datasourceInterface) {
		this.datasourceInterface = datasourceInterface;
		return this;
	}

	public VirtuosoAriadnePlusIterator virtuosoReaderAPIUrl(final String virtuosoReaderAPIUrl) {
		this.virtuosoReaderAPIUrl = virtuosoReaderAPIUrl;
		return this;
	}

	public VirtuosoAriadnePlusIterator datasourceName(final String datasourceName) {
		this.datasourceName = datasourceName;
		return this;
	}

	public VirtuosoAriadnePlusIterator errors(final Map<String, Integer> errors) {
		this.errors = errors;
		return this;
	}

	public String getDatasourceInterface() {
		return datasourceInterface;
	}

	public String getVirtuosoReaderAPIUrl() {
		return virtuosoReaderAPIUrl;
	}

	public Map<String, Integer> getErrors() {
		return errors;
	}

	public BlockingQueue<String> getElements() {
		return elements;
	}

	public RestTemplate getRestTemplate() {
		return restTemplate;
	}

	public VirtuosoAriadnePlusIterator restTemplate(final RestTemplate restTemplate) {
		this.restTemplate = restTemplate;
		return this;
	}

	public String getDatasourceName() {
		return datasourceName;
	}

	public boolean isStarted() {
		return started;
	}

	public List<String> getListForClass() {
		return listForClass;
	}

	public String getCurrentElement() {
		return currentElement;
	}

	public ExecutorService getExecutor() {
		return executor;
	}

	public void setDatasourceName(final String datasourceName) {
		this.datasourceName = datasourceName;
	}

	public void setDatasourceInterface(final String datasourceInterface) {
		this.datasourceInterface = datasourceInterface;
	}

	public void setVirtuosoReaderAPIUrl(final String virtuosoReaderAPIUrl) {
		this.virtuosoReaderAPIUrl = virtuosoReaderAPIUrl;
	}

	public void setStarted(final boolean started) {
		this.started = started;
	}

	public void setErrors(final Map<String, Integer> errors) {
		this.errors = errors;
	}

	public void setListForClass(final List<String> listForClass) {
		this.listForClass = listForClass;
	}

	public void setElements(final BlockingQueue<String> elements) {
		this.elements = elements;
	}

	public void setCurrentElement(final String currentElement) {
		this.currentElement = currentElement;
	}

	public void setExecutor(final ExecutorService executor) {
		this.executor = executor;
	}

	public void setRestTemplate(final RestTemplate restTemplate) {
		this.restTemplate = restTemplate;
	}
}
