package com.zm.order.tool.server.manager;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.zm.order.tool.server.builder.SearchRequestBuilder;
import com.zm.order.tool.server.component.BusinessHandlerHolder;
import com.zm.order.tool.server.component.ThreadContext;
import com.zm.order.tool.server.config.KafkaProducerLauncher;
import com.zm.order.tool.server.constants.CommonConstants;
import com.zm.order.tool.server.constants.ResExtFieldKeyMapping;
import com.zm.order.tool.server.constants.UpgradeSellerLists;
import com.zm.order.tool.server.dto.SearchOrderDTO;
import com.zm.order.tool.server.enums.BizTypeEnum;
import com.zm.order.tool.server.enums.ElasticsearchOperateType;
import com.zm.order.tool.server.handler.BusinessHandler;
import com.zm.order.tool.server.model.BaseIndexOrder;
import com.zm.order.tool.server.model.Order;
import com.zm.order.tool.server.model.RefundIndex;
import com.zm.order.tool.server.model.TraceIdentity;
import com.zm.order.tool.server.query.SearchOrderQuery;
import com.zm.order.tool.server.utils.BeanMapUtils;
import com.zm.util.dto.Result;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Description 搜索manager
 * @Author: ph
 * @Date: 2019/10/30 10:30
 */
@Component
public class SearchManager {

    private static final Logger logger = LoggerFactory.getLogger(SearchManager.class);

    @Autowired
    private RestHighLevelClient esClient;
    @Autowired
    private BusinessHandlerHolder businessHandlerHolder;

    private static final String START_TIME = "Start-time";

    private static final String END_TIME = "End-time";

    private static final String TRACE_ID = TraceIdentity.HEADER_TRACE_ID;

    private static final String SELLER_ID = "Seller-id";

    private static final String NEW_ORDER_INDEX_NAME = "new_order";

    private static final String NEW_ORDERS_INDEX_NAME = "orders";

    /**
     * 默认类型
     */
    private static final String DEFAULT_TYPE = "_doc";

    @Autowired
    private KafkaProducerLauncher kafkaProducer;

    /**
     * 搜索订单
     *
     * @param searchOrderQuery	搜索条件
     * @param includes	指定返回的字段
     * @param excludes	排除返回的字段
     * @return下午
     * Date: 2021/10/19 5:24
     */
    public SearchOrderDTO searchOrder(SearchOrderQuery searchOrderQuery, String[] includes,
                                      String[] excludes, Integer bizType) throws IOException {
        BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(bizType);
        SearchSourceBuilder searchSourceBuilder = businessHandler.buildSearchSourceBuilder(searchOrderQuery);
        searchSourceBuilder.fetchSource(includes, excludes);
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.source(searchSourceBuilder);
        searchRequest.indices(NEW_ORDER_INDEX_NAME);
        SearchResponse searchResponse = esClient.search(searchRequest, addHeader(searchOrderQuery, bizType));
        if (searchResponse.status() != RestStatus.OK) {
            throw new RuntimeException("搜索订单失败");
        }
        final long total = searchResponse.getHits().getTotalHits();

        SearchOrderDTO searchOrderDTO = new SearchOrderDTO();
        searchOrderDTO.setTotal(total);
        if (total == 0) {
            searchOrderDTO.setOrderList(Collections.EMPTY_LIST);
        } else {
            List<BaseIndexOrder> orderList = businessHandler.parseIndexOrder(searchResponse.getHits().getHits());
            searchOrderDTO.setOrderList(orderList);
        }
        return searchOrderDTO;
    }


    /**
     * 索引订单
     *
     * @param order
     * @return
     */
    public boolean indexOrder(Order order) throws Exception {
        resetOrderExt(order, null);
        BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(order.getBizType());
        BaseIndexOrder baseIndexOrder = businessHandler.buildIndexOrder(order);
        // 发送消息到ES写网关服务，由网关执行写入操作
        kafkaProducer.sendAsync(CommonConstants.ES_NEW_ORDER_INDEX,
                order.getOrderNo(), recast(baseIndexOrder, ElasticsearchOperateType.INDEX, null));
        return true;
    }

    /**
     * 索引订单
     *
     * @param order
     * @return
     */
    public boolean indexOrder(Order order, String topic) throws Exception {
        resetOrderExt(order, topic);
        if (StringUtils.isEmpty(topic)) {
            topic = CommonConstants.ES_NEW_ORDER_INDEX;
        }
        try {
            BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(order.getBizType());
            BaseIndexOrder baseIndexOrder = businessHandler.buildIndexOrder(order);
            String recast = recast(baseIndexOrder, ElasticsearchOperateType.INDEX, null);
            // 发送消息到ES写网关服务，由网关执行写入操作
            kafkaProducer.sendAsync(topic, order.getOrderNo(), recast);

        }catch (Exception e){
            logger.error("EsManager[indexOrder] failed, cause by:{}", e);
        }
        return true;
    }
    
