package com.ai.trackwalker.uavaccess.mqtt;

import com.ai.trackwalker.uavaccess.common.Jsons;
import com.ai.trackwalker.uavaccess.config.UavAccessProperties;
import com.ai.trackwalker.uavaccess.event.VendorEventIngestService;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
public class MqttBridgeSubscriber implements SmartLifecycle {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqttBridgeSubscriber.class);

    private final UavAccessProperties properties;
    private final VendorEventIngestService events;
    private final AtomicBoolean running = new AtomicBoolean();
    private MqttClient client;

    public MqttBridgeSubscriber(UavAccessProperties properties, VendorEventIngestService events) {
        this.properties = properties;
        this.events = events;
    }

    @Override
    public void start() {
        if (!properties.getMqtt().isEnabled() || running.get()) return;
        try {
            UavAccessProperties.Mqtt config = properties.getMqtt();
            client = new MqttClient(config.getBrokerUrl(), config.getClientId(), new MemoryPersistence());
            client.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    try {
                        client.subscribe(config.getTopicFilter(), config.getQos());
                        running.set(true);
                        LOGGER.info("Subscribed to FlightHub MQTT Bridge topic {} at {}", config.getTopicFilter(), serverURI);
                    } catch (Exception exception) {
                        running.set(false);
                        LOGGER.error("Unable to subscribe to FlightHub MQTT Bridge", exception);
                    }
                }

                @Override
                public void connectionLost(Throwable cause) {
                    running.set(false);
                    LOGGER.warn("FlightHub MQTT Bridge connection lost", cause);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    ingest(topic, message);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // Subscriber does not publish MQTT messages.
                }
            });
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setAutomaticReconnect(config.isAutomaticReconnect());
            if (config.getUsername() != null && !config.getUsername().isBlank()) {
                options.setUserName(config.getUsername());
            }
            if (config.getPassword() != null && !config.getPassword().isBlank()) {
                options.setPassword(config.getPassword().toCharArray());
            }
            client.connect(options);
            client.subscribe(config.getTopicFilter(), config.getQos());
            running.set(true);
        } catch (Exception exception) {
            running.set(false);
            LOGGER.error("Unable to start FlightHub MQTT Bridge subscriber", exception);
        }
    }

    private void ingest(String topic, MqttMessage message) {
        try {
            String content = new String(message.getPayload(), StandardCharsets.UTF_8);
            Map<String, Object> payload = Jsons.map(content);
            String eventId = findEventId(payload);
            events.ingest(
                    "flighthub2-default",
                    "MQTT",
                    eventId,
                    topic,
                    Instant.now().toString(),
                    payload
            );
        } catch (RuntimeException exception) {
            LOGGER.error("Unable to process FlightHub MQTT message from {}", topic, exception);
        }
    }

    private String findEventId(Map<String, Object> payload) {
        for (String key : java.util.List.of("event_id", "message_id", "tid", "bid", "id")) {
            Object value = payload.get(key);
            if (value != null && !String.valueOf(value).isBlank()) return String.valueOf(value);
        }
        return UUID.randomUUID().toString();
    }

    @Override
    public void stop() {
        try {
            if (client != null && client.isConnected()) client.disconnect();
            if (client != null) client.close();
        } catch (Exception exception) {
            LOGGER.warn("Unable to stop FlightHub MQTT subscriber cleanly", exception);
        } finally {
            running.set(false);
        }
    }

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

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public int getPhase() {
        return Integer.MAX_VALUE - 100;
    }
}
