package cn.axzo.workflow.core.conf;

import cn.axzo.framework.rocketmq.DefaultEventConsumer;
import cn.axzo.framework.rocketmq.Event;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.framework.rocketmq.EventHandlerRepository;
import cn.axzo.framework.rocketmq.EventProducer;
import cn.axzo.framework.rocketmq.RocketMQEventProducer;
import cn.axzo.framework.rocketmq.utils.TraceUtils;
import cn.axzo.workflow.core.common.enums.MqLogEventType;
import cn.axzo.workflow.core.common.event.MqLogEvent;
import cn.axzo.workflow.core.mq.CustomRocketMQEventProducer;
import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import static cn.axzo.framework.rocketmq.RocketMQEventProducer.MQ_MESSAGE_ID;

/**
 * RocketMQ 全局配置
 *
 * @author wangli
 * @since 2023/9/4 10:03
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RocketMQTemplate.class)
public class RocketMqEventConfiguration {

    @Value("${spring.application.name}")
    private String applicationName;
    @Value("${spring.profiles.active:dev}")
    private String activeProfile;
    @Resource
    private ApplicationEventPublisher applicationEventPublisher;

    private static final String DEFAULT_MODULE = "workflowEngine";
    private static final String DEFAULT_EVENT = "topic_workflow_engine_";

    @Bean
    EventProducer eventProducer(RocketMQTemplate rocketMQTemplate) {
        return new CustomRocketMQEventProducer(rocketMQTemplate,
                DEFAULT_MODULE,
                applicationName,
                EventProducer.Context.<RocketMQEventProducer.RocketMQMessageMeta>builder()
                        .meta(RocketMQEventProducer.RocketMQMessageMeta.builder()
                                .topic(DEFAULT_EVENT + activeProfile)
                                .build())
                        .headers(new HashMap<>())
                        .syncSending(Boolean.TRUE)
                        .exceptionHandler(context -> {
                            log.error("MQ, send event error: {}, event: {}",
                                    context.getThrowable().getCause().getMessage(),
                                    context.getEvent().toPrettyJsonString(),
                                    context.getThrowable());
                        })
                        .build(),
                getSendBeforeCallback(),
                getSendAfterCallback(),
                getTransactionRollbackHandler()
        );
    }

    /**
     * 真实执行 MQ 发送前的回调，
     * <p>
     * 将整个待发送的事件内容通过 spring 的事件分发器发送出去， 现目前主要是记录 MQ 的发送记录
     *
     * @return
     */
    private BiConsumer<Event, EventProducer.Context<RocketMQEventProducer.RocketMQMessageMeta>> getSendBeforeCallback() {
        return (event, context) -> {
            event.setEventId(IdUtil.simpleUUID());
            MqLogEvent mqLogEvent = new MqLogEvent(event.getEventId(), null,
                    event.getEventName(), event.getShardingKey(),
                    event.toPrettyJsonString(), TraceUtils.getOrCreateTraceId(),
                    MqLogEventType.INSERT);
            log.info("mq_send_Before: {}, uniqueId: {}", event.getShardingKey(), event.getEventId());
            applicationEventPublisher.publishEvent(mqLogEvent);
        };
    }

    /**
     * 真实执行 MQ 发送后的回调
     * <p>
     * 将发送前的 MQ 发送记录更新 MQ 组件自己的 MessageId 字段。
     *
     * @return
     */
    private BiConsumer<Event, EventProducer.Context<RocketMQEventProducer.RocketMQMessageMeta>> getSendAfterCallback() {
        return (event, context) -> {
            String messageId = context.getHeaders().get(MQ_MESSAGE_ID);
            MqLogEvent mqLogEvent = new MqLogEvent(event.getEventId(), messageId,
                    event.getEventName(), event.getShardingKey(),
                    event.toPrettyJsonString(), TraceUtils.getOrCreateTraceId(),
                    MqLogEventType.UPDATE);
            log.info("mq_send_after: {}, uniqueId: {}, messageId: {}", event.getShardingKey(), event.getEventId(), messageId);
            applicationEventPublisher.publishEvent(mqLogEvent);
        };
    }

    /**
     * 如果 MQ 注册的事务回滚后的回调
     * <p>
     * 将 MQ 发送记录更新为删除状态，意为这类数据可以不关注，可以物理删除，但该功能还是用逻辑删除。
     *
     * @return
     */
    private BiConsumer<Event, EventProducer.Context<RocketMQEventProducer.RocketMQMessageMeta>> getTransactionRollbackHandler() {
        return (event, context) -> {
            MqLogEvent mqLog = new MqLogEvent(event.getEventId(), null, event.getEventName(),
                    event.getShardingKey(), event.toPrettyJsonString(), TraceUtils.getOrCreateTraceId(),
                    MqLogEventType.DELETE);
            log.info("mq_transaction_rollback: {}, uniqueId: {}", event.getShardingKey(), event.getEventId());
            applicationEventPublisher.publishEvent(mqLog);
        };
    }

    @Bean
    EventConsumer eventConsumer(EventHandlerRepository eventHandlerRepository) {
        Consumer<EventConsumer.EventWrapper> callback = (eventWrapper) -> {
            if (eventWrapper.isHandled()) {
                // 只收集被App真正消费的消息.
                //String topic = (String) eventWrapper.getExt().get(EVENT_TOPIC_KEY);

            }
        };
        return new DefaultEventConsumer(applicationName, eventHandlerRepository, callback);
    }

    /**
     * 这里注释是为了处理嵌入式和微服务兼容情况,由引入方自主配置 topic 监听
     */
    //    @Configuration(proxyBeanMethods = false)
    //    @ConditionalOnClass(RocketMQTemplate.class)
    //    @RocketMQMessageListener(topic = DEFAULT_EVENT + "${spring.profiles.active}",
    //            consumerGroup = "${spring.application.name}-${spring.profiles.active}"
    //    )
    //    public static class DefaultListener extends BaseListener implements RocketMQListener<MessageExt> {
    //
    //        @Autowired
    //        private EventConsumer eventConsumer;
    //
    //        @Override
    //        public void onMessage(MessageExt message) {
    //            super.onEvent(message, eventConsumer);
    //        }
    //    }
    @Bean
    EventHandlerRepository eventHandlerRepository() {
        return new EventHandlerRepository((ex, logText) -> {
            log.warn("MQ, handle warning {} , Exception: {}", logText, JSONUtil.toJsonStr(ex));
            if (Objects.nonNull(ex)) {
                throw new RuntimeException(ex);
            }
        });
    }
}
