thehub/indexer/AddressIndexer.cpp

/*
 * This file is part of the Flowee project
 * Copyright (C) 2019-2021 Tom Zander <tom@flowee.org>
 * Copyright (C) 2025 John Galt <johngaltbch@pm.me>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "AddressIndexer.h"
#include "Indexer.h"
#include <uint256.h>
#include <APIProtocol.h>
#include <streaming/MessageParser.h>
#include <Message.h>

#include <QTime>
#include <QElapsedTimer>
#include <qsettings.h>
#include <qsqlerror.h>
#include <qtimer.h>
#include <qvariant.h>
#include <qcoreapplication.h>

#include <fstream>

namespace {

QString valueFromSettings(const QSettings &settings, const QString &key, const QString &defvalue=QString()) {
    QVariant x = settings.value(QString("addressdb/") + key);
    if (x.isNull()) x = settings.value(key);
    if (x.isNull()) return defvalue;
    return x.toString();
}

QString addressTable(int index) {
    return QString("AddressUsage%1").arg(index, 2, 10, QChar('_'));
}

}

class TableSpecification
{
public:
    inline virtual ~TableSpecification() {}

    virtual bool queryTableExists(QSqlQuery &query, const QString &tableName) const {
        return query.exec("select count(*) from " + tableName);
    }

    // we only create one index type, so this API is assuming a lot.
    virtual bool createIndexIfNotExists(QSqlQuery &query, const QString &tableName) const {
        QString createIndexString("create index %1_index on %1 (address_row)");
        return query.exec(createIndexString.arg(tableName));
    }
};

class PostgresTables : public TableSpecification
{
public:
    bool queryTableExists(QSqlQuery &query, const QString &tableName) const override {
        bool ok = query.exec("select exists (select 1 from pg_tables where tablename='" + tableName.toLower()
                   + "' and schemaname='public')");
        if (!ok)
            return false;
        query.next();
        return query.value(0).toInt() == 1;
    }

    bool createIndexIfNotExists(QSqlQuery &query, const QString &tableName) const override {
        QString createIndexString("CREATE INDEX IF NOT EXISTS %1_index ON %1 (address_row)");
        return query.exec(createIndexString.arg(tableName.toLower()));
    }
};

class MySQLTables : public TableSpecification
{
public:
    bool queryTableExists(QSqlQuery &query, const QString &tableName) const override {
        query.exec("SELECT COUNT(table_name) FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='"+tableName+"'");
        query.first();
        return query.value(0).toInt() == 1;
    }

    bool createIndexIfNotExists(QSqlQuery &query, const QString &tableName) const override {
        QString createIndexString("CREATE INDEX %1_index ON %1 (address_row)");
        return query.exec(createIndexString.arg(tableName.toLower()));
    }
};

AddressIndexer::AddressIndexer(const boost::filesystem::path &basedir, Indexer *datasource)
    : m_addresses(basedir),
      m_basedir(QString::fromStdWString(basedir.wstring())),
      m_dataSource(datasource),
      m_flushRequested(0),
      m_postgresCsv(m_basedir + "/postgres.csv")
{
    assert(!m_basedir.isEmpty());
}

AddressIndexer::~AddressIndexer()
{
    delete m_spec;
}

void AddressIndexer::loadSetting(const QSettings &settings)
{
    auto db = valueFromSettings(settings, "db_driver");
    m_insertDb = QSqlDatabase::addDatabase(db, "insertConnection");
    if (!m_insertDb.isValid()) {
        if (QSqlDatabase::drivers().contains(db)) {
            logFatal().nospace() << "Failed to open a databse (" << db << "), missing libs?";
        } else {
            logFatal() << "The configured database is not known. Installed drivers are:";
            auto drivers = QSqlDatabase::drivers();
            logFatal() << std::list<QString>(drivers.begin(), drivers.end());
        }
        logCritical() << "Error reported:" << m_insertDb.lastError().text();
        throw std::runtime_error("Failed to read database");
    }
    delete m_spec;
    m_spec = nullptr;

    m_selectDb = QSqlDatabase::addDatabase(db, "selectConnection");
    logCritical().nospace() << "AddressIndexer database(" << db << ") "
                        << valueFromSettings(settings, "db_username")
                        << "@" << valueFromSettings(settings, "db_hostname")
                        << " DB: " << valueFromSettings(settings, "db_database");
    if (db == "QPSQL") {
        m_insertDb.setDatabaseName(valueFromSettings(settings, "db_database"));
        m_insertDb.setUserName(valueFromSettings(settings, "db_username"));
        m_insertDb.setPassword(valueFromSettings(settings, "db_password"));
        m_insertDb.setHostName(valueFromSettings(settings, "db_hostname"));
        m_insertDb.setPort(valueFromSettings(settings, "db_port", "5432").toInt());
        m_selectDb.setDatabaseName(valueFromSettings(settings, "db_database"));
        m_selectDb.setUserName(valueFromSettings(settings, "db_username"));
        m_selectDb.setPassword(valueFromSettings(settings, "db_password"));
        m_selectDb.setHostName(valueFromSettings(settings, "db_hostname"));
        m_selectDb.setPort(valueFromSettings(settings, "db_port", "5432").toInt());
        m_spec = new PostgresTables();
    } else if (db == "QMYSQL") {
        m_insertDb.setDatabaseName(valueFromSettings(settings, "db_database"));
        m_insertDb.setUserName(valueFromSettings(settings, "db_username"));
        m_insertDb.setPassword(valueFromSettings(settings, "db_password"));
        m_insertDb.setHostName(valueFromSettings(settings, "db_hostname"));
        m_insertDb.setPort(valueFromSettings(settings, "db_port", "3306").toInt());
        m_selectDb.setDatabaseName(valueFromSettings(settings, "db_database"));
        m_selectDb.setUserName(valueFromSettings(settings, "db_username"));
        m_selectDb.setPassword(valueFromSettings(settings, "db_password"));
        m_selectDb.setHostName(valueFromSettings(settings, "db_hostname"));
        m_selectDb.setPort(valueFromSettings(settings, "db_port", "3306").toInt());
        m_spec = new MySQLTables();
    } else if (db == "QSQLITE") {
        m_insertDb.setDatabaseName(m_basedir + "/addresses.db");
        m_selectDb.setDatabaseName(m_basedir + "/addresses.db");
    }

    if (!m_selectDb.isValid()) {
        logFatal() << "Failed opening the database-connection" << m_insertDb.lastError().text();
        throw std::runtime_error("Failed to open database connection");
    }
}

int AddressIndexer::blockheight()
{
    return m_height;
}

void AddressIndexer::blockFinished(int blockheight)
{
    assert(blockheight > m_height);
    m_height = blockheight;
    if (++m_uncommittedCount > 150000) {
        commitAllData();
        m_uncommittedCount = 0;
    }
}

void AddressIndexer::insert(const Streaming::ConstBuffer &outScriptHashed, int outputIndex, int blockHeight, int offsetInBlock)
{
    assert(outScriptHashed.size() == 32); // a sha256
    assert(QThread::currentThread() == this);

    const uint256 *address = reinterpret_cast<const uint256*>(outScriptHashed.begin());
    auto result = m_addresses.lookup(*address);
    if (result.db == -1)
        result = m_addresses.append(*address);
    assert(result.db >= 0);
    assert(result.row >= 0);

    if (result.db >= int(m_uncommittedData.size()))
        m_uncommittedData.resize(result.db + 1);

    std::deque<Entry> &data = m_uncommittedData[result.db];
    assert(outputIndex < 0x7FFF);
    data.push_back({(short) outputIndex, blockHeight, result.row, offsetInBlock});
    ++m_uncommittedCount;
}

std::vector<AddressIndexer::TxData> AddressIndexer::find(const uint256 &address) const
{
    std::vector<TxData> answer;
    auto result = m_addresses.lookup(address);
    if (result.db == -1)
        return answer;

    QSqlQuery query(m_selectDb);
    const QString select = QString("select DISTINCT offset_in_block, block_height, out_index "
                                  "FROM AddressUsage%1 "
                                  "WHERE address_row=:row "
                                  "ORDER BY block_height DESC").arg(result.db, 2, 10, QChar('_'));
    query.prepare(select);
    query.bindValue(":row", result.row);
    if (!query.exec()) {
        logFatal() << "Failed to select" << query.lastError().text();
        logDebug() << "Failed with" << select;
        QCoreApplication::exit(1);
    }
    const int size = query.size();
    if (size > 0)
        answer.reserve(size);
    while (query.next()) {
        TxData txData;
        txData.offsetInBlock = query.value(0).toInt();
        txData.blockHeight = query.value(1).toInt();
        txData.outputIndex = static_cast<short>(query.value(2).toInt());
        answer.push_back(txData);
    }
    return answer;
}

void AddressIndexer::createTables()
{
    if (m_spec == nullptr)
        m_spec = new TableSpecification(); // generic one.

    /* Tables
     * AddressUsage_N
     *  address_row     INTEGER   (the row that the hashStorage provided us with)
     *  block_height    INTEGER    \
     *  offset_in_block INTEGER    /-- together give the transaction
     *  out_index       INTEGER
     *
     * LastKnownState
     *  blockHeight    INTEGER
     */

    QSqlQuery query(m_insertDb);
    bool doInsert = false;
    if (!m_spec->queryTableExists(query, "LastKnownState")) {
        logInfo() << "Creating tables...";
        if (!query.exec("create table LastKnownState (blockheight INTEGER)")) {
            logFatal() << "Failed to create table" << query.lastError().text();
            throw std::runtime_error("Failed to create table");
        }
        doInsert = true;
    }
    if (doInsert || (query.next() && query.value(0).toInt() < 1)) {
        if (!query.exec("insert into LastKnownState values (0)")) {
            logFatal() << "Failed to insert row" << query.lastError().text();
            throw std::runtime_error("Failed to insert row");
        }
        if (!query.exec("create table IBD (busy INTEGER)")) {
            logFatal() << "Failed to create notificatoin table IBD" << query.lastError().text();
            throw std::runtime_error("Failed to create table");
        }
        if (!query.exec("insert into IBD values (42)")) {
            logFatal() << "Failed...";
            throw std::runtime_error("Failed...");
        }
    }
}

