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

import cn.axzo.workflow.starter.WorkflowEngineStarterProperties;
import cn.axzo.workflow.starter.handler.monitor.BroadcastDLQReporter;
import cn.axzo.workflow.starter.handler.monitor.RpcDLQReporter;
import cn.axzo.workflow.starter.mq.monitor.AlertBroadcastDLQReporter;
import cn.axzo.workflow.starter.mq.monitor.AlertRcpDLQReporter;
import cn.axzo.workflow.starter.mq.monitor.WorkflowEngineStarterDefaultMQMonitor;
import cn.azxo.framework.common.model.CommonResponse;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import static cn.axzo.workflow.common.constant.StarterConstants.MQ_GID_NAME_SEGMENT;
import static cn.axzo.workflow.starter.StarterRPCInvokeMQConfiguration.DEFAULT_EVENT;
import static cn.axzo.workflow.starter.mq.monitor.WorkflowEngineStarterDefaultMQMonitor.DLQ_PREFIX;

/**
 * Workflow Engine Starter MessageQueue Monitor Controller
 *
 * @author wangli
 * @since 2024/6/6 10:04
 */
@RestController
@RequestMapping("/web/we/s")
public class WorkflowEngineStarterMQMonitorController {
    private static final Logger log = LoggerFactory.getLogger(WorkflowEngineStarterMQMonitorController.class);
    @Resource
    private WorkflowEngineStarterProperties starterProperties;
    @Resource
    private ObjectProvider<DefaultMQAdminExt> mqAdminExtObjectProvider;
    @Resource
    private ObjectProvider<WorkflowEngineStarterDefaultMQMonitor> monitorObjectProvider;
    @Resource
    private ObjectProvider<BroadcastDLQReporter> broadcastDLQProcessorObjectProvider;
    @Resource
    private ObjectProvider<RpcDLQReporter> rpcDLQProcessorObjectProvider;
    @Resource
    private Environment environment;
    @Value("${spring.application.name}")
    private String applicationName;
    @Value("${spring.profiles.active}")
    private String activeProfile;
    @Resource
    @Qualifier("serviceVersion")
    private String serviceVersion;
    public static String BROADCAST_CONSUMER_GROUP = "GID_%s_workflow_engine_%s_consumer";
    public static String RPC_RETRY_CONSUMER_GROUP = "GID_%s_workflow_engine_starter_%s_consumer";

    @GetMapping("/m")
    public CommonResponse<Map<String, Object>> monitor() {
        String topic = DEFAULT_EVENT + activeProfile;
        Map<String, Object> result = new HashMap<>();
        if (!starterProperties.getEnableDlqMonitor()) {
            result.put("Info", "未开启·死信队列·的监控，如需，请设置 workflow.engine.starter.enableDlqMonitor = true 后再重试！");
        } else {
            mqAdminExtObjectProvider.ifAvailable(defaultMQAdminExt -> {
                String segment = environment.getProperty(MQ_GID_NAME_SEGMENT);
                String broadcastConsumer = String.format(BROADCAST_CONSUMER_GROUP, applicationName, segment);
                String rpcConsumer = String.format(RPC_RETRY_CONSUMER_GROUP, applicationName, segment);
                try {
                    result.put("Engine Broadcast MQ", defaultMQAdminExt.examineConsumeStats(broadcastConsumer));
                } catch (Exception e) {
                    log.warn("monitor controller examineConsumeStats broadcast error: {}", e.getMessage(), e);
                }
                try {
                    result.put("Starter RPC MQ", defaultMQAdminExt.examineConsumeStats(rpcConsumer));
                } catch (Exception e) {
                    log.warn("monitor controller examineConsumeStats rpc error: {}", e.getMessage(), e);
                }
//                result.put("BrokerClusterInfo", defaultMQAdminExt.examineBrokerClusterInfo());
//                result.put("TopicClusterList", defaultMQAdminExt.getTopicClusterList(topic));
                try {
                    result.put("TopicRouteInfo", defaultMQAdminExt.examineTopicRouteInfo(topic));
                } catch (Exception e) {
                    log.warn("monitor controller examineTopicRouteInfo error: {}", e.getMessage(), e);
                }
//                result.put("TopicStats", defaultMQAdminExt.examineTopicStats(topic));
                try {
                    result.put("Broadcast-DLQ", defaultMQAdminExt.examineTopicStats(DLQ_PREFIX + broadcastConsumer));
                } catch (Exception e) {
                    log.warn("monitor controller examineTopicStats broadcast error: {}", e.getMessage(), e);
                }
                try {
                    result.put("RPC-DLQ", defaultMQAdminExt.examineTopicStats(DLQ_PREFIX + rpcConsumer));
                } catch (Exception e) {
                    log.warn("monitor controller examineTopicStats rpc error: {}", e.getMessage(), e);
                }

            });
        }
        return CommonResponse.success(result);
    }

    /**
     * @param status
     * @return
     */
    @GetMapping("/m/set")
    public CommonResponse<String> changeMonitorState(@RequestParam("status") Boolean status) {
        WorkflowEngineStarterDefaultMQMonitor monitor = monitorObjectProvider.getIfAvailable();
        if (Objects.isNull(monitor)) {
            return CommonResponse.success("未开启·死信队列·的监控，如需，请设置 workflow.engine.starter.enableDlqMonitor = true 后再重试！");
        }
        if (status) {
            if (!monitor.isRunning()) {
                monitor.start();
            }
            return CommonResponse.success("已临时开始 MQ 死信队列监控！" +
                    "在应用重启前，将持续监控。重启后，将根据环境变量来确定是否监控。");
        } else {
            if (monitor.isRunning()) {
                monitor.stop();
            }
            return CommonResponse.success("已临时关闭 MQ 死信队列监控！" +
                    "在应用重启前，将不再监控。重启后，将根据环境变量配置来确定是否监控。");
        }
    }

    @GetMapping("/m/a/set")
    public CommonResponse<String> changeDLQAlert(@RequestParam("status") Boolean status) {
        WorkflowEngineStarterDefaultMQMonitor monitor = monitorObjectProvider.getIfAvailable();
        if (Objects.isNull(monitor)) {
            return CommonResponse.success("未开启·死信队列·的监控，如需，请设置 workflow.engine.starter.enableDlqMonitor = true 后再重试！");
        }
        if (status) {
            monitor.setBroadcastDLQProcessor(new AlertBroadcastDLQReporter(environment));
            monitor.setRpcDLQProcessor(new AlertRcpDLQReporter(environment));
            return CommonResponse.success("开启 DLQ 钉钉通知");
        } else {
            monitor.setBroadcastDLQProcessor(broadcastDLQProcessorObjectProvider.getIfAvailable(() -> new BroadcastDLQReporter() {
            }));
            monitor.setRpcDLQProcessor(rpcDLQProcessorObjectProvider.getIfAvailable(() -> new RpcDLQReporter() {
            }));
            return CommonResponse.success("以关闭 DLQ 钉钉通知");
        }
    }

    @GetMapping("/v")
    public CommonResponse<String> consoleVersion(@RequestParam(value = "fix", required = false, defaultValue = "false") Boolean showLastFix) {
        String str = "当前 workflow-engine version: " + serviceVersion;
        if (showLastFix) {
            str += "fix: cn.axzo.workflow.starter.mq.retry.consumer.WorkflowEngineStarterRetryEventListener.multiMethodCache";
        }
        return CommonResponse.success(str);
    }
}
