1 files changed, 55 insertions, 220 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index e257857..ea59ae9 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -37,6 +37,7 @@
 #include "adaptorfactoryregistry.h"
 #include "definitions.h"
 #include "bufferutils.h"
+#include "storage/entitystore.h"
 
 SINK_DEBUG_AREA("pipeline")
 
@@ -46,31 +47,18 @@ using namespace Sink::Storage;
 class Pipeline::Private
 {
 public:
-    Private(const ResourceContext &context) : resourceContext(context), storage(Sink::storageLocation(), context.instanceId(), DataStore::ReadWrite), revisionChanged(false)
+    Private(const ResourceContext &context) : resourceContext(context), entityStore(context), revisionChanged(false)
     {
     }
 
     ResourceContext resourceContext;
-    DataStore storage;
-    DataStore::Transaction transaction;
+    Storage::EntityStore entityStore;
     QHash<QString, QVector<QSharedPointer<Preprocessor>>> processors;
     bool revisionChanged;
-    void storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid);
     QTime transactionTime;
     int transactionItemCount;
 };
 
-void Pipeline::Private::storeNewRevision(qint64 newRevision, const flatbuffers::FlatBufferBuilder &fbb, const QByteArray &bufferType, const QByteArray &uid)
-{
-    SinkTrace() << "Committing new revision: " << uid << newRevision;
-    DataStore::mainDatabase(transaction, bufferType)
-        .write(DataStore::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
-            [uid, newRevision](const DataStore::Error &error) { SinkWarning() << "Failed to write entity" << uid << newRevision; });
-    revisionChanged = true;
-    DataStore::setMaxRevision(transaction, newRevision);
-    DataStore::recordRevision(transaction, newRevision, uid, bufferType);
-}
-
 
 Pipeline::Pipeline(const ResourceContext &context) : QObject(nullptr), d(new Private(context))
 {
@@ -78,7 +66,6 @@ Pipeline::Pipeline(const ResourceContext &context) : QObject(nullptr), d(new Pri
 
 Pipeline::~Pipeline()
 {
-    d->transaction = DataStore::Transaction();
 }
 
 void Pipeline::setPreprocessors(const QString &entityType, const QVector<Preprocessor *> &processors)
@@ -98,27 +85,10 @@ void Pipeline::startTransaction()
     // for (auto processor : d->processors[bufferType]) {
     //     processor->startBatch();
     // }
-    if (d->transaction) {
-        return;
-    }
     SinkTrace() << "Starting transaction.";
     d->transactionTime.start();
     d->transactionItemCount = 0;
-    d->transaction = storage().createTransaction(DataStore::ReadWrite, [](const DataStore::Error &error) {
-        SinkWarning() << error.message;
-    });
-
-    //FIXME this is a temporary measure to recover from a failure to open the named databases correctly.
-    //Once the actual problem is fixed it will be enough to simply crash if we open the wrong database (which we check in openDatabase already).
-    //It seems like the validateNamedDatabase calls actually stops the mdb_put failures during sync...
-    if (d->storage.exists()) {
-        while (!d->transaction.validateNamedDatabases()) {
-            SinkWarning() << "Opened an invalid transaction!!!!!!";
-            d->transaction = storage().createTransaction(DataStore::ReadWrite, [](const DataStore::Error &error) {
-                SinkWarning() << error.message;
-            });
-        }
-    }
+    d->entityStore.startTransaction(DataStore::ReadWrite);
 }
 
 void Pipeline::commit()
@@ -129,34 +99,20 @@ void Pipeline::commit()
     //     processor->finalize();
     // }
     if (!d->revisionChanged) {
-        d->transaction.abort();
-        d->transaction = DataStore::Transaction();
+        d->entityStore.abortTransaction();
         return;
     }
-    const auto revision = DataStore::maxRevision(d->transaction);
+    const auto revision = d->entityStore.maxRevision();
     const auto elapsed = d->transactionTime.elapsed();
     SinkLog() << "Committing revision: " << revision << ":" << d->transactionItemCount << " items in: " << Log::TraceTime(elapsed) << " "
             << (double)elapsed / (double)qMax(d->transactionItemCount, 1) << "[ms/item]";
-    if (d->transaction) {
-        d->transaction.commit();
-    }
-    d->transaction = DataStore::Transaction();
+    d->entityStore.commitTransaction();
     if (d->revisionChanged) {
         d->revisionChanged = false;
         emit revisionUpdated(revision);
     }
 }
 
