package com.sdtool.business.service;

import com.sdtool.common.api.NoSqlKeysService;
import com.sdtool.common.api.SysMessageService;
import com.sdtool.common.base.BaseService;
import com.sdtool.common.datas.ESKeys;
import com.sdtool.common.entity.site.*;
import com.sdtool.common.tool.ERPUtils;
import com.dySweetFishPlugin.elasticsearch.ESClient;
import com.dySweetFishPlugin.sql.RMapUtils;
import com.dySweetFishPlugin.sql.TableIdService;
import com.dySweetFishPlugin.tool.lang.DateUtil;
import com.dySweetFishPlugin.tool.lang.ThreadFactoryWithNamePrefix;
import com.sweetfish.convert.json.JsonConvert;
import com.sweetfish.service.Local;
import com.sweetfish.source.PageFlipper;
import com.sweetfish.util.AnyValue;
import com.sweetfish.util.AutoLoad;
import com.sweetfish.util.Reproduce;
import com.sweetfish.util.ResourceType;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryAction;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.filter.InternalFilter;
import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCount;
import org.elasticsearch.search.sort.SortOrder;
import org.rex.RMap;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

/**
 * Created by jlutt on 2018-11-23.
 * 站内信服务
 *
 * @author jlutt
 */
@Local
@AutoLoad(false)
@ResourceType(SysMessageService.class)
public class SysMessageServiceImpl extends BaseService implements SysMessageService {

    @Resource
    ESClient esClient;

    @Resource
    JsonConvert jsonConvert;

    @Resource
    TableIdService tableIdService;

    @Resource
    NoSqlKeysService keysService;

    private ExecutorService messageExecutor;

    @Override
    public void init(AnyValue config) {
        messageExecutor = Executors.newFixedThreadPool(8, new ThreadFactoryWithNamePrefix("[站内信程池]"));
    }

    @Override
    public void start(AnyValue config) {
    }

    @Override
    public void destroy(AnyValue config) {
        if (messageExecutor != null) {
            messageExecutor.shutdown();
        }
    }

    /**
     * 用户登录时同步系统的未读群发站内信
     * 点对点的站内信直接读取源数据即可
     *
     * @param userId 用户id
     */
    public void syncSysGroupMessage(long userId, long supplierCode) {
        CompletableFuture.runAsync(() -> {
            //读取所有未过期的信息
            BoolQueryBuilder qb = QueryBuilders.boolQuery();
            qb = qb.must(QueryBuilders.termQuery("messageType", 0))
                    .must(QueryBuilders.termQuery("expireFlag", 0));

            SearchResponse childRes = esClient.getClient().prepareSearch(keysService.getESKey(ESKeys.ESERPSYSMSG_INDEX, supplierCode))
                    .setTypes(ESKeys.ESERPDEFAULT_TYPE)
                    .setQuery(qb)
                    .setSize(1000)
                    .addSort("sendTimeLong", SortOrder.ASC)
                    .setScroll(new TimeValue(60000))
                    .execute()
                    .actionGet();

            List<SysMessage> messageList = new ArrayList<>();

            do {
                for (SearchHit hit : childRes.getHits().getHits()) {
                    SysMessage sysMessage = jsonConvert.convertFrom(SysMessage.class, hit.getSourceAsString());
                    messageList.add(sysMessage);
                }

                childRes = esClient.getClient().prepareSearchScroll(childRes.getScrollId()).setScroll(new TimeValue(60000)).execute().actionGet();
            } while (childRes.getHits().getHits().length != 0);

            if (!messageList.isEmpty()) {

                BulkRequestBuilder bulkRequest = esClient.getClient().prepareBulk();

                messageList.forEach((x) -> {
                    //设置Operation Type为create表示相同id存在的情况下，不写入也不覆盖，默认的为index覆盖。
                    try {
                        IndexRequestBuilder messageRequest = esClient.getClient()
                                .prepareIndex(keysService.getESKey(ESKeys.ESERPUSERMSG_INDEX, supplierCode), ESKeys.ESERPDEFAULT_TYPE)
                                .setOpType(DocWriteRequest.OpType.CREATE)
                                .setId(userId + "_" + x.getMessageId())
                                .setSource(jsonBuilder()
                                        .startObject()
                                        .field("userId", userId)
                                        .field("messageId", x.getMessageId())
                                        .field("messageType", x.getMessageType())
                                        .field("sendUserId", x.getSendUserId())
                                        .field("sendUserName", x.getSendUserName())
                                        .field("sendTimeLong", x.getSendTimeLong())
                                        .field("voidFlag", 0)
                                        .field("readStatus", 0)
                                        .field("textId", x.getTextId())
                                        .endObject());
                        bulkRequest.add(messageRequest);
                    } catch (IOException e) {
                        logger.error(e.getMessage(), e);
                    }
                });
                BulkResponse bulkResponse = bulkRequest.execute().actionGet();

                if (bulkResponse.hasFailures()) {
                    logger.error("索引站内信出错：" + bulkResponse.buildFailureMessage());
                } else {
                    logger.info("索引站内信成功：" + userId);
                }
            }

        }, messageExecutor);
    }

