16 files changed, 267 insertions, 55 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 2f779b5..25ea667 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -34,6 +34,7 @@ set(command_SRCS
     resourceconfig.cpp
     domain/applicationdomaintype.cpp
     domain/event.cpp
+    domain/mail.cpp
     ${storage_SRCS})
 
 add_library(${PROJECT_NAME} SHARED ${command_SRCS})
@@ -51,6 +52,7 @@ generate_flatbuffers(
     commands/synchronize
     commands/notification
     domain/event
+    domain/mail
     entity
     metadata
     queuedcommand
diff --git a/common/domain/applicationdomaintype.cpp b/common/domain/applicationdomaintype.cpp
index 47ff0c3..3cc075b 100644
--- a/common/domain/applicationdomaintype.cpp
+++ b/common/domain/applicationdomaintype.cpp
@@ -40,6 +40,18 @@ QByteArray getTypeName<AkonadiResource>()
     return "akonadiresource";
 }
 
+template<>
+QByteArray getTypeName<Mail>()
+{
+    return "mail";
+}
+
+template<>
+QByteArray getTypeName<Folder>()
+{
+    return "folder";
+}
+
 }
 }
 
diff --git a/common/domain/applicationdomaintype.h b/common/domain/applicationdomaintype.h
index 29bebcf..e0a6de0 100644
--- a/common/domain/applicationdomaintype.h
+++ b/common/domain/applicationdomaintype.h
@@ -112,10 +112,14 @@ struct Calendar : public ApplicationDomainType {
     using ApplicationDomainType::ApplicationDomainType;
 };
 