-DataStore::Transaction &Pipeline::transaction()
-{
-    return d->transaction;
-}
-
-DataStore &Pipeline::storage() const
-{
-    return d->storage;
-}
-
 KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 {
     d->transactionItemCount++;
@@ -175,7 +131,7 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     QByteArray key;
     if (createEntity->entityId()) {
         key = QByteArray(reinterpret_cast<char const *>(createEntity->entityId()->Data()), createEntity->entityId()->size());
-        if (DataStore::mainDatabase(d->transaction, bufferType).contains(key)) {
+        if (d->entityStore.contains(bufferType, key)) {
             SinkError() << "An entity with this id already exists: " << key;
             return KAsync::error<qint64>(0);
         }
@@ -208,29 +164,23 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
 
     auto adaptor = adaptorFactory->createAdaptor(*entity);
     auto memoryAdaptor = QSharedPointer<Sink::ApplicationDomain::MemoryBufferAdaptor>::create(*(adaptor), adaptor->availableProperties());
-    foreach (const auto &processor, d->processors[bufferType]) {
-        processor->newEntity(key, DataStore::maxRevision(d->transaction) + 1, *memoryAdaptor, d->transaction);
-    }
-    //The maxRevision may have changed meanwhile if the entity created sub-entities
-    const qint64 newRevision = DataStore::maxRevision(d->transaction) + 1;
-
-    // Add metadata buffer
-    flatbuffers::FlatBufferBuilder metadataFbb;
-    auto metadataBuilder = MetadataBuilder(metadataFbb);
-    metadataBuilder.add_revision(newRevision);
-    metadataBuilder.add_operation(Operation_Creation);
-    metadataBuilder.add_replayToSource(replayToSource);
-    auto metadataBuffer = metadataBuilder.Finish();
-    FinishMetadataBuffer(metadataFbb, metadataBuffer);
 
-    flatbuffers::FlatBufferBuilder fbb;
-    adaptorFactory->createBuffer(memoryAdaptor, fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize());
+    d->revisionChanged = true;
+    auto revision = d->entityStore.maxRevision();
+    auto o = Sink::ApplicationDomain::ApplicationDomainType{d->resourceContext.instanceId(), key, revision, memoryAdaptor};
+    o.setChangedProperties(o.availableProperties().toSet());
 
-    d->storeNewRevision(newRevision, fbb, bufferType, key);
+    auto preprocess = [&, this](ApplicationDomain::ApplicationDomainType &newEntity) {
+        foreach (const auto &processor, d->processors[bufferType]) {
+            processor->newEntity(newEntity);
+        }
+    };
 
-    //FIXME entityStore->create(bufferType, memoryAdaptor, replayToSource)
+    if (!d->entityStore.add(bufferType, o, replayToSource, preprocess)) {
+        return KAsync::error<qint64>(0);
+    }
 
-    return KAsync::value(newRevision);
+    return KAsync::value(d->entityStore.maxRevision());
 }
 
 KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
@@ -254,6 +204,7 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
     }
     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());
     SinkTrace() << "Modified Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
@@ -269,7 +220,6 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
         }
     }
 
