Prepared new query based synchronization API

8 files changed, 258 insertions, 43 deletions

diff --git a/common/query.h b/common/query.h
index 925b014..aa2d643 100644
--- a/common/query.h
+++ b/common/query.h
@@ -55,6 +55,9 @@ public:
         bool operator==(const Filter &other) const;
     };
 
+    QueryBase() = default;
+    QueryBase(const QByteArray &type) : mType(type) {}
+
     bool operator==(const QueryBase &other) const;
 
     Comparator getFilter(const QByteArray &property) const
@@ -62,11 +65,23 @@ public:
         return mBaseFilterStage.propertyFilter.value(property);
     }
 
+    template <class T>
+    Comparator getFilter() const
+    {
+        return getFilter(T::name);
+    }
+
     bool hasFilter(const QByteArray &property) const
     {
         return mBaseFilterStage.propertyFilter.contains(property);
     }
 
+    template <class T>
+    bool hasFilter() const
+    {
+        return hasFilter(T::name);
+    }
+
     void setBaseFilters(const QHash<QByteArray, Comparator> &filter)
     {
         mBaseFilterStage.propertyFilter = filter;
diff --git a/common/remoteidmap.cpp b/common/remoteidmap.cpp
index da57cf6..a16473d 100644
--- a/common/remoteidmap.cpp
+++ b/common/remoteidmap.cpp
@@ -75,6 +75,15 @@ QByteArray RemoteIdMap::resolveLocalId(const QByteArray &bufferType, const QByte
     return remoteId;
 }
 
+QByteArrayList RemoteIdMap::resolveLocalIds(const QByteArray &bufferType, const QByteArrayList &localIds)
+{
+    QByteArrayList result;
+    for (const auto &l : localIds) {
+        result << resolveLocalId(bufferType, l);
+    }
+    return result;
+}
+
 QByteArray RemoteIdMap::readValue(const QByteArray &key)
 {
     QByteArray value;
diff --git a/common/remoteidmap.h b/common/remoteidmap.h
index 32c5efd..52e05d7 100644
--- a/common/remoteidmap.h
+++ b/common/remoteidmap.h
@@ -22,6 +22,7 @@
 #include "sink_export.h"
 
 #include "storage.h"
+#include <QByteArrayList>
 
 namespace Sink {
 
@@ -53,6 +54,7 @@ public:
      * This can fail if the entity hasn't been written back to the server yet.
      */
     QByteArray resolveLocalId(const QByteArray &bufferType, const QByteArray &localId);
+    QByteArrayList resolveLocalIds(const QByteArray &bufferType, const QByteArrayList &localId);
 
     QByteArray readValue(const QByteArray &key);
     void writeValue(const QByteArray &key, const QByteArray &value);
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 85c68e4..713387e 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -222,21 +222,71 @@ void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray
     }
 }
 
+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};
+        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)
 {
     modifyEntity(entity.identifier(), entity.revision(), ApplicationDomain::getTypeName<DomainType>(), entity);
 }
 
