/*!
 *
 * Copyright (c) 2024-2026 Diality Inc. - All Rights Reserved.
 * \copyright
 * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN
 * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER.
 *
 * \file    CloudConnectController.cpp
 * \author  (original)  Stephen Quong
 * \date    (original)  24-May-2026
 *
 */
#include <google/protobuf/descriptor.h>
#include <google/protobuf/dynamic_message.h>
#include <google/protobuf/util/json_util.h>

#include <QCoreApplication>
#include <QDebug>
#include <QFile>
#include <QFileInfo>
#include <QLoggingCategory>
#include <QSettings>
#include <QString>

#include "CloudConnectController.h"
#include "LeahiMsgDefs.pb.h"
#include "LeahiMsgProtoUtils.h"

Q_LOGGING_CATEGORY(logConfig,       "config")
Q_LOGGING_CATEGORY(logProto,        "proto")
Q_LOGGING_CATEGORY(logCanRouting,   "can.routing")
Q_LOGGING_CATEGORY(logMqtt,         "mqtt")
Q_LOGGING_CATEGORY(logStats,        "mqtt.stats")

/*!
 * \brief   CloudConnectController::CloudConnectController
 * \details Constructor. Wires the frame→dispatcher→completed-message pipeline.
 *          The CAN device itself is started by startCan() once this object is
 *          on its worker thread.
 * \param   configPath      - path to the settings file
 * \param   canRoutingPath  - path to the CAN message routing INI
 * \param   parent          - optional QObject parent
 * \note    Must be constructed on the main thread, but later moved to a worker thread.
 *          _canInterface is parented to this object so it migrates with it.
 */
CloudConnectController::CloudConnectController(QObject *parent) :
    QObject(parent),
    _canInterface(this),
    _dispatcher(this),
    _mqttClient(this)
{
    connect(&_canInterface, &Can::CanInterface::didFrameReceive, this, &CloudConnectController::onCanFrameReceive);
    connect(&_dispatcher, &Can::MessageDispatcher::didActionReceive, this, &CloudConnectController::onCanMessageReceive);
    connect(&_mqttClient, &MqttClient::didStateChanged, this, &CloudConnectController::onCloudStateChanged);
    connect(&_mqttClient, &MqttClient::didMessageStatusChanged, this, &CloudConnectController::onCloudMessageStatusChanged);
    connect(&_mqttClient, &MqttClient::didMessageReceived, this, &CloudConnectController::onCloudMessageReceived);
}

/*!
 * \brief   CloudConnectController::~CloudConnectController
 */
CloudConnectController::~CloudConnectController() = default;

bool CloudConnectController::loadConfig(const QString &configPath)
{
    static auto resolvePath = [](const QString &path) -> QString {
        const QFileInfo info(path);
        return info.isAbsolute()
            ? path
            : QString("%1/%2").arg(QCoreApplication::applicationDirPath(), path);
    };

    if (!QFileInfo::exists(configPath)) {
        qCCritical(logConfig).noquote() << QString("config file %1 does not exist").arg(configPath);
        return false;
    }

    QSettings config = QSettings(configPath, QSettings::IniFormat);
    if (config.status() != QSettings::NoError) {
        qCCritical(logConfig).noquote() << QString("could not read config file %1").arg(configPath);
        return false;
    }

    // NOTE: do not try to prefix keys in the General (i.e. "General/<key>") or beginGroup("General") or
    //       QSettings will look for a group in the INI file called "%General", "General" is a special group.
    if (!loadCanRouting(resolvePath(config.value(QStringLiteral("CanRoutingConfig")).toString()))) {
        return false;
    }

    _appServerSocketPath = config.value("App/SocketName", "/tmp/cloudconnect.sock").toString();

    config.beginGroup(QStringLiteral("Cloud"));
    MqttClient::Config mqttConfig;
    mqttConfig.endpoint = config.value(QStringLiteral("ServerAddress"), QStringLiteral("localhost")).toString();
    mqttConfig.port = config.value(QStringLiteral("Port"), 8883).toInt();
    mqttConfig.clientId = config.value(QStringLiteral("ClientId")).toString();
    mqttConfig.caPath = resolvePath(config.value(QStringLiteral("CaFile")).toString());
    mqttConfig.certPath = resolvePath(config.value(QStringLiteral("CertFile")).toString());
    mqttConfig.keyPath = resolvePath(config.value(QStringLiteral("KeyFile")).toString());
    _deviceId = config.value(QStringLiteral("SerialNumber")).toString();
    _topicPrefix = config.value(QStringLiteral("TopicPrefix")).toString();
    config.endGroup();
    if (!_mqttClient.init(mqttConfig)) {
        return false;
    }

    return true;
}

