#include "fgc/MqttControlChannel.h"

#include "fgc/ChannelTopics.h"

#include "fgc/Logger.h"

#include <chrono>

namespace fgc {

namespace {
constexpr int  kQos = 1;
const auto     kConnectTimeout = std::chrono::seconds(5);
const auto     kPublishTimeout = std::chrono::seconds(3);
constexpr size_t kOutboxMax = 64;
const auto     kOutboxFlush = std::chrono::seconds(5);
}  // namespace

MqttControlChannel::MqttControlChannel(const std::string& broker_uri, const std::string& tower,
                                       const std::string& user, const std::string& password)
    : broker_uri_(broker_uri),
      tower_(tower),
      topic_status_(towerTopic(tower, kTopicStatusCode)),
      topic_cam_event_(towerTopic(tower, kTopicCamEvent)),
      topic_health_(towerTopic(tower, kTopicHealth)),
      topic_command_(towerTopic(tower, kTopicCommand)),
      topic_command_reply_(towerTopic(tower, kTopicCommandReply)),
      client_(broker_uri, tower) {
    conn_opts_.set_keep_alive_interval(20);
    conn_opts_.set_clean_session(true);
    conn_opts_.set_automatic_reconnect(true);  // replaces the old manual reconnect loop
    if (!user.empty()) conn_opts_.set_user_name(user);
    if (!password.empty()) conn_opts_.set_password(password);
    client_.set_callback(*this);
    outbox_thread_ = std::thread(&MqttControlChannel::runOutbox, this);
}

MqttControlChannel::~MqttControlChannel() {
    disconnect();
    stopOutbox();
}

void MqttControlChannel::post(const std::string& topic, const std::string& payload) {
    {
        std::lock_guard<std::mutex> lock(outbox_mutex_);
        if (outbox_.size() >= kOutboxMax) {
            history_.record(/*outbound=*/true, Delivery::Dropped, outbox_.front().first,
                            outbox_.front().second);
            outbox_.pop_front();
        }
        outbox_.emplace_back(topic, payload);
    }
    outbox_cv_.notify_one();
}

void MqttControlChannel::runOutbox() {
    std::unique_lock<std::mutex> lock(outbox_mutex_);
    for (;;) {
        outbox_cv_.wait(lock, [this] { return outbox_stop_ || !outbox_.empty(); });
        if (outbox_.empty()) return;   // stopping, and nothing left
        auto m = std::move(outbox_.front());
        outbox_.pop_front();
        lock.unlock();
        publish(m.first, m.second);
        lock.lock();
    }
}

void MqttControlChannel::stopOutbox() {
    if (!outbox_thread_.joinable()) return;
    // Give what is waiting a few seconds to go out; drop (and record) the rest.
    const auto deadline = std::chrono::steady_clock::now() + kOutboxFlush;
    for (;;) {
        {
            std::lock_guard<std::mutex> lock(outbox_mutex_);
            if (outbox_.empty() || std::chrono::steady_clock::now() >= deadline) {
                for (const auto& m : outbox_)
                    history_.record(/*outbound=*/true, Delivery::Dropped, m.first, m.second);
                outbox_.clear();
                outbox_stop_ = true;
                break;
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(20));
    }
    outbox_cv_.notify_all();
    outbox_thread_.join();
}

bool MqttControlChannel::connect() {
    try {
        client_.connect(conn_opts_, nullptr, *this)->wait_for(kConnectTimeout);
        connected_ = client_.is_connected();
    } catch (const mqtt::exception& e) {
        LOG_ERROR << "MQTT connect failed (" << broker_uri_ << "): " << e.what();
        connected_ = false;
    }
    if (connected_)
        LOG_INFO << "MQTT connected to " << broker_uri_ << " as '" << tower_ << "'";
    return connected_;
}

void MqttControlChannel::disconnect() {
    stopOutbox();   // the last pictures' events go out before the link closes
    if (!client_.is_connected()) return;
    try {
        client_.disconnect()->wait();
    } catch (const mqtt::exception& e) {
        LOG_WARN << "MQTT disconnect: " << e.what();
    }
    connected_ = false;
}

bool MqttControlChannel::connected() const { return connected_; }

void MqttControlChannel::connected(const std::string& /*cause*/) {
    LOG_INFO << "MQTT (re)connected; subscribing";
    history_.noteConnect();
    client_.subscribe(topic_command_, kQos);
    connected_ = true;
    // Say the mode again: whoever watches StatusCode may have missed changes
    // while the link was down. Not waited on - this runs on Paho's callback
    // thread, which is the one that would deliver the acknowledgement.
    // Health likewise: a paused tower must not look healthy to a page that
    // subscribed while the link was down.
    std::string status, health;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        status = last_status_;
        health = last_health_;
    }
    auto republish = [this](const std::string& topic, const std::string& payload) {
        if (payload.empty()) return;
        try {
            auto msg = mqtt::make_message(topic, payload);
            msg->set_qos(kQos);
            msg->set_retained(true);
            client_.publish(msg);
            history_.record(/*outbound=*/true, Delivery::Ok, topic, payload);
        } catch (const mqtt::exception& e) {
            LOG_WARN << "MQTT republish of " << topic << " failed: " << e.what();
        }
    };
    republish(topic_status_, status);
    republish(topic_health_, health);
}

void MqttControlChannel::connection_lost(const std::string& cause) {
    LOG_WARN << "MQTT connection lost" << (cause.empty() ? "" : ": " + cause);
    history_.noteLoss();
    connected_ = false;
    // Paho will auto-reconnect when configured; the broker callback re-subscribes.
}

void MqttControlChannel::on_failure(const mqtt::token& /*tok*/) {
    LOG_WARN << "MQTT action failed";
}

void MqttControlChannel::message_arrived(mqtt::const_message_ptr msg) {
    const std::string topic   = msg->get_topic();
    const std::string payload = msg->to_string();
    LOG_TRACE_CAT(LogCat::Mqtt) << "RX  " << topic << ' ' << payload;
    history_.record(/*outbound=*/false, Delivery::Ok, topic, payload);
    std::lock_guard<std::mutex> lock(mutex_);
    if (topic == topic_command_) {
        // Queued for the control thread, which runs it exactly as a typed
        // command. Retained ones are queued too, flagged, so they are answered
        // (and refused) rather than silently swallowed.
        RemoteCommand rc = parseCommandPayload(payload);
        rc.received_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                             std::chrono::system_clock::now().time_since_epoch()).count();
        rc.retained = msg->is_retained();
        // A flood - a stuck publisher - must not grow without bound.
        if (commands_.size() < 256) commands_.push_back(std::move(rc));
        else LOG_WARN << "MQTT command queue full; dropped: " << payload;
    }
}

