package cn.fateverse.query.portal.service;

import cn.fateverse.query.dubbo.DubboDispatchServletPublish;
import cn.fateverse.query.portal.config.DubboServiceBean;
import cn.fateverse.query.portal.event.DispatchSyncEvent;
import com.alibaba.cloud.nacos.NacosDiscoveryProperties;
import com.alibaba.cloud.nacos.NacosServiceManager;
import com.alibaba.cloud.nacos.registry.NacosRegistration;
import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.rpc.cluster.specifyaddress.Address;
import org.apache.dubbo.rpc.cluster.specifyaddress.UserSpecifiedAddressUtil;
import org.springframework.context.event.EventListener;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author Clay
 * @date 2024/4/19 17:43
 */
@Slf4j
@Component
public class DispatchSyncService {


    private final String serviceName;

    private final String host;

    private final NamingService namingService;

    private final DubboServiceBean serviceBean;

    @DubboReference
    private DubboDispatchServletPublish dubboDispatchServletPublish;


    private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(64);


    public DispatchSyncService(NacosDiscoveryProperties nacosDiscoveryProperties, Environment environment, NacosRegistration nacosRegistration, ThreadPoolTaskExecutor taskExecuteExecutor, DubboServiceBean serviceBean) {
        //获取当前主机地址
        this.host = nacosRegistration.getHost();
        //获取到端口信息的bean
        this.serviceBean = serviceBean;
        NacosServiceManager nacosServiceManager = new NacosServiceManager();
        Properties nacosProperties = nacosDiscoveryProperties.getNacosProperties();
        //获取dubbo的命名空间
        String namespace = environment.getProperty("dubbo.registry.parameters.namespace");
        //获取到当前服务名称
        this.serviceName = environment.getProperty("dubbo.application.name");
        nacosProperties.setProperty(PropertyKeyConst.NAMESPACE, namespace);
        //获取到nacos的管理服务
        this.namingService = nacosServiceManager.getNamingService(nacosProperties);
        //启动单线程,监听阻塞队列
        new Thread(() -> {
            while (true) {
                try {
                    Runnable take = queue.take();
                    //执行同步任务
                    taskExecuteExecutor.execute(take);
                } catch (InterruptedException e) {
                    log.info("", e);
                }
            }
        }).start();
    }

    @EventListener(DispatchSyncEvent.class)
    public void syncDispatch(DispatchSyncEvent event) {
        syncDispatch(event.getPath(), event.getRequestMethod(), event.getPublish());
    }


    private void syncDispatch(String path, String requestMethod, Boolean publish) {
        try {
            //获取nacos中的服务信息
            List<Instance> allInstances = namingService.getAllInstances(serviceName);
            for (Instance instance : allInstances) {
                //判断当前服务是否为本服务
                if (!(instance.getIp().equals(host) && instance.getPort() == serviceBean.getPort())) {
                    //不是本服务则需要发起同步任务
                    Task task = new Task(instance.getIp(), instance.getPort(), path, publish, requestMethod) {
                        @Override
                        public void run() {
                            try {
                                //获取云端服务信息
                                List<Instance> allInstances = namingService.getAllInstances(serviceName);
                                //如果没有则不进行同步
                                if (ObjectUtils.isEmpty(allInstances)) {
                                    return;
                                }
                                //检查云端实例中是否还存在当前服务,不存在当前任务结束
                                Optional<Instance> optional = allInstances.stream().filter(instance -> this.ip.equals(instance.getIp()) && this.port == instance.getPort()).findAny();
                                if (optional.isEmpty()) {
                                    return;
                                }
                                Instance instance = optional.get();
                                //指定需要同步服务的ip + port
                                UserSpecifiedAddressUtil.setAddress(new Address(instance.getIp(), instance.getPort(), true));
                                //记录同步状态
                                Boolean state = null;
                                //是否为发布任务
                                if (publish) {
                                    log.info("dubboDispatchServletPublish.publish({}, {})", path, requestMethod);
                                    state = dubboDispatchServletPublish.publish(path, requestMethod);
                                } else {
                                    log.info("dubboDispatchServletPublish.unpublish({}, {})", path, requestMethod);
                                    state = dubboDispatchServletPublish.unpublish(path, requestMethod);
                                }
                                //如果为false,则将当前任务重新放入到队列中,等待下一次执行
                                if (state == null || !state) {
                                    queue.add(this);
                                }
                            } catch (NacosException e) {
                                log.error("NacosException: {}", e.getMessage());
                                //如果发生异常也是
                                //todo 后续可以添加当前任务报错次数的处理
                                queue.add(this);
                            } finally {
                                //将指定的地址恢复
                                UserSpecifiedAddressUtil.setAddress(null);
                            }
                        }
                    };
                    //像阻塞队列中添加任务
                    queue.add(task);
                }
            }
        } catch (NacosException e) {
            log.error("NacosException: {}", e.getMessage());
            throw new RuntimeException(e);
        }
    }


    private static abstract class Task implements Runnable {
        /**
         * 服务ip
         */
        protected final String ip;
        /**
         * 服务端口
         */
        protected final int port;
        /**
         * 路径
         */
        protected final String path;
        /**
         * 是否为发布
         */
        protected final Boolean publish;
        /**
         * 方法类型
         */
        protected final String requestMethod;


        private Task(String ip, int port, String path, Boolean publish, String requestMethod) {
            this.ip = ip;
            this.port = port;
            this.path = path;
            this.publish = publish;
            this.requestMethod = requestMethod;
        }


    }


}