/*!
 * \brief   CloudConnectController::initThread
 * \details Moves this object and its children onto a worker thread and starts it.
 * \param   thread  Thread to move to, owned by the caller.
 * \note    Must be called from the main thread, before the event loop runs.
 */
void CloudConnectController::initThread(QThread &thread)
{
    Q_ASSERT_X(QThread::currentThread() == qApp->thread(), __func__,
        "CloudConnectController initialization must be done in Main Thread");

    thread.setObjectName(QString("%1_Thread").arg(metaObject()->className()));
    moveToThread(&thread);
    thread.start();
}

/*!
 * \brief   CloudConnectController::startCan
 * \details Creates and connects the CAN device.
 * \return  true if the device was created and connected.
 * \note    Must run on the controller thread, after initThread(). connectDevice()
 *          installs the CAN socket notifier on the calling thread, so starting
 *          here keeps the notifier, the frame queue and the drain loop on one
 *          thread. QCanBusDevice::framesAvailable() is unguarded in Qt 5.15, so
 *          a cross-thread split of those is a data race, not just a style issue.
 */
bool CloudConnectController::startCan()
{
    Q_ASSERT_X(QThread::currentThread() == thread(), __func__,
        "startCan() must run on the controller thread");

    return _canInterface.init();
}

/*!
 * \brief   CloudConnectController::listenForApp
 * \details Creates the app socket server, then starts listening.
 * \return  true if the server bound successfully, false otherwise.
 * \note    Must run on the controller thread, after initThread(). QLocalServer ties
 *          its socket engine to the thread that calls listen(), and accepted client
 *          sockets are parented to that engine.
 */
bool CloudConnectController::listenForApp()
{
    Q_ASSERT_X(QThread::currentThread() == thread(), __func__,
        "listenForApp() must run on the controller thread"); // use Q_ASSERT only during object creation or thread move

    if (_appServer == nullptr) {
        _appServer = QSharedPointer<CloudConnectServer>::create(this);
    }

    return _appServer->listen(_appServerSocketPath);
}

/*!
 * \brief   CloudConnectController::connectToCloud
 * \details Configures the MQTT interface from the [Mqtt] settings and starts connecting.
 * \return  true if the attempt was started, or if MQTT is disabled.
 * \note    Must run on the controller thread, after initThread(). The interface's
 *          QMqttClient socket belongs to the thread that constructs it.
 */
bool CloudConnectController::connectToCloud()
{
    Q_ASSERT_X(QThread::currentThread() == thread(), __func__,
        "connectToCloud() must run on the controller thread");

    return _mqttClient.open();
}

/*!
 * \brief   CloudConnectController::loadCanRouting
 * \details Parses CAN message routing INI and populates _canRouting.
 * \param   canRoutingPath - path to the CAN message routing INI file
 * \return  true if CanRouting config successfully loads, otherwise false
 */
