package eu.dnetlib.lbs.matchers;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;

import javax.annotation.PostConstruct;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.ScrolledPage;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Component;

import eu.dnetlib.lbs.elasticsearch.Event;
import eu.dnetlib.lbs.elasticsearch.Notification;
import eu.dnetlib.lbs.elasticsearch.NotificationRepository;
import eu.dnetlib.lbs.events.output.DispatcherManager;
import eu.dnetlib.lbs.properties.ElasticSearchProperties;
import eu.dnetlib.lbs.subscriptions.MapCondition;
import eu.dnetlib.lbs.subscriptions.NotificationMode;
import eu.dnetlib.lbs.subscriptions.Subscription;
import eu.dnetlib.lbs.subscriptions.SubscriptionRepository;
import eu.dnetlib.lbs.utils.LbsQueue;
import eu.dnetlib.lbs.utils.QueueManager;
import eu.dnetlib.lbs.utils.ThreadManager;

@Component
public class SubscriptionEventMatcher implements Runnable {

	private static final int SCROLL_TIMEOUT_IN_MILLIS = 10000;

	@Autowired
	private QueueManager queueManager;
	@Autowired
	private DispatcherManager dispatcherManager;
	@Autowired
	private SubscriptionRepository subscriptionRepo;

	@Autowired
	private ElasticsearchTemplate elasticsearchTemplate;

	@Autowired
	private ElasticSearchProperties props;

	private LbsQueue<Subscription, Subscription> queue;

	@Autowired
	private NotificationRepository notificationRepository;

	@Autowired
	private ThreadManager threadManager;

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

	@PostConstruct
	public void init() {
		queue = queueManager.newQueue("subscr-events-matcher-queue", Subscription::isReady);
		threadManager.newThread("subscr-events-matcher", this);
	}

	public void startMatching(final Subscription s) {
		if (queue.offer(s)) {
			log.info("Matching of subscription " + s.getSubscriptionId() + " in queue");
		} else {
			log.info("Subscription " + s.getSubscriptionId() + " not queued");
		}
	}

	@Override
	public void run() {
		log.info("SubscriptionEventMatcher started: " + Thread.currentThread().getName());

		while (true) {
			try {
				final Subscription s = queue.takeOne();
				if (s != null && Subscription.isReady(s)) {
					startSubscriptionEventsMatcher(s);
					s.setLastNotificationDate(new Date());
					subscriptionRepo.save(s);
				}
			} catch (final Throwable e) {
				log.error("Error iterating matching queue", e);
			}
		}
	}

	private void startSubscriptionEventsMatcher(final Subscription s) {
		log.info("Start matching subscription: " + s);

		try {

			final BoolQueryBuilder mapQuery = QueryBuilders.boolQuery();
			s.getConditionsAsList()
					.stream()
					.map(MapCondition::asQueryBuilder)
					.filter(Objects::nonNull)
					.forEach(mapQuery::must);

			final long lastNotification = s.getLastNotificationDate() != null ? s.getLastNotificationDate().getTime() : 0;

			final SearchQuery searchQuery = new NativeSearchQueryBuilder()
					.withQuery(QueryBuilders.boolQuery()
							.must(QueryBuilders.matchQuery("topic", s.getTopic()))
							.must(QueryBuilders.rangeQuery("creationDate").from(lastNotification))
							.must(QueryBuilders.nestedQuery("map", mapQuery, ScoreMode.None)))
					.withSearchType(SearchType.DEFAULT)
					.withIndices(props.getEventsIndexName())
					.withTypes(props.getEventsIndexType())
					.withPageable(PageRequest.of(0, 10))
					.build();

			final List<Event> eventsToNotify = new ArrayList<>();

			ScrolledPage<Event> scrollResp = (ScrolledPage<Event>) elasticsearchTemplate.startScroll(SCROLL_TIMEOUT_IN_MILLIS, searchQuery, Event.class);

			while (scrollResp.hasContent()) {
				for (final Event e : scrollResp.getContent()) {
					final Notification n = new Notification(s, e);
					if (isNotAlreadyNotified(n)) {
						notificationRepository.save(n);
						if (s.getMode() == NotificationMode.EMAIL && eventsToNotify.size() < 20) {
							eventsToNotify.add(e);
						} else if (s.getMode() == NotificationMode.MOCK) {
							dispatcherManager.dispatch(s, e);
						}
					}
				}

				scrollResp = (ScrolledPage<Event>) elasticsearchTemplate.continueScroll(scrollResp.getScrollId(), SCROLL_TIMEOUT_IN_MILLIS, Event.class);
			}
			elasticsearchTemplate.clearScroll(scrollResp.getScrollId());

			log.info("End matching subscription: " + s.getSubscriptionId());
			if (s.getMode() == NotificationMode.EMAIL && !eventsToNotify.isEmpty()) {
				dispatcherManager.dispatch(s, eventsToNotify.toArray(new Event[eventsToNotify.size()]));
			}

		} catch (final Throwable e) {
			log.error("Error matching subscription: " + s.getSubscriptionId(), e);
		}
	}

	private boolean isNotAlreadyNotified(final Notification n) {
		return !notificationRepository.findById(n.getNotificationId()).isPresent();
	}

}