/*!
 *
 * 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    MessageSpool.cpp
 * \author  (original)  Stephen Quong
 * \date    (original)  28-Jul-2026
 *
 */
#include "MessageSpool.h"

#include <QDateTime>
#include <QDir>
#include <QFileInfo>
#include <QLoggingCategory>
#include <QMutexLocker>

#include <sqlite3.h>

Q_LOGGING_CATEGORY(lcSpool, "cloudconnect.spool")

namespace {

/*!
 * \brief   Stmt
 * \details RAII wrapper around a prepared statement. sqlite3_finalize() must
 *          run on every exit path; hand-written finalize calls are the source
 *          of the leaks seen in the legacy implementation, where an early
 *          return skipped them.
 */
class Stmt
{
public:
    Stmt(sqlite3 *db, const char *sql)
    {
        _rc = sqlite3_prepare_v2(db, sql, -1, &_stmt, nullptr);
    }

    ~Stmt()
    {
        if (_stmt != nullptr) {
            sqlite3_finalize(_stmt);
        }
    }

    Stmt(const Stmt &) = delete;
    Stmt &operator=(const Stmt &) = delete;

    bool isValid() const { return _rc == SQLITE_OK && _stmt != nullptr; }
    sqlite3_stmt *get() const { return _stmt; }

private:
    sqlite3_stmt *_stmt = nullptr;
    int _rc = SQLITE_ERROR;
};

/*!
 * \brief   columnText
 * \details Reads a TEXT column that may be NULL. sqlite3_column_text() returns
 *          nullptr for a NULL column, which would construct a QString from a
 *          null pointer if used unguarded.
 * \param   stmt    Prepared statement positioned on a row.
 * \param   column  Zero-based column index.
 * \return  The column value, or an empty QString when the column is NULL.
 */
QString columnText(sqlite3_stmt *stmt, int column)
{
    const auto *text = reinterpret_cast<const char *>(sqlite3_column_text(stmt, column));
    return (text != nullptr) ? QString::fromUtf8(text) : QString();
}

/*!
 * \brief   bindText
 * \details Binds a QString as UTF-8. SQLITE_TRANSIENT tells SQLite to copy the
 *          bytes immediately; the temporary QByteArray below dies at the end of
 *          this call, so SQLITE_STATIC would leave SQLite holding a dangling
 *          pointer.
 * \param   stmt   Prepared statement to bind against.
 * \param   index  One-based bind parameter index.
 * \param   value  String to bind.
 * \return  The sqlite3_bind_text() result code.
 */
int bindText(sqlite3_stmt *stmt, int index, const QString &value)
{
    const QByteArray utf8 = value.toUtf8();
    return sqlite3_bind_text(stmt, index, utf8.constData(), utf8.size(), SQLITE_TRANSIENT);
}

} // namespace

/*!
 * \brief   MessageSpool::MessageSpool
 * \details Constructor. Stores the configuration; the database is not touched
 *          until open() is called.
 * \param   config  Database path and retention window.
 */
MessageSpool::MessageSpool(Config config)
    : _config(std::move(config))
{
}

/*!
 * \brief   MessageSpool::~MessageSpool
 * \details Destructor. Closes the database if it is still open.
 */
MessageSpool::~MessageSpool()
{
    close();
}

/*!
 * \brief   MessageSpool::open
 * \details Creates the parent directory if needed, opens the database, applies
 *          the durability pragmas, and creates the schema.
 * \return  true when the spool is ready for use.
 */
bool MessageSpool::open()
{
    QMutexLocker locker(&_mutex);

    if (_db != nullptr) {
        return true;
    }

    const QString parentDir = QFileInfo(_config.dbPath).absolutePath();
    if (!QDir().mkpath(parentDir)) {
        qCCritical(lcSpool) << "Cannot create spool directory" << parentDir;
        return false;
    }

    if (sqlite3_open(_config.dbPath.toUtf8().constData(), &_db) != SQLITE_OK) {
        qCCritical(lcSpool) << "Cannot open" << _config.dbPath
                            << ":" << sqlite3_errmsg(_db);
        sqlite3_close(_db);
        _db = nullptr;
        return false;
    }

    // WAL keeps readers non-blocking during a write; busy_timeout absorbs the
    // contention between a drain and a concurrent insert.
    const bool pragmasOk = execSql("PRAGMA journal_mode=WAL;")
                        && execSql("PRAGMA synchronous=NORMAL;")
                        && execSql("PRAGMA busy_timeout=5000;");
    if (!pragmasOk) {
        qCCritical(lcSpool) << "Cannot apply pragmas to" << _config.dbPath;
        sqlite3_close(_db);
        _db = nullptr;
        return false;
    }

    const bool schemaOk = execSql(
        "CREATE TABLE IF NOT EXISTS spool ("
        "    id           INTEGER PRIMARY KEY AUTOINCREMENT,"
        "    timestamp_ms INTEGER NOT NULL,"
        "    topic        TEXT    NOT NULL,"
        "    payload      BLOB    NOT NULL,"
        "    sequence     INTEGER NOT NULL DEFAULT 0,"
        "    device_id    TEXT    NOT NULL DEFAULT '',"
        "    msg_id       TEXT    NOT NULL DEFAULT ''"
        ");")
        && execSql("CREATE INDEX IF NOT EXISTS idx_spool_timestamp ON spool(timestamp_ms);");
    if (!schemaOk) {
        qCCritical(lcSpool) << "Cannot create schema in" << _config.dbPath;
        sqlite3_close(_db);
        _db = nullptr;
        return false;
    }

    qCInfo(lcSpool) << "Opened" << _config.dbPath
                    << "retention" << _config.retentionHours << "hours";
    return true;
}

