File indexing completed on 2024-05-12 05:26:01

 /*
  * Copyright (C) 2014 Aaron Seigo <aseigo@kde.org>
  * Copyright (C) 2015 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 "pipeline.h"
 
 #include <QByteArray>
 #include <QVector>
 #include <QDebug>
 #include <QTime>
 #include "entity_generated.h"
 #include "metadata_generated.h"
 #include "createentity_generated.h"
 #include "modifyentity_generated.h"
 #include "deleteentity_generated.h"
 #include "entitybuffer.h"
 #include "log.h"
 #include "domain/applicationdomaintype.h"
 #include "adaptorfactoryregistry.h"
 #include "definitions.h"
 #include "bufferutils.h"
 #include "storage/entitystore.h"
 #include "store.h"
 
 using namespace Sink;
 using namespace Sink::Storage;
 
 class Pipeline::Private
 {
 public:
     Private(const ResourceContext &context, const Sink::Log::Context &ctx) : logCtx{ctx.subContext("pipeline")}, resourceContext(context), entityStore(context, ctx), revisionChanged(false)
     {
     }
 
     Sink::Log::Context logCtx;
     ResourceContext resourceContext;
     Storage::EntityStore entityStore;
     QHash<QString, QVector<QSharedPointer<Preprocessor>>> processors;
     bool revisionChanged;
     QTime transactionTime;
     int transactionItemCount;
 };
 
 
 Pipeline::Pipeline(const ResourceContext &context, const Sink::Log::Context &ctx) : QObject(nullptr), d(new Private(context, ctx))
 {
     //Create main store immediately on first start
     d->entityStore.initialize();
 }
 
 Pipeline::~Pipeline()
 {
 }
 
 void Pipeline::setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &processors)
 {
     auto &list = d->processors[entityType];
     list.clear();
     for (auto p : processors) {
         p->setup(d->resourceContext.resourceType, d->resourceContext.instanceId(), this, &d->entityStore);
         list.append(QSharedPointer<Preprocessor>(p));
     }
 }
 
 void Pipeline::startTransaction()
 {
     // TODO call for all types
     // But avoid doing it during cleanup
     // for (auto processor : d->processors[bufferType]) {
     //     processor->startBatch();
     // }
     SinkTraceCtx(d->logCtx) << "Starting transaction.";
     d->transactionTime.start();
     d->transactionItemCount = 0;
     d->entityStore.startTransaction(DataStore::ReadWrite);
 }
 
 void Pipeline::commit()
 {
     // TODO call for all types
     // But avoid doing it during cleanup
     // for (auto processor : d->processors[bufferType]) {
     //     processor->finalize();
     // }
     if (!d->revisionChanged) {
         d->entityStore.abortTransaction();
         return;
     }
     const auto revision = d->entityStore.maxRevision();
     d->entityStore.commitTransaction();
     const auto elapsed = d->transactionTime.elapsed();
     SinkTraceCtx(d->logCtx) << "Committing revision: " << revision << ":" << d->transactionItemCount << " items in: " << Log::TraceTime(elapsed) << " "
             << (double)elapsed / (double)qMax(d->transactionItemCount, 1) << "[ms/item]";
     if (d->revisionChanged) {
         d->revisionChanged = false;
         emit revisionUpdated(revision);
     }
 }
 
 KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 {
     d->transactionItemCount++;
 
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Commands::VerifyCreateEntityBuffer(verifyer)) {
             SinkWarningCtx(d->logCtx) << "invalid buffer, not a create entity buffer";
             return KAsync::error<qint64>();
         }
     }
     auto createEntity = Commands::GetCreateEntity(command);
 
     const bool replayToSource = createEntity->replayToSource();
     const QByteArray bufferType = QByteArray(reinterpret_cast<char const *>(createEntity->domainType()->Data()), createEntity->domainType()->size());
     QByteArray key;
     if (createEntity->entityId()) {
         key = QByteArray(reinterpret_cast<char const *>(createEntity->entityId()->Data()), createEntity->entityId()->size());
         if (!key.isEmpty() && d->entityStore.contains(bufferType, key)) {
             SinkErrorCtx(d->logCtx) << "An entity with this id already exists: " << key;
             return KAsync::error<qint64>();
         }
     }
 
     if (key.isEmpty()) {
         key = DataStore::generateUid();
     }
     SinkTraceCtx(d->logCtx) << "New Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
     Q_ASSERT(!key.isEmpty());
 
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
         if (!VerifyEntityBuffer(verifyer)) {
             SinkWarningCtx(d->logCtx) << "invalid buffer, not an entity buffer";
             return KAsync::error<qint64>();
         }
     }
     auto entity = GetEntity(createEntity->delta()->Data());
     if (!entity->resource()->size() && !entity->local()->size()) {
         SinkWarningCtx(d->logCtx) << "No local and no resource buffer while trying to create entity.";
         return KAsync::error<qint64>();
     }
 
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceContext.resourceType, bufferType);
     if (!adaptorFactory) {
         SinkWarningCtx(d->logCtx) << "no adaptor factory for type " << bufferType << d->resourceContext.resourceType;
         return KAsync::error<qint64>();
     }
 
     auto adaptor = adaptorFactory->createAdaptor(*entity);
     auto memoryAdaptor = QSharedPointer<Sink::ApplicationDomain::MemoryBufferAdaptor>::create();
     Sink::ApplicationDomain::copyBuffer(*adaptor, *memoryAdaptor);
 
     d->revisionChanged = true;
     auto revision = d->entityStore.maxRevision();
     auto o = Sink::ApplicationDomain::ApplicationDomainType{d->resourceContext.instanceId(), key, revision, memoryAdaptor};
     o.setChangedProperties(o.availableProperties().toSet());
 
     auto newEntity = *ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<ApplicationDomain::ApplicationDomainType>(o, o.availableProperties());
     newEntity.setChangedProperties(newEntity.availableProperties().toSet());
 
     foreach (const auto &processor, d->processors[bufferType]) {
         processor->newEntity(newEntity);
     }
 
     if (!d->entityStore.add(bufferType, newEntity, replayToSource)) {
         return KAsync::error<qint64>();
     }
 
     return KAsync::value(d->entityStore.maxRevision());
 }
 
 template <class T>
 struct CreateHelper {
     KAsync::Job<void> operator()(const ApplicationDomain::ApplicationDomainType &arg) const {
         return Sink::Store::create<T>(T{arg});
     }
 };
 
 static KAsync::Job<void> create(const QByteArray &type, const ApplicationDomain::ApplicationDomainType &newEntity)
 {
     return TypeHelper<CreateHelper>{type}.operator()<KAsync::Job<void>, const ApplicationDomain::ApplicationDomainType&>(newEntity);
 }
 
 KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 {
     d->transactionItemCount++;
 
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Commands::VerifyModifyEntityBuffer(verifyer)) {
             SinkWarningCtx(d->logCtx) << "invalid buffer, not a modify entity buffer";
             return KAsync::error<qint64>();
         }
     }
     auto modifyEntity = Commands::GetModifyEntity(command);
     Q_ASSERT(modifyEntity);
     QList<QByteArray> changeset;
     if (modifyEntity->modifiedProperties()) {
         changeset = BufferUtils::fromVector(*modifyEntity->modifiedProperties());
     } else {
         SinkWarningCtx(d->logCtx) << "No changeset available";
     }
     const qint64 baseRevision = modifyEntity->revision();
     const bool replayToSource = modifyEntity->replayToSource();
 
     const QByteArray bufferType = QByteArray(reinterpret_cast<char const *>(modifyEntity->domainType()->Data()), modifyEntity->domainType()->size());
     const QByteArray key = QByteArray(reinterpret_cast<char const *>(modifyEntity->entityId()->Data()), modifyEntity->entityId()->size());
     SinkTraceCtx(d->logCtx) << "Modified Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
     if (bufferType.isEmpty() || key.isEmpty()) {
         SinkWarningCtx(d->logCtx) << "entity type or key " << bufferType << key;
         return KAsync::error<qint64>();
     }
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(modifyEntity->delta()->Data()), modifyEntity->delta()->size());
         if (!VerifyEntityBuffer(verifyer)) {
             SinkWarningCtx(d->logCtx) << "invalid buffer, not an entity buffer";
             return KAsync::error<qint64>();
         }
     }
 
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceContext.resourceType, bufferType);
     if (!adaptorFactory) {
         SinkWarningCtx(d->logCtx) << "no adaptor factory for type " << bufferType;
         return KAsync::error<qint64>();
     }
 
     auto diffEntity = GetEntity(modifyEntity->delta()->Data());
     Q_ASSERT(diffEntity);
     Sink::ApplicationDomain::ApplicationDomainType diff{d->resourceContext.instanceId(), key, baseRevision, adaptorFactory->createAdaptor(*diffEntity)};
     diff.setChangedProperties(changeset.toSet());
 
     QByteArrayList deletions;
     if (modifyEntity->deletions()) {
         deletions = BufferUtils::fromVector(*modifyEntity->deletions());
     }
 
     Sink::ApplicationDomain::ApplicationDomainType current;
     bool alreadyRemoved = false;
 
     d->entityStore.readLatest(bufferType, diff.identifier(), [&](const QByteArray &uid, const EntityBuffer &buffer) {
         if (buffer.operation() == Sink::Operation_Removal) {
             alreadyRemoved = true;
         } else {
             auto entity = Sink::ApplicationDomain::ApplicationDomainType{d->resourceContext.instanceId(), key, baseRevision, adaptorFactory->createAdaptor(buffer.entity())};
             current = *Sink::ApplicationDomain::ApplicationDomainType::getInMemoryRepresentation<Sink::ApplicationDomain::ApplicationDomainType>(entity, entity.availableProperties());
         }
     });
 
     if (alreadyRemoved) {
         SinkWarningCtx(d->logCtx) << "Tried to modify a removed entity: " << diff.identifier();
         return KAsync::error<qint64>();
     }
 
     if (current.identifier().isEmpty()) {
         SinkWarningCtx(d->logCtx) << "Failed to read current version: " << diff.identifier();
         return KAsync::error<qint64>();
     }
 
     //We avoid overwriting local changes that haven't been played back yet with remote modifications
     QSet<QByteArray> excludeProperties;
     if (!replayToSource) { //We assume this means the change is coming from the source already
         d->entityStore.readRevisions(bufferType, diff.identifier(), baseRevision, [&] (const QByteArray &uid, qint64 revision, const Sink::EntityBuffer &entity) {
             if (entity.metadataBuffer()) {
                 if (auto metadata = GetMetadata(entity.metadataBuffer())) {
                     if (metadata->operation() == Operation_Modification && metadata->modifiedProperties()) {
                         const auto locallyModifiedProperties = BufferUtils::fromVector(*metadata->modifiedProperties()).toSet();
                         SinkTraceCtx(d->logCtx) << "Protecting locally modified properties: " << locallyModifiedProperties;
                         excludeProperties += locallyModifiedProperties;
                     }
                 }
             }
         });
     }
 
     auto newEntity = d->entityStore.applyDiff(bufferType, current, diff, deletions, excludeProperties);
 
     bool isMove = false;
     if (modifyEntity->targetResource()) {
         isMove = modifyEntity->removeEntity();
         newEntity.setResource(BufferUtils::extractBuffer(modifyEntity->targetResource()));
     }
 
     foreach (const auto &processor, d->processors[bufferType]) {
         bool exitLoop = false;
         const auto result = processor->process(Preprocessor::Modification, current, newEntity);
         switch (result.action) {
             case Preprocessor::MoveToResource:
                 isMove = true;
                 exitLoop = true;
                 break;
             case Preprocessor::CopyToResource:
                 isMove = true;
                 exitLoop = true;
                 break;
             case Preprocessor::DropModification:
                 SinkTraceCtx(d->logCtx) << "Dropping modification";
                 return KAsync::error<qint64>();
             case Preprocessor::NoAction:
             case Preprocessor::DeleteEntity:
             default:
                 break;
         }
         if (exitLoop) {
             break;
         }
     }
 
     //The entity is either being copied or moved
     if (newEntity.resourceInstanceIdentifier() != d->resourceContext.resourceInstanceIdentifier) {
         auto copy = *ApplicationDomain::ApplicationDomainType::getInMemoryCopy<ApplicationDomain::ApplicationDomainType>(newEntity, newEntity.availableProperties());
         copy.setResource(newEntity.resourceInstanceIdentifier());
         copy.setChangedProperties(copy.availableProperties().toSet());
         SinkTraceCtx(d->logCtx) << "Moving entity to new resource " << copy.identifier() << copy.resourceInstanceIdentifier();
         return create(bufferType, copy)
             .then([=](const KAsync::Error &error) {
                 if (!error) {
                     SinkTraceCtx(d->logCtx) << "Move of " << current.identifier() << "was successfull";
                     if (isMove) {
                         flatbuffers::FlatBufferBuilder fbb;
                         auto entityId = fbb.CreateString(current.identifier().toStdString());
                         auto type = fbb.CreateString(bufferType.toStdString());
                         auto location = Sink::Commands::CreateDeleteEntity(fbb, current.revision(), entityId, type, true);
                         Sink::Commands::FinishDeleteEntityBuffer(fbb, location);
                         const auto data = BufferUtils::extractBuffer(fbb);
                         deletedEntity(data, data.size()).exec();
                     }
                 } else {
                     SinkErrorCtx(d->logCtx) << "Failed to move entity " << newEntity.identifier() << " to resource " << newEntity.resourceInstanceIdentifier();
                 }
             })
             .then([this] {
                 return d->entityStore.maxRevision();
             });
     }
 
     d->revisionChanged = true;
     if (!d->entityStore.modify(bufferType, current, newEntity, replayToSource)) {
         return KAsync::error<qint64>();
     }
 
     return KAsync::value(d->entityStore.maxRevision());
 }
 
 KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
 {
     d->transactionItemCount++;
 
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
         if (!Commands::VerifyDeleteEntityBuffer(verifyer)) {
             SinkWarningCtx(d->logCtx) << "invalid buffer, not a delete entity buffer";
             return KAsync::error<qint64>();
         }
     }
     auto deleteEntity = Commands::GetDeleteEntity(command);
 
     const bool replayToSource = deleteEntity->replayToSource();
     const QByteArray bufferType = QByteArray(reinterpret_cast<char const *>(deleteEntity->domainType()->Data()), deleteEntity->domainType()->size());
     const QByteArray key = QByteArray(reinterpret_cast<char const *>(deleteEntity->entityId()->Data()), deleteEntity->entityId()->size());
     SinkTraceCtx(d->logCtx) << "Deleted Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
 
     const auto current = d->entityStore.readLatest(bufferType, key);
 
     foreach (const auto &processor, d->processors[bufferType]) {
         processor->deletedEntity(current);
     }
 
     d->revisionChanged = true;
     if (!d->entityStore.remove(bufferType, current, replayToSource)) {
         return KAsync::error<qint64>();
     }
 
     return KAsync::value(d->entityStore.maxRevision());
 }
 
 void Pipeline::cleanupRevisions(qint64 revision)
 {
     //We have to set revisionChanged, otherwise a call to commit might abort
     //the transaction when not using the implicit internal transaction
     d->revisionChanged = d->entityStore.cleanupRevisions(revision);
 }
 
 
 class Preprocessor::Private {
 public:
     QByteArray resourceType;
     QByteArray resourceInstanceIdentifier;
     Pipeline *pipeline;
     Storage::EntityStore *entityStore;
 };
 
 Preprocessor::Preprocessor() : d(new Preprocessor::Private)
 {
 }
 
 Preprocessor::~Preprocessor()
 {
 }
 
 void Preprocessor::setup(const QByteArray &resourceType, const QByteArray &resourceInstanceIdentifier, Pipeline *pipeline, Storage::EntityStore *entityStore)
 {
     d->resourceType = resourceType;
     d->resourceInstanceIdentifier = resourceInstanceIdentifier;
     d->pipeline = pipeline;
     d->entityStore = entityStore;
 }
 
 void Preprocessor::startBatch()
 {
 }
 
 void Preprocessor::finalizeBatch()
 {
 }
 
 void Preprocessor::newEntity(ApplicationDomain::ApplicationDomainType &newEntity)
 {
 
 }
 
 void Preprocessor::modifiedEntity(const ApplicationDomain::ApplicationDomainType &oldEntity, ApplicationDomain::ApplicationDomainType &newEntity)
 {
 
 }
 
 void Preprocessor::deletedEntity(const ApplicationDomain::ApplicationDomainType &oldEntity)
 {
 
 }
 
 Preprocessor::Result Preprocessor::process(Type type, const ApplicationDomain::ApplicationDomainType &current, ApplicationDomain::ApplicationDomainType &diff)
 {
     switch(type) {
         case Creation:
             newEntity(diff);
             break;
         case Modification:
             modifiedEntity(current, diff);
             break;
         case Deletion:
             deletedEntity(current);
             break;
         default:
             break;
     }
     return {NoAction};
 }
 
 QByteArray Preprocessor::resourceInstanceIdentifier() const
 {
     return d->resourceInstanceIdentifier;
 }
 
 Storage::EntityStore &Preprocessor::entityStore() const
 {
     return *d->entityStore;
 }
 
 void Preprocessor::createEntity(const Sink::ApplicationDomain::ApplicationDomainType &entity, const QByteArray &typeName, bool replayToSource)
 {
     flatbuffers::FlatBufferBuilder entityFbb;
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, typeName);
     adaptorFactory->createBuffer(entity, entityFbb);
     const auto entityBuffer = BufferUtils::extractBuffer(entityFbb);
 
     flatbuffers::FlatBufferBuilder fbb;
     auto entityId = fbb.CreateString(entity.identifier().toStdString());
     auto type = fbb.CreateString(typeName.toStdString());
     auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityBuffer.constData(), entityBuffer.size());
     auto location = Sink::Commands::CreateCreateEntity(fbb, entityId, type, delta, replayToSource);
     Sink::Commands::FinishCreateEntityBuffer(fbb, location);
 
     const auto data = BufferUtils::extractBuffer(fbb);
     d->pipeline->newEntity(data, data.size()).exec();
 }
 
 void Preprocessor::deleteEntity(const Sink::ApplicationDomain::ApplicationDomainType &entity, const QByteArray &typeName, bool replayToSource)
 {
     flatbuffers::FlatBufferBuilder fbb;
     auto entityId = fbb.CreateString(entity.identifier().toStdString());
     auto type = fbb.CreateString(typeName.toStdString());
     auto location = Sink::Commands::CreateDeleteEntity(fbb, entity.revision(), entityId, type, replayToSource);
     Sink::Commands::FinishDeleteEntityBuffer(fbb, location);
     const auto data = BufferUtils::extractBuffer(fbb);
     d->pipeline->deletedEntity(data, data.size()).exec();
 }
 
 #pragma clang diagnostic push
 #pragma clang diagnostic ignored "-Wundefined-reinterpret-cast"
 #include "moc_pipeline.cpp"
 #pragma clang diagnostic pop