-class Mail : public ApplicationDomainType {
+struct Mail : public ApplicationDomainType {
+    typedef QSharedPointer<Mail> Ptr;
+    using ApplicationDomainType::ApplicationDomainType;
 };
 
-class Folder : public ApplicationDomainType {
+struct Folder : public ApplicationDomainType {
+    typedef QSharedPointer<Folder> Ptr;
+    using ApplicationDomainType::ApplicationDomainType;
 };
 
 /**
@@ -146,6 +150,12 @@ QByteArray getTypeName<Todo>();
 template<>
 QByteArray getTypeName<AkonadiResource>();
 
+template<>
+QByteArray getTypeName<Mail>();
+
+template<>
+QByteArray getTypeName<Folder>();
+
 /**
  * Type implementation.
  * 
diff --git a/common/domain/mail.cpp b/common/domain/mail.cpp
new file mode 100644
index 0000000..230ef31
--- /dev/null
+++ b/common/domain/mail.cpp
@@ -0,0 +1,104 @@
+/*
+ *   Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ *   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 2 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, write to the
+ *   Free Software Foundation, Inc.,
+ *   51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA.
+ */
+#include "mail.h"
+
+#include <QVector>
+#include <QByteArray>
+#include <QString>
+
+#include "../resultset.h"
+#include "../index.h"
+#include "../storage.h"
+#include "../log.h"
+#include "../propertymapper.h"
+#include "../query.h"
+#include "../definitions.h"
+
+#include "mail_generated.h"
+
+using namespace Akonadi2::ApplicationDomain;
+
+ResultSet TypeImplementation<Mail>::queryIndexes(const Akonadi2::Query &query, const QByteArray &resourceInstanceIdentifier, QSet<QByteArray> &appliedFilters, Akonadi2::Storage::Transaction &transaction)
+{
+    QVector<QByteArray> keys;
+    if (query.propertyFilter.contains("uid")) {
+        Index uidIndex("mail.index.uid", transaction);
+        uidIndex.lookup(query.propertyFilter.value("uid").toByteArray(), [&](const QByteArray &value) {
+            keys << value;
+        },
+        [](const Index::Error &error) {
+            Warning() << "Error in uid index: " <<  error.message;
+        });
+        appliedFilters << "uid";
+    }
+    return ResultSet(keys);
+}
+
+void TypeImplementation<Mail>::index(const Mail &type, Akonadi2::Storage::Transaction &transaction)
+{
+    const auto uid = type.getProperty("uid");
+    if (uid.isValid()) {
+        Index uidIndex("mail.index.uid", transaction);
+        uidIndex.add(uid.toByteArray(), type.identifier());
+    }
+}
+
+QSharedPointer<ReadPropertyMapper<TypeImplementation<Mail>::Buffer> > TypeImplementation<Mail>::initializeReadPropertyMapper()
+{
+    auto propertyMapper = QSharedPointer<ReadPropertyMapper<Buffer> >::create();
+    propertyMapper->addMapping("uid", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<QString>(buffer->uid());
+    });
+    propertyMapper->addMapping("sender", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<QString>(buffer->sender());
+    });
+    propertyMapper->addMapping("senderName", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<QString>(buffer->senderName());
+    });
+    propertyMapper->addMapping("subject", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<QString>(buffer->subject());
+    });
+    propertyMapper->addMapping("date", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<QString>(buffer->date());
+    });
+    propertyMapper->addMapping("unread", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<bool>(buffer->unread());
+    });
+    propertyMapper->addMapping("important", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<bool>(buffer->important());
+    });
+    propertyMapper->addMapping("folder", [](Buffer const *buffer) -> QVariant {
+        return propertyToVariant<QString>(buffer->folder());
+    });
+    return propertyMapper;
+}
+
+QSharedPointer<WritePropertyMapper<TypeImplementation<Mail>::BufferBuilder> > TypeImplementation<Mail>::initializeWritePropertyMapper()
+{
+    auto propertyMapper = QSharedPointer<WritePropertyMapper<BufferBuilder> >::create();
+    // propertyMapper->addMapping("summary", [](const QVariant &value, flatbuffers::FlatBufferBuilder &fbb) -> std::function<void(BufferBuilder &)> {
+    //     auto offset = variantToProperty<QString>(value, fbb);
+    //     return [offset](BufferBuilder &builder) { builder.add_summary(offset); };
+    // });
+    propertyMapper->addMapping("uid", [](const QVariant &value, flatbuffers::FlatBufferBuilder &fbb) -> std::function<void(BufferBuilder &)> {
+        auto offset = variantToProperty<QString>(value, fbb);
+        return [offset](BufferBuilder &builder) { builder.add_uid(offset); };
+    });
+    return propertyMapper;
+}
diff --git a/common/domain/mail.fbs b/common/domain/mail.fbs
new file mode 100644
index 0000000..13aa36d
--- /dev/null
+++ b/common/domain/mail.fbs
@@ -0,0 +1,15 @@
+namespace Akonadi2.ApplicationDomain.Buffer;
+
+table Mail {
+  uid:string;
+  folder:string;
+  sender:string;
+  senderName:string;
+  subject:string;
+  date:string;
+  unread:bool = false;
+  important:bool = false;
+}
+
+root_type Mail;
+file_identifier "AKFB";
diff --git a/common/domain/mail.h b/common/domain/mail.h
new file mode 100644
index 0000000..b58ce44
--- /dev/null
+++ b/common/domain/mail.h
@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ *   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 2 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, write to the
+ *   Free Software Foundation, Inc.,
+ *   51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA.
+ */
+#pragma once
+
+#include "applicationdomaintype.h"
+
+#include "storage.h"
+
+class ResultSet;
+class QByteArray;
+
+template<typename T>
+class ReadPropertyMapper;
+template<typename T>
+class WritePropertyMapper;
+
+namespace Akonadi2 {
+    class Query;
+
+namespace ApplicationDomain {
+    namespace Buffer {
+        struct Mail;
+        struct MailBuilder;
+    }
+
+template<>
+class TypeImplementation<Akonadi2::ApplicationDomain::Mail> {
+public:
+    typedef Akonadi2::ApplicationDomain::Buffer::Mail Buffer;
+    typedef Akonadi2::ApplicationDomain::Buffer::MailBuilder BufferBuilder;
+    static QSet<QByteArray> indexedProperties();
+    /**
+     * Returns the potential result set based on the indexes.
+     * 
+     * An empty result set indicates that a full scan is required.
+     */
+    static ResultSet queryIndexes(const Akonadi2::Query &query, const QByteArray &resourceInstanceIdentifier, QSet<QByteArray> &appliedFilters, Akonadi2::Storage::Transaction &transaction);
+    static void index(const Mail &type, Akonadi2::Storage::Transaction &transaction);
+    static QSharedPointer<ReadPropertyMapper<Buffer> > initializeReadPropertyMapper();
+    static QSharedPointer<WritePropertyMapper<BufferBuilder> > initializeWritePropertyMapper();
+};
+
+}
+}
diff --git a/common/domainadaptor.h b/common/domainadaptor.h
index f9dcc79..4943cc0 100644
--- a/common/domainadaptor.h
+++ b/common/domainadaptor.h
@@ -25,6 +25,7 @@
 
 #include "domain/applicationdomaintype.h"
 #include "domain/event.h"