    /**
     * 系统给所有用户发送群发站内信
     *
     * @return 1发送成功
     */
    private int sendSysGroupMessage(SysMessage message, SysMessageText messageText, ERPTokenUser currentUser, long supplierCode) {
        CompletableFuture.runAsync(() -> {

            messageText.setTextId(tableIdService.getArticleId(1));

            message.setMessageId(tableIdService.getArticleId(1));
            if (currentUser != null) {
                SysMessage.create(message, currentUser.getId());
                message.setSendUserId(currentUser.getId());
                message.setSendUserName(currentUser.getUserName());
            } else {
                message.setCreateBy(0);
                message.setCreateTime(new Date());
                message.setSendUserId(0);
                message.setSendUserName("");
            }
            message.setSendTimeLong(message.getCreateTime().getTime());

            message.setMessageType(0);
            message.setRecvUserId(0);
            message.setRecvUserName("");
            message.setReadStatus(0);
            message.setExpireFlag(0);
            message.setTextId(messageText.getTextId());

            BulkRequestBuilder bulkRequest = esClient.getClient().prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            IndexRequestBuilder messageRequest = esClient.getClient()
                    .prepareIndex(keysService.getESKey(ESKeys.ESERPSYSMSG_INDEX, supplierCode), ESKeys.ESERPDEFAULT_TYPE)
                    .setId(String.valueOf(message.getMessageId()))
                    .setSource(jsonConvert.convertTo(message), XContentType.JSON);
            bulkRequest.add(messageRequest);
            IndexRequestBuilder textRequest = esClient.getClient()
                    .prepareIndex(keysService.getESKey(ESKeys.ESERPSYSMSGTEXT_INDEX, supplierCode), ESKeys.ESERPDEFAULT_TYPE)
                    .setId(String.valueOf(messageText.getTextId()))
                    .setSource(jsonConvert.convertTo(messageText), XContentType.JSON);
            bulkRequest.add(textRequest);

            BulkResponse bulkResponse = bulkRequest.execute().actionGet();

            if (bulkResponse.hasFailures()) {
                logger.error("发送群发站内信错误：" + bulkResponse.buildFailureMessage());
            } else {
                logger.info("发送群发站内信成功：" + message.getMessageId());
            }

        }, messageExecutor);

        return 1;
    }

    /**
     * 系统给所有用户发送群发站内信
     *
     * @param title        信件标题
     * @param content      信件内容
     * @param expireDate   过期日期
     * @param currentUser  发送人，可为空
     * @param supplierCode 分表
     * @return 发送成功
     */
    public int sendSysGroupMessage(String title, String content,
                                   Date expireDate,
                                   ERPTokenUser currentUser, long supplierCode) {
        SysMessage sysMessage = new SysMessage();
        sysMessage.setExpireTime(expireDate.getTime());

        SysMessageText sysMessageText = new SysMessageText();
        sysMessageText.setMessageTitle(title);
        sysMessageText.setMessageContent(content);
        sysMessageText.setMessageLink("");

        return sendSysGroupMessage(sysMessage, sysMessageText, currentUser, supplierCode);
    }