bool CloudConnectController::loadCanRouting(const QString &canRoutingPath)
{
    static const QHash<QString, CanAction> actionMap = {
        { QStringLiteral("SendAlways"), CanAction::SendAlways },
        { QStringLiteral("SendDelta"),  CanAction::SendDelta  },
        { QStringLiteral("Drop"),       CanAction::Drop       },
    };
    static const QHash<QString, CloudConnectFrame::Topic> topicMap = {
        { QStringLiteral("HighPriority"),       CloudConnectFrame::Topic::HighPriority       },
        { QStringLiteral("NormalPriority"),     CloudConnectFrame::Topic::NormalPriority     },
        { QStringLiteral("DeviceLogFile"),      CloudConnectFrame::Topic::DeviceLogFile      },
        { QStringLiteral("TreatmentLogFile"),   CloudConnectFrame::Topic::TreatmentLogFile   },
        { QStringLiteral("CloudSyncLogFile"),   CloudConnectFrame::Topic::CloudSyncLogFile   },
    };

    if (!QFile::exists(canRoutingPath)) {
        qCWarning(logCanRouting).noquote() << QString("CAN mesg routing config file %1 does not exist").arg(canRoutingPath);
        return false;
    }

    QSettings canRoutingConfig(canRoutingPath, QSettings::IniFormat);
    if (canRoutingConfig.status() != QSettings::NoError) {
        qCWarning(logCanRouting).noquote() << QString("could not read CAN message routing config file %1").arg(canRoutingPath);
        return false;
    }

    int msgCount = 0;
    for (const QString &group : canRoutingConfig.childGroups()) {
        bool ok = false;
        const Can::MsgId msgId = static_cast<Can::MsgId>(group.toUInt(&ok, 16));
        if (!ok) {
            qCWarning(logCanRouting).noquote() << QString("could not convert group \"%1\" to MsgId in CAN mesg routing config file %2")
                                                  .arg(group, canRoutingPath);
            continue;
        }

        canRoutingConfig.beginGroup(group);
        const QString msgIdStr  = canRoutingConfig.value(QStringLiteral("msg_id")).toString();
        const QString actionStr = canRoutingConfig.value(QStringLiteral("action")).toString().trimmed();
        const QString topicStr  = canRoutingConfig.value(QStringLiteral("topic")).toString().trimmed();
        // qCInfo(logCanRouting).noquote() << QString("%1 (0x%2): action=%3, topic=%4")
        //     .arg(msgIdStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0')).toUpper(), actionStr, topicStr);
        canRoutingConfig.endGroup();

        CanRouting canRouting;
        // canRouting.action  = actionMap.value(actionStr, CanAction::Drop);
        canRouting.action   = actionMap.value(actionStr, CanAction::SendAlways);
        canRouting.topic    = topicMap.value(topicStr, CloudConnectFrame::Topic::NormalPriority);

        if (actionStr.length() > 0 && !actionMap.contains(actionStr)) {
            qCWarning(logCanRouting).noquote() << QString("unknown message action \"%1\" for msgId=0x%2 — defaulting to Drop")
                .arg(actionStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper()));
        }

        if (topicStr.length() > 0 && !topicMap.contains(topicStr)) {
            qCWarning(logCanRouting).noquote() << QString("unknown message topic \"%1\" for msgId=0x%2 — defaulting to NormalPriority")
                .arg(topicStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper()));
        }

        _canRouting.insert(msgId, canRouting);
        msgCount++;
    }
    qCInfo(logCanRouting).noquote() << QString("loaded CAN message routing %1 (%2 entries)").arg(canRoutingPath).arg(msgCount);
    return true;
}

/*!
 * \brief   CloudConnectController::onCanFrameReceive
 * \details Unpacks a CAN frame and sends it to the dispatcher to reassemble the entire CAN message.
 * \param   frame - the received CAN frame
 */
void CloudConnectController::onCanFrameReceive(const QCanBusFrame &frame)
{
    _dispatcher.onFrameReceive(Can::CanId(frame.frameId()), frame.payload());
}

/*!
 * \brief   CloudConnectController::onCanMessageReceive
 * \details Applies the CAN mesg routing policy from CanRouting.ini.
 * \param   msg - the reassembled message
 */