/*!
 * \brief   MessageSpool::close
 * \details Closes the database connection. Safe to call when already closed.
 */
void MessageSpool::close()
{
    QMutexLocker locker(&_mutex);

    if (_db != nullptr) {
        sqlite3_close(_db);
        _db = nullptr;
    }
}

/*!
 * \brief   MessageSpool::isOpen
 * \details Reports whether the database is currently open.
 * \return  true if open() has succeeded and close() has not run since.
 */
bool MessageSpool::isOpen() const
{
    QMutexLocker locker(&_mutex);
    return _db != nullptr;
}

/*!
 * \brief   MessageSpool::insert
 * \details Persists one message. Stamps the current time when the caller left
 *          timestampMs at 0.
 * \param   msg  Message to persist.
 * \return  The assigned rowid, or -1 on failure.
 */
qint64 MessageSpool::insert(const SpooledMessage &msg)
{
    QMutexLocker locker(&_mutex);

    if (_db == nullptr) {
        qCWarning(lcSpool) << "Insert on a closed spool";
        return -1;
    }

    Stmt stmt(_db,
        "INSERT INTO spool(timestamp_ms, topic, payload, sequence, device_id, msg_id)"
        " VALUES(?,?,?,?,?,?);");
    if (!stmt.isValid()) {
        qCCritical(lcSpool) << "Insert prepare failed:" << sqlite3_errmsg(_db);
        return -1;
    }

    const qint64 timestampMs =
        (msg.timestampMs != 0) ? msg.timestampMs : QDateTime::currentMSecsSinceEpoch();

    sqlite3_bind_int64(stmt.get(), 1, timestampMs);
    bindText(stmt.get(), 2, msg.topic);
    sqlite3_bind_blob(stmt.get(), 3, msg.payload.constData(), msg.payload.size(), SQLITE_TRANSIENT);
    sqlite3_bind_int64(stmt.get(), 4, static_cast<qint64>(msg.sequence));
    bindText(stmt.get(), 5, msg.deviceId);
    bindText(stmt.get(), 6, msg.msgId);

    if (sqlite3_step(stmt.get()) != SQLITE_DONE) {
        qCCritical(lcSpool) << "Insert failed:" << sqlite3_errmsg(_db);
        return -1;
    }

    return sqlite3_last_insert_rowid(_db);
}

/*!
 * \brief   MessageSpool::fetchBatch
 * \details Returns the oldest messages first. Ordering is by the AUTOINCREMENT
 *          id rather than by timestamp, so insertion order is preserved even
 *          when two messages share a millisecond or the clock steps backwards.
 * \param   limit  Maximum number of messages to return.
 * \return  Up to limit messages, oldest first, each with spoolId populated.
 */
QVector<SpooledMessage> MessageSpool::fetchBatch(quint32 limit)
{
    QMutexLocker locker(&_mutex);

    QVector<SpooledMessage> batch;
    if (_db == nullptr || limit == 0) {
        return batch;
    }

    Stmt stmt(_db,
        "SELECT id, timestamp_ms, topic, payload, sequence, device_id, msg_id"
        " FROM spool ORDER BY id ASC LIMIT ?;");
    if (!stmt.isValid()) {
        qCCritical(lcSpool) << "Fetch prepare failed:" << sqlite3_errmsg(_db);
        return batch;
    }

    sqlite3_bind_int(stmt.get(), 1, static_cast<int>(limit));

    while (sqlite3_step(stmt.get()) == SQLITE_ROW) {
        SpooledMessage msg;
        msg.spoolId     = sqlite3_column_int64(stmt.get(), 0);
        msg.timestampMs = sqlite3_column_int64(stmt.get(), 1);
        msg.topic       = columnText(stmt.get(), 2);

        const auto *blob = static_cast<const char *>(sqlite3_column_blob(stmt.get(), 3));
        const int blobSize = sqlite3_column_bytes(stmt.get(), 3);
        if (blob != nullptr && blobSize > 0) {
            msg.payload = QByteArray(blob, blobSize);
        }

        msg.sequence = static_cast<quint16>(sqlite3_column_int64(stmt.get(), 4));
        msg.deviceId = columnText(stmt.get(), 5);
        msg.msgId    = columnText(stmt.get(), 6);

        batch.append(msg);
    }

    return batch;
}

