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.framework.rocketmq.EventProducer;
import cn.axzo.workflow.starter.common.condition.NonContainerEnvironmentCondition;
import cn.axzo.workflow.starter.handler.filter.global.BroadcastMessageQueueFilter;
import cn.axzo.workflow.starter.mq.broadcast.consumer.WorkflowEngineBroadcastEventListener;
import cn.axzo.workflow.starter.mq.broadcast.consumer.WorkflowListener;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterDefinitionKeyInner;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterExtension;
import cn.axzo.workflow.starter.mq.broadcast.filter.InnerFilterMQOwnerShipInner;
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.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.Objects;
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)
public class StarterBroadcastMQConfiguration {
    private final Logger log = LoggerFactory.getLogger(StarterBroadcastMQConfiguration.class);

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

    //=================================  Workflow Engine Broadcast MQ  =================================//
    @Bean
    @ConditionalOnMissingBean(EventHandlerRepository.class)
    public EventHandlerRepository eventHandlerRepository() {
        return new EventHandlerRepository((ex, logText) -> {
            log.warn("MQ, handle warning {}", logText, ex);
            if (Objects.nonNull(ex)) {
                throw new RuntimeException(ex);
            }
        });
    }

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

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

        @Resource(name = "eventConsumer")
        private EventConsumer eventConsumer;
        @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) {
            for (InnerMessageQueueHandleBeforeFilter filter : filters) {
                if (filter.doFilter(message)) {
                    log.info("message has been filtered");
                    return;
                }
            }
            super.onEvent(message, eventConsumer);
        }

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

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