package cn.axzo.workflow.starter;

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.workflow.starter.common.condition.NonContainerEnvironmentCondition;
import cn.axzo.workflow.starter.common.exception.WorkflowEngineStarterException;
import cn.axzo.workflow.starter.handler.filter.global.BroadcastMessageQueueFilter;
import cn.axzo.workflow.starter.mq.broadcast.consumer.InnerWorkflowListener;
import cn.axzo.workflow.starter.mq.broadcast.consumer.WorkflowEngineBroadcastEventListener;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterDefinitionKey;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterExtension;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterMQOwnerShip;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerMessageQueueHandleBeforeFilter;
import com.google.common.collect.ImmutableList;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.function.Consumer;

import static cn.axzo.workflow.starter.StarterRPCInvokeMQConfiguration.DEFAULT_EVENT;

/**
 * 配置监听流程引擎服务广播消息的 RocketMQ 相关配置
 *
 * @author wangli
 * @since 2024/6/5 17:39
 */
@Configuration(proxyBeanMethods = false)
@ConditionalOnProperty(name = "rocketmq.name-server")
public class StarterBroadcastMQConfiguration {
    private final Logger log = LoggerFactory.getLogger(StarterBroadcastMQConfiguration.class);
    public static final String BROADCAST_EVENT_HANDLER_REPOSITORY_BEAN_NAME = "broadcastEventHandlerRepository";
    public static final String BROADCAST_EVENT_CONSUMER_BEAN_NAME = "broadcastEventConsumer";
    @Value("${spring.application.name}")
    private String applicationName;
    @Value("${spring.profiles.active:dev}")
    private String activeProfile;

    //=================================  Workflow Engine Broadcast MQ  =================================//
    @Bean(BROADCAST_EVENT_HANDLER_REPOSITORY_BEAN_NAME)
    @ConditionalOnMissingBean(name = BROADCAST_EVENT_HANDLER_REPOSITORY_BEAN_NAME)
    public EventHandlerRepository broadcastEventHandlerRepository() {
        return new EventHandlerRepository((ex, logText) -> {
            // 不管 cn.axzo.workflow.starter.BroadcastListenerProperties.failHandleType 设置是哪种策略，这里最终都会打印 WARN 级别日志，框架底层会对所有异常的消息打印 ERROR 日志
            // 如果不想打印需要扩展内部逻辑，请主动注册自己扩展的 Bean，且名称为 StarterBroadcastMQConfiguration.BROADCAST_EVENT_HANDLER_REPOSITORY_BEAN_NAME
            log.warn("Workflow Engine Starter MQ, an exception occurred during processing {}", logText, ex);
            // 让其进入死信队列
            throw new WorkflowEngineStarterException(ex);
        });
    }

    @Bean(BROADCAST_EVENT_CONSUMER_BEAN_NAME)
    @ConditionalOnMissingBean(name = BROADCAST_EVENT_CONSUMER_BEAN_NAME)
    public EventConsumer broadcastEventConsumer(@Qualifier(BROADCAST_EVENT_HANDLER_REPOSITORY_BEAN_NAME) EventHandlerRepository eventHandlerRepository) {
        Consumer<EventConsumer.EventWrapper> callback = eventWrapper -> {
            if (eventWrapper.isHandled()) {
                // 只收集被App真正消费的消息.
                if (log.isDebugEnabled()) {
                    Event event = eventWrapper.getEvent();
                    log.debug("WorkflowEngine Broadcast MQ, handled event:  {}", event.toPrettyJsonString());
                }
            }
        };
        return new DefaultEventConsumer(applicationName, eventHandlerRepository, callback);
    }

    @Component
    @ConditionalOnProperty(name = "rocketmq.name-server")
    @Conditional(NonContainerEnvironmentCondition.class)
    @RocketMQMessageListener(topic = DEFAULT_EVENT + "${spring.profiles.active}",
        consumerGroup = "GID_${spring.application.name}_workflow_engine_${GID_SEGMENT}_consumer",
        consumeMode = ConsumeMode.ORDERLY,
        maxReconsumeTimes = 0,
        nameServer = "${rocketmq.name-server}"
    )
    public static class WorkflowEngineBroadcastConsumer extends BaseListener implements RocketMQListener<MessageExt>, InitializingBean {
        private final Logger log = LoggerFactory.getLogger(WorkflowEngineBroadcastConsumer.class);

        @Resource(name = BROADCAST_EVENT_CONSUMER_BEAN_NAME)
        private EventConsumer broadcastEventConsumer;
        @Value("${spring.application.name}")
        private String applicationName;
        @Resource
        private WorkflowEngineStarterProperties starterProperties;
        @Resource
        private ObjectProvider<List<BroadcastMessageQueueFilter>> businessFilterProvider;
        private List<InnerMessageQueueHandleBeforeFilter> filters;

        @Override
        public void onMessage(MessageExt message) {
            // 处理 properties 配置进行事件过滤
            for (InnerMessageQueueHandleBeforeFilter filter : filters) {
                if (filter.doFilter(message)) {
                    if (log.isDebugEnabled()) {
                        log.debug("【{}】message has been filtered, messageId: {}", this.getClass().getSimpleName(), message.getMsgId());
                    }
                    return;
                }
            }
            super.onEvent(message, broadcastEventConsumer);
        }

        @Override
        public void afterPropertiesSet() {
            this.filters = ImmutableList.of(
                new InnerFilterMQOwnerShip(starterProperties, applicationName),
                new InnerFilterDefinitionKey(starterProperties),
                new InnerFilterExtension(businessFilterProvider));
        }
    }

    @Bean("workflowEngineBroadcastEventListener")
    public WorkflowEngineBroadcastEventListener workflowEngineBroadcastEventListener(@Qualifier(BROADCAST_EVENT_CONSUMER_BEAN_NAME) EventConsumer broadcastEventConsumer,
                                                                                     WorkflowEngineStarterProperties workflowEngineStarterProperties,
                                                                                     List<InnerWorkflowListener> listenerProvider) {
        return new WorkflowEngineBroadcastEventListener(broadcastEventConsumer, workflowEngineStarterProperties, listenerProvider);
    }
}
