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

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;
import com.github.dozermapper.core.Mapper;
import com.google.common.base.Throwables;
import com.zm.order.tool.server.constants.AppointFlagConstants;
import com.zm.order.tool.server.constants.ExceptionConstants;
import com.zm.order.tool.server.dto.OrderOperateDTO;
import com.zm.order.tool.server.enums.OperateTypeEnum;
import com.zm.order.tool.server.message.*;
import com.zm.order.tool.server.model.*;
import com.zm.util.dto.Result;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;

/**
 * @Description 消息队列manager
 * @Author: ph
 * @Date: 2019/11/12 19:58
 */
@Component
public class AsyncMessageManager {

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

    @Autowired
    private Producer producer;

    @Autowired
    private OrderProducer orderProducer;

    @Value("${order.message.push.queue.topic}")
    private String orderMessagePushQueueTopic;

    @Value("${order.message.push.queue.tag}")
    private String orderMessagePushQueueTag;

    @Value("${order.sync.es.topic}")
    private String orderSyncEsTopic;

    @Value("${order.sync.es.tag}")
    private String orderSyncEsTag;

    @Value("${order.batch.sync.es.topic}")
    private String orderBatchSyncEsTopic;
    @Value("${order.common.topic}")
    private String commonTopic;
    @Value("${order.batch.sync.es.tag}")
    private String orderBatchSyncEsTag;

    @Autowired
    private Mapper mapper;
    private static final long THRESHOLD = 100L;

