/**
 * mqtt_helper.hpp
 *
 * simplified apis for mqtt
 *
 * Bruce.Lu @20200415
 */

#ifndef __MQTT_HELPER__
#define __MQTT_HELPER__

#pragma once

#include <string>
#include <future>
#include <map>
#include <httplib.h>
#include <fmt/format.h>
#include <spdlog/spdlog.h>
#include <jsoncons/json.hpp>
#include "evutils.hpp"

extern "C" {
#include <MQTTAsync.h>
}

using namespace std;
using namespace jsoncons;
using namespace evutils;

#define EV_MAX_PRINTABLE_SIZE 512

extern void on_connected(void* context, MQTTAsync_successData* response);
extern void on_connlost(void *context, char *cause);
extern int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message);
extern void on_disconn(void* context, MQTTAsync_successData* response);
extern void on_subscribed(void* context, MQTTAsync_successData* response);
extern void on_conn_fail(void* context, MQTTAsync_failureData* response);
extern void on_subscribed_fail(void* context, MQTTAsync_failureData* response);
extern void on_sent(void* context, MQTTAsync_successData* response);
extern void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response);
extern 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);


class MqttPub {
private:
    MQTTAsync client = nullptr;
    string topic;
protected:
public:
    MqttPub(MQTTAsync client, string topic):client(client), topic(topic) {}

    int pub(const char *msg, int len, int qos = 0, bool retained = false)
    {
        MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
        MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
        int rc;

        opts.onSuccess = on_sent;
        opts.context = client;
        pubmsg.payload = (void *)msg;
        pubmsg.payloadlen =len;
        pubmsg.qos = qos;
        pubmsg.retained = retained;

        if ((rc = MQTTAsync_sendMessage(client, topic.c_str(), &pubmsg, &opts)) != MQTTASYNC_SUCCESS) {
            string msg = fmt::format("Failed to start sendMessage: {}", MQTTAsync_strerror(rc));
            spdlog::error(msg);
            throw msg;
        }
        return 0;
    }

    /// alias for pub
    int operator()(const char *msg, int len, int qos = 0, bool retained = false)
    {
        return pub(msg, len, qos, retained);
    }

    ~MqttPub()
    {
    }
};

class MqttHelper {
private:
    map<string, on_msg_fun_ptr_t> topics;
    recursive_mutex mut;
protected:
public:
    string mqtt_url;
    string id;
    string topic_lastwill;
    string topic_report;
    string addr;
    MQTTAsync client;
    enum class State {
        None,
        Ready,
        ConnOK,
        ConnFail,
        SubscribeOK,
        SubScribeFail,
        UnsubscribeOk,
        UnsubscribeFail,
        Disconnected,
        Destroyed
    };

    struct AsyncResult {
        State state;
        int rc;
        void *extra;
    };

    /// State state = State::None; // 0: initial ; 1: ready; 2: disconnected; 3: destroyed
    promise<AsyncResult> state;
    MqttHelper(string mqtt_url, string id, string topic_lastwill, string topic_report, int kai = 20, int cs = 1): mqtt_url(mqtt_url), id(id), topic_lastwill(topic_lastwill), topic_report(topic_report)
    {
        // make connection, throw excpetions
        MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
        MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
        int rc;
        int ch;
        string msg;
        auto uri = httplib::Uri::Parse(mqtt_url);
        if(!uri.Host.empty() && !uri.Port.empty()){
            addr = "tcp://" + uri.Host + ":" + uri.Port;
        }

        spdlog::debug("mqtt url: proto: {}, host: {}, port: {}, user: {}, pass: {}", uri.Protocol, uri.Host, uri.Port, uri.User, uri.Password);

        if(MQTTASYNC_SUCCESS != MQTTAsync_create(&client, addr.c_str(), ("EVC"+id).c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL)) {
            msg = "failed to async create mqtt";
            spdlog::error(msg);
            throw StrException(msg);
        }

        if(MQTTASYNC_SUCCESS != MQTTAsync_setCallbacks(client, this, on_connlost, on_msgarrvd, NULL)) {
            msg = "failed to async setcallbacks";
            spdlog::error(msg);
            throw StrException(msg);
        }
        conn_opts.keepAliveInterval = 20;
        conn_opts.cleansession = 1;
        /// TODO:
        if(!uri.User.empty() && !uri.Password.empty()){
            conn_opts.username = uri.User.c_str();
            conn_opts.password = uri.Password.c_str();
        }
        conn_opts.onSuccess = on_connected;
        conn_opts.onFailure = on_conn_fail;
        MQTTAsync_willOptions lwm = MQTTAsync_willOptions_initializer;
        /// template for lastwill message
        auto js = make_lastwill_msg(id);
        lwm.message = js.to_string().c_str();
        lwm.qos = 1;
        lwm.retained = 1;
        lwm.topicName = topic_lastwill.c_str();
        conn_opts.context = this;
        conn_opts.will = &lwm;
        spdlog::info("trying to connect to {}", addr);
        if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS) {
            msg = fmt::format("Failed to start connect: {}", MQTTAsync_strerror(rc));
            spdlog::error(msg);
            throw StrException(msg);
        }

