path: root/common/synchronizer.cpp

/*
 * Copyright (C) 2016 Christian Mollekopf <mollekopf@kolabsys.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) version 3, or any
 * later version accepted by the membership of KDE e.V. (or its
 * successor approved by the membership of KDE e.V.), which shall
 * act as a proxy defined in Section 6 of version 3 of the license.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "synchronizer.h"

#include "definitions.h"
#include "commands.h"
#include "bufferutils.h"
#include "synchronizerstore.h"
#include "datastorequery.h"
#include "createentity_generated.h"
#include "modifyentity_generated.h"
#include "deleteentity_generated.h"
#include "flush_generated.h"
#include "notification_generated.h"

SINK_DEBUG_AREA("synchronizer")

using namespace Sink;

Synchronizer::Synchronizer(const Sink::ResourceContext &context)
    : ChangeReplay(context),
    mLogCtx{"synchronizer"},
    mResourceContext(context),
    mEntityStore(Storage::EntityStore::Ptr::create(mResourceContext, mLogCtx)),
    mSyncStorage(Sink::storageLocation(), mResourceContext.instanceId() + ".synchronization", Sink::Storage::DataStore::DataStore::ReadWrite),
    mSyncInProgress(false)
{
    SinkTrace() << "Starting synchronizer: " << mResourceContext.resourceType << mResourceContext.instanceId();
}

Synchronizer::~Synchronizer()
{

}

void Synchronizer::setup(const std::function<void(int commandId, const QByteArray &data)> &enqueueCommandCallback, MessageQueue &mq)
{
    mEnqueue = enqueueCommandCallback;
    mMessageQueue = &mq;
}

void Synchronizer::enqueueCommand(int commandId, const QByteArray &data)
{
    Q_ASSERT(mEnqueue);
    mEnqueue(commandId, data);
}

Storage::EntityStore &Synchronizer::store()
{
    mEntityStore->startTransaction(Sink::Storage::DataStore::ReadOnly);
    return *mEntityStore;
}

SynchronizerStore &Synchronizer::syncStore()
{
    if (!mSyncStore) {
        mSyncStore = QSharedPointer<SynchronizerStore>::create(syncTransaction());
    }
    return *mSyncStore;
}

void Synchronizer::createEntity(const QByteArray &sinkId, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject)
{
    // These changes are coming from the source
    const auto replayToSource = false;
    flatbuffers::FlatBufferBuilder entityFbb;
    mResourceContext.adaptorFactory(bufferType).createBuffer(domainObject, entityFbb);
    flatbuffers::FlatBufferBuilder fbb;
    auto entityId = fbb.CreateString(sinkId.toStdString());
    auto type = fbb.CreateString(bufferType.toStdString());
    auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityFbb.GetBufferPointer(), entityFbb.GetSize());
    auto location = Sink::Commands::CreateCreateEntity(fbb, entityId, type, delta, replayToSource);
    Sink::Commands::FinishCreateEntityBuffer(fbb, location);
    enqueueCommand(Sink::Commands::CreateEntityCommand, BufferUtils::extractBuffer(fbb));
}

void Synchronizer::modifyEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType, const Sink::ApplicationDomain::ApplicationDomainType &domainObject, const QByteArray &newResource, bool remove)
{
    // FIXME removals
    QByteArrayList deletedProperties;
    // These changes are coming from the source
    const auto replayToSource = false;
    flatbuffers::FlatBufferBuilder entityFbb;
    mResourceContext.adaptorFactory(bufferType).createBuffer(domainObject, entityFbb);
    flatbuffers::FlatBufferBuilder fbb;
    auto entityId = fbb.CreateString(sinkId.toStdString());
    auto modifiedProperties = BufferUtils::toVector(fbb, domainObject.changedProperties());
    auto deletions = BufferUtils::toVector(fbb, deletedProperties);
    auto type = fbb.CreateString(bufferType.toStdString());
    auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityFbb.GetBufferPointer(), entityFbb.GetSize());
    auto resource = newResource.isEmpty() ? 0 : fbb.CreateString(newResource.constData());
    auto location = Sink::Commands::CreateModifyEntity(fbb, revision, entityId, deletions, type, delta, replayToSource, modifiedProperties, resource, remove);
    Sink::Commands::FinishModifyEntityBuffer(fbb, location);
    enqueueCommand(Sink::Commands::ModifyEntityCommand, BufferUtils::extractBuffer(fbb));
}

void Synchronizer::deleteEntity(const QByteArray &sinkId, qint64 revision, const QByteArray &bufferType)
{
    // These changes are coming from the source
    const auto replayToSource = false;
    flatbuffers::FlatBufferBuilder fbb;
    auto entityId = fbb.CreateString(sinkId.toStdString());
    // This is the resource type and not the domain type
    auto type = fbb.CreateString(bufferType.toStdString());
    auto location = Sink::Commands::CreateDeleteEntity(fbb, revision, entityId, type, replayToSource);
    Sink::Commands::FinishDeleteEntityBuffer(fbb, location);
    enqueueCommand(Sink::Commands::DeleteEntityCommand, BufferUtils::extractBuffer(fbb));
}

void Synchronizer::scanForRemovals(const QByteArray &bufferType, const std::function<void(const std::function<void(const QByteArray &key)> &callback)> &entryGenerator, std::function<bool(const QByteArray &remoteId)> exists)
{
    entryGenerator([this, bufferType, &exists](const QByteArray &sinkId) {
        const auto remoteId = syncStore().resolveLocalId(bufferType, sinkId);
        SinkTrace() << "Checking for removal " << sinkId << remoteId;
        // If we have no remoteId, the entity hasn't been replayed to the source yet
        if (!remoteId.isEmpty()) {
            if (!exists(remoteId)) {
                SinkTrace() << "Found a removed entity: " << sinkId;
                deleteEntity(sinkId, mEntityStore->maxRevision(), bufferType);
            }
        }
    });
}

void Synchronizer::scanForRemovals(const QByteArray &bufferType, std::function<bool(const QByteArray &remoteId)> exists)
{
    scanForRemovals(bufferType,
        [this, &bufferType](const std::function<void(const QByteArray &)> &callback) {
            store().readAllUids(bufferType, [callback](const QByteArray &uid) {
                callback(uid);
            });
        },
        exists
    );
}

void Synchronizer::modifyIfChanged(Storage::EntityStore &store, const QByteArray &bufferType, const QByteArray &sinkId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
{
    store.readLatest(bufferType, sinkId, [&, this](const Sink::ApplicationDomain::ApplicationDomainType &current) {
        bool changed = false;
        for (const auto &property : entity.changedProperties()) {
            if (entity.getProperty(property) != current.getProperty(property)) {
                SinkTrace() << "Property changed " << sinkId << property;
                changed = true;
            }
        }
        if (changed) {
            SinkTrace() << "Found a modified entity: " << sinkId;
            modifyEntity(sinkId, store.maxRevision(), bufferType, entity);
        }
    });
}

void Synchronizer::modify(const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
{
    const auto sinkId = syncStore().resolveRemoteId(bufferType, remoteId);
    Storage::EntityStore store(mResourceContext, mLogCtx);
    modifyIfChanged(store, bufferType, sinkId, entity);
}

void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
{
    SinkTrace() << "Create or modify" << bufferType << remoteId;
    Storage::EntityStore store(mResourceContext, mLogCtx);
    const auto sinkId = syncStore().resolveRemoteId(bufferType, remoteId);
    const auto found = store.contains(bufferType, sinkId);
    if (!found) {
        SinkTrace() << "Found a new entity: " << remoteId;
        createEntity(sinkId, bufferType, entity);
    } else { // modification
        modify(bufferType, remoteId, entity);
    }
}

template<typename DomainType>
void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const DomainType &entity, const QHash<QByteArray, Sink::Query::Comparator> &mergeCriteria)
{

    SinkTrace() << "Create or modify" << bufferType << remoteId;
    const auto sinkId = syncStore().resolveRemoteId(bufferType, remoteId);
    Storage::EntityStore store(mResourceContext, mLogCtx);
    const auto found = store.contains(bufferType, sinkId);
    if (!found) {
        if (!mergeCriteria.isEmpty()) {
            Sink::Query query;
            for (auto it = mergeCriteria.constBegin(); it != mergeCriteria.constEnd(); it++) {
                query.filter(it.key(), it.value());
            }
            bool merge = false;
            Storage::EntityStore store{mResourceContext, mLogCtx};
            DataStoreQuery dataStoreQuery{query, ApplicationDomain::getTypeName<DomainType>(), store};
            auto resultSet = dataStoreQuery.execute();
            resultSet.replaySet(0, 1, [this, &merge, bufferType, remoteId](const ResultSet::Result &r) {
                merge = true;
                SinkTrace() << "Merging local entity with remote entity: " << r.entity.identifier() << remoteId;
                syncStore().recordRemoteId(bufferType, r.entity.identifier(), remoteId);
            });

            if (!merge) {
                SinkTrace() << "Found a new entity: " << remoteId;
                createEntity(sinkId, bufferType, entity);
            }
        } else {
            SinkTrace() << "Found a new entity: " << remoteId;
            createEntity(sinkId, bufferType, entity);
        }
    } else { // modification
        modifyIfChanged(store, bufferType, sinkId, entity);
    }
}

QByteArrayList Synchronizer::resolveFilter(const QueryBase::Comparator &filter)
{
    QByteArrayList result;
    if (filter.value.canConvert<QByteArray>()) {
        result << filter.value.value<QByteArray>();
    } else if (filter.value.canConvert<QueryBase>()) {
        auto query = filter.value.value<QueryBase>();
        Storage::EntityStore store{mResourceContext, mLogCtx};
        DataStoreQuery dataStoreQuery{query, query.type(), store};
        auto resultSet = dataStoreQuery.execute();
        resultSet.replaySet(0, 0, [this, &result](const ResultSet::Result &r) {
            result << r.entity.identifier();
        });
    } else {
        SinkWarning() << "unknown filter type: " << filter;
        Q_ASSERT(false);
    }
    return result;
}

template<typename DomainType>
void Synchronizer::modify(const DomainType &entity, const QByteArray &newResource, bool remove)
{
    modifyEntity(entity.identifier(), entity.revision(), ApplicationDomain::getTypeName<DomainType>(), entity, newResource, remove);
}

QList<Synchronizer::SyncRequest> Synchronizer::getSyncRequests(const Sink::QueryBase &query)
{
    QList<Synchronizer::SyncRequest> list;
    list << Synchronizer::SyncRequest{query, "sync"};
    return list;
}

void Synchronizer::synchronize(const Sink::QueryBase &query)
{
    SinkTrace() << "Synchronizing";
    mSyncRequestQueue << getSyncRequests(query);
    processSyncQueue().exec();
}

void Synchronizer::flush(int commandId, const QByteArray &flushId)
{
    Q_ASSERT(!flushId.isEmpty());
    SinkTrace() << "Flushing the synchronization queue " << flushId;
    mSyncRequestQueue << Synchronizer::SyncRequest{Synchronizer::SyncRequest::Flush, commandId, flushId};
    processSyncQueue().exec();
}

void Synchronizer::flushComplete(const QByteArray &flushId)
{
    SinkTraceCtx(mLogCtx) << "Flush complete: " << flushId;
    if (mPendingSyncRequests.contains(flushId)) {
        const auto requests = mPendingSyncRequests.values(flushId);
        for (const auto &r : requests) {
            //We want to process the pending request before any others in the queue
            mSyncRequestQueue.prepend(r);
        }
        mPendingSyncRequests.remove(flushId);
        processSyncQueue().exec();
    }
}

KAsync::Job<void> Synchronizer::processSyncQueue()
{
    if (mSyncRequestQueue.isEmpty() || mSyncInProgress) {
        SinkTrace() << "Sync still in progress or nothing to do.";
        return KAsync::null<void>();
    }
    //Don't process any new requests until we're done with the pending ones.
    //Otherwise we might process a flush before the previous request actually completed.
    if (!mPendingSyncRequests.isEmpty()) {
        SinkTrace() << "We still have pending sync requests. Not executing next request.";
        return KAsync::null<void>();
    }

    auto job = KAsync::syncStart<void>([this] {
        mMessageQueue->startTransaction();
        mSyncInProgress = true;
    });
    while (!mSyncRequestQueue.isEmpty()) {
        const auto request = mSyncRequestQueue.takeFirst();
        if (request.options & SyncRequest::RequestFlush) {
            job = job.then([=] {
                //Trigger a flush and record original request without flush option
                auto modifiedRequest = request;
                modifiedRequest.options = SyncRequest::NoOptions;
                //Normally we won't have a requestId here
                if (modifiedRequest.requestId.isEmpty()) {
                    modifiedRequest.requestId = QUuid::createUuid().toRfc4122();
                }

                //The sync request will be executed once the flush has completed
                mPendingSyncRequests.insert(modifiedRequest.requestId, modifiedRequest);

                flatbuffers::FlatBufferBuilder fbb;
                auto flushId = fbb.CreateString(modifiedRequest.requestId.toStdString());
                auto location = Sink::Commands::CreateFlush(fbb, flushId, static_cast<int>(Sink::Flush::FlushSynchronization));
                Sink::Commands::FinishFlushBuffer(fbb, location);
                enqueueCommand(Sink::Commands::FlushCommand, BufferUtils::extractBuffer(fbb));
            });
        } else if (request.requestType == Synchronizer::SyncRequest::Synchronization) {
            job = job.then([this, request] {
                Sink::Notification n;
                n.id = request.requestId;
                n.type = Notification::Status;
                n.message = "Synchronization has started.";
                n.code = ApplicationDomain::BusyStatus;
                emit notify(n);
                SinkLogCtx(mLogCtx) << "Synchronizing: " << request.query;
            }).then(synchronizeWithSource(request.query)).then([this] {
                //Commit after every request, so implementations only have to commit more if they add a lot of data.
                commit();
            }).then<void>([this, request](const KAsync::Error &error) {
                if (error) {
                    //Emit notification with error
                    SinkWarningCtx(mLogCtx) << "Synchronization failed: " << error.errorMessage;
                    Sink::Notification n;
                    n.id = request.requestId;
                    n.type = Notification::Status;
                    n.message = "Synchronization has ended.";
                    n.code = ApplicationDomain::ErrorStatus;
                    emit notify(n);
                    return KAsync::error(error);
                } else {
                    SinkLogCtx(mLogCtx) << "Done Synchronizing";
                    Sink::Notification n;
                    n.id = request.requestId;
                    n.type = Notification::Status;
                    n.message = "Synchronization has ended.";
                    n.code = ApplicationDomain::ConnectedStatus;
                    emit notify(n);
                    return KAsync::null();
                }
            });
        } else if (request.requestType == Synchronizer::SyncRequest::Flush) {
            job = job.then([=] {
                Q_ASSERT(!request.requestId.isEmpty());
                if (request.flushType == Flush::FlushReplayQueue) {
                    SinkTrace() << "Emitting flush completion.";
                    Sink::Notification n;
                    n.type = Sink::Notification::FlushCompletion;
                    n.id = request.requestId;
                    emit notify(n);
                } else {
                    flatbuffers::FlatBufferBuilder fbb;
                    auto flushId = fbb.CreateString(request.requestId.toStdString());
                    auto location = Sink::Commands::CreateFlush(fbb, flushId, static_cast<int>(Sink::Flush::FlushSynchronization));
                    Sink::Commands::FinishFlushBuffer(fbb, location);
                    enqueueCommand(Sink::Commands::FlushCommand, BufferUtils::extractBuffer(fbb));
                }
            });
        } else if (request.requestType == Synchronizer::SyncRequest::ChangeReplay) {
            job = job.then(replayNextRevision());
        } else {
            SinkWarning() << "Unknown request type: " << request.requestType;
            return KAsync::error(KAsync::Error{"Unknown request type."});
        }
    }
    return job.then<void>([this](const KAsync::Error &error) {
        SinkTrace() << "Sync request processed";
        mSyncTransaction.abort();
        mMessageQueue->commit();
        mSyncStore.clear();
        mSyncInProgress = false;
        if (allChangesReplayed()) {
            emit changesReplayed();
        }
        if (error) {
            SinkWarning() << "Error during sync: " << error.errorMessage;
            return KAsync::error(error);
        }
        //In case we got more requests meanwhile.
        return processSyncQueue();
    });
}

void Synchronizer::commit()
{
    mMessageQueue->commit();
    mEntityStore->abortTransaction();
    mSyncTransaction.commit();
    mSyncStore.clear();
    if (mSyncInProgress) {
        mMessageQueue->startTransaction();
    }
}

Sink::Storage::DataStore::DataStore::Transaction &Synchronizer::syncTransaction()
{
    if (!mSyncTransaction) {
        SinkTrace() << "Starting transaction on sync store.";
        mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::DataStore::DataStore::ReadWrite);
    }
    return mSyncTransaction;
}

void Synchronizer::revisionChanged()
{
    //One replay request is enough
    for (const auto &r : mSyncRequestQueue) {
        if (r.requestType == Synchronizer::SyncRequest::ChangeReplay) {
            return;
        }
    }
    mSyncRequestQueue << Synchronizer::SyncRequest{Synchronizer::SyncRequest::ChangeReplay};
    processSyncQueue().exec();
}

bool Synchronizer::canReplay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
{
    Sink::EntityBuffer buffer(value);
    const Sink::Entity &entity = buffer.entity();
    const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
    Q_ASSERT(metadataBuffer);
    if (!metadataBuffer->replayToSource()) {
        SinkTrace() << "Change is coming from the source";
    }
    return metadataBuffer->replayToSource();
}

KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
{
    SinkTrace() << "Replaying" << type << key;

    Sink::EntityBuffer buffer(value);
    const Sink::Entity &entity = buffer.entity();
    const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
    if (!metadataBuffer) {
        SinkError() << "No metadata buffer available.";
        return KAsync::error("No metadata buffer");
    }
    if (mSyncTransaction) {
        SinkError() << "Leftover sync transaction.";
        mSyncTransaction.abort();
    }
    if (mSyncStore) {
        SinkError() << "Leftover sync store.";
        mSyncStore.clear();
    }
    Q_ASSERT(metadataBuffer);
    Q_ASSERT(!mSyncStore);
    Q_ASSERT(!mSyncTransaction);
    mEntityStore->startTransaction(Storage::DataStore::ReadOnly);

    const auto operation = metadataBuffer ? metadataBuffer->operation() : Sink::Operation_Creation;
    const auto uid = Sink::Storage::DataStore::uidFromKey(key);
    const auto modifiedProperties = metadataBuffer->modifiedProperties() ? BufferUtils::fromVector(*metadataBuffer->modifiedProperties()) : QByteArrayList();
    QByteArray oldRemoteId;

    if (operation != Sink::Operation_Creation) {
        oldRemoteId = syncStore().resolveLocalId(type, uid);
        //oldRemoteId can be empty if the resource implementation didn't return a remoteid
    }
    SinkTrace() << "Replaying " << key << type << uid << oldRemoteId;

    KAsync::Job<QByteArray> job = KAsync::null<QByteArray>();
    //TODO This requires supporting every domain type here as well. Can we solve this better so we can do the dispatch somewhere centrally?
    if (type == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) {
        auto folder = store().readEntity<ApplicationDomain::Folder>(key);
        job = replay(folder, operation, oldRemoteId, modifiedProperties);
    } else if (type == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) {
        auto mail = store().readEntity<ApplicationDomain::Mail>(key);
        job = replay(mail, operation, oldRemoteId, modifiedProperties);
    } else {
        SinkError() << "Replayed unknown type: " << type;
    }

    return job.then([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
        if (operation == Sink::Operation_Creation) {
            SinkTrace() << "Replayed creation with remote id: " << remoteId;
            if (remoteId.isEmpty()) {
                SinkWarning() << "Returned an empty remoteId from the creation";
            } else {
                syncStore().recordRemoteId(type, uid, remoteId);
            }
        } else if (operation == Sink::Operation_Modification) {
            SinkTrace() << "Replayed modification with remote id: " << remoteId;
            if (remoteId.isEmpty()) {
                SinkWarning() << "Returned an empty remoteId from the creation";
            } else {
               syncStore().updateRemoteId(type, uid, remoteId);
            }
        } else if (operation == Sink::Operation_Removal) {
            SinkTrace() << "Replayed removal with remote id: " << oldRemoteId;
            syncStore().removeRemoteId(type, uid, oldRemoteId);
        } else {
            SinkError() << "Unkown operation" << operation;
        }
    })
    .then([this](const KAsync::Error &error) {
        if (error) {
            SinkWarning() << "Failed to replay change: " << error.errorMessage;
        }
        mSyncStore.clear();
        mSyncTransaction.commit();
        mEntityStore->abortTransaction();
    });
}

KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Mail &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
{
    return KAsync::null<QByteArray>();
}

KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Folder &, Sink::Operation, const QByteArray &, const QList<QByteArray> &)
{
    return KAsync::null<QByteArray>();
}

bool Synchronizer::allChangesReplayed()
{
    if (!mSyncRequestQueue.isEmpty()) {
        SinkTrace() << "Queue is not empty";
        return false;
    }
    return ChangeReplay::allChangesReplayed();
}

#define REGISTER_TYPE(T)                                                          \
    template void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const T &entity, const QHash<QByteArray, Sink::Query::Comparator> &mergeCriteria); \
    template void Synchronizer::modify(const T &entity, const QByteArray &newResource, bool remove);

SINK_REGISTER_TYPES()

#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
#include "synchronizer.moc"
#include "moc_synchronizer.cpp"
#pragma clang diagnostic pop