void AddressIndexer::run()
{
    // wait for database to come online.
    while (!isInterruptionRequested()) {
        assert(m_selectDb.isValid());
        if (m_insertDb.open() && m_selectDb.open()) {
            createTables();

            QSqlQuery query(m_selectDb);
            if (!query.exec("select blockheight from LastKnownState")) {
                logFatal() << "Failed to select blockHeight" << query.lastError().text();
                QCoreApplication::exit(1);
                return;
            }
            query.next();
            m_height = query.value(0).toInt();
            m_topOfChain = m_spec->queryTableExists(query, "IBD") ? InInitialSync : InitialSyncFinished;
            break;
        } else {
            sleep(3);
            logCritical() << "Waiting for SQL DB to come online.";
        }
    }
    logCritical() << "Address DB connected. Starting at block-height:" << m_height << (m_topOfChain ? "" : "in initial-sync");

    assert(m_dataSource);
    while (!isInterruptionRequested()) {
        int tipHeight;
        Message message = m_dataSource->nextBlock(blockheight() + 1, &tipHeight, m_uncommittedData.empty() ? ULONG_MAX : 20000);

        if (m_flushRequested.loadRelaxed() == 1) {
            commitAllData();
            m_flushRequested = 0;
        }
        if (message.body().size() == 0) // typically true if the flush was requested
            continue;

        int txOffsetInBlock = 0;
        int outputIndex = -1;
        int blockHeight = -1;

        Streaming::MessageParser parser(message.body());
        while (parser.next() == Streaming::FoundTag) {
            if (parser.tag() == Api::BlockChain::BlockHeight) {
                blockHeight = parser.intData();
                assert(blockHeight == m_height + 1);
            } else if (parser.tag() == Api::BlockChain::Separator) {
                txOffsetInBlock = 0;
                outputIndex = -1;
            } else if (parser.tag() == Api::BlockChain::Tx_OffsetInBlock) {
                txOffsetInBlock = parser.intData();
            } else if (parser.tag() == Api::BlockChain::Tx_Out_Index) {
                outputIndex = parser.intData();
            } else if (parser.tag() == Api::BlockChain::Tx_Out_ScriptHash) {
                assert(parser.dataLength() == 32);
                assert(outputIndex >= 0);
                assert(blockHeight >= 0);
                assert(txOffsetInBlock > 0);
                insert(parser.bytesDataBuffer(), outputIndex, blockHeight, txOffsetInBlock);
            }
        }
        assert(blockHeight >= 0);
        blockFinished(blockHeight);
        if (blockHeight == tipHeight) { // immediately flush when we processed the tip of the chain
            m_topOfChain.testAndSetAcquire(InInitialSync, FlushRequested);
            commitAllData();
        }
    }
}