        auto ar = state.get_future().get();
        if(ar.state != State::Ready) {
            string msg = fmt::format("ailed to initilaze mqtt: {}", MQTTAsync_strerror(ar.rc));
            spdlog::error(msg);
            throw StrException(msg);
        }
        else {
            spdlog::info("initialze mqtt {} successfully", addr);
        }


        // reset state
        state = promise<AsyncResult>();
    }

    // can be called only state
    int subscribe(string topic, on_msg_fun_ptr_t on_msg, bool force = false)
    {
        // make sure other thread can't access state while this is in working
        lock_guard<recursive_mutex> lk(mut);
        MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
        int rc;

        if(topics.count(topic) > 0) {
            string msg = fmt::format("already subscribed to {}", topic, addr);
            if(force) {
                MQTTAsync_responseOptions opts;
                opts.context = this;
                opts.onFailure = on_unsubscribe_fail;
                opts.onSuccess = on_unsubscribe_ok;
                MQTTAsync_unsubscribe(client, topic.c_str(), &opts);
                auto st = state.get_future().get();
                state= promise<AsyncResult>();
                if(st.state == State::UnsubscribeOk) {
                    msg += ", forced uninstall ok! ";
                    spdlog::info(msg);
                }
                else if(st.state == State::UnsubscribeFail) {
                    msg += ", froce unsintall failed! " + string(MQTTAsync_strerror(st.rc));
                    spdlog::error(msg);
                    return -1;
                }
                else {
                    spdlog::error("unkown uninstall status");
                }
            }
            else {
                msg += ", ingored since force = false";
                spdlog::warn(msg);
                return 0;
            }
        }

        topics[topic] = on_msg;
        opts.onSuccess = on_subscribed;
        opts.onFailure = on_subscribed_fail;
        opts.context = this;

        if ((rc = MQTTAsync_subscribe(client, topic.c_str(), 1, &opts)) != MQTTASYNC_SUCCESS) {
            spdlog::error("failed to subscribe {}: {} at {}",topic, MQTTAsync_strerror(rc), addr);
        }
        else {
            spdlog::info("subscribe to {} success at {}", topic, addr);
        }

        auto st = state.get_future().get();
        // TODO
        if(st.state == State::SubscribeOK) {
        }
        else if(st.state == State::SubScribeFail) {
            string msg = fmt::format("failed to subscribe topic {}: {}", topic, MQTTAsync_strerror(st.rc));
            spdlog::error(msg);
            throw StrException(msg);
        }
        // reset state
        state= promise<AsyncResult>();
        return 0;
    }

    /// for create pub topic
    MqttPub operator[](string topic)
    {
        MqttPub pub(client, topic);
        return pub;
    }

    /// process messages when on_msgarrvd
    int operator()(string topic, const void* const data, int len)
    {
        if(topics.count(topic) >0 && topics[topic] != nullptr) {
            topics[topic](this, data, len, topic);
        }
        else {
            spdlog::warn("no handler installed for topic: {}", topic);
        }
        return 0;
    }

    ~MqttHelper()
    {
        int rc;
        MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
        opts.onSuccess = on_disconn;
        opts.context = this;

        // TODO: how to avoid the pending future state?
        auto fu = state.get_future();
        // wait for all other future state finished
        while(is_ready(fu) && fu.valid());
        state = promise<AsyncResult>();
        if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS) {
            spdlog::error("failed to start disconnect: {}", MQTTAsync_strerror(rc));
        }

        state.get_future().get();
        MQTTAsync_destroy(&client);
    }

};


/// just a simple wrapper around MqttHelper
class MqttMgr {
    public:
    static map<string, MqttHelper *> insts;
    static mutex mut;
    public:
    
