colnod-connector/src/colnod/syncmanager.cpp

#include "syncmanager.h"


Q_LOGGING_CATEGORY(syncmanager, "sync.manager");


SyncManager::SyncManager(bool dryMode)
    :mDryMode(dryMode)
{

}

bool SyncManager::init()
{
    database = std::make_shared<DatabaseAPI>(&dbMan, mDryMode);

    if (database->open()){
        mlastSync.set_timestamp(database->getLastSync());

        colnod = std::make_shared<ColnodAPI>(&reqMan, mlastSync.timestamp(), mDryMode);

        // -1 --> some error with getting data from DB
        if (mlastSync.timestamp() >= 0){
            return true;
        }
    }
    return false;
}


void SyncManager::run()
{
    qint64 syncBegin = database->getNow();

    qInfo(syncmanager) << EQUALS;
    qInfo(syncmanager) << "SYNCHRONIZATION BEGIN";

    database->initNewTokens();

    downloadData();
    connect(this, &SyncManager::dataReady, [=](){
        prepareData();

        connect(this, &SyncManager::dataProcessFinished, [=](){
            updateLastUpdate(syncBegin);
            connect(this, &SyncManager::syncFinished, [=](){
                database->setLastSync(syncBegin);

                qInfo(syncmanager) << "Update last synchronization time to time of this sync ("
                                   <<  QDateTime::fromMSecsSinceEpoch(syncBegin).toString("yyyy-MM-dd hh:mm:ss.zzz")
                                   << ")";
                qInfo(syncmanager) << EQUALS;
                qInfo(syncmanager) << "SYNCHRONIZATION FINISHED";
                qInfo(syncmanager) << EQUALS;

                QCoreApplication::exit();
            });
        });

        processData();
    });
}

void SyncManager::downloadData()
{
    qInfo(syncmanager) << EQUALS;
    qInfo(syncmanager) << "DOWNLOADING DATA -> BEGIN";
    qInfo(syncmanager) << DASH;
    qInfo(syncmanager) << "Modified data since last synchronization (" << mlastSync.toDateString() << "):";
    qInfo(syncmanager) << DASH;

    connect(colnod.get(), &ColnodAPI::downloadDataFinished, this, &SyncManager::saveColnodData);
    connect(database.get(), &DatabaseAPI::downloadDataFinished, this, &SyncManager::saveDatabaseData);
    database->downloadData();
    colnod->downloadData();
}

// finds conflicts -> changes on both sides (colnod and DB), apply newer change
void SyncManager::prepareData()
{
    qInfo(syncmanager) << EQUALS;
    qInfo(syncmanager) << "PREPARING DATA -> BEGIN";
    qInfo(syncmanager) << DASH;
    qInfo(syncmanager) << "Resolve conflicts ( changes on same item on both Colnod and DB ):";
    qInfo(syncmanager) << DASH;


    colnodDataToProcess = std::make_shared<Data>(*colnodData);
    databaseDataToProcess = std::make_shared<Data>(*databaseData);

    QList<QString> conflicts;
    QList<QString> alreadySynced;

    for (const auto &colnodToken : colnodDataToProcess->updatedTokens) {
        if (databaseDataToProcess->updatedTokens.contains(colnodToken->id())) {
            if (databaseDataToProcess->updatedTokens.value(colnodToken->id()) > colnodToken) {
                conflicts.append(colnodToken->id());
            }
            continue;
        }
        auto token = database->getLastUpdate(colnodToken->id());
        if ( !token )
            if (token == colnodToken)
                alreadySynced.append(colnodToken->id());
    }
    for (const auto &id : conflicts)
        colnodDataToProcess->updatedTokens.remove(id);
    for (const auto &id : alreadySynced)
        colnodDataToProcess->updatedTokens.remove(id);

    qInfo(syncmanager) << TAB << "-> COLNOD DATA - CHANGES:";
    qInfo(syncmanager) << TAB << TAB << conflicts.size() << "TOKENS WERE MODIFIED IN DATABASE";
    qInfo(syncmanager) << TAB << TAB << alreadySynced.size() << "TOKENS WERE ALREADY SYNCED";
    qInfo(syncmanager) << DASH;

    conflicts.clear();

    for (const auto &databaseToken : databaseData->updatedTokens) {
        if (colnodDataToProcess->updatedTokens.contains(databaseToken->id())) {
            if (databaseToken > colnodDataToProcess->updatedTokens.value(databaseToken->id()))
                conflicts.append(databaseToken->id());
        }
    }
    for (const auto &id : conflicts)
        colnodDataToProcess->updatedTokens.remove(id);

    qInfo(syncmanager) << TAB << "-> DATABASE DATA - CHANGES:";
    qInfo(syncmanager) << TAB << TAB << conflicts.size() << "TOKENS WERE MODIFIED IN COLNOD";
    qInfo(syncmanager) << DASH;

    qInfo(syncmanager) << "PREPARING DATA -> FINISHED";
    qInfo(syncmanager) << EQUALS;
}

void SyncManager::processData()
{
    qInfo(syncmanager) << EQUALS;
    qInfo(syncmanager) << "PROCESSING DATA -> BEGIN";
    qInfo(syncmanager) << DASH;
    qInfo(syncmanager) << "Push changes to both Colnod ad Database:";
    qInfo(syncmanager) << DASH;

    database->processData(colnodDataToProcess);
    connect(colnod.get(), &ColnodAPI::dataProcessFinished, this, &SyncManager::dataProcessFinished);
    colnod->processData(databaseDataToProcess);

    qInfo(syncmanager) << "PROCESSING DATA -> FINISHED";
    qInfo(syncmanager) << EQUALS;
}

void SyncManager::updateLastUpdate(qint64 syncBegin)
{
    Timestamp time;
    time.set_timestamp(syncBegin);
    colnod->getTokensUpdated(time);
    connect(colnod.get(), &ColnodAPI::partOfDataReady, [=](std::shared_ptr<Data> data){
        database->setLastUpdate(data->updatedTokens);
        emit syncFinished();
    });
}

void SyncManager::areDataReady()
{
    if (++dataAmount == 2) {

        qInfo(syncmanager) << "DOWNLOADING DATA -> FINISHED";
        qInfo(syncmanager) << EQUALS;

        emit dataReady();
    }
}

void SyncManager::saveColnodData(const std::shared_ptr<Data> &data)
{
    colnodData = data;
    areDataReady();
}

void SyncManager::saveDatabaseData(const std::shared_ptr<Data> &data)
{
    databaseData = data;
    areDataReady();
}