package cn.axzo.workflow.starter.mq.monitor;

import cn.axzo.workflow.starter.WorkflowEngineStarterProperties;
import cn.axzo.workflow.starter.handler.monitor.BroadcastDLQReporter;
import cn.axzo.workflow.starter.handler.monitor.RpcDLQReporter;
import lombok.SneakyThrows;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;

import static cn.axzo.workflow.common.constant.StarterConstants.K8S_POD_NAME_SPACE;
import static cn.axzo.workflow.common.constant.StarterConstants.MQ_GID_NAME_SEGMENT;
import static cn.axzo.workflow.starter.mq.monitor.console.WorkflowEngineStarterMQMonitorController.BROADCAST_CONSUMER_GROUP;
import static cn.axzo.workflow.starter.mq.monitor.console.WorkflowEngineStarterMQMonitorController.RPC_RETRY_CONSUMER_GROUP;

/**
 * 用于监控 Starter 关注的 Topic 相关的 Producer 与 Consumer 信息
 *
 * @author wangli
 * @since 2024/6/6 10:04
 */
public class WorkflowEngineStarterDefaultMQMonitor implements SmartLifecycle {

    private static final Logger log = LoggerFactory.getLogger(WorkflowEngineStarterDefaultMQMonitor.class);
    public static final String DLQ_PREFIX = "%DLQ%";
    private static final String BROADCAST = "broadcast";
    private static final String RPC = "rpc";
    /**
     * 可能为 null
     */
    private final DefaultMQAdminExt defaultMQAdminExt;
    private final Environment environment;
    private final ThreadPoolTaskScheduler taskScheduler;
    private BroadcastDLQReporter broadcastDLQProcessor;
    private RpcDLQReporter rpcDLQProcessor;
    private final WorkflowEngineStarterProperties starterProperties;
    private final AtomicBoolean running = new AtomicBoolean(false);

    public WorkflowEngineStarterDefaultMQMonitor(ObjectProvider<DefaultMQAdminExt> mqAdminExtObjectProvider,
                                                 ObjectProvider<BroadcastDLQReporter> broadcastDLQProcessorObjectProvider,
                                                 ObjectProvider<RpcDLQReporter> rpcDLQProcessorObjectProvider,
                                                 WorkflowEngineStarterProperties workflowEngineStarterProperties,
                                                 Environment environment) {
        this.defaultMQAdminExt = mqAdminExtObjectProvider.getIfAvailable();
        this.environment = environment;
        if (workflowEngineStarterProperties.getAlert()) {
            this.broadcastDLQProcessor = new AlertBroadcastDLQReporter(environment);
            this.rpcDLQProcessor = new AlertRcpDLQReporter(environment);
        } else {
            this.broadcastDLQProcessor = broadcastDLQProcessorObjectProvider.getIfAvailable(() -> new BroadcastDLQReporter() {
            });
            this.rpcDLQProcessor = rpcDLQProcessorObjectProvider.getIfAvailable(() -> new RpcDLQReporter() {
            });
        }
        this.starterProperties = workflowEngineStarterProperties;
        this.taskScheduler = init();
    }

    private ThreadPoolTaskScheduler init() {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setBeanName("Workflow-Engine-Watch-MQ-Scheduler");
        taskScheduler.initialize();
        return taskScheduler;
    }

    @SneakyThrows
    public void mqWatch() {
        mqWatch(BROADCAST);
        mqWatch(RPC);
    }

    @SneakyThrows
    public void mqWatch(String type) {
        String segment = environment.getProperty(MQ_GID_NAME_SEGMENT);
        String applicationName = environment.getProperty("spring.application.name");
        // 周期性比对
        if (Objects.equals(type, BROADCAST)) {
            compareDLQ(String.format(BROADCAST_CONSUMER_GROUP, applicationName, segment), type);
        } else if (Objects.equals(type, RPC)) {
            compareDLQ(String.format(RPC_RETRY_CONSUMER_GROUP, applicationName, segment), type);
        }
    }

    private void compareDLQ(String consumerGroupName, String dqlType) {
        log.info("current broadcast consumer group is :{}", consumerGroupName);
        String myPodNamespace = environment.getProperty(K8S_POD_NAME_SPACE);
        if (!StringUtils.hasText(myPodNamespace)) {
            return;
        }
        try {
            TopicStatsTable table = defaultMQAdminExt.examineTopicStats(DLQ_PREFIX + consumerGroupName);
            if (Objects.isNull(table) || CollectionUtils.isEmpty(table.getOffsetTable())) {
                return;
            }
            table.getOffsetTable().forEach((k, v) -> {
                if (Objects.equals(DLQ_PREFIX + consumerGroupName, k.getTopic())) {
                    if (BROADCAST.equals(dqlType)) {
                        broadcastDLQProcessor.process(v, DLQ_PREFIX + consumerGroupName);
                    } else if (RPC.equals(dqlType)) {
                        rpcDLQProcessor.process(v, DLQ_PREFIX + consumerGroupName);
                    }
                }
            });
        } catch (Exception e) {
            log.warn("该异常信息是正常的，不影响服务启动！");
        }
    }

    public void setBroadcastDLQProcessor(BroadcastDLQReporter broadcastDLQProcessor) {
        this.broadcastDLQProcessor = broadcastDLQProcessor;
    }

    public void setRpcDLQProcessor(RpcDLQReporter rpcDLQProcessor) {
        this.rpcDLQProcessor = rpcDLQProcessor;
    }

    @Override
    public boolean isAutoStartup() {
        if (starterProperties.getEnableDlqMonitor() && starterProperties.getRpcMonitorStatus()) {
            taskScheduler.scheduleWithFixedDelay(() -> mqWatch(RPC), starterProperties.getDlqMonitorIntervalInMs());
        }
        if (starterProperties.getEnableDlqMonitor() && starterProperties.getBroadcast().getMonitorStatus()) {
            taskScheduler.scheduleWithFixedDelay(() -> mqWatch(BROADCAST), starterProperties.getDlqMonitorIntervalInMs());
        }
        return false;
    }

    @Override
    public void start() {
        if (running.compareAndSet(false, true)) {
            log.info("starting workflow engine mq monitor");
            taskScheduler.scheduleWithFixedDelay(this::mqWatch, 30000);
        }
    }

    @Override
    public void stop() {
        if (running.compareAndSet(true, false)) {
            log.info("destroying workflow engine mq monitor");
            taskScheduler.shutdown();
        }
    }

    @Override
    public boolean isRunning() {
        return this.running.get();
    }
}