std::vector<RemoteCommand> MqttControlChannel::takeCommands() {
    std::lock_guard<std::mutex> lock(mutex_);
    std::vector<RemoteCommand> out;
    out.swap(commands_);
    return out;
}

void MqttControlChannel::publishCommandReply(const CommandReply& reply) {
    // Not retained: a reply answers one request, and a retained one would be
    // handed to every web page that subscribes later, as if it were new.
    publish(topic_command_reply_, commandReplyJson(reply), /*retained=*/false);
}

void MqttControlChannel::publish(const std::string& topic, const std::string& payload,
                                 bool retained) {
    if (!client_.is_connected()) {
        // Discarded: there is no queue and no retry, so this message is simply
        // gone. Recording it is the only trace it ever existed.
        history_.record(/*outbound=*/true, Delivery::Dropped, topic, payload);
        return;
    }
    LOG_TRACE_CAT(LogCat::Mqtt) << "PUB " << topic << ' ' << payload;
    try {
        auto msg = mqtt::make_message(topic, payload);
        msg->set_qos(kQos);
        msg->set_retained(retained);
        client_.publish(msg)->wait_for(kPublishTimeout);
        history_.record(/*outbound=*/true, Delivery::Ok, topic, payload);
    } catch (const mqtt::exception& e) {
        LOG_WARN << "MQTT publish failed: " << e.what();
        history_.record(/*outbound=*/true, Delivery::Failed, topic, payload);
    }
}

void MqttControlChannel::publishStatus(int code) {
    const std::string payload = std::to_string(code);
    {
        std::lock_guard<std::mutex> lock(mutex_);
        last_status_ = payload;
    }
    history_.noteStatus(payload);
    publish(topic_status_, payload);
}

void MqttControlChannel::publishCamEvent(const CamEvent& e) {
    post(topic_cam_event_, camEventJson(e));
}

void MqttControlChannel::publishHealth(const HealthReport& h) {
    const std::string payload = healthJson(h);
    {
        std::lock_guard<std::mutex> lock(mutex_);
        last_health_ = payload;
    }
    post(topic_health_, payload);
}

}  // namespace fgc