+#include "domain/mail.h"
 #include "entity_generated.h"
 #include "metadata_generated.h"
 #include "entitybuffer.h"
diff --git a/common/entitystorage.cpp b/common/entitystorage.cpp
index 8a3391e..bcc3562 100644
--- a/common/entitystorage.cpp
+++ b/common/entitystorage.cpp
@@ -19,9 +19,9 @@
 
 #include "entitystorage.h"
 
-static void scan(const Akonadi2::Storage::Transaction &transaction, const QByteArray &key, std::function<bool(const QByteArray &key, const Akonadi2::Entity &entity)> callback)
+static void scan(const Akonadi2::Storage::Transaction &transaction, const QByteArray &key, std::function<bool(const QByteArray &key, const Akonadi2::Entity &entity)> callback, const QByteArray &bufferType)
 {
-    transaction.openDatabase().scan(key, [=](const QByteArray &key, const QByteArray &value) -> bool {
+    transaction.openDatabase(bufferType + ".main").scan(key, [=](const QByteArray &key, const QByteArray &value) -> bool {
         //Skip internals
         if (Akonadi2::Storage::isInternalKey(key)) {
             return true;
@@ -58,17 +58,17 @@ void EntityStorageBase::readValue(const Akonadi2::Storage::Transaction &transact
         auto domainObject = create(key, revision, mDomainTypeAdaptorFactory->createAdaptor(entity));
         resultCallback(domainObject);
         return true;
-    });
+    }, mBufferType);
 }
 
-static ResultSet fullScan(const Akonadi2::Storage::Transaction &transaction)
+static ResultSet fullScan(const Akonadi2::Storage::Transaction &transaction, const QByteArray &bufferType)
 {
     //TODO use a result set with an iterator, to read values on demand
     QVector<QByteArray> keys;
     scan(transaction, QByteArray(), [=, &keys](const QByteArray &key, const Akonadi2::Entity &) {
         keys << key;
         return true;
-    });
+    }, bufferType);
     Trace() << "Full scan found " << keys.size() << " results";
     return ResultSet(keys);
 }
@@ -99,7 +99,7 @@ ResultSet EntityStorageBase::getResultSet(const Akonadi2::Query &query, Akonadi2
 
     //We do a full scan if there were no indexes available to create the initial set.
     if (appliedFilters.isEmpty()) {
-        resultSet = fullScan(transaction);
+        resultSet = fullScan(transaction, mBufferType);
     }
 
     auto filter = [remainingFilters, query, baseRevision, topRevision](const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject) -> bool {
diff --git a/common/entitystorage.h b/common/entitystorage.h
index 8256938..9d928b8 100644
--- a/common/entitystorage.h
+++ b/common/entitystorage.h
@@ -52,6 +52,7 @@ protected:
 
 protected:
     QByteArray mResourceInstanceIdentifier;
+    QByteArray mBufferType;
     DomainTypeAdaptorFactoryInterface::Ptr mDomainTypeAdaptorFactory;
 };
 
@@ -60,10 +61,10 @@ class EntityStorage : public EntityStorageBase
 {
 
 public:
-    EntityStorage(const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &adaptorFactory)
+    EntityStorage(const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &adaptorFactory, const QByteArray &bufferType)
         : EntityStorageBase(instanceIdentifier, adaptorFactory)
     {
-
+        mBufferType = bufferType;
     }
 
 protected:
@@ -84,7 +85,7 @@ protected:
 
 public:
 
-    virtual void read(const Akonadi2::Query &query, const QPair<qint64, qint64> &revisionRange, const QSharedPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > &resultProvider) 
+    virtual void read(const Akonadi2::Query &query, const QPair<qint64, qint64> &revisionRange, const QSharedPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > &resultProvider)
     {
         Akonadi2::Storage storage(Akonadi2::storageLocation(), mResourceInstanceIdentifier);
         storage.setDefaultErrorHandler([](const Akonadi2::Storage::Error &error) {
diff --git a/common/facade.h b/common/facade.h
index be053f6..d53ec4a 100644
--- a/common/facade.h
+++ b/common/facade.h
@@ -109,7 +109,7 @@ public:
     GenericFacade(const QByteArray &resourceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &adaptorFactory = DomainTypeAdaptorFactoryInterface::Ptr(), const QSharedPointer<EntityStorage<DomainType> > storage = QSharedPointer<EntityStorage<DomainType> >(), const QSharedPointer<Akonadi2::ResourceAccessInterface> resourceAccess = QSharedPointer<Akonadi2::ResourceAccessInterface>())
         : Akonadi2::StoreFacade<DomainType>(),
         mResourceAccess(resourceAccess),
-        mStorage(storage ? storage : QSharedPointer<EntityStorage<DomainType> >::create(resourceIdentifier, adaptorFactory)),
+        mStorage(storage ? storage : QSharedPointer<EntityStorage<DomainType> >::create(resourceIdentifier, adaptorFactory, bufferTypeForDomainType())),
         mDomainTypeAdaptorFactory(adaptorFactory),
         mResourceInstanceIdentifier(resourceIdentifier)
     {
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 8ef6187..33e5d5c 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -143,7 +143,7 @@ KAsync::Job<void> Pipeline::newEntity(void const *command, size_t size)
     auto createEntity = Akonadi2::Commands::GetCreateEntity(command);
 
     //TODO rename createEntitiy->domainType to bufferType
-    const QString entityType = QString::fromUtf8(reinterpret_cast<char const*>(createEntity->domainType()->Data()), createEntity->domainType()->size());
+    const QByteArray bufferType = QByteArray(reinterpret_cast<char const*>(createEntity->domainType()->Data()), createEntity->domainType()->size());
     {
         flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
         if (!Akonadi2::VerifyEntityBuffer(verifyer)) {
@@ -152,6 +152,10 @@ KAsync::Job<void> Pipeline::newEntity(void const *command, size_t size)
         }
     }
     auto entity = Akonadi2::GetEntity(createEntity->delta()->Data());
+    if (!entity->resource()->size() && !entity->local()->size()) {
+        Warning() << "No local and no resource buffer while trying to create entity.";
+        return KAsync::error<void>();
+    }
 
     //Add metadata buffer
     flatbuffers::FlatBufferBuilder metadataFbb;
@@ -165,14 +169,14 @@ KAsync::Job<void> Pipeline::newEntity(void const *command, size_t size)
     flatbuffers::FlatBufferBuilder fbb;
     EntityBuffer::assembleEntityBuffer(fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize(), entity->resource()->Data(), entity->resource()->size(), entity->local()->Data(), entity->local()->size());
 
-    d->transaction.openDatabase().write(key, QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
+    d->transaction.openDatabase(bufferType + ".main").write(key, QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
     Akonadi2::Storage::setMaxRevision(d->transaction, newRevision);
-    Log() << "Pipeline: wrote entity: " << key << newRevision;
+    Log() << "Pipeline: wrote entity: " << key << newRevision << bufferType;
 
-    return KAsync::start<void>([this, key, entityType, newRevision](KAsync::Future<void> &future) {
-        PipelineState state(this, NewPipeline, key, d->newPipeline[entityType], newRevision, [&future]() {
+    return KAsync::start<void>([this, key, bufferType, newRevision](KAsync::Future<void> &future) {
+        PipelineState state(this, NewPipeline, key, d->newPipeline[bufferType], newRevision, [&future]() {
             future.setFinished();
-        });
+        }, bufferType);
         d->activePipelines << state;
         state.step();
     });
@@ -195,10 +199,10 @@ KAsync::Job<void> Pipeline::modifiedEntity(void const *command, size_t size)
     Q_ASSERT(modifyEntity);
 
     //TODO rename modifyEntity->domainType to bufferType
-    const QByteArray entityType = QByteArray(reinterpret_cast<char const*>(modifyEntity->domainType()->Data()), modifyEntity->domainType()->size());
+    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());
-    if (entityType.isEmpty() || key.isEmpty()) {
-        Warning() << "entity type or key " << entityType << key;
+    if (bufferType.isEmpty() || key.isEmpty()) {
+        Warning() << "entity type or key " << bufferType << key;
         return KAsync::error<void>();
     }
     {
@@ -209,9 +213,9 @@ KAsync::Job<void> Pipeline::modifiedEntity(void const *command, size_t size)
         }
     }
 
-    auto adaptorFactory = d->adaptorFactory.value(entityType);
+    auto adaptorFactory = d->adaptorFactory.value(bufferType);
     if (!adaptorFactory) {
-        Warning() << "no adaptor factory for type " << entityType;
+        Warning() << "no adaptor factory for type " << bufferType;
         return KAsync::error<void>();
     }
 
@@ -220,7 +224,7 @@ KAsync::Job<void> Pipeline::modifiedEntity(void const *command, size_t size)
     auto diff = adaptorFactory->createAdaptor(*diffEntity);
 
     QSharedPointer<Akonadi2::ApplicationDomain::BufferAdaptor> current;
-    storage().createTransaction(Akonadi2::Storage::ReadOnly).openDatabase().scan(key, [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
+    storage().createTransaction(Akonadi2::Storage::ReadOnly).openDatabase(bufferType + ".main").scan(key, [&current, adaptorFactory](const QByteArray &key, const QByteArray &data) -> bool {
         Akonadi2::EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
         if (!buffer.isValid()) {
             Warning() << "Read invalid buffer from disk";
@@ -228,6 +232,9 @@ KAsync::Job<void> Pipeline::modifiedEntity(void const *command, size_t size)
             current = adaptorFactory->createAdaptor(buffer.entity());
         }
         return false;
+    },
+    [](const Storage::Error &error) {
+        Warning() << "Failed to read value from storage: " << error.message;
     });
     //TODO error handler
 
@@ -265,13 +272,13 @@ KAsync::Job<void> Pipeline::modifiedEntity(void const *command, size_t size)
     adaptorFactory->createBuffer(*newObject, fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize());
 
     //TODO don't overwrite the old entry, but instead store a new revision
-    d->transaction.openDatabase().write(key, QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
+    d->transaction.openDatabase(bufferType + ".main").write(key, QByteArray::fromRawData(reinterpret_cast<char const *>(fbb.GetBufferPointer()), fbb.GetSize()));
     Akonadi2::Storage::setMaxRevision(d->transaction, newRevision);
 
-    return KAsync::start<void>([this, key, entityType, newRevision](KAsync::Future<void> &future) {
-        PipelineState state(this, ModifiedPipeline, key, d->modifiedPipeline[entityType], newRevision, [&future]() {
+    return KAsync::start<void>([this, key, bufferType, newRevision](KAsync::Future<void> &future) {
+        PipelineState state(this, ModifiedPipeline, key, d->modifiedPipeline[bufferType], newRevision, [&future]() {
             future.setFinished();
-        });
+        }, bufferType);
         d->activePipelines << state;
         state.step();
     });
@@ -292,18 +299,18 @@ KAsync::Job<void> Pipeline::deletedEntity(void const *command, size_t size)
     }
     auto deleteEntity = Akonadi2::Commands::GetDeleteEntity(command);
 
-    const QByteArray entityType = QByteArray(reinterpret_cast<char const*>(deleteEntity->domainType()->Data()), deleteEntity->domainType()->size());
+    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());
 
     //TODO instead of deleting the entry, a new revision should be created that marks the entity as deleted
-    d->transaction.openDatabase().remove(key);
+    d->transaction.openDatabase(bufferType + ".main").remove(key);
     Akonadi2::Storage::setMaxRevision(d->transaction, newRevision);
     Log() << "Pipeline: deleted entity: "<< newRevision;
 
-    return KAsync::start<void>([this, key, entityType, newRevision](KAsync::Future<void> &future) {
-        PipelineState state(this, DeletedPipeline, key, d->deletedPipeline[entityType], newRevision, [&future](){
+    return KAsync::start<void>([this, key, bufferType, newRevision](KAsync::Future<void> &future) {
+        PipelineState state(this, DeletedPipeline, key, d->deletedPipeline[bufferType], newRevision, [&future](){
             future.setFinished();
-        });
+        }, bufferType);
         d->activePipelines << state;
         state.step();
     });
@@ -354,14 +361,15 @@ void Pipeline::pipelineCompleted(PipelineState state)
 class PipelineState::Private : public QSharedData
 {
 public:
-    Private(Pipeline *p, Pipeline::Type t, const QByteArray &k, QVector<Preprocessor *> filters, const std::function<void()> &c, qint64 r)
+    Private(Pipeline *p, Pipeline::Type t, const QByteArray &k, QVector<Preprocessor *> filters, const std::function<void()> &c, qint64 r, const QByteArray &b)
         : pipeline(p),
           type(t),
           key(k),
           filterIt(filters),
           idle(true),
           callback(c),
-          revision(r)
+          revision(r),
+          bufferType(b)
     {}
 
     Private()
@@ -378,6 +386,7 @@ public:
     bool idle;
     std::function<void()> callback;
     qint64 revision;
+    QByteArray bufferType;
 };
 
 PipelineState::PipelineState()
@@ -386,8 +395,8 @@ PipelineState::PipelineState()
 
 }
 
-PipelineState::PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<Preprocessor *> &filters, qint64 revision, const std::function<void()> &callback)
-    : d(new Private(pipeline, type, key, filters, callback, revision))
+PipelineState::PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<Preprocessor *> &filters, qint64 revision, const std::function<void()> &callback, const QByteArray &bufferType)
+    : d(new Private(pipeline, type, key, filters, callback, revision, bufferType))
 {
 }
 
@@ -431,6 +440,11 @@ qint64 PipelineState::revision() const
     return d->revision;
 }
 
+QByteArray PipelineState::bufferType() const
+{
+    return d->bufferType;
+}
+
 void PipelineState::step()
 {
     if (!d->pipeline) {
diff --git a/common/pipeline.h b/common/pipeline.h
index a3b3735..573af73 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -85,7 +85,7 @@ class AKONADI2COMMON_EXPORT PipelineState
 {
 public:
     PipelineState();
-    PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<Preprocessor *> &filters, qint64 revision, const std::function<void()> &callback);
+    PipelineState(Pipeline *pipeline, Pipeline::Type type, const QByteArray &key, const QVector<Preprocessor *> &filters, qint64 revision, const std::function<void()> &callback, const QByteArray &bufferType);
     PipelineState(const PipelineState &other);
     ~PipelineState();
 
@@ -96,7 +96,7 @@ public:
     QByteArray key() const;
     Pipeline::Type type() const;
     qint64 revision() const;
-    //TODO expose command
+    QByteArray bufferType() const;
 
     void step();
     void processingCompleted(Preprocessor *filter);
@@ -114,7 +114,6 @@ public:
     Preprocessor();
     virtual ~Preprocessor();
 
-    //TODO pass actual command as well, for changerecording
     virtual void process(const PipelineState &state, Akonadi2::Storage::Transaction &transaction);
     //TODO to record progress
     virtual QString id() const;
@@ -142,7 +141,7 @@ public:
 
     void process(const PipelineState &state, Akonadi2::Storage::Transaction &transaction) Q_DECL_OVERRIDE
     {
-        transaction.openDatabase().scan(state.key(), [this, &state, &transaction](const QByteArray &key, const QByteArray &value) -> bool {
+        transaction.openDatabase(state.bufferType() + ".main").scan(state.key(), [this, &state, &transaction](const QByteArray &key, const QByteArray &value) -> bool {
             auto entity = Akonadi2::GetEntity(value);
             mFunction(state, *entity, transaction);
             processingCompleted(state);
diff --git a/common/propertymapper.cpp b/common/propertymapper.cpp
index 89495ae..7ff072a 100644
--- a/common/propertymapper.cpp
+++ b/common/propertymapper.cpp
@@ -38,3 +38,9 @@ QVariant propertyToVariant<QString>(const flatbuffers::String *property)
     return QVariant();
 }
 
+template <>
+QVariant propertyToVariant<bool>(uint8_t property)
+{
+    return static_cast<bool>(property);
+}
+
diff --git a/common/propertymapper.h b/common/propertymapper.h
index 0c6c16f..72468e2 100644
--- a/common/propertymapper.h
+++ b/common/propertymapper.h
@@ -35,6 +35,8 @@ flatbuffers::uoffset_t variantToProperty(const QVariant &, flatbuffers::FlatBuff
  */
 template <typename T>
 QVariant propertyToVariant(const flatbuffers::String *);
+template <typename T>
+QVariant propertyToVariant(uint8_t);
 
 
 /**
diff --git a/common/storage.h b/common/storage.h
index 191f535..d186b2e 100644
--- a/common/storage.h
+++ b/common/storage.h
@@ -161,8 +161,6 @@ public:
     qint64 diskUsage() const;
     void removeFromDisk() const;
 
-    qint64 maxRevision();
-    void setMaxRevision(qint64 revision);
     static qint64 maxRevision(const Akonadi2::Storage::Transaction &);
     static void setMaxRevision(Akonadi2::Storage::Transaction &, qint64 revision);
 
diff --git a/common/storage_common.cpp b/common/storage_common.cpp
index a506cf8..f22150a 100644
--- a/common/storage_common.cpp
+++ b/common/storage_common.cpp
@@ -54,23 +54,11 @@ std::function<void(const Storage::Error &error)> Storage::defaultErrorHandler()
     return basicErrorHandler();
 }
 
-void Storage::setMaxRevision(qint64 revision)
-{
-    auto transaction = createTransaction(Akonadi2::Storage::ReadWrite);
-    setMaxRevision(transaction, revision);
-}
-
 void Storage::setMaxRevision(Akonadi2::Storage::Transaction &transaction, qint64 revision)
 {
     transaction.openDatabase().write("__internal_maxRevision", QByteArray::number(revision));
 }
 
-qint64 Storage::maxRevision()
-{
-    auto transaction = createTransaction(Akonadi2::Storage::ReadOnly);
-    return maxRevision(transaction);
-}
-
 qint64 Storage::maxRevision(const Akonadi2::Storage::Transaction &transaction)
 {
     qint64 r = 0;