/*!
 * \brief   MessageSpool::remove
 * \details Deletes one row. Call only once delivery of that message has been
 *          confirmed; removing earlier forfeits the store-and-forward guarantee.
 * \param   spoolId  Rowid returned by insert().
 * \return  true when the row was deleted.
 */
bool MessageSpool::remove(qint64 spoolId)
{
    QMutexLocker locker(&_mutex);

    if (_db == nullptr) {
        return false;
    }

    Stmt stmt(_db, "DELETE FROM spool WHERE id = ?;");
    if (!stmt.isValid()) {
        qCCritical(lcSpool) << "Remove prepare failed:" << sqlite3_errmsg(_db);
        return false;
    }

    sqlite3_bind_int64(stmt.get(), 1, spoolId);

    if (sqlite3_step(stmt.get()) != SQLITE_DONE) {
        qCCritical(lcSpool) << "Remove failed for id" << spoolId
                            << ":" << sqlite3_errmsg(_db);
        return false;
    }

    return sqlite3_changes(_db) > 0;
}

/*!
 * \brief   MessageSpool::depth
 * \details Counts the messages currently held.
 * \return  Number of spooled messages.
 */
quint64 MessageSpool::depth() const
{
    return scalarQuery("SELECT COUNT(*) FROM spool;");
}

/*!
 * \brief   MessageSpool::sizeBytes
 * \details Sums the stored payload bytes.
 * \return  Total payload size in bytes, excluding SQLite overhead.
 */
quint64 MessageSpool::sizeBytes() const
{
    return scalarQuery("SELECT COALESCE(SUM(LENGTH(payload)),0) FROM spool;");
}

/*!
 * \brief   MessageSpool::purgeExpired
 * \details Deletes messages older than the configured retention window. Time
 *          is the only bound on spool growth, so this must be called
 *          periodically for the disk budget to hold.
 * \return  The number of messages purged, or -1 on failure.
 */
int MessageSpool::purgeExpired()
{
    QMutexLocker locker(&_mutex);

    if (_db == nullptr) {
        return -1;
    }

    const qint64 cutoffMs = QDateTime::currentMSecsSinceEpoch()
        - static_cast<qint64>(_config.retentionHours) * 3600LL * 1000LL;

    Stmt stmt(_db, "DELETE FROM spool WHERE timestamp_ms < ?;");
    if (!stmt.isValid()) {
        qCCritical(lcSpool) << "Purge prepare failed:" << sqlite3_errmsg(_db);
        return -1;
    }

    sqlite3_bind_int64(stmt.get(), 1, cutoffMs);

    if (sqlite3_step(stmt.get()) != SQLITE_DONE) {
        qCCritical(lcSpool) << "Purge failed:" << sqlite3_errmsg(_db);
        return -1;
    }

    const int purged = sqlite3_changes(_db);
    if (purged > 0) {
        qCInfo(lcSpool) << "Purged" << purged << "messages older than"
                        << _config.retentionHours << "hours";
    }

    return purged;
}

/*!
 * \brief   MessageSpool::execSql
 * \details Runs a statement that returns no rows, logging any error.
 * \param   sql  Statement to execute.
 * \return  true when the statement succeeded.
 * \note    Callers must already hold the mutex.
 */
bool MessageSpool::execSql(const char *sql)
{
    char *error = nullptr;
    if (sqlite3_exec(_db, sql, nullptr, nullptr, &error) != SQLITE_OK) {
        qCCritical(lcSpool) << "SQL error:" << (error != nullptr ? error : "unknown");
        sqlite3_free(error);
        return false;
    }
    return true;
}

/*!
 * \brief   MessageSpool::scalarQuery
 * \details Runs a statement whose first row and column hold a single count.
 * \param   sql  Statement to execute.
 * \return  The scalar value, or 0 when the spool is closed or the query fails.
 */
quint64 MessageSpool::scalarQuery(const char *sql) const
{
    QMutexLocker locker(&_mutex);

    if (_db == nullptr) {
        return 0;
    }

    Stmt stmt(_db, sql);
    if (!stmt.isValid()) {
        return 0;
    }

    quint64 value = 0;
    if (sqlite3_step(stmt.get()) == SQLITE_ROW) {
        value = static_cast<quint64>(sqlite3_column_int64(stmt.get(), 0));
    }

    return value;
}