    private void resetOrderExt(Order order, String topic) {
        if (!UpgradeSellerLists.LIST.contains(order.getSellerId())) {
            return;
        }
        Map<String, Object> extraMap = order.getExtraMap();
        if (Objects.equals(CommonConstants.ES_NEW_INDEX_TOPIC, topic)) {
            ResExtFieldKeyMapping.MAPPING.forEach((k, v) -> {
                if (extraMap.get(k) != null) {
                    extraMap.put(v, extraMap.get(k));
                }
            });
        }
        order.setExtraMap(extraMap);
    }

    /**
     * 索引订单
     *
     * @param refundIndex
     * @return
     */
    public boolean indexRefund(List<IndexRequest> refundIndex) throws Exception {
        // 发送消息到ES写网关服务，由网关执行写入操作
        BulkRequest request = new BulkRequest();
        refundIndex.forEach(request::add);
        try {
            esClient.bulk(request, RequestOptions.DEFAULT);
            return true;
        } catch (Throwable throwable) {
            logger.error("EsManager[bulkIndex] failed, cause by:{}",Throwables.getStackTraceAsString(throwable));
            throw new RuntimeException(throwable.getMessage());
        }
    }

    public boolean indexRefunds(List<RefundIndex> refundIndex) throws Exception {
        if (CollectionUtils.isEmpty(refundIndex)) {
            return true;
        }
        List<IndexRequest> indexRequestList = Lists.newArrayListWithCapacity(refundIndex.size());
        refundIndex.forEach(order -> {
            IndexRequest indexRequest = new IndexRequest();
            indexRequest.index("order_refund_209_1711962856");
            indexRequest.id(order.getDocId().toString());
            indexRequest.type("_doc");
            indexRequest.source(JSON.toJSONString(order), XContentType.JSON);
            indexRequestList.add(indexRequest);
        });
        return this.indexRefund(indexRequestList);
    }

    public void indexRefundMsg(List<RefundIndex> refundIndex) throws Exception {
        for (RefundIndex index : refundIndex) {
            kafkaProducer.sendAsync(CommonConstants.ORDER_CENTER_REFUND_TOPIC,
                    index.getRefundNo(), recast(index, ElasticsearchOperateType.INDEX, null));
        }
    }

    public boolean indexOrder(List<Order> orderList) {
        if (CollectionUtils.isEmpty(orderList)) {
            return true;
        }
        orderList.forEach(order -> {
            try {
                indexOrder(order);
            } catch (Exception ex) {
                logger.error("SearchManager[batch_index_order]failed! order:{}, cause by:{}", order, ex);
            }
        });
        return true;
    }

    public boolean indexOrder(List<Order> orderList, String indexName) {
        if (CollectionUtils.isEmpty(orderList)) {
            return true;
        }
        orderList.forEach(order -> {
            try {
                indexOrder(order, indexName);
            } catch (Exception ex) {
                logger.error("SearchManager[batch_index_order]failed! order:{}, cause by:{}", order, ex);
            }
        });
        return true;
    }

    /**
     * 搜索订单
     *
     * @param searchOrderQuery
     * @return
     * @throws IOException
     */
    public SearchOrderDTO searchOrder(SearchOrderQuery searchOrderQuery) throws IOException {
        SearchOrderDTO searchOrderDTO = new SearchOrderDTO();
        BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(5);
        SearchSourceBuilder searchSourceBuilder = businessHandler.buildSearchSourceBuilder(searchOrderQuery);
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.source(searchSourceBuilder);
        searchRequest.indices(NEW_ORDER_INDEX_NAME);
        SearchResponse searchResponse = esClient.search(searchRequest, addHeader(searchOrderQuery, 5));
        if (searchResponse.status() != RestStatus.OK) {
            throw new RuntimeException("搜索订单失败");
        }
        final long total = searchResponse.getHits().getTotalHits();

        searchOrderDTO.setTotal(total);
        if (total == 0) {
            searchOrderDTO.setOrderList(Collections.EMPTY_LIST);
        } else {
            List<BaseIndexOrder> orderList = businessHandler.parseIndexOrder(searchResponse.getHits().getHits());
            searchOrderDTO.setOrderList(orderList);
        }
        return searchOrderDTO;
    }