    /**
     * 订单消息推送
     *
     * @return
     */
    public Result orderMessagePush(OrderPush orderPush) {
        Result result = new Result();
        try {
            Message message = new Message();
            message.setTopic(orderMessagePushQueueTopic);
            message.setTag(orderMessagePushQueueTag);
            message.setKey(orderPush.getOrderNo());
            message.setShardingKey(orderPush.getOrderNo());
            message.setBody(JSON.toJSONString(orderPush).getBytes());
            // 发消息
            SendResult sendResult = orderProducer.send(message, orderPush.getOrderNo());
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[orderMessagePush]failed, cause by:{}",
                    Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    @Value("${order.message.queue.topic}")
    private String orderMessageQueueTopic;

    @Value("${order.message.queue.tag}")
    private String orderMessageQueueTag;

    /**
     * 发送订单推送队列
     *
     * @return
     */
    public Result sendOrderMessageQueue(OrderPushMessage before, OrderPushMessage after, Order order) {
        Result result = new Result();
        try {
            OrderMessageQueue orderMessageQueue = new OrderMessageQueue();
            orderMessageQueue.setBefore(before);
            orderMessageQueue.setAfter(after);
            orderMessageQueue.setOrderNo(order.getOrderNo());
            orderMessageQueue.setUserId(order.getUserId());
            orderMessageQueue.setBizType(order.getBizType());
            Message message = new Message();
            message.setTopic(orderMessageQueueTopic);
            message.setTag(orderMessageQueueTag);
            message.setKey(order.getOrderNo());
            message.setShardingKey(order.getOrderNo());
            message.setBody(JSON.toJSONString(orderMessageQueue).getBytes());
            // 发消息
            long time1 = System.currentTimeMillis();
            SendResult sendResult = orderProducer.send(message, order.getOrderNo());
            long time2 = System.currentTimeMillis();
            long delay = time2 - time1;
            if (delay > THRESHOLD) {
                logger.info("[delayOns]orderMessagePush send ons message delay, msgId:{}, delayTime:{}",
                        sendResult.getMessageId(), delay);
            }
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendOrderMessageQueue]failed, orderNo:{}, cause by:{}",
                    order.getOrderNo(), Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    /**
     * 发送同步es消息
     *
     * @param orderNo
     * @param userId
     * @param version
     * @return
     */
    public Result sendOrderSyncEsMessage(String orderNo, Long userId, Integer version) {
        Result result = new Result();
        try {
            OrderSyncEsMessage orderSyncEsMessage = new OrderSyncEsMessage();
            orderSyncEsMessage.setOrderNo(orderNo);
            orderSyncEsMessage.setUserId(userId);
            orderSyncEsMessage.setVersion(version);
            Message message = new Message();
            message.setTopic(orderSyncEsTopic);
            message.setTag(orderSyncEsTag);
            message.setBody(JSON.toJSONString(orderSyncEsMessage).getBytes());
            SendResult sendResult = producer.send(message);
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendOrderSyncEsMessage]failed, orderNo:{} cause by:{}",
                    orderNo, Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SEND_SYNC_ES_MESSAGE_ERROR);
        }
        return result;
    }

    /**
     * 订单消息推送
     *
     * @return
     */
    public Result sendBatchSyncMessage(List<BatchSyncEsMessage> messageList) {
        Result result = new Result();
        try {
            Message message = new Message();
            message.setTopic(commonTopic);
            message.setTag(orderBatchSyncEsTag);
            message.setBody(JSON.toJSONString(messageList).getBytes());
            SendResult sendResult = producer.send(message);
//            logger.info("[orderMessagePush]success! request:{}", messageList);
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendBatchSyncMessage]failed, cause by:{}",
                    Throwables.getStackTraceAsString(throwable));
        }
        return result;
    }

    /**
     * 订单消息推送
     *
     * @return
     */
    public Result sendBatchSyncMessageNewOrder(List<BatchSyncEsMessage> messageList) {
        Result result = new Result();
        try {
            Message message = new Message();
            message.setTopic(commonTopic);
            message.setTag(orderBatchSyncEsTag);
            message.setBody(JSON.toJSONString(messageList).getBytes());
            Properties properties = new Properties();
            properties.setProperty("qm-trace-store-id", "209138");
            message.setUserProperties(properties);
            SendResult sendResult = producer.send(message);
            logger.info("[sendBatchSyncMessageNewOrder]success! res:{}", sendResult);
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendBatchSyncMessage]failed, cause by:{}",
                    Throwables.getStackTraceAsString(throwable));
        }
        return result;
    }

    @Value("${order.comment.count.topic}")
    private String orderCommentCountTopic;

    @Value("${order.comment.count.tag}")
    private String orderCommentCountTag;

    /**
     * 发送评论消息
     *
     * @param orderComment orderComment
     * @return
     * Date: 2021/6/22 5:02 下午
     */
    public Result<Integer> sendCommentCountMessage(OrderComment orderComment) {
        Result<Integer> result = new Result();
        try {
            OrderCommentCountMessage commentMessage = mapper.map(orderComment, OrderCommentCountMessage.class);
            commentMessage.setFirstCreate(1);
            Message message = new Message();
            message.setTopic(orderCommentCountTopic);
            message.setTag(orderCommentCountTag);
            message.setBody(JSON.toJSONString(commentMessage).getBytes());
            SendResult sendResult = producer.send(message);
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendCommentCountMessage]failed, orderNo:{}, sellerId:{}, cause by:{}",
                    orderComment.getOrderNo(), orderComment.getSellerId(), Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    @Value("${order.index.topic}")
    private String orderIndexTopic;

    @Value("${order.index.tag}")
    private String orderIndexTag;
    /**
     * 发送订单索引消息
     *
     * @return
     */
    public Result sendOrderIndexMessage(Order order, OperateTypeEnum operateTypeEnum) {
        Result result = new Result();
        Message message = new Message();
        try {
            OrderIndexMessage orderIndexMessage = new OrderIndexMessage();
            orderIndexMessage.setBizType(order.getBizType());
            orderIndexMessage.setOrderId(order.getId());
            orderIndexMessage.setOrderNo(order.getOrderNo());
            orderIndexMessage.setUserId(order.getUserId());
            orderIndexMessage.setSourceNo(order.getSourceNo());
            orderIndexMessage.setSellerId(order.getSellerId());
            orderIndexMessage.setOperateType(operateTypeEnum.getType());
            // 构建默认值
            if (Objects.isNull(orderIndexMessage.getUserId())) {
                orderIndexMessage.setUserId(0L);
            }
            if (Objects.isNull(orderIndexMessage.getOrderId())) {
                orderIndexMessage.setOrderId(0L);
            }
            if (Objects.isNull(orderIndexMessage.getBizType())) {
                orderIndexMessage.setUserId(0L);
            }
            message.setTopic(orderIndexTopic);
            message.setTag(orderIndexTag);
            message.setBody(JSON.toJSONString(orderIndexMessage).getBytes());
            // 发消息
            long time1 = System.currentTimeMillis();
            SendResult sendResult = producer.send(message);
            long time2 = System.currentTimeMillis();
            long delay = time2 - time1;
            if (delay > THRESHOLD) {
            }
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendOrderIndexMessage] failed, message:{}, cause by:{}",
                    message.toString(), Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SEND_ORDER_INDEX_MESSAGE_ERROR);
        }
        return result;
    }

    @Value("${order.comment.message.topic}")
    private String orderCommentMessageTopic;

    @Value("${order.comment.message.tag}")
    private String orderCommentMessageTag;

    /**
     * 发送评论消息
     *
     * @param body message
     * @return
     * Date: 2021/6/22 5:02 下午
     */
    public Result<Integer> sendCommentMessage(OrderCommentMessage body) {
        Result<Integer> result = new Result();
        try {
            Message message = new Message();
            message.setTopic(orderCommentMessageTopic);
            message.setTag(orderCommentMessageTag);
            message.setBody(JSON.toJSONString(body).getBytes());
            // 发消息
            producer.send(message);
            result.setStatus(true);
        } catch (Exception e) {
            logger.error("AsyncMessageManager[sendCommentMessage]failed! request:{}, cause by:{}", body, e);
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    @Value("${sync.sub.orders.topic}")
    private String syncSubOrdersTopic;

    @Value("${sync.sub.orders.tag}")
    private String syncSubOrdersTag;

    public Result sendSyncSubOrdersMessage(OrderOperateDTO operateDTO) {
        Result result = new Result();
        try {
            SyncSubOrdersMessage syncSubOrdersMessage = new SyncSubOrdersMessage();
            syncSubOrdersMessage.setOrderNo(operateDTO.getOrderNo());
            syncSubOrdersMessage.setUserId(operateDTO.getUserId());
            Map<String, Object> operateInfo = operateDTO.getOperateInfo();
            Integer multiUserFlag = null;
            if (Objects.nonNull(operateInfo.get(AppointFlagConstants.MULTI_USER_FLAG))) {
                multiUserFlag = Integer.valueOf(operateInfo.get(AppointFlagConstants.MULTI_USER_FLAG).toString());
            }
            syncSubOrdersMessage.setMultiUserFlag(multiUserFlag);
            Message message = new Message();
            message.setTopic(syncSubOrdersTopic);
            message.setTag(syncSubOrdersTag);
            message.setBody(JSON.toJSONString(syncSubOrdersMessage).getBytes());
            // 发消息
            long time1 = System.currentTimeMillis();
            SendResult sendResult = producer.send(message);
            long time2 = System.currentTimeMillis();
            long delay = time2 - time1;
            if (delay > THRESHOLD) {
                logger.info("[delayOns]sendSyncSubOrdersMessage send ons message delay, msgId:{}, delayTime:{}",
                        sendResult.getMessageId(), delay);
            }
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendSyncSubOrdersMessage]massage send failed, cause by:{}", throwable);
            result.setException(ExceptionConstants.SEND_SYNC_SUB_ORDERS_MESSAGE_ERROR);
        }
        return result;
    }


    /**
     * 订单取消消息
     * @date 2022/7/5 14:41
     * @param order
     * @param orderCancelTag
     * @return com.zm.util.dto.Result<java.lang.Integer>
     **/
    public Result<Integer> sendOrderCancelNoticeMessage(Order order, String orderCancelTag) {
        Result<Integer> result = new Result<>();
        try {
            OrderNoticeMessage body = mapper.map(order, OrderNoticeMessage.class);
            Message message = new Message();
            message.setTopic(commonTopic);
            message.setTag(orderCancelTag);
            message.setBody(JSON.toJSONString(body).getBytes());
            sendMessageAndTime(message, "sendOrderCancelNoticeMessage");
            result.setStatus(true);
        } catch (Throwable e) {
            logger.error("AsyncMessageManager[sendOrderCancelNoticeMessage] failed! order:{}, cause by:{}",
                    order, e);
        }
        return result;
    }

    /**
     * 订单完成消息
     * @date 2022/7/5 14:41
     * @param order
     * @param orderCompleteTag
     * @return com.zm.util.dto.Result<java.lang.Integer>
     **/
    public Result<Integer> sendOrderCompleteNoticeMessage(Order order, String orderCompleteTag) {
        Result<Integer> result = new Result<>();
        try {
            OrderNoticeMessage body = mapper.map(order, OrderNoticeMessage.class);
            Message message = new Message();
            message.setTopic(commonTopic);
            message.setTag(orderCompleteTag);
            message.setBody(JSON.toJSONString(body).getBytes());
            sendMessageAndTime(message, "sendOrderCompleteNoticeMessage");
            result.setStatus(true);
        } catch (Throwable e) {
            logger.error("AsyncMessageManager[sendOrderCompleteNoticeMessage] failed! order:{}, cause by:{}",
                    order, e);
        }
        return result;
    }

    private void sendMessageAndTime(Message message, String methodName) {
        long time1 = System.currentTimeMillis();
        SendResult sendResult = producer.send(message);
        long time2 = System.currentTimeMillis();
        long delay = time2 - time1;
        if (delay > THRESHOLD) {
            logger.info("[delayOns]methodName:{} send ons message delay, msgId:{}, delayTime:{}",
                    methodName, sendResult.getMessageId(), delay);
        }
    }

    /**
     * 发送消息
     *
     * @param body message
     * @return
     * Date: 2021/6/22 5:02 下午
     */
    public Result<Integer> sendMallMessage(MallTransferMessage body) {
        Result<Integer> result = new Result();
        try {
            Message message = new Message();
            message.setTopic("mall-apiserver-common");
            message.setTag("migration-order-tag");
            message.setBody(JSON.toJSONString(body).getBytes());
            message.setMsgID(body.getOrderNo());
            message.setShardingKey(body.getOrderNo());
            // 发消息
            SendResult send = producer.send(message);
            logger.info("AsyncMessageManager[sendMallMessage] body:{}, SendResult:{}", body, send);
            result.setStatus(true);
        } catch (Exception e) {
            logger.error("AsyncMessageManager[sendMallMessage]failed! request:{}, cause by:{}", body, e);
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    /**
     * 发送评论同步ES消息
     *
     * @param commentToEsDTOList commentToEsDTOList
     * @return
     * Date: 2021/6/22 5:02 下午
     */
    public Result<Integer> sendCommentSyncEsMessage(List<CommentDocumentMessage> commentToEsDTOList) {
        Result<Integer> result = new Result();
        try {
            Message message = new Message();
            message.setTopic("order-center-item-comment-sync-es");
            message.setTag("order_item_comment_es");
            message.setBody(JSON.toJSONString(commentToEsDTOList).getBytes());
            // 发消息
            sendMessageAndTime(message, "sendCommentSyncEsMessage");
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("AsyncMessageManager[sendCommentSyncEsMessage]failed, cause by:{}",
                    Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }
}
