/*!
 *
 * Copyright (c) 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    MqttClient.cpp
 * \author  (original)  Stephen Quong
 * \date    (original)  30-Jul-2026
 *
 */
#include <algorithm>

#include <QAbstractSocket>
#include <QFile>
#include <QLoggingCategory>
#include <QSslConfiguration>
#include <QSslKey>
#include <QtMqtt/QMqttClient>
#include <QtMqtt/QMqttSubscription>
#include <QtMqtt/QMqttTopicFilter>
#include <QtMqtt/QMqttTopicName>

#include "MqttClient.h"

Q_LOGGING_CATEGORY(logMqtt, "cloudconnect.mqtt")

/*!
 * \brief   MqttClient::MqttClient
 * \details Constructor
 * \param   parent - optional QObject parent
 */
MqttClient::MqttClient(QObject *parent) :
    QObject(parent)
{
}

/*!
 * \brief   MqttClient::~MqttClient
 * \details Ends any open session.
 */
MqttClient::~MqttClient()
{
    close();
}

/*!
 * \brief   MqttClient::init
 * \details Validates the configuration and builds the TLS configuration from
 *          the certificate and key paths. Connection to server is not attempted
 *          until open() is called.
 * \note    This function does nothing if a client connection already exists.
 * \note    keyPath is required. caPath and certPath are optional: without
 *          caPath the system CA set verifies the broker, and without certPath
 *          the session is server-authenticated only rather than mTLS.
 * \param   config - parameters for the session, retained for the next open()
 * \return  true if the configuration is usable, otherwise false
 */
bool MqttClient::init(const Config &config)
{
    if (_client != nullptr) {
        qCCritical(logMqtt).noquote() << "client already initialized";
        return false;
    }

    QSslConfiguration ssl = QSslConfiguration::defaultConfiguration();
    if (QFile::exists(config.caPath)) {
        const QList<QSslCertificate> cas = QSslCertificate::fromPath(config.caPath);
        if (cas.isEmpty()) {
            qCCritical(logMqtt).noquote() << QString("%1 is not a CA certificate").arg(config.caPath);
            return false;
        }
        ssl.setCaCertificates(cas);
    }
    else {
        qCCritical(logMqtt).noquote() << QString("CA certificate file %1 does not exists").arg(config.caPath);
        return false;
    }

    if (QFile::exists(config.certPath)) {
        const QList<QSslCertificate> certs = QSslCertificate::fromPath(config.certPath);
        if (certs.isEmpty()) {
            qCCritical(logMqtt).noquote() << QString("%1 is not a certificate").arg(config.certPath);
            return false;
        }
        ssl.setLocalCertificate(certs.first());
    }
    else {
        qCCritical(logMqtt).noquote() << QString("certificate file %1 does not exists").arg(config.certPath);
        return false;
    }

    if (QFile::exists(config.keyPath)) {
        QFile keyFile(config.keyPath);
        if (!keyFile.open(QIODevice::ReadOnly)) {
            qCCritical(logMqtt).noquote() << QString("could not read private key %1").arg(config.keyPath);
            return false;
        }
        const QByteArray pem = keyFile.readAll();
        keyFile.close();
        QSslKey key(pem, QSsl::Rsa, QSsl::Pem);
        if (key.isNull()) {
            key = QSslKey(pem, QSsl::Ec, QSsl::Pem);
        }
        if (key.isNull()) {
            qCCritical(logMqtt).noquote() << QString("private key %1 is not a PEM, RSA, or EC key").arg(config.keyPath);
            return false;
        }
        ssl.setPrivateKey(key);
    }
    else {
        qCCritical(logMqtt).noquote() << QString("key file %1 does not exist").arg(config.keyPath);
        return false;
    }

    _sslConfig = ssl;
    _config = config;

    return true;
}

/*!
 * \brief   MqttClient::open
 * \details Creates the QMqttClient from the configuration validated by init(),
 *          wires its signals and starts a TLS connection attempt.
 * \return  true if the attempt was started
 * \note    Must be called on this object's thread; the client's socket belongs
 *          to the thread that constructs it.
 */
bool MqttClient::open()
{
    if (_client != nullptr) {
        qCCritical(logMqtt).noquote() << "client connection already opened";
        return false;
    }

    _client = new QMqttClient(this);
    _client->setProtocolVersion(QMqttClient::MQTT_3_1_1);
    _client->setHostname(_config.endpoint);
    _client->setPort(_config.port);
    _client->setClientId(_config.clientId);
    _client->setKeepAlive(_config.keepAliveSecs);
    _client->setCleanSession(_config.cleanSession);

    connect(_client, &QMqttClient::stateChanged, this, &MqttClient::onStateChanged);
    connect(_client, &QMqttClient::messageStatusChanged, this, &MqttClient::didMessageStatusChanged);
    // Fires for every delivery on this connection, whichever subscription matched.
    connect(_client, &QMqttClient::messageReceived, this,
        [this](const QByteArray &message, const QMqttTopicName &topic) {
            Q_EMIT didMessageReceived(topic.name(), message);
        }
    );
    connect(_client, &QMqttClient::errorChanged, this,
        [](QMqttClient::ClientError error) {
            if (error != QMqttClient::NoError) {
                qCWarning(logMqtt).noquote() << "client error" << error;
            }
        }
    );

    qCInfo(logMqtt).noquote() << QString("attempting to connect to %1:%2 as %3")
        .arg(_config.endpoint).arg(_config.port).arg(_config.clientId);
    _client->connectToHostEncrypted(_sslConfig);

    return true;
}

