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

import com.alibaba.fastjson.JSON;
import com.zm.order.tool.server.component.CommentDocHandleHolder;
import com.zm.order.tool.server.config.KafkaProducerLauncher;
import com.zm.order.tool.server.constants.CommonConstants;
import com.zm.order.tool.server.enums.ElasticsearchOperateType;
import com.zm.order.tool.server.handler.CommentDocHandler;
import com.zm.order.tool.server.model.BaseIndexComment;
import com.zm.order.tool.server.model.OrderItemComment;
import com.zm.order.tool.server.model.TraceIdentity;
import com.zm.order.tool.server.utils.BeanMapUtils;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.List;
import java.util.Map;


/**
 * 搜索manager
 *
 * @Author: zhuxiaofeng
 * @Date: 2019/10/30 10:30
 */
@Component
public class SearchCommentManager {

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

    @Autowired
    private RestHighLevelClient esClient;

    @Autowired
    private CommentDocHandleHolder handle;

    @Autowired
    private KafkaProducerLauncher kafkaProducer;

    private static final String DEFAULT_TYPE = "_doc";

    private static final String INDEX_NAME = "comment";

    private static final String NEW_INDEX_NAME = "order_comments";

    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";

    /**
     * 构建评价文档
     *
     * @param comment 评价参数
     * @return 文档
     */
    public BaseIndexComment buildDocument(OrderItemComment comment) {
        Assert.notNull(comment.getSellerId(), "seller id is empty!");
        CommentDocHandler docHandler = handle.obtainHandler(comment.getBizType());
        return docHandler.buildCommentDoc(comment);
    }

    /**
     * 构建请求参数
     *
     * @param document 文档
     * @return 是否成功
     */
    public IndexRequest buildRequest(BaseIndexComment document) {
        IndexRequest request = new IndexRequest();
        request.index(INDEX_NAME);
        request.type(DEFAULT_TYPE);
        request.id(document.getCommentId().toString());
        request.source(JSON.toJSONString(document), XContentType.JSON);
        return request;
    }

    /**
     * 新增文档
     *
     * @param comment 评价参数
     * @return 是否成功
     */
    public boolean indexDocument(OrderItemComment comment) throws IOException {
        try {
            BaseIndexComment document = buildDocument(comment);
            // 发送消息到ES写网关服务，由网关执行写入操作
            kafkaProducer.sendAsync(CommonConstants.ES_ORDER_COMMENT_INDEX, comment.getOrderNo(), recast(document));
        } catch (Throwable e) {
            logger.error("SearchCommentManager[indexDocument]failed!orderNo:{},cause by:{}", comment.getOrderNo(), e);
            return false;
        }
        return true;
    }

    /**
     * 新增文档
     *
     * @param comment 评价参数
     * @return 是否成功
     */
    public boolean indexDocument(OrderItemComment comment, String topic) throws IOException {
        try {
            BaseIndexComment document = buildDocument(comment);
            // 发送消息到ES写网关服务，由网关执行写入操作
            kafkaProducer.sendAsync(topic, comment.getOrderNo(), recast(document));
        } catch (Throwable e) {
            logger.error("SearchCommentManager[indexDocument]failed!orderNo:{},cause by:{}", comment.getOrderNo(), e);
            return false;
        }
        return true;
    }

    /**
     * 批量插入
     *
     * @return
     */
    public boolean batchIndex(List<OrderItemComment> list) throws IOException {
        if (CollectionUtils.isEmpty(list)) {
            return true;
        }
        for (OrderItemComment comment : list) {
            indexDocument(comment);
        }
        return true;
    }

    /**
     * 批量插入
     *
     * @return
     */
    public boolean batchIndex(List<OrderItemComment> list, String topic) throws IOException {
        if (CollectionUtils.isEmpty(list)) {
            return true;
        }
        for (OrderItemComment comment : list) {
            indexDocument(comment, topic);
        }
        return true;
    }

    /**
     * 重建body
     *
     * <p>
     *     全量写入（index）
     * </p>
     *
     * @param bean bean
     *
     * @return json for body
     */
    private String recast(Object bean) {
        if (bean == null) {
            return null;
        }
        Map<String, Object> body = BeanMapUtils.beanToMap(bean);
        body.put("_rt_changeFlg", ElasticsearchOperateType.INDEX.getType());
        return JSON.toJSONString(body);
    }
}