-    // TODO use only readPropertyMapper and writePropertyMapper
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceContext.resourceType, bufferType);
     if (!adaptorFactory) {
         SinkWarning() << "no adaptor factory for type " << bufferType;
@@ -278,72 +228,26 @@ KAsync::Job<qint64> Pipeline::modifiedEntity(void const *command, size_t size)
 
     auto diffEntity = GetEntity(modifyEntity->delta()->Data());
     Q_ASSERT(diffEntity);
-    auto diff = adaptorFactory->createAdaptor(*diffEntity);
-
-    QSharedPointer<ApplicationDomain::BufferAdaptor> current;
-    DataStore::mainDatabase(d->transaction, bufferType)
-        .findLatest(key,
-            [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
-                EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
-                if (!buffer.isValid()) {
-                    SinkWarning() << "Read invalid buffer from disk";
-                } else {
-                    current = adaptorFactory->createAdaptor(buffer.entity());
-                }
-                return false;
-            },
-            [baseRevision](const DataStore::Error &error) { SinkWarning() << "Failed to read old revision from storage: " << error.message << "Revision: " << baseRevision; });
-
-    if (!current) {
-        SinkWarning() << "Failed to read local value " << key;
-        return KAsync::error<qint64>(0);
-    }
-
-    auto newAdaptor = QSharedPointer<Sink::ApplicationDomain::MemoryBufferAdaptor>::create(*(current), current->availableProperties());
-
-    // Apply diff
-    // FIXME only apply the properties that are available in the buffer
-    SinkTrace() << "Applying changed properties: " << changeset;
-    for (const auto &property : changeset) {
-        const auto value = diff->getProperty(property);
-        if (value.isValid()) {
-            newAdaptor->setProperty(property, value);
-        }
-    }
+    Sink::ApplicationDomain::ApplicationDomainType diff{d->resourceContext.instanceId(), key, baseRevision, adaptorFactory->createAdaptor(*diffEntity)};
+    diff.setChangedProperties(changeset.toSet());
 
-    // Remove deletions
+    QByteArrayList deletions;
     if (modifyEntity->deletions()) {
-        for (const flatbuffers::String *property : *modifyEntity->deletions()) {
-            newAdaptor->setProperty(BufferUtils::extractBuffer(property), QVariant());
-        }
+        deletions = BufferUtils::fromVector(*modifyEntity->deletions());
     }
 
-    newAdaptor->resetChangedProperties();
-    foreach (const auto &processor, d->processors[bufferType]) {
-        processor->modifiedEntity(key, DataStore::maxRevision(d->transaction) + 1, *current, *newAdaptor, d->transaction);
-    }
-    //The maxRevision may have changed meanwhile if the entity created sub-entities
-    const qint64 newRevision = DataStore::maxRevision(d->transaction) + 1;
+    auto preprocess = [&, this](const ApplicationDomain::ApplicationDomainType &oldEntity, ApplicationDomain::ApplicationDomainType &newEntity) {
+        foreach (const auto &processor, d->processors[bufferType]) {
+            processor->modifiedEntity(oldEntity, newEntity);
+        }
+    };
 
-    // Add metadata buffer
-    flatbuffers::FlatBufferBuilder metadataFbb;
-    {
-        //We add availableProperties to account for the properties that have been changed by the preprocessors
-        auto modifiedProperties = BufferUtils::toVector(metadataFbb, changeset + newAdaptor->changedProperties());
-        auto metadataBuilder = MetadataBuilder(metadataFbb);
-        metadataBuilder.add_revision(newRevision);
-        metadataBuilder.add_operation(Operation_Modification);
-        metadataBuilder.add_replayToSource(replayToSource);
-        metadataBuilder.add_modifiedProperties(modifiedProperties);
-        auto metadataBuffer = metadataBuilder.Finish();
-        FinishMetadataBuffer(metadataFbb, metadataBuffer);
+    d->revisionChanged = true;
+    if (!d->entityStore.modify(bufferType, diff, deletions, replayToSource, preprocess)) {
+        return KAsync::error<qint64>(0);
     }
 
-    flatbuffers::FlatBufferBuilder fbb;
-    adaptorFactory->createBuffer(newAdaptor, fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize());
-
-    d->storeNewRevision(newRevision, fbb, bufferType, key);
-    return KAsync::value(newRevision);
+    return KAsync::value(d->entityStore.maxRevision());
 }
 
 KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
@@ -364,106 +268,38 @@ KAsync::Job<qint64> Pipeline::deletedEntity(void const *command, size_t size)
     const QByteArray key = QByteArray(reinterpret_cast<char const *>(deleteEntity->entityId()->Data()), deleteEntity->entityId()->size());
     SinkTrace() << "Deleted Entity. Type: " << bufferType << "uid: "<< key << " replayToSource: " << replayToSource;
 
-    bool found = false;
-    bool alreadyRemoved = false;
-    DataStore::mainDatabase(d->transaction, bufferType)
-        .findLatest(key,
-            [&found, &alreadyRemoved](const QByteArray &key, const QByteArray &data) -> bool {
-                auto entity = GetEntity(data.data());
-                if (entity && entity->metadata()) {
-                    auto metadata = GetMetadata(entity->metadata()->Data());
-                    found = true;
-                    if (metadata->operation() == Operation_Removal) {
-                        alreadyRemoved = true;
-                    }
-                }
-                return false;
-            },
-            [](const DataStore::Error &error) { SinkWarning() << "Failed to read old revision from storage: " << error.message; });
-
-    if (!found) {
-        SinkWarning() << "Failed to find entity " << key;
-        return KAsync::error<qint64>(0);
-    }
-    if (alreadyRemoved) {
-        SinkWarning() << "Entity is already removed " << key;
-        return KAsync::error<qint64>(0);
-    }
-
-    const qint64 newRevision = DataStore::maxRevision(d->transaction) + 1;
-
-    // Add metadata buffer
-    flatbuffers::FlatBufferBuilder metadataFbb;
-    auto metadataBuilder = MetadataBuilder(metadataFbb);
-    metadataBuilder.add_revision(newRevision);
-    metadataBuilder.add_operation(Operation_Removal);
-    metadataBuilder.add_replayToSource(replayToSource);
-    auto metadataBuffer = metadataBuilder.Finish();
-    FinishMetadataBuffer(metadataFbb, metadataBuffer);
-
-    flatbuffers::FlatBufferBuilder fbb;
-    EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), 0, 0, 0, 0);
+    auto preprocess = [&, this](const ApplicationDomain::ApplicationDomainType &oldEntity) {
+        foreach (const auto &processor, d->processors[bufferType]) {
+            processor->deletedEntity(oldEntity);
+        }
+    };
 
-    auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceContext.resourceType, bufferType);
-    if (!adaptorFactory) {
-        SinkWarning() << "no adaptor factory for type " << bufferType;
+    d->revisionChanged = true;
+    if (!d->entityStore.remove(bufferType, key, replayToSource, preprocess)) {
         return KAsync::error<qint64>(0);
     }
 
-    QSharedPointer<ApplicationDomain::BufferAdaptor> current;
-    DataStore::mainDatabase(d->transaction, bufferType)
-        .findLatest(key,
-            [this, bufferType, newRevision, adaptorFactory, key, &current](const QByteArray &, const QByteArray &data) -> bool {
-                EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
-                if (!buffer.isValid()) {
-                    SinkWarning() << "Read invalid buffer from disk";
-                } else {
-                    current = adaptorFactory->createAdaptor(buffer.entity());
-                }
-                return false;
-            },
-            [this](const DataStore::Error &error) { SinkError() << "Failed to find value in pipeline: " << error.message; });
-
-    d->storeNewRevision(newRevision, fbb, bufferType, key);
-
-    foreach (const auto &processor, d->processors[bufferType]) {
-        processor->deletedEntity(key, newRevision, *current, d->transaction);
-    }
-
-    return KAsync::value(newRevision);
+    return KAsync::value(d->entityStore.maxRevision());
 }
 
 void Pipeline::cleanupRevision(qint64 revision)
 {
+    d->entityStore.cleanupRevision(revision);
     d->revisionChanged = true;
-    const auto uid = DataStore::getUidFromRevision(d->transaction, revision);
-    const auto bufferType = DataStore::getTypeFromRevision(d->transaction, revision);
-    SinkTrace() << "Cleaning up revision " << revision << uid << bufferType;
-    DataStore::mainDatabase(d->transaction, bufferType)
-        .scan(uid,
-            [&](const QByteArray &key, const QByteArray &data) -> bool {
-                EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
-                if (!buffer.isValid()) {
-                    SinkWarning() << "Read invalid buffer from disk";
-                } else {
-                    const auto metadata = flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer());
-                    const qint64 rev = metadata->revision();
-                    // Remove old revisions, and the current if the entity has already been removed
-                    if (rev < revision || metadata->operation() == Operation_Removal) {
-                        DataStore::removeRevision(d->transaction, rev);
-                        DataStore::mainDatabase(d->transaction, bufferType).remove(key);
-                    }
-                }
-
-                return true;
-            },
-            [](const DataStore::Error &error) { SinkWarning() << "Error while reading: " << error.message; }, true);
-    DataStore::setCleanedUpRevision(d->transaction, revision);
 }
 
 qint64 Pipeline::cleanedUpRevision()
 {
-    return DataStore::cleanedUpRevision(d->transaction);
+    /* return d->entityStore.cleanedUpRevision(); */
+    /* return DataStore::cleanedUpRevision(d->transaction); */
+    //FIXME Just move the whole cleanup revision iteration into the entitystore
+    return 0;
+}
+
+qint64 Pipeline::revision()
+{
+    //FIXME Just move the whole cleanup revision iteration into the entitystore
+    return 0;
 }
 
 class Preprocessor::Private {
@@ -492,7 +328,7 @@ void Preprocessor::startBatch()
 {
 }
 
-void Preprocessor::finalize()
+void Preprocessor::finalizeBatch()
 {
 }
 
@@ -510,7 +346,6 @@ void Preprocessor::createEntity(const Sink::ApplicationDomain::ApplicationDomain
 
     flatbuffers::FlatBufferBuilder fbb;
     auto entityId = fbb.CreateString(entity.identifier());
-    // This is the resource buffer type and not the domain type
     auto type = fbb.CreateString(typeName);
     auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityBuffer.constData(), entityBuffer.size());
     auto location = Sink::Commands::CreateCreateEntity(fbb, entityId, type, delta);