package io.delphiplatform.api.v3.search.service;

import org.elasticsearch.ElasticsearchStatusException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.UncategorizedElasticsearchException;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.delphiplatform.api.v3.constant.SearchIndexNames;
import io.delphiplatform.api.v3.model.ads.LinkfireLink;
import io.delphiplatform.api.v3.rdb.service.ads.LinkfireDimLinkService;
import io.delphiplatform.api.v3.search.entity.LinkfireLinkDocumentEntity;
import io.delphiplatform.api.v3.search.processing.LinkfireLinkDocumentConverter;
import io.delphiplatform.api.v3.search.repository.LinkfireLinkDocumentRepository;

import static org.elasticsearch.index.query.QueryBuilders.queryStringQuery;

@Service
public class LinkfireLinkSearchService {

    private final LinkfireDimLinkService linkfireLinkService;
    private final LinkfireLinkDocumentConverter linkfireLinkDocumentConverter;
    private final LinkfireLinkDocumentRepository linkfireLinkDocumentRepository;
    private final ElasticsearchOperations elasticsearchOperations;

    @Autowired
    public LinkfireLinkSearchService(
        LinkfireDimLinkService linkfireLinkService,
        LinkfireLinkDocumentConverter linkfireLinkDocumentConverter,
        LinkfireLinkDocumentRepository linkfireLinkDocumentRepository,
        ElasticsearchOperations elasticsearchOperations) {
        this.linkfireLinkService = linkfireLinkService;
        this.linkfireLinkDocumentConverter = linkfireLinkDocumentConverter;
        this.linkfireLinkDocumentRepository = linkfireLinkDocumentRepository;
        this.elasticsearchOperations = elasticsearchOperations;
    }

    /**
     * Simple query Elasticsearch linkfire-link index by link_url.
     */
    public List<LinkfireLinkDocumentEntity> findAllByLinkUrl(String queryString) {
        return linkfireLinkDocumentRepository.findAllByLinkUrl(queryString);
    }

    /**
     * Simple query Elasticsearch linkfire-link index by link_url.
     */
    public SearchHits<LinkfireLinkDocumentEntity> searchAllByLinkUrl(String queryString) {
        return linkfireLinkDocumentRepository.searchAllByLinkUrl(queryString);
    }

    /**
     * Query Elasticsearch linkfire-link index by link_url with support for pagination.
     *
     * @throws ResponseStatusException for certain Elasticsearch exceptions
     */
    public SearchHits<LinkfireLinkDocumentEntity> searchAllByLinkUrl(String queryString, Integer limit, Integer page) {
        IndexCoordinates index = IndexCoordinates.of(SearchIndexNames.LINKFIRE_LINK);
        NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
            .withQuery(queryStringQuery(queryString).field("link_url"))
            .withPageable(PageRequest.of(page, limit))
            .build();
        try {
            return elasticsearchOperations.search(searchQuery, LinkfireLinkDocumentEntity.class, index);
        } catch (UncategorizedElasticsearchException e) {
            // ensure we get the correct status instead of a generic 500 in the event of bad input (400)
            if (e.getCause() instanceof ElasticsearchStatusException) {
                ElasticsearchStatusException exception = (ElasticsearchStatusException) e.getCause();
                String message = "Potentially invalid query provided. Make sure to escape any reserved characters. "
                    + "Details: " + e.getMessage();
                throw new ResponseStatusException(HttpStatus.valueOf(exception.status().getStatus()), message);
            }
            throw e;
        }
    }

    /**
     * Query Elasticsearch linkfire-link index by link_url and convert results to standard {@link LinkfireLink}
     * objects.
     * <p>
     * A naïve but faster implementation that does not make a call to Postgres. Assumes all fields are contained within
     * the Elasticsearch document.
     */
    public List<LinkfireLink> searchAndConvertAllByLinkUrl(String queryString, Integer limit, Integer page) {
        SearchHits<LinkfireLinkDocumentEntity> searchHits = searchAllByLinkUrl(queryString, limit, page);
        if (searchHits.getTotalHits() == 0) {
            return Collections.emptyList();
        }
        Stream<LinkfireLinkDocumentEntity> linkfireLinkDocuments = searchHits.getSearchHits().stream()
            .map(SearchHit::getContent);
        return linkfireLinkDocumentConverter.convert(linkfireLinkDocuments).collect(Collectors.toList());
    }

    /**
     * Query Elasticsearch linkfire-link index by link_url and convert results to standard {@link LinkfireLink}
     * objects.
     * <p>
     * Makes a call to Postgres to populate fields to ensure indexed data exists.
     */
    public List<LinkfireLink> searchAndConvertAllByLinkUrlWithRdb(String queryString, Integer limit, Integer page) {
        Map<String, Float> linkIdToScore = new HashMap<>();
        Map<String, LinkfireLink> linkIdToLinkfireLink = new HashMap<>();
        Set<UUID> linkIds = new HashSet<>(); // for searching postgres

        SearchHits<LinkfireLinkDocumentEntity> searchHits = searchAllByLinkUrl(queryString, limit, page);
        if (searchHits.getTotalHits() == 0) {
            return Collections.emptyList();
        }
        searchHits.getSearchHits()
            .forEach(hit -> {
                String linkfireLinkId = hit.getContent().getLinkId();
                linkIdToScore.put(linkfireLinkId, hit.getScore());
                linkIds.add(UUID.fromString(linkfireLinkId));
            });

        // for custom PQ order
        Comparator<Entry<String, Float>> idScoreComparator = (a, b) -> Float.compare(b.getValue(), a.getValue());
        // PQ to keep sorted by score
        PriorityBlockingQueue<Entry<String, Float>> maxQueue =
            new PriorityBlockingQueue<>(linkIdToScore.entrySet().size(), idScoreComparator);
        maxQueue.addAll(linkIdToScore.entrySet());
        // get linkfireLink entities from postgres
        linkfireLinkService.findAllById(linkIds)
            .forEach(linkfireLink -> linkIdToLinkfireLink.put(linkfireLink.getLinkId(), linkfireLink));
        // use PQ to order results by score
        return Stream.generate(maxQueue::poll)
            .takeWhile(Objects::nonNull)
            .map(entry -> linkIdToLinkfireLink.get(entry.getKey()))
            .filter(Objects::nonNull)
            .collect(Collectors.toList());
    }

}