    /**
     * 搜索订单
     *
     * @param searchOrderQuery
     * @return
     * @throws IOException
     */
    public SearchOrderDTO searchOrder(SearchOrderQuery searchOrderQuery, String[] includes, String[] excludes) throws IOException {
        SearchOrderDTO searchOrderDTO = new SearchOrderDTO();
        BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(5);
        SearchSourceBuilder searchSourceBuilder = businessHandler.buildSearchSourceBuilder(searchOrderQuery);
        searchSourceBuilder.fetchSource(includes, excludes);
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.source(searchSourceBuilder);
        searchRequest.indices(NEW_ORDER_INDEX_NAME);
        SearchResponse searchResponse = esClient.search(searchRequest, addHeader(searchOrderQuery, 5));
        if (searchResponse.status() != RestStatus.OK) {
            throw new RuntimeException("搜索订单失败");
        }
        final long total = searchResponse.getHits().getTotalHits();

        searchOrderDTO.setTotal(total);
        if (total == 0) {
            searchOrderDTO.setOrderList(Collections.EMPTY_LIST);
        } else {
            List<BaseIndexOrder> orderList = businessHandler.parseIndexOrder(searchResponse.getHits().getHits());
            searchOrderDTO.setOrderList(orderList);
        }
        return searchOrderDTO;
    }

    /**
     * 搜索订单
     *
     * @param searchOrderQuery
     * @return
     * @throws IOException
     */
    public SearchOrderDTO searchOrder(SearchOrderQuery searchOrderQuery, Integer bizType) throws IOException {
        SearchOrderDTO searchOrderDTO = new SearchOrderDTO();
        BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(bizType);
        SearchSourceBuilder searchSourceBuilder = businessHandler.buildSearchSourceBuilder(searchOrderQuery);
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.source(searchSourceBuilder);
        searchRequest.indices(NEW_ORDER_INDEX_NAME);
        SearchResponse searchResponse = esClient.search(searchRequest, addHeader(searchOrderQuery, bizType));
        if (searchResponse.status() != RestStatus.OK) {
            throw new RuntimeException("搜索订单失败");
        }
        final long total = searchResponse.getHits().getTotalHits();


        searchOrderDTO.setTotal(total);
        if (total == 0) {
            searchOrderDTO.setOrderList(Collections.EMPTY_LIST);
        } else {
            List<BaseIndexOrder> orderList = businessHandler.parseIndexOrder(searchResponse.getHits().getHits());
            searchOrderDTO.setOrderList(orderList);
        }
        return searchOrderDTO;
    }

    /**
     * 统计数量
     *
     * @param searchOrderQuery
     * @return
     */
    public int countOrder(SearchOrderQuery searchOrderQuery) throws IOException {
        return 0;
    }

    /**
     * 删除索引
     *
     * @param id id
     * @return
     */
    public boolean deleteOrder(Long id, Long sellerId) {
        try {
            this.delete(NEW_ORDER_INDEX_NAME, DEFAULT_TYPE, id.toString(), sellerId);
            return true;
        } catch (Throwable throwable) {
            logger.error("EsManager[deleteOrder] failed, request:[id:{}], cause by:{}",
                    id, Throwables.getStackTraceAsString(throwable));
            throw new RuntimeException(throwable.getMessage());
        }
    }

    /**
     * 删除索引
     *
     * @param id id
     * @return
     */
    public void delete(String indexName, String typeName, String id, Long sellerId) {
        try {
            // 发送消息到ES写网关服务，由网关执行写入操作
            Map<String, Object> body = new HashMap<>();
            body.put("sellerId", sellerId);
            kafkaProducer.sendAsync(CommonConstants.ES_NEW_ORDER_INDEX,
                    id, recast(body, ElasticsearchOperateType.DELETE, Long.valueOf(id)));
        } catch (Throwable throwable) {
            logger.error("EsManager[delete] failed, request:[id:{}], cause by:{}",
                    id, Throwables.getStackTraceAsString(throwable));
            throw new RuntimeException(throwable.getMessage());
        }
    }

    public boolean indexAmountOrderList(List<Order> orderList, String formatDate) {
        return true;
    }

    /**
     * 索引订单
     *
     * @param order
     * @return
     */
    public Result indexAmountOrder(Order order) {
        return Result.successNullData();
    }


    /**
     * 删除时间范围内的订单
     *
     * @param searchOrderQuery
     * @return
     */
    public void delOrder(SearchOrderQuery searchOrderQuery) {
//        try {
//            QueryBuilder queryBuilder = SearchRequestBuilder.buildDelEsQuery(searchOrderQuery);
//            DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(NEW_ORDER_INDEX_NAME);
//            //可限制删除记录数 setSize()
//            //每批次删除数量 默认即1000
//            deleteByQueryRequest.setBatchSize(CommonConstants.BATCH_SIZE_OF_DEL);
//            deleteByQueryRequest.setQuery(queryBuilder);
//            ActionListener listener = new DelEsListener();
//            esClient.deleteByQueryAsync(deleteByQueryRequest, RequestOptions.DEFAULT, listener);
//            logger.info("delOrder-record-info[send] queryBuilder:{} ", queryBuilder.toString());
//        } catch (Throwable e) {
//            logger.error("SearchManager[delOrder] failed!, cause by:{}", e);
//        }
    }