    static MqttHelper* get_instance(string mqtt_url, string devsn, string topic_lastwill="", string topic_report=""){
        auto key = mqtt_url + "/" + devsn;
        MqttHelper *ret = nullptr;
        lock_guard<mutex> lg(MqttMgr::mut);
        try{
            if(MqttMgr::insts.count(key) == 0){
                if(topic_lastwill.empty()){
                    topic_lastwill = evutils::consts::pub_topic_lastwill + devsn;
                }

                if(topic_report.empty()){
                    topic_report = evutils::consts::pub_topic_report;
                }

                auto inst = new MqttHelper(mqtt_url, devsn, topic_lastwill, topic_report);
                auto js = make_online_msg(devsn);
                auto msg = js.to_string();
                auto rc = (*inst)[topic_report].pub(msg.c_str(), msg.size(), 2, false);
                ret = inst;
                MqttMgr::insts[key] = inst;
            }else{
                ret = MqttMgr::insts[key];
            }
        }catch(exception &e) {
            spdlog::error("failed to create mqtt adaptor: {}", e.what());
        }
        
        return ret;
    }

    static void remove(string mqtt_url, string devsn){
        auto key = mqtt_url + "/" + devsn;
        lock_guard<mutex> lg(MqttMgr::mut);
        if(MqttMgr::insts.count(key) == 0){

        }else{
            delete MqttMgr::insts[key];
            MqttMgr::insts.erase(key);
        }
    }

    MqttMgr(){}
    ~MqttMgr(){
        for(auto &x:insts){
            if(x.second){
                delete x.second;
            }
            insts.erase(x.first);
        }
    }

        /// low level api for report and response
    static void _report_response(MqttHelper *client, string topic, string message)
    {
        // auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
        if(client == nullptr) {
            spdlog::error("invalid mqtt client {}");
            return;
        }
        auto rc = (*client)[topic].pub(message.c_str(), message.size(), 2, false);
        if(message.size() > EV_MAX_PRINTABLE_SIZE){
            string ellipsis = " <...> ";
            message = message.substr(0, EV_MAX_PRINTABLE_SIZE/2 - ellipsis.size()) + ellipsis + message.substr(message.size() -1 - EV_MAX_PRINTABLE_SIZE/2, EV_MAX_PRINTABLE_SIZE/2);
        }
        if(rc < 0) {
            spdlog::error("failed to pub mqtt message on {}: {}", topic, message);
        }
        else {
            spdlog::info("successfully pub message at {}:{}", topic, message);
        }
    }

    /// having data field along with code and msg
    /// signature {"rid", "code", "msg", "cata",  "data":{}}
    static void report_response_json(MqttHelper *client, string topic, json& message)
    {
        // time
        auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
        message.insert_or_assign(consts::kMsgTime, ts);
        // rid
        auto rid = message["rid"].as<string>();
        message.insert_or_assign(consts::kMsgRid, rid);
        // type
        if(rid.empty()) {
            message.insert_or_assign(consts::kMsgType, consts::kMsgTypeReport);
        }
        else {
            message.insert_or_assign(consts::kMsgType, consts::kMsgTypeResponse);
        }

        // sn
        message.insert_or_assign(consts::kMsgSn, client->id);

        _report_response(client, topic, message.to_string());
    }

    /// no data field. only code and msg field
    static void report_response_args(MqttHelper *client, string topic, int code, string message, string cata, string rid, json &&data)
    {
        json js;
        /// current timestamp
        auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
        js.insert_or_assign(consts::kMsgTime, ts);
        /// rid
        js.insert_or_assign(consts::kMsgRid, rid);
        /// type
        if(rid.empty()) {
            js.insert_or_assign(consts::kMsgType, consts::kMsgTypeReport);
        }
        else {
            js.insert_or_assign(consts::kMsgType, consts::kMsgTypeResponse);
        }
        js.insert_or_assign(consts::kMsgCata, cata);
        ///
        js.insert_or_assign(consts::kMsgCode, code);
        js.insert_or_assign(consts::kMsgMsg, message);
        js.insert_or_assign(consts::kMsgSn, client->id);
        if(!data.empty()){
            js.insert_or_assign(consts::kMsgData, data);
        }

        _report_response(client, topic, js.to_string());
    }

    static void report_response_args(MqttHelper *client, string topic, int code, string message, string cata, string rid, json &data)
    {
        json js;
        /// current timestamp
        auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
        js.insert_or_assign(consts::kMsgTime, ts);
        /// rid
        js.insert_or_assign(consts::kMsgRid, rid);
        /// type
        if(rid.empty()) {
            js.insert_or_assign(consts::kMsgType, consts::kMsgTypeReport);
        }
        else {
            js.insert_or_assign(consts::kMsgType, consts::kMsgTypeResponse);
        }
        js.insert_or_assign(consts::kMsgCata, cata);
        ///
        js.insert_or_assign(consts::kMsgCode, code);
        js.insert_or_assign(consts::kMsgMsg, message);
        js.insert_or_assign(consts::kMsgSn, client->id);
        if(!data.empty()){
            js.insert_or_assign(consts::kMsgData, data);
        }

        _report_response(client, topic, js.to_string());
    }
};

#endif