/*!
 *
 * 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 <QAbstractSocket>
#include <QFile>
#include <QLoggingCategory>
#include <QSslConfiguration>
#include <QSslKey>
#include <QtMqtt/QMqttClient>
#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 paths it names. No client is created and no
 *          connection is attempted here; that is open()'s job.
 * \param   config - parameters for the session, retained for the next open()
 * \return  true if the configuration is usable, otherwise false
 * \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.
 */
bool MqttClient::init(const Config &config)
{
    if (_client != nullptr) {
        qCCritical(logMqtt).noquote() << "client already initialized";
        return false;
    }

    QSslConfiguration ssl = QSslConfiguration::defaultConfiguration();
    if (!config.caPath.isEmpty()) {
        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);
    }

    if (!config.certPath.isEmpty()) {
        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());
    }

    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();
    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);

    _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.
 * \note    The attempt is asynchronous, so a true return only means the client
 *          accepted the request. The outcome arrives via onStateChanged().
 */
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::messageSent, this, &MqttClient::onMessageSent);
    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;

    failPending();
    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   topic    - fully resolved MQTT topic
 * \param   payload  - serialised message bytes
 * \param   messageId  - caller token echoed back on acknowledgement, or -1 for none
 * \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, qint64 messageId)
{
    if (_client == nullptr || _client->state() != QMqttClient::Connected) {
        return false;
    }

    const qint32 id = _client->publish(QMqttTopicName(topic), payload, 1, false);
    if (id == -1) {
        qCWarning(logMqtt).noquote() << QString("client rejected publishing to topic %1").arg(topic);
        return false;
    }

    _pending.insert(id, messageId);
    // TODO: determine if messages waiting on a PUBACK need to be re-sent
    //       set a timeout interval to indicate when a message needs to be
    //       resent and have a timer periodically check _pending for messages
    //       that have met that threshold

    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: Receiving state change to Disconnected here means that connection
                //       was not done by user of this class calling close() and is most
                //       likely caused by externally factors, like the server side closing
                //       the connection.
                qCInfo(logMqtt).noquote() << "client disconnected";
                failPending();
                _client->deleteLater();
                _client = nullptr;
                // TODO: reconnect on disconnect
                //       set a flag reconnect flag indicating if connection should try to
                //       be re-established
                break;
            case QMqttClient::Connecting:
                qCInfo(logMqtt).noquote() << "client connecting";
                break;
            case QMqttClient::Connected:
                qCInfo(logMqtt).noquote() << "client connected";
                break;
        }
    }
}

/*!
 * \brief   MqttClient::onMessageSent
 * \details Clears one QoS 1 publish from _pending on PUBACK.
 * \param   id - QMqttClient's identifier for the acknowledged publish
 */
void MqttClient::onMessageSent(qint32 id)
{
    if (!_pending.contains(id)) {
        qCWarning(logMqtt).noquote() << QString("received PUBACK for an unknown published message with id %1").arg(id);
        return;
    }
    _pending.take(id);
}

/*!
 * \brief   MqttClient::failPending
 * \details Abandons every publish still awaiting a PUBACK.
 */
void MqttClient::failPending()
{
    if (_pending.isEmpty()) {
        return;
    }

    qCWarning(logMqtt).noquote() << QString("Failing %1 unacknowledged publishes").arg(_pending.size());
    _pending.clear();
}