void AddressIndexer::commitAllData()
{
    assert(QThread::currentThread() == this);
    if (m_height == -1) {
        assert(m_uncommittedData.empty());
        return;
    }

    QElapsedTimer time;
    time.start();
    int rowsInserted = 0;
    logCritical() << "AddressDB sending data to SQL DB";
    // create tables outside of transaction
    for (size_t db = 0; db < m_uncommittedData.size(); ++db) {
        const std::deque<Entry> &list = m_uncommittedData.at(db);
        if (!list.empty()) {
            const QString table = addressTable(db);
            QSqlQuery query(m_insertDb);
            if (!m_spec->queryTableExists(query, table)) {
                static QString q("create table %1 ("
                          "address_row INTEGER, "
                          "block_height INTEGER, offset_in_block INTEGER, out_index INTEGER)");
                if (!query.exec(q.arg(table))) {
                    logFatal() << "Failed to create table" << query.lastError().text();
                    QCoreApplication::exit(1);
                }
                // when creating a new table, set the index on the previous table.
                if (db > 0 && m_topOfChain.loadRelaxed() == InitialSyncFinished) {
                    if (!m_spec->createIndexIfNotExists(query, addressTable(db - 1))) {
                        logFatal() << "Failed to create index" << query.lastError().text();
                        QCoreApplication::exit(1);
                    }
                }
            }
        }
    }

    m_insertDb.transaction();
    for (size_t db = 0; db < m_uncommittedData.size(); ++db) {
        const std::deque<Entry> &list = m_uncommittedData.at(db);
        if (!list.empty()) {
            const QString table = addressTable(db);

            std::ofstream outFile(m_postgresCsv.toStdString(), std::ios_base::trunc);  // write mode

            if (outFile.is_open()) {
                for (auto entry : list) {
                    outFile << entry.row << "," << entry.height << "," << entry.offsetInBlock << "," << entry.outIndex << std::endl;
                }
                outFile.close();
            } else {
                logFatal() << "Unable to open file for writing: " << m_postgresCsv;
                return;
            }

            QSqlQuery sqlQuery(m_insertDb);
            QString query("COPY %1 (address_row, block_height, offset_in_block, out_index) FROM '%2' WITH (FORMAT csv)");
            query = query.arg(table, m_postgresCsv);
            if (!sqlQuery.exec(query)) {
                logFatal() << "Failed to start COPY command:" << sqlQuery.lastError().text();
                return;
            }

            rowsInserted += list.size();
        }
    }
    m_uncommittedData.clear();
    QSqlQuery query(m_insertDb);
    query.prepare("update LastKnownState set blockHeight=:bh");
    query.bindValue(":bh", m_height);
    if (!query.exec()) {
        logFatal() << "Failed to update blockheight" << query.lastError().text();
        logDebug() << " q" << query.lastQuery();
    }

    m_insertDb.commit();
    logCritical().nospace() << "AddressDB: SQL-DB took " << time.elapsed() << "ms to insert " << rowsInserted << " rows";

    if (m_topOfChain == FlushRequested) { // only ever run this code once per DB
        logCritical() << "Reached top of chain, creating indexes on our tables";
        for (int db = 0;; ++db) {
            if (!m_spec->queryTableExists(query, addressTable(db))) // found the last DB
                break;
            const QString tableName = addressTable(db);
            if (m_spec->createIndexIfNotExists(query, tableName))
                logInfo() << "Created index on SQL table" << tableName;
        }
        logCritical() << "Dropping table 'IBD' which was our indicator of initial sync";
        query.exec("drop table IBD");
        m_topOfChain = InitialSyncFinished;
    }
}