+QList<Synchronizer::SyncRequest> Synchronizer::getSyncRequests(const Sink::QueryBase &query)
+{
+    QList<Synchronizer::SyncRequest> list;
+    list << Synchronizer::SyncRequest{query};
+    return list;
+}
+
 KAsync::Job<void> Synchronizer::synchronize(const Sink::QueryBase &query)
 {
     SinkTrace() << "Synchronizing";
+    mSyncRequestQueue << getSyncRequests(query);
+    return processSyncQueue();
+}
+
+KAsync::Job<void> Synchronizer::processSyncQueue()
+{
+    if (mSyncRequestQueue.isEmpty() || mSyncInProgress) {
+        return KAsync::null<void>();
+    }
     mSyncInProgress = true;
     mMessageQueue->startTransaction();
-    return synchronizeWithSource(query).syncThen<void>([this]() {
+
+    auto job = KAsync::null<void>();
+    while (!mSyncRequestQueue.isEmpty()) {
+        auto request = mSyncRequestQueue.takeFirst();
+        job = job.then(synchronizeWithSource(request.query)).syncThen<void>([this] {
+            //Commit after every request, so implementations only have to commit more if they add a lot of data.
+            commit();
+        });
+    }
+    return job.then<void>([this](const KAsync::Error &error) {
         mSyncStore.clear();
         mMessageQueue->commit();
         mSyncInProgress = false;
+        if (error) {
+            SinkWarning() << "Error during sync: " << error.errorMessage;
+            return KAsync::error(error);
+        }
+        return KAsync::null<void>();
     });
 }
 
diff --git a/common/synchronizer.h b/common/synchronizer.h
index c03c425..47518ee 100644
--- a/common/synchronizer.h
+++ b/common/synchronizer.h
@@ -90,11 +90,42 @@ protected:
     void modify(const DomainType &entity);
     // template <typename DomainType>
     // void remove(const DomainType &entity);
+    QByteArrayList resolveFilter(const QueryBase::Comparator &filter);
 
     virtual KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) = 0;
 
+    struct SyncRequest {
+        SyncRequest(const Sink::QueryBase &q)
+            : flushQueue(false),
+            query(q)
+        {
+        }
+
+        bool flushQueue;
+        Sink::QueryBase query;
+    };
+
+    /**
+     * This allows the synchronizer to turn a single query into multiple synchronization requests.
+     *
+     * The idea is the following;
+     * The input query is a specification by the application of what data needs to be made available.
+     * Requests could be:
+     * * Give me everything (signified by the default constructed/empty query)
+     * * Give me all mails of folder X
+     * * Give me all mails of folders matching some constraints
+     *
+     * getSyncRequests allows the resource implementation to apply it's own defaults to that request;
+     * * While a maildir resource might give you always all emails of a folder, an IMAP resource might have a date limit, to i.e. only retrieve the last 14 days worth of data.
+     * * A resource get's to define what "give me everything" means. For email that may be turned into first a requests for folders, and then a request for all emails in those folders.
+     *
+     * This will allow synchronizeWithSource to focus on just getting to the content.
+     */
+    virtual QList<Synchronizer::SyncRequest> getSyncRequests(const Sink::QueryBase &query);
+
 private:
     void modifyIfChanged(Storage::EntityStore &store, const QByteArray &bufferType, const QByteArray &sinkId, const Sink::ApplicationDomain::ApplicationDomainType &entity);
+    KAsync::Job<void> processSyncQueue();
 
     Sink::ResourceContext mResourceContext;
     Sink::Storage::EntityStore::Ptr mEntityStore;
@@ -102,6 +133,7 @@ private:
     Sink::Storage::DataStore mSyncStorage;
     Sink::Storage::DataStore::Transaction mSyncTransaction;
     std::function<void(int commandId, const QByteArray &data)> mEnqueue;
+    QList<SyncRequest> mSyncRequestQueue;
     MessageQueue *mMessageQueue;
     bool mSyncInProgress;
 };
diff --git a/examples/imapresource/imapresource.cpp b/examples/imapresource/imapresource.cpp
index d5a59b9..302d681 100644
--- a/examples/imapresource/imapresource.cpp
+++ b/examples/imapresource/imapresource.cpp
@@ -272,41 +272,102 @@ public:
 
     }
 
-    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
+    QList<Synchronizer::SyncRequest> getSyncRequests(const Sink::QueryBase &query) Q_DECL_OVERRIDE
     {
-        SinkLog() << " Synchronizing";
-        return KAsync::start<void>([this]() {
-            SinkTrace() << "Connecting to:" << mServer << mPort;
-            SinkTrace() << "as:" << mUser;
-            auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort);
+        QList<Synchronizer::SyncRequest> list;
+        if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) {
+            list << Synchronizer::SyncRequest{query};
+        } else if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) {
+            list << Synchronizer::SyncRequest{query};
+        } else {
+            list << Synchronizer::SyncRequest{Sink::QueryBase(ApplicationDomain::getTypeName<ApplicationDomain::Folder>())};
+            list << Synchronizer::SyncRequest{Sink::QueryBase(ApplicationDomain::getTypeName<ApplicationDomain::Mail>())};
+        }
+        return list;
+    }
 
-            return imap->login(mUser, mPassword)
-            .addToContext(imap)
-            .onError([](const KAsync::Error &error) {
-                SinkWarning() << "Login failed.";
-            })
-            .then<QVector<Folder>>([this, imap]() {
-                auto folderList = QSharedPointer<QVector<Folder>>::create();
-                SinkLog() << "Login was successful";
-                return  imap->fetchFolders([folderList](const Folder &folder) {
-                    *folderList << folder;
-                })
-                .onError([](const KAsync::Error &error) {
-                    SinkWarning() << "Folder list sync failed.";
+    KAsync::Job<void> login(QSharedPointer<ImapServerProxy> imap)
+    {
+        SinkTrace() << "Connecting to:" << mServer << mPort;
+        SinkTrace() << "as:" << mUser;
+        return imap->login(mUser, mPassword)
+        .addToContext(imap)
+        .onError([](const KAsync::Error &error) {
+            SinkWarning() << "Login failed.";
+        });
+    }
+
+    KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
+    {
+        if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) {
+            return KAsync::start<void>([this]() {
+                auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort);
+                auto job = login(imap);
+                job = job.then<QVector<Folder>>([this, imap]() {
+                    auto folderList = QSharedPointer<QVector<Folder>>::create();
+                    return  imap->fetchFolders([folderList](const Folder &folder) {
+                        *folderList << folder;
+                    })
+                    .onError([](const KAsync::Error &error) {
+                        SinkWarning() << "Folder list sync failed.";
+                    })
+                    .syncThen<QVector<Folder>>([this, folderList]() {
+                        synchronizeFolders(*folderList);
+                        commit();
+                        return *folderList;
+                    });
+                });
+                return job;
+            });
+        } else if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) {
+            //TODO
+            //if we have a folder filter:
+            //* execute the folder query and resolve the results to the remote identifier
+            //* query only those folders
+            //if we have a date filter:
+            //* apply the date filter to the fetch
+            //if we have no folder filter:
+            //* fetch list of folders from server directly and sync (because we have no guarantee that the folder sync was already processed by the pipeline).
+            return KAsync::start<void>([this, query]() {
+                auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort);
+                auto job = login(imap);
+                job = job.then<QVector<Folder>>([this, imap, query]() {
+                    SinkLog() << "Login was successful";
+                    //FIXME If we were able to to flush in between we could just query the local store for the folder list.
+                    //
+                    if (query.hasFilter<ApplicationDomain::Mail::Folder>()) {
+                        QVector<Folder> folders;
+                        auto folderFilter = query.getFilter<ApplicationDomain::Mail::Folder>();
+                        auto localIds = resolveFilter(folderFilter);
+                        auto folderRemoteIds = syncStore().resolveLocalIds(ApplicationDomain::getTypeName<ApplicationDomain::Folder>(), localIds);
+                        for (const auto &r : folderRemoteIds) {
+                            folders << Folder{r};
+                        }
+                        return KAsync::value(folders);
+                    } else {
+                        auto folderList = QSharedPointer<QVector<Folder>>::create();
+                        return  imap->fetchFolders([folderList](const Folder &folder) {
+                            *folderList << folder;
+                        })
+                        .onError([](const KAsync::Error &error) {
+                            SinkWarning() << "Folder list sync failed.";
+                        })
+                        .syncThen<QVector<Folder>>([this, folderList]() {
+                            return *folderList;
+                        });
+                    }
                 })
-                .syncThen<QVector<Folder>>([this, folderList]() {
-                    synchronizeFolders(*folderList);
-                    commit();
-                    return *folderList;
+                .serialEach<void>([this, imap](const Folder &folder) {
+                    if (folder.noselect) {
+                        return KAsync::null<void>();
+                    }
+                    return synchronizeFolder(imap, folder);
                 });
-            })
-            .serialEach<void>([this, imap](const Folder &folder) {
-                if (folder.noselect) {
-                    return KAsync::null<void>();
-                }
-                return synchronizeFolder(imap, folder);
+
+                return job;
             });
-        });
+        }
+        return KAsync::error<void>("Nothing to do");
     }
 
 public:
diff --git a/examples/imapresource/imapserverproxy.h b/examples/imapresource/imapserverproxy.h
index 8b39f23..6e6626e 100644
--- a/examples/imapresource/imapserverproxy.h
+++ b/examples/imapresource/imapserverproxy.h
@@ -58,6 +58,20 @@ struct Message {
 };
 
 struct Folder {
+    Folder() = default;
+    Folder(QList<QString> pathParts_, const QString &path_, const QChar &separator_, bool noselect_)
+        : pathParts(pathParts_),
+        path(path_),
+        separator(separator_),
+        noselect(noselect_)
+    {
+    }
+
+    Folder(const QString &path_)
+        : path(path_)
+    {
+    }
+
     QString normalizedPath() const
     {
         return pathParts.join('/');
@@ -73,7 +87,7 @@ struct Folder {
     QList<QString> pathParts;
     QString path;
     QChar separator;
-    bool noselect;
+    bool noselect = false;
 };
 
 struct SelectResult {
diff --git a/examples/maildirresource/maildirresource.cpp b/examples/maildirresource/maildirresource.cpp
index 820ec2f..fc77315 100644
--- a/examples/maildirresource/maildirresource.cpp
+++ b/examples/maildirresource/maildirresource.cpp
@@ -330,25 +330,57 @@ public:
         SinkLog() << "Synchronized " << count << " mails in " << listingPath << Sink::Log::TraceTime(elapsed) << " " << elapsed/qMax(count, 1) << " [ms/mail]";
     }
 
+    QList<Synchronizer::SyncRequest> getSyncRequests(const Sink::QueryBase &query) Q_DECL_OVERRIDE
+    {
+        QList<Synchronizer::SyncRequest> list;
+        if (!query.type().isEmpty()) {
+            //We want to synchronize something specific
+            list << Synchronizer::SyncRequest{query};
+        } else {
+            //We want to synchronize everything
+            list << Synchronizer::SyncRequest{Sink::QueryBase(ApplicationDomain::getTypeName<ApplicationDomain::Folder>())};
+            //FIXME we can't process the second synchronization before the pipeline of the first one is processed, otherwise we can't execute a query on the local data.
+            /* list << Synchronizer::SyncRequest{Flush}; */
+            list << Synchronizer::SyncRequest{Sink::QueryBase(ApplicationDomain::getTypeName<ApplicationDomain::Mail>())};
+        }
+        return list;
+    }
+
     KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE
     {
-        SinkLog() << " Synchronizing";
-        return KAsync::start<void>([this]() {
+        auto job = KAsync::start<void>([this] {
             KPIM::Maildir maildir(mMaildirPath, true);
             if (!maildir.isValid(false)) {
                 return KAsync::error<void>(1, "Maildir path doesn't point to a valid maildir: " + mMaildirPath);
             }
-            synchronizeFolders();
-            //The next sync needs the folders available
-            commit();
-            for (const auto &folder : listAvailableFolders()) {
-                synchronizeMails(folder);
-                //Don't let the transaction grow too much
-                commit();
-            }
-            SinkLog() << "Done Synchronizing";
             return KAsync::null<void>();
         });
+
+        if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) {
+            job = job.syncThen<void>([this] {
+                synchronizeFolders();
+            });
+        } else if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Mail>()) {
+            job = job.syncThen<void>([this, query] {
+                QStringList folders;
+                if (query.hasFilter<ApplicationDomain::Mail::Folder>()) {
+                    auto folderFilter = query.getFilter<ApplicationDomain::Mail::Folder>();
+                    auto localIds = resolveFilter(folderFilter);
+                    auto folderRemoteIds = syncStore().resolveLocalIds(ApplicationDomain::getTypeName<ApplicationDomain::Folder>(), localIds);
+                    for (const auto &r : folderRemoteIds) {
+                        folders << r;
+                    }
+                } else {
+                    folders = listAvailableFolders();
+                }
+                for (const auto &folder : folders) {
+                    synchronizeMails(folder);
+                    //Don't let the transaction grow too much
+                    commit();
+                }
+            });
+        }
+        return job;
     }
 
 public: