New synchronization algorithm that only fetches the last 14 days.

3 files changed, 139 insertions, 98 deletions

diff --git a/examples/imapresource/imapresource.cpp b/examples/imapresource/imapresource.cpp
index 63ae07b..aa0fb94 100644
--- a/examples/imapresource/imapresource.cpp
+++ b/examples/imapresource/imapresource.cpp
@@ -40,10 +40,14 @@
 #include "entitystore.h"
 #include "remoteidmap.h"
 #include "query.h"
+
+#include <QtGlobal>
 #include <QDate>
 #include <QUuid>
 #include <QDir>
 #include <QDirIterator>
+#include <QDateTime>
+#include <QtAlgorithms>
 
 #include "imapserverproxy.h"
 #include "entityreader.h"
@@ -149,7 +153,7 @@ public:
         }
     }
 
-    void synchronizeMails(const QString &path, const QVector<Message> &messages)
+    void synchronizeMails(const QString &path, const Message &message)
     {
         auto time = QSharedPointer<QTime>::create();
         time->start();
@@ -159,23 +163,20 @@ public:
 
         const auto folderLocalId = syncStore().resolveRemoteId(ENTITY_TYPE_FOLDER, path.toUtf8());
 
-        int count = 0;
-        for (const auto &message : messages) {
-            count++;
-            const auto remoteId = assembleMailRid(folderLocalId, message.uid);
+        const auto remoteId = assembleMailRid(folderLocalId, message.uid);
 
-            SinkTrace() << "Found a mail " << remoteId << message.msg->subject(true)->asUnicodeString() << message.flags;
+        Q_ASSERT(message.msg);
+        SinkTrace() << "Found a mail " << remoteId << message.msg->subject(true)->asUnicodeString() << message.flags;
 
-            auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier);
-            mail.setFolder(folderLocalId);
-            mail.setMimeMessage(message.msg->encodedContent());
-            mail.setUnread(!message.flags.contains(Imap::Flags::Seen));
-            mail.setImportant(message.flags.contains(Imap::Flags::Flagged));
+        auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier);
+        mail.setFolder(folderLocalId);
+        mail.setMimeMessage(message.msg->encodedContent());
+        mail.setUnread(!message.flags.contains(Imap::Flags::Seen));
+        mail.setImportant(message.flags.contains(Imap::Flags::Flagged));
 
-            createOrModify(bufferType, remoteId, mail);
-        }
-        const auto elapsed = time->elapsed();
-        SinkTrace() << "Synchronized " << count << " mails in " << path << Sink::Log::TraceTime(elapsed) << " " << elapsed/qMax(count, 1) << " [ms/mail]";
+        createOrModify(bufferType, remoteId, mail);
+        // const auto elapsed = time->elapsed();
+        // SinkTrace() << "Synchronized " << count << " mails in " << path << Sink::Log::TraceTime(elapsed) << " " << elapsed/qMax(count, 1) << " [ms/mail]";
     }
 
     void synchronizeRemovals(const QString &path, const QSet<qint64> &messages)
@@ -220,29 +221,71 @@ public:
         auto capabilities = imap->getCapabilities();
         bool canDoIncrementalRemovals = false;
         return KAsync::start<void>([=]() {
+            //First we fetch flag changes for all messages. Since we don't know which messages are locally available we just get everything and only apply to what we have.
+            SinkLog() << "About to update flags" << folder.normalizedPath();
             auto uidNext = syncStore().readValue(folder.normalizedPath().toUtf8() + "uidnext").toLongLong();
             const auto changedsince = syncStore().readValue(folder.normalizedPath().toUtf8() + "changedsince").toLongLong();
-            return imap->fetchFlags(folder, KIMAP2::ImapSet(1, qMax(uidNext, qint64(1))), changedsince, [this, folder](const QVector<Message> &messages) {
-                // synchronizeMails(folder.normalizedPath(), messages);
+            return imap->fetchFlags(folder, KIMAP2::ImapSet(1, qMax(uidNext, qint64(1))), changedsince, [this, folder](const Message &message) {
                 const auto folderLocalId = syncStore().resolveRemoteId(ENTITY_TYPE_FOLDER, folder.normalizedPath().toUtf8());
-                for (const auto &message : messages) {
-                    const auto remoteId = assembleMailRid(folderLocalId, message.uid);
+                const auto remoteId = assembleMailRid(folderLocalId, message.uid);
 
+                SinkLog() << "Updating mail flags " << remoteId << message.flags;
 
-                    auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier);
-                    mail.setUnread(!message.flags.contains(Imap::Flags::Seen));
-                    mail.setImportant(message.flags.contains(Imap::Flags::Flagged));
+                auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier);
+                mail.setUnread(!message.flags.contains(Imap::Flags::Seen));
+                mail.setImportant(message.flags.contains(Imap::Flags::Flagged));
 
-                    modify(ENTITY_TYPE_MAIL, remoteId, mail);
-                }
+                modify(ENTITY_TYPE_MAIL, remoteId, mail);
             })
             .syncThen<void, SelectResult>([this, folder](const SelectResult &selectResult) {
                 syncStore().writeValue(folder.normalizedPath().toUtf8() + "changedsince", QByteArray::number(selectResult.highestModSequence));
             });
         })
         .then<void>([=]() {
+            //Get the range we're interested in. This is what we're going to download.
+            return imap->fetchUidsSince(imap->mailboxFromFolder(folder), QDate::currentDate().addDays(-14))
+            .then<void, QVector<qint64>>([this, folder, imap](const QVector<qint64> &uidsToFetch) {
+                SinkTrace() << "Received result set " << uidsToFetch;
+                SinkTrace() << "About to fetch mail" << folder.normalizedPath();
+                const auto uidNext = syncStore().readValue(folder.normalizedPath().toUtf8() + "uidnext").toLongLong();
+                QVector<qint64> filteredAndSorted = uidsToFetch;
+                qSort(filteredAndSorted.begin(), filteredAndSorted.end(), qGreater<qint64>());
+                auto lowerBound = qLowerBound(filteredAndSorted.begin(), filteredAndSorted.end(), uidNext, qGreater<qint64>());
+                if (lowerBound != filteredAndSorted.end()) {
+                    filteredAndSorted.erase(lowerBound, filteredAndSorted.end());
+                }
+
+                auto maxUid = QSharedPointer<qint64>::create(0);
+                if (!filteredAndSorted.isEmpty()) {
+                    *maxUid = filteredAndSorted.first();
+                }
+                SinkTrace() << "Uids to fetch: " << filteredAndSorted;
+                return imap->fetchMessages(folder, filteredAndSorted, [this, folder, maxUid](const Message &m) {
+                    if (*maxUid < m.uid) {
+                        *maxUid = m.uid;
+                    }
+                    synchronizeMails(folder.normalizedPath(), m);
+                },
+                [this, maxUid, folder](int progress, int total) {
+                    SinkLog() << "Progress: " << progress << " out of " << total;
+                    //commit every 10 messages
+                    if ((progress % 10) == 0) {
+                        commit();
+                    }
+                })
+                .syncThen<void>([this, maxUid, folder]() {
+                    SinkLog() << "UIDMAX: " << *maxUid << folder.normalizedPath();
+                    if (*maxUid > 0) {
+                        syncStore().writeValue(folder.normalizedPath().toUtf8() + "uidnext", QByteArray::number(*maxUid));
+                    }
+                    commit();
+                });
+            });
+        })
+        .then<void>([=]() {
             //TODO Remove what's no longer existing
             if (canDoIncrementalRemovals) {
+                //TODO do an examine with QRESYNC and remove VANISHED messages
             } else {
                 return imap->fetchUids(folder).syncThen<void, QVector<qint64>>([this, folder](const QVector<qint64> &uids) {
                     SinkTrace() << "Syncing removals";
@@ -251,32 +294,6 @@ public:
                 });
             }
             return KAsync::null<void>();
-        })
-        .then<void>([this, folder, imap]() {
-            SinkTrace() << "About to fetch mail";
-            const auto uidNext = syncStore().readValue(folder.normalizedPath().toUtf8() + "uidnext").toLongLong();
-            auto maxUid = QSharedPointer<qint64>::create(0);
-            return imap->fetchMessages(folder, uidNext, [this, folder, maxUid](const QVector<Message> &messages) {
-                SinkTrace() << "Got mail";
-                for (const auto &m : messages) {
-                    if (*maxUid < m.uid) {
-                        *maxUid = m.uid;
-                    }
-                }
-                synchronizeMails(folder.normalizedPath(), messages);
-            },
-            [this, maxUid, folder](int progress, int total) {
-                SinkLog() << "Progress: " << progress << " out of " << total;
-                //commit every 10 messages
-                if ((progress % 10) == 0) {
-                    commit();
-                }
-            })
-            .syncThen<void>([this, maxUid, folder]() {
-                syncStore().writeValue(folder.normalizedPath().toUtf8() + "uidnext", QByteArray::number(*maxUid));
-                SinkLog() << "UIDMAX: " << *maxUid << folder.normalizedPath();
-                commit();
-            });
         });
 
 
@@ -578,10 +595,8 @@ KAsync::Job<void> ImapResource::inspect(int inspectionType, const QByteArray &in
         auto inspectionJob = imap->login(mUser, mPassword)
             .then<Imap::SelectResult>(imap->select(folderRemoteId))
             .syncThen<void, Imap::SelectResult>([](Imap::SelectResult){})
-            .then<void>(imap->fetch(set, scope, [imap, messageByUid](const QVector<Imap::Message> &messages) {
-                for (const auto &m : messages) {
-                    messageByUid->insert(m.uid, m);
-                }
+            .then<void>(imap->fetch(set, scope, [imap, messageByUid](const Imap::Message &message) {
+                messageByUid->insert(message.uid, message);
             }));
 
         if (inspectionType == Sink::ResourceControl::Inspection::PropertyInspectionType) {
@@ -641,10 +656,8 @@ KAsync::Job<void> ImapResource::inspect(int inspectionType, const QByteArray &in
             auto messageByUid = QSharedPointer<QHash<qint64, Imap::Message>>::create();
             return imap->login(mUser, mPassword)
                 .then<void>(imap->select(remoteId).syncThen<void>([](){}))
-                .then<void>(imap->fetch(set, scope, [=](const QVector<Imap::Message> &messages) {
-                    for (const auto &m : messages) {
-                        messageByUid->insert(m.uid, m);
-                    }
+                .then<void>(imap->fetch(set, scope, [=](const Imap::Message message) {
+                    messageByUid->insert(message.uid, message);
                 }))
                 .then<void>([imap, messageByUid, expectedCount]() {
                     if (messageByUid->size() != expectedCount) {
diff --git a/examples/imapresource/imapserverproxy.cpp b/examples/imapresource/imapserverproxy.cpp
index a75825e..58164d3 100644
--- a/examples/imapresource/imapserverproxy.cpp
+++ b/examples/imapresource/imapserverproxy.cpp
@@ -263,21 +263,24 @@ KAsync::Job<void> ImapServerProxy::fetch(const KIMAP2::ImapSet &set, KIMAP2::Fet
 
 KAsync::Job<QVector<qint64>> ImapServerProxy::search(const KIMAP2::ImapSet &set)
 {
+    return search(KIMAP2::Term(KIMAP2::Term::Uid, set));
+}
+
+KAsync::Job<QVector<qint64>> ImapServerProxy::search(const KIMAP2::Term &term)
+{
     auto search = new KIMAP2::SearchJob(mSession);
-    search->setTerm(KIMAP2::Term(KIMAP2::Term::Uid, set));
+    search->setTerm(term);
     search->setUidBased(true);
     return runJob<QVector<qint64>>(search, [](KJob *job) -> QVector<qint64> {
         return static_cast<KIMAP2::SearchJob*>(job)->results();
     });
 }
 
-KAsync::Job<void> ImapServerProxy::fetch(const KIMAP2::ImapSet &set, KIMAP2::FetchJob::FetchScope scope, const std::function<void(const QVector<Message> &)> &callback)
+KAsync::Job<void> ImapServerProxy::fetch(const KIMAP2::ImapSet &set, KIMAP2::FetchJob::FetchScope scope, const std::function<void(const Message &)> &callback)
 {
     return fetch(set, scope,
                     [callback](const KIMAP2::FetchJob::Result &result) {
-                        QVector<Message> list;
-                        list << Message{result.uid, result.size, result.attributes, result.flags, result.message};
-                        callback(list);
+                        callback(Message{result.uid, result.size, result.attributes, result.flags, result.message});
                     });
 }
 
@@ -286,9 +289,9 @@ QStringList ImapServerProxy::getCapabilities() const
     return mCapabilities;
 }
 
-KAsync::Job<QList<qint64>> ImapServerProxy::fetchHeaders(const QString &mailbox, const qint64 minUid)
+KAsync::Job<QVector<qint64>> ImapServerProxy::fetchHeaders(const QString &mailbox, const qint64 minUid)
 {
-    auto list = QSharedPointer<QList<qint64>>::create();
+    auto list = QSharedPointer<QVector<qint64>>::create();
     KIMAP2::FetchJob::FetchScope scope;
     scope.mode = KIMAP2::FetchJob::FetchScope::Flags;
 
@@ -304,14 +307,19 @@ KAsync::Job<QList<qint64>> ImapServerProxy::fetchHeaders(const QString &mailbox,
 
                 list->append(result.uid);
             })
-    .syncThen<QList<qint64>>([list](){
+    .syncThen<QVector<qint64>>([list](){
         return *list;
     });
 }
 
 KAsync::Job<QVector<qint64>> ImapServerProxy::fetchUids(const QString &mailbox)
 {
-    return select(mailbox).then<QVector<qint64>>(search(KIMAP2::ImapSet(1, 0)));
+    return select(mailbox).then<QVector<qint64>>(search(KIMAP2::Term(KIMAP2::Term::Uid, KIMAP2::ImapSet(1, 0))));
+}
+
+KAsync::Job<QVector<qint64>> ImapServerProxy::fetchUidsSince(const QString &mailbox, const QDate &since)
+{
+    return select(mailbox).then<QVector<qint64>>(search(KIMAP2::Term(KIMAP2::Term::Since, since)));
 }
 
 KAsync::Job<void> ImapServerProxy::list(KIMAP2::ListJob::Option option, const std::function<void(const KIMAP2::MailBoxDescriptor &mailboxes, const QList<QByteArray> &flags)> &callback)
@@ -390,7 +398,7 @@ QString ImapServerProxy::mailboxFromFolder(const Folder &folder) const
     return folder.pathParts.join(mPersonalNamespaceSeparator);
 }
 
-KAsync::Job<SelectResult> ImapServerProxy::fetchFlags(const Folder &folder, const KIMAP2::ImapSet &set, qint64 changedsince, std::function<void(const QVector<Message> &)> callback)
+KAsync::Job<SelectResult> ImapServerProxy::fetchFlags(const Folder &folder, const KIMAP2::ImapSet &set, qint64 changedsince, std::function<void(const Message &)> callback)
 {
     SinkTrace() << "Fetching flags " << folder.normalizedPath();
     return select(mailboxFromFolder(folder)).then<SelectResult, SelectResult>([=](const SelectResult &selectResult) -> KAsync::Job<SelectResult> {
@@ -413,7 +421,7 @@ KAsync::Job<SelectResult> ImapServerProxy::fetchFlags(const Folder &folder, cons
     });
 }
 
-KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, qint64 uidNext, std::function<void(const QVector<Message> &)> callback, std::function<void(int, int)> progress)
+KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, qint64 uidNext, std::function<void(const Message &)> callback, std::function<void(int, int)> progress)
 {
     auto time = QSharedPointer<QTime>::create();
     time->start();
@@ -426,40 +434,56 @@ KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, qint64 ui
         }
 
         SinkTrace() << "Fetching messages from  " << folder.normalizedPath() << selectResult.uidNext << uidNext;
-        return fetchHeaders(mailboxFromFolder(folder), (uidNext + 1)).then<void, QList<qint64>>([this, callback, time, progress, folder](const QList<qint64> &uidsToFetch){
+        return fetchHeaders(mailboxFromFolder(folder), (uidNext + 1)).then<void, QVector<qint64>>([this, callback, time, progress, folder](const QVector<qint64> &uidsToFetch){
             SinkTrace() << "Fetched headers" << folder.normalizedPath();
             SinkTrace() << "  Total: " << uidsToFetch.size();
             SinkTrace() << "  Uids to fetch: " << uidsToFetch;
             SinkTrace() << "  Took: " << Sink::Log::TraceTime(time->elapsed());
-            auto totalCount = uidsToFetch.size();
+            return fetchMessages(folder, uidsToFetch, callback, progress);
+        });
+
+    });
+}
+
+KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, const QVector<qint64> &uidsToFetch, std::function<void(const Message &)> callback, std::function<void(int, int)> progress)
+{
+    auto time = QSharedPointer<QTime>::create();
+    time->start();
+    Q_ASSERT(!mPersonalNamespaceSeparator.isNull());
+    return select(mailboxFromFolder(folder)).then<void, SelectResult>([this, callback, folder, time, progress, uidsToFetch](const SelectResult &selectResult) -> KAsync::Job<void> {
+
+        SinkTrace() << "Fetching messages" << folder.normalizedPath();
+        SinkTrace() << "  Total: " << uidsToFetch.size();
+        SinkTrace() << "  Uids to fetch: " << uidsToFetch;
+        auto totalCount = uidsToFetch.size();
+        if (progress) {
+            progress(0, totalCount);
+        }
+        if (uidsToFetch.isEmpty()) {
+            SinkTrace() << "Nothing to fetch";
+            return KAsync::null<void>();
+        }
+        KIMAP2::FetchJob::FetchScope scope;
+        scope.parts.clear();
+        scope.mode = KIMAP2::FetchJob::FetchScope::Full;
+
+        KIMAP2::ImapSet set;
+        set.add(uidsToFetch);
+        auto count = QSharedPointer<int>::create();
+        return fetch(set, scope, [=](const Message &message) {
+            *count += 1;
             if (progress) {
-                progress(0, totalCount);
-            }
-            if (uidsToFetch.isEmpty()) {
-                SinkTrace() << "Nothing to fetch";
-                callback(QVector<Message>());
-                return KAsync::null<void>();
+                progress(*count, totalCount);
             }
-            KIMAP2::FetchJob::FetchScope scope;
-            scope.parts.clear();
-            scope.mode = KIMAP2::FetchJob::FetchScope::Full;
-
-            KIMAP2::ImapSet set;
-            set.add(uidsToFetch.toVector());
-            auto count = QSharedPointer<int>::create();
-            return fetch(set, scope, [=](const QVector<Message> &messages) {
-                *count += messages.size();
-                if (progress) {
-                    progress(*count, totalCount);
-                }
-                callback(messages);
-            });
+            callback(message);
         });
-
+    })
+    .syncThen<void>([time]() {
+        SinkTrace() << "The fetch took: " << Sink::Log::TraceTime(time->elapsed());
     });
 }
 
-KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, std::function<void(const QVector<Message> &)> callback, std::function<void(int, int)> progress)
+KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, std::function<void(const Message &)> callback, std::function<void(int, int)> progress)
 {
     return fetchMessages(folder, 0, callback, progress);
 }
diff --git a/examples/imapresource/imapserverproxy.h b/examples/imapresource/imapserverproxy.h
index 452b479..8b39f23 100644
--- a/examples/imapresource/imapserverproxy.h
+++ b/examples/imapresource/imapserverproxy.h
@@ -25,6 +25,7 @@
 #include <KIMAP2/ListJob>
 #include <KIMAP2/Session>
 #include <KIMAP2/FetchJob>
+#include <KIMAP2/SearchJob>
 
 namespace Imap {
 
@@ -110,30 +111,33 @@ public:
     KAsync::Job<void> expunge(const KIMAP2::ImapSet &set);
     KAsync::Job<void> copy(const KIMAP2::ImapSet &set, const QString &newMailbox);
     KAsync::Job<QVector<qint64>> search(const KIMAP2::ImapSet &set);
+    KAsync::Job<QVector<qint64>> search(const KIMAP2::Term &term);
 
     typedef std::function<void(const KIMAP2::FetchJob::Result &)> FetchCallback;
 
     KAsync::Job<void> fetch(const KIMAP2::ImapSet &set, KIMAP2::FetchJob::FetchScope scope, FetchCallback callback);
-    KAsync::Job<void> fetch(const KIMAP2::ImapSet &set, KIMAP2::FetchJob::FetchScope scope, const std::function<void(const QVector<Message> &)> &callback);
+    KAsync::Job<void> fetch(const KIMAP2::ImapSet &set, KIMAP2::FetchJob::FetchScope scope, const std::function<void(const Message &)> &callback);
     KAsync::Job<void> list(KIMAP2::ListJob::Option option, const std::function<void(const KIMAP2::MailBoxDescriptor &mailboxes,const QList<QByteArray> &flags)> &callback);
 
     QStringList getCapabilities() const;
 
     //Composed calls that do login etc.
-    KAsync::Job<QList<qint64>> fetchHeaders(const QString &mailbox, qint64 minUid = 1);
+    KAsync::Job<QVector<qint64>> fetchHeaders(const QString &mailbox, qint64 minUid = 1);
     KAsync::Job<void> remove(const QString &mailbox, const KIMAP2::ImapSet &set);
     KAsync::Job<void> remove(const QString &mailbox, const QByteArray &imapSet);
     KAsync::Job<void> move(const QString &mailbox, const KIMAP2::ImapSet &set, const QString &newMailbox);
     KAsync::Job<QString> createSubfolder(const QString &parentMailbox, const QString &folderName);
     KAsync::Job<QString> renameSubfolder(const QString &mailbox, const QString &newName);
     KAsync::Job<QVector<qint64>> fetchUids(const QString &mailbox);
+    KAsync::Job<QVector<qint64>> fetchUidsSince(const QString &mailbox, const QDate &since);
 
     QString mailboxFromFolder(const Folder &) const;
 
     KAsync::Job<void> fetchFolders(std::function<void(const Folder &)> callback);
-    KAsync::Job<void> fetchMessages(const Folder &folder, std::function<void(const QVector<Message> &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>());
-    KAsync::Job<void> fetchMessages(const Folder &folder, qint64 uidNext, std::function<void(const QVector<Message> &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>());
-    KAsync::Job<SelectResult> fetchFlags(const Folder &folder, const KIMAP2::ImapSet &set, qint64 changedsince, std::function<void(const QVector<Message> &)> callback);
+    KAsync::Job<void> fetchMessages(const Folder &folder, std::function<void(const Message &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>());
+    KAsync::Job<void> fetchMessages(const Folder &folder, qint64 uidNext, std::function<void(const Message &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>());
+    KAsync::Job<void> fetchMessages(const Folder &folder, const QVector<qint64> &uidsToFetch, std::function<void(const Message &)> callback, std::function<void(int, int)> progress);
+    KAsync::Job<SelectResult> fetchFlags(const Folder &folder, const KIMAP2::ImapSet &set, qint64 changedsince, std::function<void(const Message &)> callback);
     KAsync::Job<QVector<qint64>> fetchUids(const Folder &folder);
 
 private: