package org.matrix.socket.queue;

import cn.hutool.log.LogFactory;
import cn.hutool.log.level.Level;
import com.alibaba.fastjson.JSONObject;
import org.matrix.actuators.sql.SqlExpActuator;
import org.matrix.actuators.util.ThreadUtil;
import org.matrix.database.entity.ExecutionRecord;
import org.matrix.database.service.IExecutionRecordService;
import org.matrix.enums.ModuleType;
import org.matrix.exception.ExecutionRunningException;
import org.matrix.exception.GlobalException;
import org.matrix.socket.ExecutionSocketHandler;
import org.matrix.socket.vo.TestExecuteLog;
import org.matrix.socket.enums.TestExecuteType;
import org.matrix.socket.pool.TestCaseExecuteSocketPool;
import org.matrix.util.SpringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


/**
 * 运行态日志消息队列
 *
 * @author huangxiahao
 */
public class LogQueueRuntime {

    private static final ConcurrentHashMap<Long, TestExecuteLog> LOG_MAP = new ConcurrentHashMap<>();

    private static IExecutionRecordService EXECUTION_RECORD_SERVICE;

    public static IExecutionRecordService getExecutionRecordService() {
        if (EXECUTION_RECORD_SERVICE == null) {
            EXECUTION_RECORD_SERVICE = SpringUtils.getBean("executionRecordImpl");
        }
        return EXECUTION_RECORD_SERVICE;
    }

    public static ExecutionRecord addExecutionRecord(TestExecuteLog testExecuteLog, String log) {
        return getExecutionRecordService().addExecutionRecord(testExecuteLog, log);
    }


    /**
     * 如果返回false,表示该用例已经在线程中执行了
     */
    public static void put(Long threadId, TestExecuteLog testExecuteLog) {
        LOG_MAP.put(threadId, testExecuteLog);
    }

    public static void setTestData(Long dataId) {
        Long currentThreadId = ThreadUtil.currentThreadId();
        LOG_MAP.get(currentThreadId).setTestDataId(dataId);
    }

    public static void clearTestData() {
        Long currentThreadId = ThreadUtil.currentThreadId();
        LOG_MAP.get(currentThreadId).setTestDataId(-1L);
    }

    public static void addNewLog(Class<?> clazz, ModuleType type, String logString) {
        Logger classLog = LoggerFactory.getLogger(clazz);
        LogMessage logMsg = new LogMessage(LocalDateTime.now(), type, logString);
        classLog.info(logMsg.toLogString());
        Long currentThreadId = ThreadUtil.currentThreadId();
        TestExecuteLog testExecuteLog = LOG_MAP.get(currentThreadId);
        if (testExecuteLog != null) {
            testExecuteLog.addLog(logMsg);
            String messageToDb = JSONObject.toJSONString(logMsg);
            ExecutionRecord executionRecord = addExecutionRecord(testExecuteLog, messageToDb);
            sendMessage(currentThreadId, testExecuteLog.getUniqueKey(), executionRecord.getId(), messageToDb);
        } else {
            throw new GlobalException("该线程中的TestCaseLog对象未初始化，请先调用put方法初始化对象");
        }
    }

    public static void remove(Long threadId) {
        LOG_MAP.remove(threadId);
    }

    public static void sendMessage(Long threadId, String uniqueKey, Long logId, String log) {
        TestExecuteLog testExecuteLog = LOG_MAP.get(threadId);
        // 测试任务ID.测试用例ID.测试数据ID
        String message = JSONObject.toJSONString(log);
        String logMessage =
                String.format(
                        "%s.%s.%s.%s.%s"
                        , testExecuteLog.getTestJobId()
                        , testExecuteLog.getTestCaseId()
                        , testExecuteLog.getTestDataId()
                        , logId
                        , message
                );
        //如果uniqueKey中存在监听的socket则向该socket发送消息
        List<WebSocketSession> webSocketSessions = ExecutionSocketHandler.EXECUTE_MONITOR_SOCKET_POOL.get(uniqueKey);
        if (webSocketSessions != null) {
            for (WebSocketSession webSocketSession : webSocketSessions) {
                socketSendMessage(webSocketSession, logMessage);
            }
        }
        //如果线程中存在socket则向该socket发送消息
        WebSocketSession webSocketSession = TestCaseExecuteSocketPool.get(threadId);
        if (webSocketSession != null) {
            socketSendMessage(webSocketSession, logMessage);
        }

    }

    private static void socketSendMessage(WebSocketSession socketSession, String message) {
        try {
            if (socketSession != null && socketSession.isOpen()) {
                socketSession.sendMessage(new TextMessage(
                        message
                ));
            }
        } catch (IOException e) {
            LogFactory.get().log(Level.INFO, "发送socket消息失败,socket已经断开连接");
        }
    }

    public static void initTestCaseLog(Long jobId, Long userId, Long caseId, TestExecuteType type, String uniqueKey) {
        TestExecuteLog testExecuteLog = new TestExecuteLog();
        testExecuteLog.setTestJobId(jobId);
        testExecuteLog.setTestCaseId(caseId);
        testExecuteLog.setUserId(userId);
        testExecuteLog.setType(type);
        testExecuteLog.setUniqueKey(uniqueKey);
        if (checkIsInRun(testExecuteLog)) {
            throw new ExecutionRunningException("当前 任务/用例 正在执行中");
        }
        LogQueueRuntime.put(ThreadUtil.currentThreadId(), testExecuteLog);
    }

    public static Boolean checkIsInRun(TestExecuteLog testExecuteLog) {
        for (TestExecuteLog value : LOG_MAP.values()) {
            if (value.equals(testExecuteLog)) {
                return true;
            }
        }
        return false;
    }

    public static TestExecuteLog getCurrentTestExecute(Long threadId) {
        return LOG_MAP.get(threadId);
    }

}
