package cn.axzo.workflow.starter;

import cn.axzo.framework.domain.data.IdHelper;
import cn.axzo.framework.rocketmq.BaseListener;
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.workflow.starter.api.WorkflowCoreService;
import cn.axzo.workflow.starter.common.condition.NonContainerEnvironmentCondition;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterMQOwnerShipInner;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerMessageQueueHandleBeforeFilter;
import cn.axzo.workflow.starter.mq.retry.consumer.WorkflowEngineStarterRetryEventListener;
import cn.axzo.workflow.starter.mq.retry.producer.RpcInvokeEventProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

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;

/**
 * 配置 RPC 动作的 RocketMQ 消息的发送方和消息方等配置信息
 *
 * @author wangli
 * @since 2024/5/30 14:05
 */
@Configuration(proxyBeanMethods = false)
public class StarterRPCInvokeMQConfiguration {
    private final Logger log = LoggerFactory.getLogger(StarterRPCInvokeMQConfiguration.class);

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

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

    //=======================================  RPC Invoke MQ  ========================================//

    /**
     * 客户端 RPC Retry 事件生产者
     *
     * @param rocketMQTemplate
     * @return
     */
    @Bean
    public EventProducer workflowEngineStarterEventProducer(RocketMQTemplate rocketMQTemplate) {
        return new RpcInvokeEventProducer(rocketMQTemplate,
                DEFAULT_MODULE,
                applicationName + "_WE_Starter",
                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(IdHelper.get32UUID());
            log.info("mq_send_Before: {}, uniqueId: {}", event.getShardingKey(), event.getEventId());
        };
    }

    /**
     * 真实执行 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);
            log.info("mq_send_after: {}, uniqueId: {}, messageId: {}", event.getShardingKey(), event.getEventId(), messageId);
        };
    }

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

    @Bean
    @ConditionalOnMissingBean(name = "workflowEngineStarterEventHandlerRepository")
    public EventHandlerRepository workflowEngineStarterEventHandlerRepository() {
        return new EventHandlerRepository((ex, logText) -> {
            log.warn("MQ, handle warning {}", logText, ex);
            if (Objects.nonNull(ex)) {
                throw new RuntimeException(ex);
            }
        });
    }

    @Bean
    @ConditionalOnMissingBean(name = "workflowEngineStarterEventConsumer")
    public EventConsumer workflowEngineStarterEventConsumer(@Qualifier("workflowEngineStarterEventHandlerRepository") EventHandlerRepository workflowEngineStarterEventHandlerRepository) {
        Consumer<EventConsumer.EventWrapper> callback = eventWrapper -> {
            if (eventWrapper.isHandled()) {
                // 只收集被App真正消费的消息.
                Event event = eventWrapper.getEvent();
                log.info("WorkflowEngineStarter RPC MQ, handled event:  {}", event.toPrettyJsonString());
            }
        };
        return new DefaultEventConsumer(applicationName + "Starter", workflowEngineStarterEventHandlerRepository, callback);
    }

    @Component
    @Conditional(NonContainerEnvironmentCondition.class)
    @RocketMQMessageListener(topic = DEFAULT_EVENT + "${spring.profiles.active}",
            consumerGroup = "GID_${spring.application.name}_workflow_engine_starter_${GID_SEGMENT}_consumer",
            consumeMode = ConsumeMode.CONCURRENTLY,
            maxReconsumeTimes = 3, // TODO 发布时需调整为 7, 总共耗时在 15min 内
            replyTimeout = 10000,
            nameServer = "${rocketmq.name-server}"
    )
    public static class WorkflowEngineStarterRetryConsumer extends BaseListener implements RocketMQListener<MessageExt>, InitializingBean {
        private final Logger log = LoggerFactory.getLogger(WorkflowEngineStarterRetryConsumer.class);

        @Resource(name = "workflowEngineStarterEventConsumer")
        private EventConsumer workflowEngineStarterEventConsumer;
        @Value("${spring.application.name}")
        private String applicationName;
        @Resource
        private WorkflowEngineStarterProperties starterProperties;
        private InnerMessageQueueHandleBeforeFilter filter;

        @Override
        public void onMessage(MessageExt message) {
            if (filter.doFilter(message)) {
                log.info("message has been filtered");
                return;
            }
            super.onEvent(message, workflowEngineStarterEventConsumer);
        }

        @Override
        public void afterPropertiesSet() {
            this.filter = new InnerFilterMQOwnerShipInner(starterProperties, applicationName);
        }
    }

    @Bean
    public WorkflowEngineStarterRetryEventListener workflowEngineClientRetryEventListener(@Qualifier("workflowEngineStarterEventConsumer") EventConsumer workflowEngineStarterEventConsumer,
                                                                                          Environment environment,
                                                                                          WorkflowCoreService workflowCoreService) {
        return new WorkflowEngineStarterRetryEventListener(workflowEngineStarterEventConsumer, environment, workflowCoreService);
    }

}