    /**
     * 发送P2P站内信
     *
     * @param message     消息结构
     * @param messageText 内容结构
     * @param currentUser 发送人
     * @param userIds     接收人
     * @return 1发送成功
     */
    private int sendP2PMessage(SysMessage message, SysMessageText messageText, ERPTokenUser currentUser, long supplierCode, String... userIds) {
        CompletableFuture.runAsync(() -> {

            messageText.setTextId(tableIdService.getArticleId(1));

            if (currentUser != null) {
                SysMessage.create(message, currentUser.getId());
                message.setSendUserId(currentUser.getId());
                message.setSendUserName(currentUser.getUserName());
            } else {
                message.setCreateBy(0);
                message.setCreateTime(new Date());
                message.setSendUserId(0);
                message.setSendUserName("系统自动");
            }
            message.setSendTimeLong(message.getCreateTime().getTime());
            message.setReadStatus(0);
            message.setExpireFlag(0);
            message.setTextId(messageText.getTextId());

            BulkRequestBuilder bulkRequest = esClient.getClient().prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);

            Reproduce<SysMessage, SysMessage> messageReproduce = Reproduce.create(SysMessage.class, SysMessage.class);
            Arrays.stream(userIds).forEach((x) -> {
                SysMessage saveMessage = new SysMessage();
                saveMessage = messageReproduce.apply(saveMessage, message);
                saveMessage.setMessageId(tableIdService.getArticleId(1));
                saveMessage.setRecvUserId(Integer.parseInt(x));

                IndexRequestBuilder messageRequest = esClient.getClient()
                        .prepareIndex(keysService.getESKey(ESKeys.ESERPP2PMSG_INDEX, supplierCode), ESKeys.ESERPDEFAULT_TYPE)
                        .setId(String.valueOf(saveMessage.getMessageId()))
                        .setSource(jsonConvert.convertTo(saveMessage), XContentType.JSON);
                bulkRequest.add(messageRequest);
            });

            IndexRequestBuilder textRequest = esClient.getClient()
                    .prepareIndex(keysService.getESKey(ESKeys.ESERPP2PMSGTEXT_INDEX, supplierCode), ESKeys.ESERPDEFAULT_TYPE)
                    .setId(String.valueOf(messageText.getTextId()))
                    .setSource(jsonConvert.convertTo(messageText), XContentType.JSON);
            bulkRequest.add(textRequest);

            BulkResponse bulkResponse = bulkRequest.execute().actionGet();

            if (bulkResponse.hasFailures()) {
                logger.error("发送P2P消息错误：" + bulkResponse.buildFailureMessage());
            } else {
                logger.info("发送P2P消息成功：" + userIds.length);
            }

        }, messageExecutor);

        return 1;
    }

    /**
     * 发送P2P站内信
     *
     * @param messageKind  消息类型
     * @param title        信件标题
     * @param content      信件内容
     * @param currentUser  发送人
     * @param supplierCode 分表
     * @param userIds      接受用户id数组
     * @return 1成功
     */
    public int sendP2PMessage(int messageKind, String title, String content,
                              ERPTokenUser currentUser, long supplierCode, String... userIds) {
        SysMessage sysMessage = new SysMessage();
        sysMessage.setMessageType(messageKind);

        SysMessageText sysMessageText = new SysMessageText();
        sysMessageText.setMessageTitle(title);
        sysMessageText.setMessageContent(content);
        sysMessageText.setMessageLink("");

        return sendP2PMessage(sysMessage, sysMessageText, currentUser, supplierCode, userIds);
    }

    /**
     * 获取某个用户未读站内信的数量
     *
     * @param userId 用户id
     * @return 未读站内信数量
     */
    public int queryUnReadMessageCount(long userId, long supplierCode) {
        //1、查询群发消息未读的数据
        BoolQueryBuilder qbGroup = QueryBuilders.boolQuery();
        qbGroup = qbGroup.must(QueryBuilders.termQuery("userId", userId))
                .must(QueryBuilders.termQuery("messageType", 0))
                .must(QueryBuilders.termQuery("readStatus", 0));

        FilterAggregationBuilder groupAggregation = AggregationBuilders.filter("message", qbGroup);
        groupAggregation.subAggregation(AggregationBuilders.count("message_Count").field("messageId"));
        SearchResponse groupAggRes = esClient.getClient().prepareSearch(keysService.getESKey(ESKeys.ESERPUSERMSG_INDEX, supplierCode))
                .setTypes(ESKeys.ESERPDEFAULT_TYPE)
                .setQuery(qbGroup)
                .addAggregation(groupAggregation)
                //.setSize(0)
                .execute()
                .actionGet();

        InternalFilter groupGenders = groupAggRes.getAggregations().get("message");
        ValueCount sysCount = groupGenders.getAggregations().get("message_Count");

        //2、查询p2p消息未读的数据
        BoolQueryBuilder qbP2P = QueryBuilders.boolQuery();
        qbP2P = qbP2P.must(QueryBuilders.termQuery("recvUserId", userId))
                .must(QueryBuilders.termQuery("readStatus", 0))
                .must(QueryBuilders.termQuery("expireFlag", 0));

        FilterAggregationBuilder p2pAggregation = AggregationBuilders.filter("message", qbP2P);
        p2pAggregation.subAggregation(AggregationBuilders.count("message_Count").field("messageId"));
        SearchResponse p2pAggRes = esClient.getClient().prepareSearch(keysService.getESKey(ESKeys.ESERPP2PMSG_INDEX, supplierCode))
                .setTypes(ESKeys.ESERPDEFAULT_TYPE)
                .setQuery(qbP2P)
                .addAggregation(p2pAggregation)
                //.setSize(0)
                .execute()
                .actionGet();

        InternalFilter p2pGenders = p2pAggRes.getAggregations().get("message");
        ValueCount p2pCount = p2pGenders.getAggregations().get("message_Count");

        return (int) (sysCount.getValue() + p2pCount.getValue());
    }

    /**
     * 处理消息列表，获取消息内容
     */
    private void processMessageText(List<SysMessageMap> messageMapList, List<String> textIds, String index) {
        SearchRequestBuilder textBuilder = esClient.getClient().prepareSearch(index)
                .setTypes(ESKeys.ESERPDEFAULT_TYPE)
                .setQuery(QueryBuilders.idsQuery().addIds(textIds.toArray(new String[0])));

        SearchHits textHits = textBuilder.execute().actionGet().getHits();
        textHits.forEach(x -> {
            String json = x.getSourceAsString();
            SysMessageText messageText = jsonConvert.convertFrom(SysMessageText.class, json);

            messageMapList.forEach((y) -> {
                if (y.getTextId() == messageText.getTextId()) {
                    y.setMessageText(messageText);
                }
            });
        });
    }

    /**
     * 用户获取个人能读取的站内信列表
     *
     * @param userId      用户id
     * @param messageType 站内信类型 0表示系统群发消息
     * @param pageFlipper 分页参数
     * @return 站内信列表
     */
    private List<SysMessageMap> queryMessageList(long userId, int messageType, PageFlipper pageFlipper, long supplierCode) {
        BoolQueryBuilder qb = QueryBuilders.boolQuery();

        if (messageType == 0) {
            qb = qb.must(QueryBuilders.termQuery("userId", userId))
                    .must(QueryBuilders.termQuery("messageType", 0));
        } else {
            qb = qb.must(QueryBuilders.termQuery("recvUserId", userId))
                    .must(QueryBuilders.termQuery("messageType", messageType));
        }

        String index = (messageType == 0) ? keysService.getESKey(ESKeys.ESERPUSERMSG_INDEX, supplierCode) : keysService.getESKey(ESKeys.ESERPP2PMSG_INDEX, supplierCode);

        SearchRequestBuilder requestBuilder = esClient.getClient().prepareSearch(index)
                .setTypes(ESKeys.ESERPDEFAULT_TYPE)
                .setQuery(qb)
                .addSort("sendTimeLong", SortOrder.DESC)
                .setSize(pageFlipper.getPageSize())
                .setFrom(pageFlipper.getPageSize() * (pageFlipper.getPageIndex() - 1));

        SearchHits searchHits = requestBuilder.execute().actionGet().getHits();

        List<SysMessageMap> messageMapList = new ArrayList<>();
        List<String> textIds = new ArrayList<>();
        searchHits.forEach(x -> {
            String json = x.getSourceAsString();

            if (messageType == 0) {
                SysMessageStatus messageStatus = jsonConvert.convertFrom(SysMessageStatus.class, json);
                messageMapList.add(new SysMessageMap((messageStatus)));
                textIds.add(String.valueOf(messageStatus.getTextId()));
            } else {
                SysMessage message = jsonConvert.convertFrom(SysMessage.class, json);
                messageMapList.add(new SysMessageMap((message)));
                textIds.add(String.valueOf(message.getTextId()));
            }
        });

        //获取消息内容
        if (!textIds.isEmpty()) {
            String messageIndex = (messageType == 0) ? keysService.getESKey(ESKeys.ESERPSYSMSGTEXT_INDEX, supplierCode) : keysService.getESKey(ESKeys.ESERPP2PMSGTEXT_INDEX, supplierCode);

            processMessageText(messageMapList, textIds, messageIndex);
        }

        pageFlipper.setTotal(searchHits.getTotalHits());

        return messageMapList;
    }

    /**
     * 标记站内信为已读
     *
     * @param userId      用户id
     * @param messageIds  站内信ids
     * @param messageType 站内信类型 0表示系统群发消息
     */
    public void readMessage(long userId, String messageIds, int messageType, long supplierCode) {
        String[] ids;

        if (messageType == 0) {
            ids = Arrays.stream(messageIds.split(",")).map((x) -> userId + "_" + x).toArray(String[]::new);
        } else {
            ids = messageIds.split(",");
        }

        String index = (messageType == 0) ? keysService.getESKey(ESKeys.ESERPUSERMSG_INDEX, supplierCode) : keysService.getESKey(ESKeys.ESERPP2PMSG_INDEX, supplierCode);
        BoolQueryBuilder qb = QueryBuilders.boolQuery();
        qb = qb.must(QueryBuilders.idsQuery().addIds(ids))
                .must(QueryBuilders.typeQuery(ESKeys.ESERPDEFAULT_TYPE));

        //批量更新
        UpdateByQueryAction.INSTANCE.newRequestBuilder(esClient.getClient())
                .refresh(true)
                .abortOnVersionConflict(false)
                .source(index)
                .script(new Script(
                        ScriptType.INLINE,
                        "painless",
                        "ctx._source.readStatus = 1",
                        Collections.emptyMap()
                ))
                .filter(qb)
                .execute(new ActionListener<>() {
                    //回调监听
                    @Override
                    public void onResponse(BulkByScrollResponse response) {
                        logger.info("修改已读：" + response.getUpdated());
                    }

                    @Override
                    public void onFailure(Exception e) {
                        // Handle the exception
                        logger.error(e.getMessage(), e);
                    }
                });
    }

    /**
     * 管理-站内信获取列表
     *
     * @param searchData   查询条件
     * @param currentUser  当前用户
     * @param pageFlipper  分页参数
     * @param messageIndex es站内信索引库index
     * @param textIndex    es站内信内容index
     * @return 站内信列表
     */
    @SuppressWarnings("rawtypes")
    private RMap manageMessageList(RMap searchData, ERPTokenUser currentUser, PageFlipper pageFlipper,
                                   String messageIndex, String textIndex) {
        BoolQueryBuilder qb = QueryBuilders.boolQuery();

        qb = qb.must(QueryBuilders.typeQuery(ESKeys.ESERPDEFAULT_TYPE))
                .must(QueryBuilders.termQuery("sendUserId", currentUser.getId()));

        if (searchData != null) {
            if (ERPUtils.mapContainKey(searchData, "beginTime")) {
                Date beginDate = DateUtil.asDate(searchData.getString("beginTime"), "yyyy-MM-dd");
                qb = qb.must(QueryBuilders.rangeQuery("sendTimeLong").gte(DateUtil.getStartOfDay(beginDate).getTime()));
            }

            if (ERPUtils.mapContainKey(searchData, "endTime")) {
                Date endDate = DateUtil.asDate(searchData.getString("endTime"), "yyyy-MM-dd");
                qb = qb.must(QueryBuilders.rangeQuery("sendTimeLong").lte(DateUtil.getEndOfDay(endDate).getTime()));
            }
        }

        SearchRequestBuilder requestBuilder = esClient.getClient().prepareSearch(messageIndex)
                .setQuery(qb)
                .addSort("sendTimeLong", SortOrder.DESC)
                .setSize(pageFlipper.getPageSize())
                .setFrom(pageFlipper.getPageSize() * (pageFlipper.getPageIndex() - 1));

        SearchHits searchHits = requestBuilder.execute().actionGet().getHits();

        List<SysMessageMap> messageMapList = new ArrayList<>();
        List<String> textIds = new ArrayList<>();
        searchHits.forEach(x -> {
            String json = x.getSourceAsString();

            SysMessage message = jsonConvert.convertFrom(SysMessage.class, json);
            messageMapList.add(new SysMessageMap((message)));
            textIds.add(String.valueOf(message.getTextId()));
        });

        //获取消息内容
        if (!textIds.isEmpty()) {
            processMessageText(messageMapList, textIds, textIndex);
        }

        pageFlipper.setTotal(searchHits.getTotalHits());

        return RMapUtils.successV2(messageMapList, null, pageFlipper);

    }

    /**
     * 管理系统群发站内信--获取列表
     *
     * @param searchData  查询条件
     * @param currentUser 当前用户
     * @param pageFlipper 分页
     * @return 站内信列表
     */
    @SuppressWarnings("rawtypes")
    public RMap manageSysMessageList(RMap searchData, ERPTokenUser currentUser, PageFlipper pageFlipper, long supplierCode) {
        return manageMessageList(searchData, currentUser, pageFlipper,
                keysService.getESKey(ESKeys.ESERPSYSMSG_INDEX, supplierCode),
                keysService.getESKey(ESKeys.ESERPSYSMSGTEXT_INDEX, supplierCode));
    }

    /**
     * 管理当前登录人员发送的点对点消息--获取列表
     *
     * @param searchData  查询条件
     * @param currentUser 当前用户
     * @param pageFlipper 分页
     * @return 站内信列表
     */
    @SuppressWarnings("rawtypes")
    public RMap manageP2PMessageList(RMap searchData, ERPTokenUser currentUser, PageFlipper pageFlipper, long supplierCode) {
        return manageMessageList(searchData, currentUser, pageFlipper, keysService.getESKey(ESKeys.ESERPP2PMSG_INDEX, supplierCode), keysService.getESKey(ESKeys.ESERPP2PMSGTEXT_INDEX, supplierCode));
    }

    /**
     * 用户获取个人能读取的站内信列表
     *
     * @param currentUser 当前用户
     * @param messageType 站内信类型 0表示系统群发消息
     * @param pageFlipper 分页参数
     * @return 站内信列表
     */
    @SuppressWarnings("rawtypes")
    public RMap manageRecvMessageList(int messageType, ERPTokenUser currentUser, PageFlipper pageFlipper, long supplierCode) {
        List<SysMessageMap> list = queryMessageList(currentUser.getId(), messageType, pageFlipper, supplierCode);
        return RMapUtils.successV2(list, null, pageFlipper);
    }
}
