#include "mqtt_client.h"

#include <mosquitto.h>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <memory>
#include <mutex>
#include <sstream>
#include <utility>
#include <vector>
#include <unistd.h>

namespace mqtt {
namespace {

constexpr int kBrokerPort = 1883;
constexpr int kKeepaliveSeconds = 60;
constexpr const char *kUsername = "admin";
constexpr const char *kPassword = "admin";

std::string make_client_id(std::size_t index){
    const auto ticks = std::chrono::steady_clock::now().time_since_epoch().count();
    std::ostringstream id;
    id << "rv1106-" << std::hex << static_cast<unsigned long>(getpid())
       << '-' << static_cast<unsigned long long>(ticks)
       << '-' << index;
    return id.str();
}

class BrokerConnection final {
public:
    BrokerConnection(std::string host, std::string client_id,
                     const Topics &topics, MessageRouter &router)
        : host_(std::move(host)), client_id_(std::move(client_id)),
          topics_(topics), router_(router) {}

    ~BrokerConnection(){
        stop();
    }

    bool start(){
        mosq_ = mosquitto_new(client_id_.c_str(), true, this);
        if(!mosq_){
            std::cerr << "[mqtt] cannot allocate client for " << host_ << "\n";
            return false;
        }

        int protocol = MQTT_PROTOCOL_V5;
        int rc = mosquitto_int_option(mosq_, MOSQ_OPT_PROTOCOL_VERSION, protocol);
        if(rc == MOSQ_ERR_SUCCESS)
            rc = mosquitto_username_pw_set(mosq_, kUsername, kPassword);
        mosquitto_reconnect_delay_set(mosq_, 2, 10, true);
        mosquitto_connect_callback_set(mosq_, on_connect_callback);
        mosquitto_disconnect_callback_set(mosq_, on_disconnect_callback);
        mosquitto_message_callback_set(mosq_, on_message_callback);
        if(rc == MOSQ_ERR_SUCCESS)
            rc = mosquitto_connect_async(mosq_, host_.c_str(), kBrokerPort,
                                         kKeepaliveSeconds);
        if(rc == MOSQ_ERR_SUCCESS) rc = mosquitto_loop_start(mosq_);
        if(rc != MOSQ_ERR_SUCCESS){
            std::cerr << "[mqtt] cannot start " << host_ << ':' << kBrokerPort
                      << ": " << mosquitto_strerror(rc) << "\n";
            mosquitto_destroy(mosq_);
            mosq_ = nullptr;
            return false;
        }
        started_ = true;
        std::cerr << "[mqtt] connecting " << host_ << ':' << kBrokerPort
                  << " client_id=" << client_id_ << " protocol=v5\n";
        return true;
    }

    void stop(){
        if(!mosq_) return;
        started_ = false;
        mosquitto_disconnect(mosq_);
        mosquitto_loop_stop(mosq_, true);
        connected_ = false;
        mosquitto_destroy(mosq_);
        mosq_ = nullptr;
    }

    int send(const std::string &topic, const std::string &payload,
             int qos, bool retain){
        if(!started_ || !mosq_) return MOSQ_ERR_NO_CONN;
        return mosquitto_publish(mosq_, nullptr, topic.c_str(),
                                 static_cast<int>(payload.size()), payload.data(),
                                 qos, retain);
    }

    bool is_started() const{
        return started_;
    }

private:
    static void on_connect_callback(mosquitto *, void *userdata, int rc){
        if(userdata) static_cast<BrokerConnection *>(userdata)->on_connect(rc);
    }

    static void on_disconnect_callback(mosquitto *, void *userdata, int rc){
        if(userdata) static_cast<BrokerConnection *>(userdata)->on_disconnect(rc);
    }

    static void on_message_callback(mosquitto *, void *userdata,
                                    const mosquitto_message *message){
        if(userdata)
            static_cast<BrokerConnection *>(userdata)->on_message(message);
    }

    void on_connect(int rc){
        if(rc != 0){
            connected_ = false;
            std::cerr << "[mqtt] connection rejected by " << host_ << ':'
                      << kBrokerPort << " rc=" << rc << "\n";
            return;
        }
        connected_ = true;
        int app_rc = mosquitto_subscribe(mosq_, nullptr,
                                         topics_.app_to_device.c_str(), 0);
        int server_rc = mosquitto_subscribe(mosq_, nullptr,
                                            topics_.server_to_device.c_str(), 0);
        std::cerr << "[mqtt] connected " << host_ << ':' << kBrokerPort
                  << " subscribed=" << topics_.app_to_device << ','
                  << topics_.server_to_device
                  << " rc=" << app_rc << '/' << server_rc << "\n";
    }

    void on_disconnect(int rc){
        connected_ = false;
        if(started_)
            std::cerr << "[mqtt] disconnected " << host_ << ':' << kBrokerPort
                      << " rc=" << rc << "; reconnect is automatic\n";
    }

    void on_message(const mosquitto_message *message){
        if(!message || !message->topic || !message->payload ||
           message->payloadlen < 0) return;
        router_.route(message->topic,
                      std::string(static_cast<const char *>(message->payload),
                                  static_cast<std::size_t>(message->payloadlen)));
    }

    mosquitto *mosq_ = nullptr;
    std::string host_;
    std::string client_id_;
    Topics topics_;
    MessageRouter &router_;
    std::atomic<bool> started_{false};
    std::atomic<bool> connected_{false};
};

} // namespace

class Client::Impl {
public:
    Impl(Topics topics, MessageRouter &router)
        : topics(std::move(topics)), router(router) {}

    Topics topics;
    MessageRouter &router;
    std::vector<std::unique_ptr<BrokerConnection>> connections;
    std::mutex publish_mutex;
    bool library_initialized = false;
};

Client::Client(Topics topics, MessageRouter &router)
    : impl_(std::make_unique<Impl>(std::move(topics), router)) {}

Client::~Client(){
    stop();
}

bool Client::start(const RemoteConfig &config){
    stop();
    if(mosquitto_lib_init() != MOSQ_ERR_SUCCESS){
        std::cerr << "[mqtt] Mosquitto library initialization failed\n";
        return false;
    }
    impl_->library_initialized = true;

    std::size_t started = 0;
    for(std::size_t i = 0; i < config.brokers.size() && i < 15; ++i){
        auto connection = std::make_unique<BrokerConnection>(
            config.brokers[i], make_client_id(i), impl_->topics, impl_->router);
        if(connection->start()) ++started;
        impl_->connections.emplace_back(std::move(connection));
    }
    std::cerr << "[mqtt] clients started=" << started
              << '/' << impl_->connections.size() << "\n";
    return started > 0;
}

void Client::stop(){
    for(auto &connection : impl_->connections) connection->stop();
    impl_->connections.clear();
    if(impl_->library_initialized){
        mosquitto_lib_cleanup();
        impl_->library_initialized = false;
    }
}

int Client::retry_failed(){
    std::lock_guard<std::mutex> lock(impl_->publish_mutex);
    int recovered = 0;
    for(auto &connection : impl_->connections){
        if(!connection->is_started() && connection->start()) ++recovered;
    }
    return recovered;
}

int Client::publish_all(const std::string &topic, const std::string &payload,
                        int qos, bool retain){
    std::lock_guard<std::mutex> lock(impl_->publish_mutex);
    int sent = 0;
    for(auto &connection : impl_->connections){
        if(connection->send(topic, payload, qos, retain) == MOSQ_ERR_SUCCESS)
            ++sent;
    }
    return sent;
}

} // namespace mqtt