    private static final List<Integer> BIZ_TYPE_LIST = new ArrayList<>(4);

    static {
        BIZ_TYPE_LIST.add(BizTypeEnum.BAKING.getType());
        BIZ_TYPE_LIST.add(BizTypeEnum.CATERING.getType());
        BIZ_TYPE_LIST.add(BizTypeEnum.ASSISTANT.getType());
        BIZ_TYPE_LIST.add(BizTypeEnum.MEAL_MATE.getType());
    }

    private RequestOptions addHeader(SearchOrderQuery query, Integer bizType) {
        RequestOptions.Builder builder =  RequestOptions.DEFAULT.toBuilder();
        // 设置header头
        // 分片维度按照商户数据量进行处理，top20所有商户均在[BIZ_TYPE_LIST]中，其他数据统一归纳至Base索引中
        // 餐饮1.0，零售请求中存零售请求中存在无法透传商家id场景在无法透传商家id场景，所以隔离出这部分请求，设置分片为-1，网关自动路由至Base索引中
        boolean absent = checkOrderIndexRoute(bizType, query.getSellerId(), query.getSellerIdList());
        if (absent) {
            builder.addHeader(SELLER_ID, "-1");
        }
        // 设置header头
        if (!CollectionUtils.isEmpty(query.getIndexRouteSeller())) {
            builder.addHeader(SELLER_ID, query.getIndexRouteSeller().stream()
                    .map(Object::toString).collect(Collectors.joining(",")));
        } else if (!CollectionUtils.isEmpty(query.getSellerIdList())) {
            builder.addHeader(SELLER_ID, query.getSellerIdList().stream()
                    .map(Object::toString).collect(Collectors.joining(",")));
        } else if (query.getSellerId() != null) {
            builder.addHeader(SELLER_ID, query.getSellerId().toString());
        }
        if (query.getOrderAtStart() != null) {
            builder.addHeader(START_TIME, String.valueOf(query.getOrderAtStart().getTime()));
        }
        if (query.getOrderAtEnd() != null) {
            builder.addHeader(END_TIME, String.valueOf(query.getOrderAtEnd().getTime()));
        }
        builder.addHeader(TRACE_ID, ThreadContext.getTraceIdentity().getTraceId());
        return builder.build();
    }

    private boolean checkOrderIndexRoute(Integer bizType, Long sellerId, List<Long> sellerIdList) {
        return !BIZ_TYPE_LIST.contains(bizType) && sellerId == null && CollectionUtils.isEmpty(sellerIdList);
    }


    /**
     * 重建body，设置操作类型
     *
     * <p>
     *     全量写入（index）方式，不需要传id参数
     *     更新操作（update）方式，必须传入id
     * </p>
     *
     * @param bean bean
     *
     * @return json for body
     */
    @SuppressWarnings("unchecked")
    private String recast(Object bean, ElasticsearchOperateType operateType, Long id) {
        if (bean == null) {
            return null;
        }
        Map body;
        switch (operateType) {
            case INDEX:
                body = BeanMapUtils.beanToMap(bean);
                break;
            case UPDATE:
                body = (Map) bean;
                body.put("orderId", id);
                break;
            case DELETE:
                body = (Map) bean;
                body.put("orderId", id);
                break;
            default:
                throw new IllegalArgumentException("Wrong update elasticsearch operation type.");
        }
        body.put("_rt_changeFlg", operateType.getType());
        return JSON.toJSONString(body);
    }

    /**
     * 搜索订单
     *
     * @param searchOrderQuery
     * @return
     * @throws IOException
     */
    public SearchOrderDTO searchStdOrder(SearchOrderQuery searchOrderQuery, String[] includes, String[] excludes) throws IOException {
        SearchOrderDTO searchOrderDTO = new SearchOrderDTO();
        BusinessHandler businessHandler = businessHandlerHolder.orderGenerator(5);
        SearchSourceBuilder searchSourceBuilder = businessHandler.buildSearchSourceBuilder(searchOrderQuery);
        searchSourceBuilder.fetchSource(includes, excludes);
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.source(searchSourceBuilder);
        searchRequest.indices(NEW_ORDERS_INDEX_NAME);
        SearchResponse searchResponse = esClient.search(searchRequest, addHeader(searchOrderQuery, 5));
        if (searchResponse.status() != RestStatus.OK) {
            throw new RuntimeException("搜索订单失败");
        }
        final long total = searchResponse.getHits().getTotalHits();

        searchOrderDTO.setTotal(total);
        if (total == 0) {
            searchOrderDTO.setOrderList(Collections.EMPTY_LIST);
        } else {
            List<BaseIndexOrder> orderList = businessHandler.parseIndexOrder(searchResponse.getHits().getHits());
            searchOrderDTO.setOrderList(orderList);
        }
        return searchOrderDTO;
    }
}