/*!
 * \brief   MqttClient::close
 * \details Ends the session and discards the client. Sends a DISCONNECT only
 *          when the socket drains, otherwise aborts it.
 */
void MqttClient::close()
{
    if (_client == nullptr) {
        return;
    }

    // clear the handle and disconnect signals first so when client state changes
    // to Disconnected it does not trigger a reconnect in onStateChanged.
    QMqttClient *client = _client;
    _client = nullptr;

    // stop any signals coming from client before modifying it
    client->disconnect(this);

    // flush to force the socket to send the Disconnect so bytesToWrite hopefully
    // does not have any pending data to send (bytesToWrite() == 0)
    auto *socket = qobject_cast<QAbstractSocket *>(client->transport());
    if (socket != nullptr) {
        socket->flush();
    }

    // if bytesToWrite is zero and state is not Disconnected, then send Disconnect
    if (socket == nullptr || socket->bytesToWrite() == 0) {
        if (client->state() != QMqttClient::Disconnected) {
            client->disconnectFromHost();
        }
    }
    else {
        // if there are pending bytes to write, then disconnectFromHost() blocks 30s
        // waiting for the bytes to be sent, just abort, which forcees the write
        // to fail immediately and skip the wait
        socket->abort();
    }

    // clean up the client
    client->deleteLater();
}

/*!
 * \brief   MqttClient::publish
 * \details Hands one message to the client for QoS 1 delivery.
 * \param[in]   topic   - fully resolved MQTT topic
 * \param[in]   payload - serialised message bytes
 * \param[out]  msgId   - identifier assigned by QMqttClient::publish to track message
 * \param[in]   qos     - QoS level (default=1)
 * \return  true if the client accepted the message.
 * \note    Return status does NOT indicate delivery, only that the client
 *          accepted the message for sending. The PUBACK that confirms delivery
 *          arrives later in onMessageSent().
 */
bool MqttClient::publish(const QString &topic, const QByteArray &payload, qint32 &msgId, quint8 qos)
{
    if (_client == nullptr || _client->state() != QMqttClient::Connected) {
        msgId = -1;
        return false;
    }

    msgId = _client->publish(QMqttTopicName(topic), payload, qos, false);
    if (msgId != -1) {
        // qCDebug(logMqtt).noquote() << QString("publish accepted (msgId=%1): topic=%2, qos=%3")
        //     .arg(msgId).arg(topic).arg(qos);
        return true;
    }
    else {
        qCWarning(logMqtt).noquote() << QString("client rejected publish: topic=%1, qos=%2")
            .arg(topic).arg(qos);
        return false;
    }
}

/*!
 * \brief   MqttClient::subscribe
 * \details Subscribes this session to a topic filter. Deliveries arrive on
 *          didMessageReceived(), which is not per-subscription: it carries the
 *          topic so the receiver can demultiplex.
 * \param   topicFilter - MQTT topic filter, wildcards allowed
 * \param   qos         - requested maximum QoS for deliveries on this filter
 * \return  true if the client accepted the request.
 * \note    Return status does NOT mean the subscription is active. The SUBACK
 *          arrives later; until it does, the broker may drop matching messages.
 *          The granted QoS can also be lower than requested, which is not an error.
 * \note    Subscriptions live on the QMqttClient and die with it, so they must be
 *          re-established after every reconnect (see didConnectionChange()).
 */
bool MqttClient::subscribe(const QString &topicFilter, quint8 qos)
{
    if (_client == nullptr || _client->state() != QMqttClient::Connected) {
        qCWarning(logMqtt).noquote() << QString("cannot subscribe to %1 while disconnected").arg(topicFilter);
        return false;
    }

    QMqttSubscription *subscription = _client->subscribe(QMqttTopicFilter(topicFilter), qos);
    if (subscription == nullptr) {
        qCWarning(logMqtt).noquote() << QString("client rejected subscribing to topic %1").arg(topicFilter);
        return false;
    }

    // A subscription AWS IoT refuses (a filter outside the policy's iot:Receive or
    // iot:Subscribe resources) fails here rather than at publish time, and without
    // this log the only symptom is that no message ever arrives.
    connect(subscription, &QMqttSubscription::stateChanged, this,
        [topicFilter](QMqttSubscription::SubscriptionState state) {
            if (state == QMqttSubscription::Subscribed) {
                qCInfo(logMqtt).noquote() << QString("subscribed to %1").arg(topicFilter);
            }
            else if (state == QMqttSubscription::Error) {
                qCWarning(logMqtt).noquote() << QString("subscription to topic %1 failed").arg(topicFilter);
            }
        }
    );

    return true;
}

/*!
 * \brief   MqttClient::onStateChanged
 * \details Handle QMqttClient state changes and perform additional actions as
 *          a result of state transition, if applicable.
 */
void MqttClient::onStateChanged()
{
    if (_client != nullptr) {
        switch (_client->state()) {
            case QMqttClient::Disconnected:
                // NOTE: close() disconnects all signals from _client so further signals,
                //       like stateChanged, will no longer be received
                qCInfo(logMqtt).noquote() << "client disconnected";
                _client->deleteLater();
                _client = nullptr;
                // TODO: reconnect on disconnect
                Q_EMIT didStateChanged(QMqttClient::Disconnected);
                break;
            case QMqttClient::Connecting:
                qCInfo(logMqtt).noquote() << "client connecting";
                Q_EMIT didStateChanged(_client->state());
                break;
            case QMqttClient::Connected:
                qCInfo(logMqtt).noquote() << "client connected";
                // Subscriptions did not survive the previous client, so this is where
                // a receiver re-establishes them.
                Q_EMIT didStateChanged(_client->state());
                break;
        }
    }
}