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

import com.alibaba.fastjson.JSON;
import com.google.common.base.Throwables;
import com.zm.order.tool.server.component.RefundBusHandlerHolder;
import com.zm.order.tool.server.component.RefundDocHandleHolder;
import com.zm.order.tool.server.config.KafkaProducerLauncher;
import com.zm.order.tool.server.dto.SearchRefundDTO;
import com.zm.order.tool.server.enums.ElasticsearchOperateType;
import com.zm.order.tool.server.handler.refund.RefundBusHandler;
import com.zm.order.tool.server.handler.refund.RefundDocHandler;
import com.zm.order.tool.server.model.Refund;
import com.zm.order.tool.server.model.TraceIdentity;
import com.zm.order.tool.server.model.refund.BaseRefundIndex;
import com.zm.order.tool.server.query.SearchRefundQuery;
import com.zm.order.tool.server.utils.BeanMapUtils;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
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.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
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 java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * 搜索退款manager
 *
 * @Author: zhuxiaofeng
 * @Date: 2022-11-10 14:23:35
 */
@Component
public class SearchRefundManager {

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

    @Autowired
    private RestHighLevelClient esClient;

    @Autowired
    private RefundDocHandleHolder handle;

    @Autowired
    private KafkaProducerLauncher kafkaProducer;

    private static final String DEFAULT_TYPE = "_doc";

    private static final String INDEX_NAME = "refund";

    /**
     * 构建退款文档
     *
     * @param refund 退款数据
     * @return 文档
     */
    public BaseRefundIndex buildDocument(Refund refund) {
        BaseRefundIndex index = new BaseRefundIndex();
        try {
            RefundDocHandler docHandler = handle.obtainHandler(refund.getBizType());
            index = docHandler.buildRefundDoc(refund);
        } catch (Throwable e) {
            logger.error("buildDocument[BaseRefundIndex] failed!, cause by:{}", e);
        }
        return index;

    }

    /**
     * 构建退款文档
     *
     * @param refund 退款数据
     * @return 文档
     */
    public void indexRefund(Refund refund, String indexName) {
        try {
            RefundDocHandler docHandler = handle.obtainHandler(4);
            BaseRefundIndex document = docHandler.buildRefundDoc(refund);
            kafkaProducer.sendAsync(indexName,
                    refund.getRefundNo(), recast(document, ElasticsearchOperateType.INDEX, null));
        } catch (Throwable e) {
            logger.error("buildDocument[BaseRefundIndex] failed! cause by:{0}", e);
        }

    }

    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:
            case PART_UPDATE:
                body = (Map) bean;
                body.put("docId", id);
                break;
            default:
                throw new IllegalArgumentException("Wrong update elasticsearch operation type from refund.");
        }
        body.put("_rt_changeFlg", operateType.getType());
        return JSON.toJSONString(body);
    }

    /**
     * 构建请求参数
     *
     * @param document 文档
     * @return 是否成功
     */
    public IndexRequest buildRequest(BaseRefundIndex document) {
        IndexRequest request = new IndexRequest();
        try {
            request.type(DEFAULT_TYPE);
            request.index(INDEX_NAME);
            request.id(document.getRefundId().toString());
            request.source(JSON.toJSONString(document), XContentType.JSON);
        } catch (Throwable e) {
            logger.error("BaseRefundIndex[buildRequest] failed!, cause by:{}", e);
        }
        return request;
    }

    /**
     * 新增文档
     *
     * @param refund 退款模型
     * @return 是否成功
     */
    public boolean indexDocument(Refund refund) throws IOException {
        try {
            BaseRefundIndex document = buildDocument(refund);
            IndexResponse index = esClient.index(buildRequest(document), RequestOptions.DEFAULT);
        } catch (Throwable e) {
            logger.error("SearchRefundManager[indexDocument]failed!orderNo:{},cause by:{}", refund.getOrderNo(), e);
            return false;
        }
        return true;
    }

    /**
     * 批量插入
     *
     * @return
     */
    public boolean batchIndex(List<Refund> list) throws IOException {
        if (CollectionUtils.isEmpty(list)) {
            return true;
        }
        BulkRequest request = new BulkRequest();
        for (Refund refund : list) {
            BaseRefundIndex document = buildDocument(refund);
            request.add(buildRequest(document));
        }
        try {
            esClient.bulk(request, RequestOptions.DEFAULT);
        } catch (Throwable throwable) {
            logger.error("RefundManager[batchIndex] failed, request:[{}], cause by:{}",
                    request, Throwables.getStackTraceAsString(throwable));
            throw new RuntimeException(throwable.getMessage());
        }
        return true;
    }

    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 INDEX_NAMES = "order_refund";

    /**
     * 新
     */
    private static final String NEW_INDEX_NAME = "order_refunds";

    @Autowired
    private RefundBusHandlerHolder handler;

    /**
     * 退款单检索
     *
     * <p>
     *      适用于退款单导出场景，只返回指定字段
     *      注意：如果根据id推进分页，请设置respCount字段为false
     * </p>
     *
     * @param query query
     * @param includes 期望只返回的字段
     * @param excludes 期望排除的字段
     *
     * @return SearchRefundDTO
     */
    public SearchRefundDTO searchRefund(SearchRefundQuery query, String[] includes, String[] excludes) throws IOException {
        RefundBusHandler busHandler = handler.refundBusHandler(5);
        if (busHandler == null) {
            throw new RuntimeException("Absent biz type for refund search.");
        }
        SearchSourceBuilder searchSourceBuilder = busHandler.buildSearchSourceBuilder(query);
        searchSourceBuilder.fetchSource(includes, excludes);
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(INDEX_NAMES);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = esClient.search(searchRequest, addHeader(query));
        if (searchResponse.status() != RestStatus.OK) {
            throw new RuntimeException("搜索退款单失败");
        }
        SearchHit[] hits = searchResponse.getHits().getHits();
        final long total = searchResponse.getHits().getTotalHits();
        return (total == 0 || hits.length == 0) ?
                new SearchRefundDTO(0L, Collections.emptyList()) :
                new SearchRefundDTO(total, busHandler.parseIndex(hits));
    }

    private RequestOptions addHeader(SearchRefundQuery query) {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        // 设置header头
        if (query.getSellerId() != null) {
            builder.addHeader(SELLER_ID, query.getSellerId().toString());
        }
        if (query.getApplyAtStart() != null) {
            builder.addHeader(START_TIME, String.valueOf(query.getApplyAtStart().getTime()));
        }
        if (query.getApplyAtEnd() != null) {
            builder.addHeader(END_TIME, String.valueOf(query.getApplyAtEnd().getTime()));
        }
        return builder.build();
    }

}
