/**
 * @file mqtt_helper.cc
 *
 * @brief this is the main file impelents mqtt helpers
 *
 * @ingroup evcamera
 *
 * @author Bruce Lu
 * Contact: lzbgt@icloud.com
 *
 */

#include <mqtt_helper.hpp>

#include <string>
#include <chrono>
#include <thread>
#include <set>
#include <map>
#include <future>
#include <mutex>
#include <httplib.h>

// #define SPDLOG_ACTIVE_LEVEL SPDLOG_LEVEL_TRACE
// #define SPDLOG_DEBUG_ON
// #define SPDLOG_TRACE_ON
#include <spdlog/spdlog.h>
#include <fmt/format.h>
extern "C" {
#include <MQTTAsync.h>
}

using namespace std;

void on_connected(void* context, MQTTAsync_successData* response);
void on_connlost(void *context, char *cause);
int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message);
void on_disconn(void* context, MQTTAsync_successData* response);
void on_subscribed(void* context, MQTTAsync_successData* response);
void on_conn_fail(void* context, MQTTAsync_failureData* response);
void on_subscribed_fail(void* context, MQTTAsync_failureData* response);
void on_sent(void* context, MQTTAsync_successData* response);
void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response);
void on_unsubscribe_ok(void* context, MQTTAsync_successData* response);

class MqttHelper;
typedef void (*on_res_fun_ptr_t)(void *res);
typedef void (*on_msg_fun_ptr_t)(MqttHelper *hlp, const void * const data, int len, string topic);

template<typename R>
bool is_ready(std::future<R> const& f)
{
    return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
}

void on_connlost(void *context, char *cause)
{
    MqttHelper *self = (MqttHelper *) context;
    MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
    int rc;

    spdlog::error("mqtt connection lost: {}", cause? cause: "unkown reason");

    spdlog::warn("reconnecting {}", self->addr);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTAsync_connect(self->client, &conn_opts)) != MQTTASYNC_SUCCESS) {
        string msg = fmt::format("Failed to start connect:", MQTTAsync_strerror(rc));
        spdlog::error(msg);
        //self->state.set_exception(StrException(msg));
    }
}

int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message)
{
    int i;
    char* payloadptr;
    MqttHelper *self = (MqttHelper *) context;
    (*self)(string(topic), message->payload, message->payloadlen);
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topic);
    return 1;
}


void on_disconn(void* context, MQTTAsync_successData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    // self->state = MqttHelper::State::Disconnected;
    // if(self->state.get_future().valid()){
    //     self->state = std::promise<MqttHelper::State>();
    // }

    auto js = make_lastwill_msg(self->id);
    auto str = js.to_string();
    (*self)["hello"].pub(str.c_str(), str.size(), 1, true);

    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::Disconnected;
    self->state.set_value(as);
    spdlog::info("successful disconnected {}", self->addr);
}

void on_subscribed(void* context, MQTTAsync_successData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    //spdlog::debug("subscribe succeeded");

    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::SubscribeOK;
    self->state.set_value(as);
}

void on_subscribed_fail(void* context, MQTTAsync_failureData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    //string msg = fmt::format("subscribe failed: {}", MQTTAsync_strerror(response ? response->code : 0));
    // spdlog::error(msg);
    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::SubScribeFail;
    as.rc = response?response->code:0;
    self->state.set_value(as);
}


void on_unsubscribe_ok(void* context, MQTTAsync_successData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::UnsubscribeOk;
    self->state.set_value(as);
}

void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    string msg = fmt::format("unsubscribe failed: {}", MQTTAsync_strerror(response ? response->code : 0));
    spdlog::error(msg);
    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::UnsubscribeFail;
    as.rc = response?response->code:0;
    self->state.set_value(as);
}

void on_conn_fail(void* context, MQTTAsync_failureData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    string msg = fmt::format("Connect failed: {}", MQTTAsync_strerror(response ? response->code : 0));
    spdlog::error(msg);
    //self->state.set_exception(StrException(msg));
    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::ConnFail;
    as.rc = response?response->code:0;
    self->state.set_value(as);
}

void on_sent(void* context, MQTTAsync_successData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    spdlog::debug("Message delivery confirmed, tok: {}", response->token);
}

void on_connected(void* context, MQTTAsync_successData* response)
{
    MqttHelper *self = (MqttHelper *) context;
    MqttHelper::AsyncResult as;
    as.state = MqttHelper::State::Ready;
    self->state.set_value(as);
    spdlog::info("connected to mqtt {}", self->addr);
}

map<string, MqttHelper *> MqttMgr::insts;
mutex MqttMgr::mut;

#ifdef TEST_MQTT

#include <httplib.h>

void my_on_msg1(MqttHelper *hlp, const void * const data, int len, string topic)
{
    string msg;
    msg.resize(len+1);
    memcpy((void*)&msg[0], data, len);
    *(char*)(&msg[0] +len) = 0;
    spdlog::info("I1 recv on {}: {}", topic, msg);
    (*hlp)[to_string(stoi(topic) + 1)]("hello ack1", 11);
}

void my_on_msg2(MqttHelper *hlp, const void * const data, int len, string topic)
{
    string msg;
    msg.resize(len+1);
    memcpy((void*)&msg[0], data, len);
    *(char*)(&msg[0] +len) = 0;
    spdlog::info("I2 recv on {}: {}", topic, msg);
    (*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}

int main()
{
    string mqtt_url = "tcp://admin:vJ3zHqWrHbrqxVMT@evcloudsvc.ilabservice.cloud:11883";
    string mqtt_cid = "testaabbc";

    auto tmpEnv = getenv("MQTT_URL");
    if(tmpEnv) {
        mqtt_url = string(tmpEnv);
    }

    tmpEnv = getenv("CID");
    if(tmpEnv) {
        mqtt_cid = string(tmpEnv);
    }

    spdlog::set_level(spdlog::level::debug);
    {
        try {
            MqttHelper hlp(mqtt_url, mqtt_cid);
            hlp["hello"].pub("hello", 6, 1);
            hlp.subscribe("no_on_msg", NULL);
            auto x = thread([&hlp] {
                for(int i = 0; i < 20; i ++)
                {
                    hlp.subscribe(to_string(i), my_on_msg1);
                }
                hlp["hello"].pub("hello1", 7, 1);
            });

            auto y = thread([&hlp] {
                for(int i = 0; i < 10; i ++)
                {
                    hlp.subscribe(to_string(i), my_on_msg2);
                }
                for(int i = 10; i < 30; i ++)
                {
                    hlp.subscribe(to_string(i), my_on_msg2, true);
                }
                hlp["hello"].pub("hello2", 7, 1);
            });

            x.join();
            y.join();
            this_thread::sleep_for(chrono::seconds(20));
        }
        catch(exception &e) {
            spdlog::error("failed to connect mqtt: {}, {}", mqtt_url, e.what());
            exit(1);
        }
    }

    this_thread::sleep_for(chrono::seconds(2));
    return 0;
}

#endif
