import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.*;
import java.nio.channels.FileLock;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class MqttToZabbix {

    // ---------- Конфигурация ----------
    private static String mqttBroker;
    private static String mqttTopic;
    private static String mqttUser;
    private static String mqttPassword;
    private static String zabbixServer;
    private static int zabbixPort;
    private static int corePoolSize;
    private static int maxPoolSize;
    private static int queueCapacity;
    private static int maxRetries;
    private static long initialDelayMs;
    private static double backoffMultiplier;

    // Параметры мониторинга
    private static String metricsHost;
    private static int metricsInterval;

    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static ExecutorService executorService;
    private static ScheduledExecutorService metricsScheduler;
    private static List<MappingRule> mappingRules = new ArrayList<>();

    // MQTT клиент
    private static MqttClient mqttClient;

    // Счётчики для мониторинга
    private static final AtomicLong processedCounter = new AtomicLong(0);
    private static final AtomicLong errorCounter = new AtomicLong(0);
    private static final AtomicLong retryCounter = new AtomicLong(0);

    // ---------- Вспомогательный класс правила ----------
    static class MappingRule {
        Pattern topicPattern;
        String hostTemplate;
        String keyTemplate;
        String valueField;
        String subscribeTopic;

        MappingRule(String topicPattern, String hostTemplate, String keyTemplate,
                    String valueField, String subscribeTopic) {
            this.topicPattern = Pattern.compile(topicPattern);
            this.hostTemplate = hostTemplate;
            this.keyTemplate = keyTemplate;
            this.valueField = valueField;
            this.subscribeTopic = subscribeTopic;
        }

        Matcher matcher(String topic) {
            return topicPattern.matcher(topic);
        }
    }

    // ---------- Точка входа ----------
    public static void main(String[] args) {
        // Блокировка от двойного запуска
        try (FileOutputStream lockFile = new FileOutputStream("app.lock");
             FileLock lock = lockFile.getChannel().tryLock()) {
            if (lock == null) {
                System.err.println("Another instance is already running. Exiting.");
                System.exit(1);
            }
        } catch (IOException e) {
            System.err.println("Failed to acquire lock: " + e.getMessage());
            System.exit(1);
        }

        if (!loadConfig("config.properties")) {
            System.err.println("Failed to load configuration. Exiting.");
            System.exit(1);
        }
        if (!loadMapping("mapping.json")) {
            System.err.println("Failed to load mapping. Exiting.");
            System.exit(1);
        }

        // Валидация окружения
        validateEnvironment();

        // Создаём пул потоков
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(queueCapacity);
        executorService = new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                60L, TimeUnit.SECONDS,
                workQueue,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        // Запускаем планировщик метрик
        metricsScheduler = Executors.newSingleThreadScheduledExecutor();
        metricsScheduler.scheduleAtFixedRate(() -> {
            try {
                long processed = processedCounter.get();
                long errors = errorCounter.get();
                long retries = retryCounter.get();
                int queueSize = ((ThreadPoolExecutor) executorService).getQueue().size();

                sendMetric("mqtt.bridge.processed", processed);
                sendMetric("mqtt.bridge.errors", errors);
                sendMetric("mqtt.bridge.retries", retries);
                sendMetric("mqtt.bridge.queue_size", queueSize);

                System.out.println("Metrics sent: processed=" + processed +
                        ", errors=" + errors +
                        ", retries=" + retries +
                        ", queue=" + queueSize);
            } catch (Exception e) {
                System.err.println("Failed to send metrics: " + e.getMessage());
            }
        }, metricsInterval, metricsInterval, TimeUnit.SECONDS);

        // Подключаемся к MQTT
        connectMQTT();
    }

    // ---------- Валидация окружения ----------
    private static void validateEnvironment() {
        // Проверка zabbix_sender
        try {
            ProcessBuilder pb = new ProcessBuilder("zabbix_sender", "--help");
            pb.redirectErrorStream(true);
            Process p = pb.start();
            int exitCode = p.waitFor();
            if (exitCode != 0) {
                System.err.println("WARNING: zabbix_sender not found or not executable. Sending will fail.");
            } else {
                System.out.println("zabbix_sender: OK");
            }
        } catch (Exception e) {
            System.err.println("WARNING: Failed to check zabbix_sender: " + e.getMessage());
        }

        // Проверка обязательных параметров (дополнительно)
        if (zabbixServer == null || zabbixServer.trim().isEmpty()) {
            System.err.println("ERROR: zabbix.server is not set.");
            System.exit(1);
        }
        if (zabbixPort <= 0 || zabbixPort > 65535) {
            System.err.println("ERROR: Invalid zabbix.port.");
            System.exit(1);
        }
        if (mqttBroker == null || mqttBroker.trim().isEmpty()) {
            System.err.println("ERROR: mqtt.broker is not set.");
            System.exit(1);
        }

        // Проверка маппинга на дублирующиеся topicPattern (опционально)
        Set<String> patterns = new HashSet<>();
        for (MappingRule rule : mappingRules) {
            String patternStr = rule.topicPattern.pattern();
            if (!patterns.add(patternStr)) {
                System.err.println("WARNING: Duplicate topicPattern: " + patternStr);
            }
            if (rule.hostTemplate == null || rule.hostTemplate.trim().isEmpty()) {
                System.err.println("ERROR: host is empty for pattern: " + patternStr);
                System.exit(1);
            }
            if (rule.keyTemplate == null || rule.keyTemplate.trim().isEmpty()) {
                System.err.println("ERROR: key is empty for pattern: " + patternStr);
                System.exit(1);
            }
            if (rule.subscribeTopic != null && rule.subscribeTopic.trim().isEmpty()) {
                System.err.println("WARNING: subscribeTopic is empty for pattern: " + patternStr + ", will use default.");
            }
        }
        System.out.println("Environment validation completed.");
    }

    // ---------- Подключение к MQTT ----------
    private static void connectMQTT() {
        try {
            mqttClient = new MqttClient(mqttBroker, MqttClient.generateClientId(), new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            if (mqttUser != null && !mqttUser.trim().isEmpty()) {
                options.setUserName(mqttUser);
                options.setPassword(mqttPassword.toCharArray());
            }

            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.err.println("MQTT connection lost: " + cause.getMessage());
                    new Thread(MqttToZabbix::reconnect).start();
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    String payload = new String(message.getPayload());
                    System.out.println("Received from MQTT topic '" + topic + "': " + payload);

                    executorService.submit(() -> {
                        try {
                            MappingRule rule = findRule(topic);
                            if (rule != null) {
                                processWithRule(topic, payload, rule);
                            } else {
                                processLegacy(payload);
                            }
                            processedCounter.incrementAndGet();
                        } catch (Exception e) {
                            System.err.println("Error processing MQTT message: " + e.getMessage());
                            errorCounter.incrementAndGet();
                        }
                    });
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {}
            });

            mqttClient.connect(options);
            subscribeTopics();
            System.out.println("Connected to MQTT broker.");
            System.out.println("Zabbix target: " + zabbixServer + ":" + zabbixPort);

            // Shutdown hook
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                System.out.println("Shutting down...");
                executorService.shutdown();
                metricsScheduler.shutdown();
                try {
                    if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
                        executorService.shutdownNow();
                    }
                    if (!metricsScheduler.awaitTermination(10, TimeUnit.SECONDS)) {
                        metricsScheduler.shutdownNow();
                    }
                } catch (InterruptedException e) {
                    executorService.shutdownNow();
                    metricsScheduler.shutdownNow();
                }
                try {
                    if (mqttClient != null && mqttClient.isConnected()) {
                        mqttClient.disconnect();
                        mqttClient.close();
                    }
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }));

        } catch (MqttException e) {
            System.err.println("MQTT initialization error: " + e.getMessage());
            reconnect();
        }
    }

    // ---------- Переподключение к MQTT (экспоненциальная задержка) ----------
    private static void reconnect() {
        long delay = 1000;
        int maxDelay = 60000;
        while (!Thread.currentThread().isInterrupted()) {
            try {
                System.out.println("Attempting to reconnect to MQTT...");
                if (mqttClient != null && !mqttClient.isConnected()) {
                    MqttConnectOptions options = new MqttConnectOptions();
                    options.setCleanSession(true);
                    if (mqttUser != null && !mqttUser.trim().isEmpty()) {
                        options.setUserName(mqttUser);
                        options.setPassword(mqttPassword.toCharArray());
                    }
                    mqttClient.connect(options);
                    subscribeTopics();
                    System.out.println("Reconnected to MQTT successfully.");
                    return;
                }
            } catch (MqttException e) {
                System.err.println("Reconnect attempt failed: " + e.getMessage());
            }
            try {
                Thread.sleep(delay);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                return;
            }
            delay = Math.min(delay * 2, maxDelay);
        }
    }

    // ---------- Подписка на топики (из маппинга или глобальный) ----------
    private static void subscribeTopics() throws MqttException {
        Set<String> topics = new HashSet<>();
        for (MappingRule rule : mappingRules) {
            if (rule.subscribeTopic != null && !rule.subscribeTopic.trim().isEmpty()) {
                topics.add(rule.subscribeTopic.trim());
            }
        }
        if (topics.isEmpty()) {
            String defaultTopic = (mqttTopic != null && !mqttTopic.trim().isEmpty()) ? mqttTopic : "#";
            topics.add(defaultTopic);
        }
        for (String topic : topics) {
            mqttClient.subscribe(topic);
            System.out.println("Subscribed to: " + topic);
        }
    }

    // ---------- Загрузка конфигурации ----------
    private static boolean loadConfig(String fileName) {
        Properties props = new Properties();
        try (InputStream input = getConfigInputStream(fileName)) {
            if (input == null) {
                System.err.println("Configuration file not found: " + fileName);
                return false;
            }
            props.load(input);

            mqttBroker = props.getProperty("mqtt.broker");
            mqttTopic = props.getProperty("mqtt.topic");
            mqttUser = props.getProperty("mqtt.user", "");
            mqttPassword = props.getProperty("mqtt.password", "");
            zabbixServer = props.getProperty("zabbix.server");
            String portStr = props.getProperty("zabbix.port", "10051");
            try {
                zabbixPort = Integer.parseInt(portStr);
            } catch (NumberFormatException e) {
                System.err.println("Invalid zabbix.port: " + portStr);
                return false;
            }

            corePoolSize = Integer.parseInt(props.getProperty("thread.pool.core", "2"));
            maxPoolSize = Integer.parseInt(props.getProperty("thread.pool.max", "4"));
            queueCapacity = Integer.parseInt(props.getProperty("thread.pool.queue", "1000"));
            maxRetries = Integer.parseInt(props.getProperty("retry.max.attempts", "3"));
            initialDelayMs = Long.parseLong(props.getProperty("retry.initial.delay.ms", "1000"));
            backoffMultiplier = Double.parseDouble(props.getProperty("retry.backoff.multiplier", "2.0"));

            metricsHost = props.getProperty("metrics.host", "MQTT-Bridge");
            metricsInterval = Integer.parseInt(props.getProperty("metrics.interval", "60"));

            return true;
        } catch (IOException e) {
            System.err.println("Error reading configuration file: " + e.getMessage());
            return false;
        }
    }

    // ---------- Загрузка маппинга ----------
    private static boolean loadMapping(String fileName) {
        try (InputStream input = getConfigInputStream(fileName)) {
            if (input == null) {
                System.err.println("Mapping file not found: " + fileName + " - using legacy mode.");
                return true;
            }
            JsonNode root = objectMapper.readTree(input);
            if (root.isArray()) {
                for (JsonNode node : root) {
                    String pattern = node.get("topicPattern").asText();
                    String host = node.get("host").asText();
                    String key = node.get("key").asText();
                    String valueField = node.has("valueField") ? node.get("valueField").asText() : null;
                    String subscribeTopic = node.has("subscribeTopic") ? node.get("subscribeTopic").asText() : null;
                    mappingRules.add(new MappingRule(pattern, host, key, valueField, subscribeTopic));
                }
                System.out.println("Loaded " + mappingRules.size() + " mapping rules.");
                return true;
            } else {
                System.err.println("Mapping file must contain a JSON array.");
                return false;
            }
        } catch (IOException e) {
            System.err.println("Error reading mapping file: " + e.getMessage());
            return false;
        }
    }

    private static InputStream getConfigInputStream(String fileName) throws IOException {
        File file = new File(fileName);
        if (file.exists() && file.isFile()) {
            return new FileInputStream(file);
        }
        InputStream is = MqttToZabbix.class.getClassLoader().getResourceAsStream(fileName);
        if (is != null) {
            return is;
        }
        return null;
    }

    // ---------- Поиск правила по топику ----------
    private static MappingRule findRule(String topic) {
        for (MappingRule rule : mappingRules) {
            Matcher m = rule.matcher(topic);
            if (m.matches()) {
                return rule;
            }
        }
        return null;
    }

    // ---------- Обработка по правилу ----------
    private static void processWithRule(String topic, String payload, MappingRule rule) {
        Matcher matcher = rule.matcher(topic);
        if (!matcher.matches()) return;

        String host = interpolate(rule.hostTemplate, matcher);
        String key = interpolate(rule.keyTemplate, matcher);

        String valueStr;
        if (rule.valueField != null && !rule.valueField.isEmpty() && !"null".equalsIgnoreCase(rule.valueField)) {
            try {
                JsonNode jsonNode = objectMapper.readTree(payload);
                if (jsonNode.isObject()) {
                    JsonNode fieldNode = jsonNode.path(rule.valueField);
                    if (!fieldNode.isMissingNode()) {
                        valueStr = fieldNode.asText();
                    } else {
                        System.err.println("Field '" + rule.valueField + "' not found in JSON payload: " + payload);
                        return;
                    }
                } else {
                    System.err.println("Payload is not a JSON object, using entire payload for topic " + topic);
                    valueStr = payload;
                }
            } catch (IOException e) {
                System.err.println("Failed to parse JSON for topic " + topic + ": " + e.getMessage() + ", using entire payload");
                valueStr = payload;
            }
        } else {
            valueStr = payload;
        }

        sendToZabbixWithRetry(host, key, valueStr, System.currentTimeMillis() / 1000);
    }

    // ---------- Legacy обработка (старый формат) ----------
    private static void processLegacy(String payload) {
        if (payload == null || payload.trim().isEmpty() || !payload.trim().startsWith("{")) {
            System.err.println("Legacy processing skipped: payload is not a JSON object: " + payload);
            return;
        }
        try {
            Map<String, Object> data = objectMapper.readValue(payload, Map.class);
            String host = (String) data.get("host");
            String key = (String) data.get("key");
            Object value = data.get("value");
            Long clock = data.containsKey("clock")
                    ? ((Number) data.get("clock")).longValue()
                    : System.currentTimeMillis() / 1000;
            sendToZabbixWithRetry(host, key, String.valueOf(value), clock);
        } catch (Exception e) {
            System.err.println("Legacy processing failed: " + e.getMessage());
        }
    }

    // ---------- Интерполяция строк с $1, $2 ... ----------
    private static String interpolate(String template, Matcher matcher) {
        if (template == null) return null;
        String result = template;
        for (int i = 1; i <= matcher.groupCount(); i++) {
            String group = matcher.group(i);
            if (group != null) {
                result = result.replace("$" + i, group);
            }
        }
        return result;
    }

    // ---------- Отправка с повторными попытками ----------
    private static void sendToZabbixWithRetry(String host, String key, String value, long clock) {
        long delayMs = initialDelayMs;
        for (int attempt = 1; attempt <= maxRetries; attempt++) {
            try {
                boolean success = sendToZabbix(host, key, value, clock);
                if (success) {
                    return;
                }
                if (attempt < maxRetries) {
                    System.err.println("Retry " + attempt + "/" + maxRetries + " for " + host + ":" + key +
                            " - waiting " + delayMs + " ms");
                    retryCounter.incrementAndGet();
                    Thread.sleep(delayMs);
                    delayMs = (long) (delayMs * backoffMultiplier);
                } else {
                    System.err.println("All retries failed for " + host + ":" + key);
                    errorCounter.incrementAndGet();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("Retry interrupted for " + host + ":" + key);
                return;
            } catch (Exception e) {
                if (attempt < maxRetries) {
                    System.err.println("Attempt " + attempt + "/" + maxRetries + " failed for " + host + ":" + key +
                            " - error: " + e.getMessage() + ", retrying in " + delayMs + " ms");
                    retryCounter.incrementAndGet();
                    try {
                        Thread.sleep(delayMs);
                        delayMs = (long) (delayMs * backoffMultiplier);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                } else {
                    System.err.println("All retries failed for " + host + ":" + key + " - final error: " + e.getMessage());
                    errorCounter.incrementAndGet();
                }
            }
        }
    }

    // ---------- Отправка в Zabbix (через внешний zabbix_sender) ----------
    private static boolean sendToZabbix(String host, String key, String value, long clock) {
        try {
            ProcessBuilder pb = new ProcessBuilder(
                    "zabbix_sender",
                    "-z", zabbixServer,
                    "-p", String.valueOf(zabbixPort),
                    "-s", host,
                    "-k", key,
                    "-o", String.valueOf(value)
            );
            pb.redirectErrorStream(true);
            Process p = pb.start();
            int exitCode = p.waitFor();
            if (exitCode == 0) {
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.contains("processed:")) {
                            System.out.println("zabbix_sender: " + line);
                        }
                    }
                }
                return true;
            } else {
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        System.err.println("zabbix_sender error: " + line);
                    }
                }
                return false;
            }
        } catch (Exception e) {
            System.err.println("zabbix_sender execution error: " + e.getMessage());
            return false;
        }
    }

    // ---------- Отправка метрик (без учёта счётчиков) ----------
    private static void sendMetric(String key, long value) {
        try {
            ProcessBuilder pb = new ProcessBuilder(
                    "zabbix_sender",
                    "-z", zabbixServer,
                    "-p", String.valueOf(zabbixPort),
                    "-s", metricsHost,
                    "-k", key,
                    "-o", String.valueOf(value)
            );
            pb.redirectErrorStream(true);
            Process p = pb.start();
            int exitCode = p.waitFor();
            if (exitCode != 0) {
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        System.err.println("metric send error: " + line);
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("Failed to send metric " + key + ": " + e.getMessage());
        }
    }
}