void CloudConnectController::onCanMessageReceive(const Can::Message &msg)
{
    const auto it = _canRouting.constFind(msg.msgId);
    if (it == _canRouting.constEnd()) {
        qCInfo(logCanRouting).noquote() << QString("no action defined for %1 (0x%2), dropping")
            .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0'));
        return;
    }

    if (it->action == CanAction::Drop) {
        qCInfo(logCanRouting).noquote() << QString("%1 (0x%2) dropped")
            .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0'));
        return;
    }

    auto &[received, cachedMsg] = _canCache[msg.msgId];
    if (it->action == CanAction::SendDelta && received &&
        cachedMsg.data.chopped(1) == msg.data.chopped(1)) { // move chops to payload sending
        qCInfo(logCanRouting).noquote() << QString("%1 (0x%2) did not changed from previous, dropping")
            .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0'));
        return;
    }

    QByteArray payload;
    if (leahi::canMessageToProtobufByteArray(QDateTime::currentDateTime(), msg, payload)) {
        // Read the token out before it is advanced: the value handed to publish() is
        // what the delivery ledger reports, so logging the post-increment value made
        // every trace-back off by one.
        qint32 msgId;
        if (_mqttClient.publish(mqttTopic(it->topic), payload, msgId)) {
            // Logs the token so an unacknowledged publish reported by the MQTT delivery
            // ledger can be traced back to the message it carried.
            qCWarning(logMqtt).noquote() << QString("%1 (0x%2) published (msgId=%3)")
                .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')).arg(msgId);
        }
        else {
            // TODO: spool the message if publish fails
            qCWarning(logMqtt).noquote() << QString("could not publish %1 (0x%2), message lost")
                .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0'));
        }

        // NOTE: this records that the client ACCEPTED the message, not that AWS
        //       acknowledged it. Only the PUBACK counted in the MqttClient ledger
        //       means delivered.
        // received  = published;
    }
    else {
        qCWarning(logProto).noquote() << QString("could not serialize %1 (0x%2) message to protobuf, message dropped")
            .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0'));
    }

    // Cache the message only after it has been successfully sent.
    // Note: saving before successfully send may prevent a message of with that msgId from being
    //       sent until a delta message is received if SendDelta is specified for the msgId.
    cachedMsg = msg;
}

/*!
 * \brief   CloudConnectController::mqttTopic
 * \details Builds the MQTT topic for a message class.
 * \param   topic - message class from the CAN message routing INI
 * \return  Topic of the form {prefix}/{deviceId}/{suffix}.
 */
QString CloudConnectController::mqttTopic(CloudConnectFrame::Topic topic) const
{
    static const QMap<CloudConnectFrame::Topic, QString> suffixes = {
        { CloudConnectFrame::Topic::HighPriority,     QStringLiteral("high")      },
        { CloudConnectFrame::Topic::NormalPriority,   QStringLiteral("normal")    },
        { CloudConnectFrame::Topic::DeviceLogFile,    QStringLiteral("log")       },
        { CloudConnectFrame::Topic::TreatmentLogFile, QStringLiteral("tx_log")    },
        { CloudConnectFrame::Topic::CloudSyncLogFile, QStringLiteral("cs_log")    },
    };

    // return suffixes.value(topic, QStringLiteral("normal"));
    Q_UNUSED(topic)
    return QString("%1/%2/%3").arg(_topicPrefix, _deviceId, "clinical");
}

/*!
 * \brief   CloudConnectController::onCloudStateChanged
 * \details Re-establishes the echo subscriptions on every new session, and
 *          retires anything still awaiting an echo when one ends.
 * \param   connected - true once the broker has accepted the session
 * \note    Subscribing here rather than in connectToCloud() is required, not
 *          stylistic: open() only starts the attempt, and a SUBSCRIBE sent
 *          before the CONNACK is rejected by the client.
 */
void CloudConnectController::onCloudStateChanged(QMqttClient::ClientState state)
{
    switch (state) {
        case QMqttClient::Disconnected:
            // clear any pending messages
            // expireInFlight(true);
            break;
        case QMqttClient::Connecting:
            break;
        case QMqttClient::Connected:
            (void)_mqttClient.subscribe(mqttTopic(CloudConnectFrame::Topic::NormalPriority));
            break;
        default:
            break;
    }
}

/*!
 * \brief   CloudConnectController::onCloudMessageStatusChanged
 * \details Correlates one message delivered by the broker back to the publish
 *          that produced it and credits that msgId.
 * \param   msgId      - message identifier
 * \param   status      - new message status
 * \param   properties  - additional properties specified by the server/broker
 */
