A first draft of the threading algorithm.

1 files changed, 102 insertions, 15 deletions

diff --git a/common/domain/mail.cpp b/common/domain/mail.cpp
index bb5ad58..859ebef 100644
--- a/common/domain/mail.cpp
+++ b/common/domain/mail.cpp
@@ -41,6 +41,7 @@ SINK_DEBUG_AREA("mail");
 
 static QMutex sMutex;
 
+using namespace Sink;
 using namespace Sink::ApplicationDomain;
 
 static TypeIndex &getIndex()
@@ -49,26 +50,54 @@ static TypeIndex &getIndex()
     static TypeIndex *index = 0;
     if (!index) {
         index = new TypeIndex("mail");
-        index->addProperty<QByteArray>("uid");
-        index->addProperty<QByteArray>("sender");
-        index->addProperty<QByteArray>("senderName");
-        index->addProperty<QString>("subject");
-        index->addProperty<QDateTime>("date");
-        index->addProperty<QByteArray>("folder");
-        index->addPropertyWithSorting<QByteArray, QDateTime>("folder", "date");
+        index->addProperty<QByteArray>(Mail::Uid::name);
+        index->addProperty<QByteArray>(Mail::Sender::name);
+        index->addProperty<QByteArray>(Mail::SenderName::name);
+        index->addProperty<QString>(Mail::Subject::name);
+        index->addProperty<QDateTime>(Mail::Date::name);
+        index->addProperty<QByteArray>(Mail::Folder::name);
+        index->addPropertyWithSorting<QByteArray, QDateTime>(Mail::Folder::name, Mail::Date::name);
+        index->addProperty<QByteArray>(Mail::MessageId::name);
+        index->addProperty<QByteArray>(Mail::ParentMessageId::name);
     }
     return *index;
 }
 
-ResultSet TypeImplementation<Mail>::queryIndexes(const Sink::Query &query, const QByteArray &resourceInstanceIdentifier, QSet<QByteArray> &appliedFilters, QByteArray &appliedSorting, Sink::Storage::Transaction &transaction)
+static void updateThreadingIndex(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Sink::Storage::Transaction &transaction)
 {
-    return getIndex().query(query, appliedFilters, appliedSorting, transaction);
+    auto messageId = bufferAdaptor.getProperty(Mail::MessageId::name).toByteArray();
+    auto parentMessageId = bufferAdaptor.getProperty(Mail::ParentMessageId::name).toByteArray();
+
+    Index msgIdIndex("msgId", transaction);
+    Index msgIdThreadIdIndex("msgIdThreadId", transaction);
+
+    //Add the message to the index
+    Q_ASSERT(msgIdIndex.lookup(messageId).isEmpty());
+    msgIdIndex.add(messageId, identifier);
+
+    //If parent is already available, add to thread of parent
+    QByteArray thread;
+    if (!parentMessageId.isEmpty() && !msgIdIndex.lookup(parentMessageId).isEmpty()) {
+        thread = msgIdThreadIdIndex.lookup(parentMessageId);
+        msgIdThreadIdIndex.add(messageId, thread);
+    } else {
+        thread = QUuid::createUuid().toByteArray();
+        if (!parentMessageId.isEmpty()) {
+            //Register parent with thread for when it becomes available
+            msgIdThreadIdIndex.add(parentMessageId, thread);
+        }
+    }
+    Q_ASSERT(!thread.isEmpty());
+    msgIdThreadIdIndex.add(messageId, thread);
+
+    //Look for parentMessageId and resolve to local id if available
 }
 
 void TypeImplementation<Mail>::index(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Sink::Storage::Transaction &transaction)
 {
     SinkTrace() << "Indexing " << identifier;
     getIndex().add(identifier, bufferAdaptor, transaction);
+    updateThreadingIndex(identifier, bufferAdaptor, transaction);
 }
 
 void TypeImplementation<Mail>::removeIndex(const QByteArray &identifier, const BufferAdaptor &bufferAdaptor, Sink::Storage::Transaction &transaction)
@@ -91,6 +120,8 @@ QSharedPointer<ReadPropertyMapper<TypeImplementation<Mail>::Buffer> > TypeImplem
     propertyMapper->addMapping<Mail::Draft, Buffer>(&Buffer::draft);
     propertyMapper->addMapping<Mail::Trash, Buffer>(&Buffer::trash);
     propertyMapper->addMapping<Mail::Sent, Buffer>(&Buffer::sent);
+    propertyMapper->addMapping<Mail::MessageId, Buffer>(&Buffer::messageId);
+    propertyMapper->addMapping<Mail::ParentMessageId, Buffer>(&Buffer::parentMessageId);
     return propertyMapper;
 }
 
@@ -110,16 +141,72 @@ QSharedPointer<WritePropertyMapper<TypeImplementation<Mail>::BufferBuilder> > Ty
     propertyMapper->addMapping<Mail::Draft>(&BufferBuilder::add_draft);
     propertyMapper->addMapping<Mail::Trash>(&BufferBuilder::add_trash);
     propertyMapper->addMapping<Mail::Sent>(&BufferBuilder::add_sent);
+    propertyMapper->addMapping<Mail::MessageId>(&BufferBuilder::add_messageId);
+    propertyMapper->addMapping<Mail::ParentMessageId>(&BufferBuilder::add_parentMessageId);
     return propertyMapper;
 }
 
-DataStoreQuery TypeImplementation<Mail>::prepareQuery(const Sink::Query &query, Sink::Storage::Transaction &transaction)
+class ThreadedDataStoreQuery : public DataStoreQuery
+{
+public:
+    typedef QSharedPointer<ThreadedDataStoreQuery> Ptr;
+    using DataStoreQuery::DataStoreQuery;
+
+protected:
+    ResultSet postSortFilter(ResultSet &resultSet) Q_DECL_OVERRIDE
+    {
+        auto query = mQuery;
+        if (query.threadLeaderOnly) {
+            auto rootCollection = QSharedPointer<QMap<QByteArray, QDateTime>>::create();
+            auto filter = [this, query, rootCollection](const QByteArray &uid, const Sink::EntityBuffer &entity) -> bool {
+                //TODO lookup thread
+                //if we got thread already in the result set compare dates and if newer replace
+                //else insert
+
+                const auto messageId = getProperty(entity.entity(), ApplicationDomain::Mail::MessageId::name).toByteArray();
+
+                Index msgIdIndex("msgId", mTransaction);
+                Index msgIdThreadIdIndex("msgIdThreadId", mTransaction);
+                auto thread = msgIdThreadIdIndex.lookup(messageId);
+                SinkTrace() << "MsgId: " << messageId << " Thread: " << thread << getProperty(entity.entity(), ApplicationDomain::Mail::Date::name).toDateTime();
+
+                if (rootCollection->contains(thread)) {
+                    auto date = rootCollection->value(thread);
+                    //The mail we have in our result already is newer, so we can ignore this one
+                    if (date > getProperty(entity.entity(), ApplicationDomain::Mail::Date::name).toDateTime()) {
+                        return false;
+                    }
+                    qWarning() << "############################################################################";
+                    qWarning() << "Found a newer mail, remove the old one";
+                    qWarning() << "############################################################################";
+                }
+                rootCollection->insert(thread, getProperty(entity.entity(), ApplicationDomain::Mail::Date::name).toDateTime());
+                return true;
+            };
+            return createFilteredSet(resultSet, filter);
+        } else {
+            return resultSet;
+        }
+    }
+};
+
+DataStoreQuery::Ptr TypeImplementation<Mail>::prepareQuery(const Sink::Query &query, Sink::Storage::Transaction &transaction)
 {
-    auto mapper = initializeReadPropertyMapper();
-    return DataStoreQuery(query, ApplicationDomain::getTypeName<Mail>(), transaction, getIndex(), [mapper](const Sink::Entity &entity, const QByteArray &property) {
+    if (query.threadLeaderOnly) {
+        auto mapper = initializeReadPropertyMapper();
+        return ThreadedDataStoreQuery::Ptr::create(query, ApplicationDomain::getTypeName<Mail>(), transaction, getIndex(), [mapper](const Sink::Entity &entity, const QByteArray &property) {
+
+            const auto localBuffer = Sink::EntityBuffer::readBuffer<Buffer>(entity.local());
+            return mapper->getProperty(property, localBuffer);
+        });
 
-        const auto localBuffer = Sink::EntityBuffer::readBuffer<Buffer>(entity.local());
-        return mapper->getProperty(property, localBuffer);
-    });
+    } else {
+        auto mapper = initializeReadPropertyMapper();
+        return DataStoreQuery::Ptr::create(query, ApplicationDomain::getTypeName<Mail>(), transaction, getIndex(), [mapper](const Sink::Entity &entity, const QByteArray &property) {
+
+            const auto localBuffer = Sink::EntityBuffer::readBuffer<Buffer>(entity.local());
+            return mapper->getProperty(property, localBuffer);
+        });
+    }
 }