void CloudConnectController::onCloudMessageStatusChanged(qint32 id, QMqtt::MessageStatus status,
    const QMqttMessageStatusProperties &properties)
{
    Q_UNUSED(id) // SQ
    Q_UNUSED(properties) // SQ
    switch (status) {
        case QMqtt::MessageStatus::Unknown:
            qCInfo(logMqtt).noquote() << "message status changed to Unknown"; // SQ
            break;
        case QMqtt::MessageStatus::Published:
            qCInfo(logMqtt).noquote() << "message status changed to Published"; // SQ
            break;
        case QMqtt::MessageStatus::Acknowledged:
            qCInfo(logMqtt).noquote() << "message status changed to Acknowledged"; // SQ
            break;
        case QMqtt::MessageStatus::Received:
            qCInfo(logMqtt).noquote() << "message status changed to Received"; // SQ
            break;
        case QMqtt::MessageStatus::Released:
            qCInfo(logMqtt).noquote() << "message status changed to Released"; // SQ
            break;
        case QMqtt::MessageStatus::Completed:
            qCInfo(logMqtt).noquote() << "message status changed to Completed"; // SQ
            break;
        default:
            break;
    }
}

/*!
 * \brief   CloudConnectController::onCloudMessageReceived
 * \details Correlates one message delivered by the broker back to the publish
 *          that produced it and credits that msgId.
 * \param   topic   - topic the message was delivered on
 * \param   payload - serialised protobuf message
 * \note    The msgId is read from the message itself: every Leahi message carries
 *          Header at field 1, and Envelope exists to parse exactly that much
 *          without knowing the concrete message type.
 */
void CloudConnectController::onCloudMessageReceived(const QString &topic, const QByteArray &payload)
{
    qCInfo(logMqtt).noquote() << QString("received message with topic=%1").arg(topic);  // SQ

    // TODO: filter by topic

    leahi::messages::Envelope envelope;
    if (!envelope.ParseFromArray(payload.constData(), payload.size()) || !envelope.has_header()) {
        // Unknown fields parse cleanly in proto3, so a foreign payload usually
        // lands here as a header-less Envelope rather than as a parse failure.
        qCWarning(logMqtt).noquote() << QString("received mesg (size=%1) for topic %2 do not contain a valid protobuf payload, ignoring")
            .arg(payload.size()).arg(topic);
        return;
    }

    const leahi::messages::Header &header = envelope.header();
    const Can::MsgId msgId = Can::MsgId(header.msgid());

    // No serial here: Header carries only timestamp/sequence/msgId, and the device
    // identity travels in the topic (logged by handlePublishPacket) rather than the
    // payload. Printing the timestamp instead also proves updateHeader populated it.
    const qint64 headerMsecs = header.timestamp().seconds() * 1000 + header.timestamp().nanos() / 1000000;
    qCInfo(logProto).noquote() << QString("header: msgId=0x%1 (%2), seq=%3, timestamp=%4")
        .arg(QString("%1").arg(msgId, 4, 16, QChar('0')).toUpper(), leahi::msgIdString(static_cast<leahi::MsgId>(msgId)))
        .arg(header.sequence())
        .arg(QDateTime::fromMSecsSinceEpoch(headerMsecs).toString(Qt::ISODateWithMs));

    const std::string &typeName = leahi::msgIdToProtoName(msgId);
    if (typeName.empty()) {
        qCWarning(logProto).noquote() << QString("no protobuf mapping for message with msgId=0x%1, ignoring").arg(msgId, 4, 16, QChar('0'));
        return;
    }

    const google::protobuf::Descriptor *descriptor =
        google::protobuf::DescriptorPool::generated_pool()->FindMessageTypeByName(typeName);
    if (descriptor == nullptr) {
        qCWarning(logProto).noquote() << QString("no descriptor for protobuf message %1, ignoring").arg(QString::fromStdString(typeName));
        return;
    }

    google::protobuf::DynamicMessageFactory factory;
    std::unique_ptr<google::protobuf::Message> body(factory.GetPrototype(descriptor)->New());
    if (!body->ParseFromArray(payload.constData(), static_cast<int>(payload.size()))) {
        qCWarning(logProto).noquote() << QString("could not parse protobuf message %1, ignoring").arg(QString::fromStdString(typeName));
        return;
    }

    std::string json;
    google::protobuf::util::JsonPrintOptions options;
    options.add_whitespace = true;
    options.always_print_primitive_fields = true;
    google::protobuf::util::MessageToJsonString(*body, &json, options);
    qCDebug(logProto).noquote() << QString::fromStdString(typeName) << ":" << Qt::endl << QString::fromStdString(json);
}